websms/freeipa
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Import Upstream version 4.12.4

Browse Source
This commit is contained in:
geos_one
2025-08-12 22:28:56 +02:00
parent 03a8170b15
commit 9181ee2487
1629 changed files with 874094 additions and 554378 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
ipatests/test_ipaserver/httptest.py
View File

@@ -36,7 +36,7 @@ class Unauthorized_HTTP_test:
content_type = 'application/x-www-form-urlencoded'
accept_language = 'en-us'
def send_request(self, method='POST', params=None):
def send_request(self, method='POST', params=None, host=None):
"""
Send a request to HTTP server
@@ -44,10 +44,11 @@ class Unauthorized_HTTP_test:
"""
if params is not None:
if self.content_type == 'application/x-www-form-urlencoded':
# urlencode *can* take two arguments
# pylint: disable=too-many-function-args
params = urllib.parse.urlencode(params, True)
url = 'https://' + self.host + self.app_uri
if host:
url = 'https://' + host + self.app_uri
else:
url = 'https://' + self.host + self.app_uri
headers = {'Content-Type': self.content_type,
'Accept-Language': self.accept_language,

12
ipatests/test_ipaserver/test_changepw.py
View File

@@ -53,10 +53,11 @@ class test_changepw(XMLRPC_test, Unauthorized_HTTP_test):
request.addfinalizer(fin)
def _changepw(self, user, old_password, new_password):
def _changepw(self, user, old_password, new_password, host=None):
return self.send_request(params={'user': str(user),
'old_password' : str(old_password),
'new_password' : str(new_password)},
host=host
)
def _checkpw(self, user, password):
@@ -89,6 +90,15 @@ class test_changepw(XMLRPC_test, Unauthorized_HTTP_test):
# make sure that password is NOT changed
self._checkpw(testuser, old_password)
def test_invalid_referer(self):
response = self._changepw(testuser, old_password, new_password,
'attacker.test')
assert_equal(response.status, 400)
# make sure that password is NOT changed
self._checkpw(testuser, old_password)
def test_pwpolicy_error(self):
response = self._changepw(testuser, old_password, '1')

10
ipatests/test_ipaserver/test_install/test_installer.py
View File

@@ -3,9 +3,7 @@
#
from __future__ import absolute_import
import six
from abc import ABCMeta, abstractproperty
from abc import ABCMeta, abstractmethod
from collections import namedtuple
import itertools
@@ -17,11 +15,12 @@ from ipaserver.install.ipa_replica_install import ReplicaInstall
Keyval = namedtuple('Keyval', ['option', 'value'])
class InstallerTestBase(six.with_metaclass(ABCMeta, object)):
class InstallerTestBase(metaclass=ABCMeta):
OPTS_DICT = {}
# don't allow creating classes with tested_cls unspecified
@abstractproperty
@property
@abstractmethod
def tested_cls(self):
return None
@@ -47,7 +46,6 @@ class InstallerTestBase(six.with_metaclass(ABCMeta, object)):
"class first.")
# add all options from the option groups
# pylint: disable=no-member
for opt_group in cls.tested_cls.option_parser.option_groups:
for opt in opt_group.option_list:
cls.OPTS_DICT[opt.dest] = opt._short_opts + opt._long_opts

99
ipatests/test_ipaserver/test_install/test_installutils.py
View File

@@ -55,21 +55,28 @@ def gpgkey(request, tempdir):
f.write("allow-preset-passphrase\n")
# daemonize agent (detach from the console and run in the background)
subprocess.Popen(
[paths.GPG_AGENT, '--batch', '--daemon'],
env=env, stdout=devnull, stderr=devnull
subprocess.run(
[paths.SYSTEMD_RUN, '--service-type=forking',
'--property', 'SELinuxContext=system_u:system_r:initrc_t:s0',
'--setenv=GNUPGHOME={}'.format(gnupghome),
'--setenv=LC_ALL=C.UTF-8',
'--setenv=LANGUAGE=C',
'--unit=gpg-agent', '/bin/bash',
'-c', ' '.join([paths.GPG_AGENT, '--daemon', '--batch'])],
check=True,
env=env,
)
def fin():
subprocess.run(
[paths.SYSTEMCTL, 'stop', 'gpg-agent'],
check=True,
env=env,
)
if orig_gnupghome is not None:
os.environ['GNUPGHOME'] = orig_gnupghome
else:
os.environ.pop('GNUPGHOME', None)
subprocess.run(
[paths.GPG_CONF, '--kill', 'all'],
check=True,
env=env,
)
request.addfinalizer(fin)
@@ -205,7 +212,7 @@ RAM_NOT_OK = str(10 * 1000 * 1000)
@patch('ipaserver.install.installutils.in_container')
@patch('builtins.open', mock_open_multi(RAM_NOT_OK, "0"))
@patch('os.path.exists')
def test_in_container_insufficient_ram(mock_exists, mock_in_container):
def test_cgroup_v1_insufficient_ram(mock_exists, mock_in_container):
"""In a container with insufficient RAM and zero used"""
mock_in_container.return_value = True
mock_exists.side_effect = [True, True]
@@ -217,7 +224,7 @@ def test_in_container_insufficient_ram(mock_exists, mock_in_container):
@patch('ipaserver.install.installutils.in_container')
@patch('builtins.open', mock_open_multi(RAM_OK, RAM_CA_USED))
@patch('os.path.exists')
def test_in_container_ram_ok_no_ca(mock_exists, mock_in_container):
def test_cgroup_v1_ram_ok_no_ca(mock_exists, mock_in_container):
"""In a container with just enough RAM to install w/o a CA"""
mock_in_container.return_value = True
mock_exists.side_effect = [True, True]
@@ -228,7 +235,7 @@ def test_in_container_ram_ok_no_ca(mock_exists, mock_in_container):
@patch('ipaserver.install.installutils.in_container')
@patch('builtins.open', mock_open_multi(RAM_OK, RAM_MOSTLY_USED))
@patch('os.path.exists')
def test_in_container_insufficient_ram_with_ca(mock_exists, mock_in_container):
def test_cgroup_v1_insufficient_ram_with_ca(mock_exists, mock_in_container):
"""In a container and just miss the minimum RAM required"""
mock_in_container.return_value = True
mock_exists.side_effect = [True, True]
@@ -238,8 +245,74 @@ def test_in_container_insufficient_ram_with_ca(mock_exists, mock_in_container):
@patch('ipaserver.install.installutils.in_container')
@patch('builtins.open', mock_open_multi(RAM_NOT_OK, "0"))
@patch('os.path.exists')
def test_cgroup_v2_insufficient_ram(mock_exists, mock_in_container):
"""In a container with insufficient RAM and zero used"""
mock_in_container.return_value = True
mock_exists.side_effect = [False, True, True]
with pytest.raises(ScriptError):
installutils.check_available_memory(True)
@patch('ipaserver.install.installutils.in_container')
@patch('builtins.open', mock_open_multi(RAM_OK, RAM_CA_USED))
@patch('os.path.exists')
def test_cgroup_v2_ram_ok_no_ca(mock_exists, mock_in_container):
"""In a container with just enough RAM to install w/o a CA"""
mock_in_container.return_value = True
mock_exists.side_effect = [False, True, True]
installutils.check_available_memory(False)
@patch('ipaserver.install.installutils.in_container')
@patch('builtins.open', mock_open_multi(RAM_OK, RAM_MOSTLY_USED))
@patch('os.path.exists')
def test_cgroup_v2_insufficient_ram_with_ca(mock_exists, mock_in_container):
"""In a container and just miss the minimum RAM required"""
mock_in_container.return_value = True
mock_exists.side_effect = [False, True, True]
with pytest.raises(ScriptError):
installutils.check_available_memory(True)
@patch('ipaserver.install.installutils.in_container')
@patch('builtins.open', mock_open_multi('max', RAM_MOSTLY_USED))
@patch('os.path.exists')
@patch('psutil.virtual_memory')
def test_not_container_insufficient_ram_with_ca(mock_psutil, mock_in_container):
def test_cgroup_v2_no_limit_ok(mock_psutil, mock_exists, mock_in_container):
"""In a container and just miss the minimum RAM required"""
mock_in_container.return_value = True
fake_memory = psutil._pslinux.svmem
fake_memory.available = int(RAM_OK)
mock_psutil.return_value = fake_memory
mock_exists.side_effect = [False, True, True]
installutils.check_available_memory(True)
@patch('ipaserver.install.installutils.in_container')
@patch('builtins.open', mock_open_multi('max', RAM_MOSTLY_USED))
@patch('os.path.exists')
@patch('psutil.virtual_memory')
def test_cgroup_v2_no_limit_not_ok(mock_psutil, mock_exists, mock_in_container):
"""In a container and just miss the minimum RAM required"""
mock_in_container.return_value = True
fake_memory = psutil._pslinux.svmem
fake_memory.available = int(RAM_NOT_OK)
mock_psutil.return_value = fake_memory
mock_exists.side_effect = [False, True, True]
with pytest.raises(ScriptError):
installutils.check_available_memory(True)
@patch('ipaserver.install.installutils.in_container')
@patch('psutil.virtual_memory')
def test_bare_insufficient_ram_with_ca(mock_psutil, mock_in_container):
"""Not a container and insufficient RAM"""
mock_in_container.return_value = False
fake_memory = psutil._pslinux.svmem
@@ -252,7 +325,7 @@ def test_not_container_insufficient_ram_with_ca(mock_psutil, mock_in_container):
@patch('ipaserver.install.installutils.in_container')
@patch('psutil.virtual_memory')
def test_not_container_ram_ok(mock_psutil, mock_in_container):
def test_bare_ram_ok(mock_psutil, mock_in_container):
"""Not a container and sufficient RAM"""
mock_in_container.return_value = False
fake_memory = psutil._pslinux.svmem

73
ipatests/test_ipaserver/test_jsplugins.py Normal file
View File

@@ -0,0 +1,73 @@
# Copyright (C) 2020 FreeIPA Contributors see COPYING for license
import os
import pytest
from ipalib import facts
from ipatests.test_ipaserver.httptest import Unauthorized_HTTP_test
from ipatests.util import assert_equal, assert_not_equal
from ipaplatform.paths import paths
@pytest.mark.tier1
@pytest.mark.skipif(
not facts.is_ipa_configured(),
reason="Requires configured IPA server",
)
class test_jsplugins(Unauthorized_HTTP_test):
app_uri = '/ipa/ui/js/freeipa/plugins.js'
jsplugins = (('foo', 'foo.js'), ('bar', ''))
content_type = 'application/javascript'
def test_jsplugins(self):
empty_response = "define([],function(){return[];});"
# Step 1: make sure default response has no additional plugins
response = self.send_request(method='GET')
assert_equal(response.status, 200)
response_data = response.read().decode(encoding='utf-8')
assert_equal(response_data, empty_response)
# Step 2: add fake plugins
try:
for (d, f) in self.jsplugins:
dir = os.path.join(paths.IPA_JS_PLUGINS_DIR, d)
if not os.path.exists(dir):
os.mkdir(dir, 0o755)
if f:
with open(os.path.join(dir, f), 'w') as js:
js.write("/* test js plugin */")
except OSError as e:
pytest.skip(
'Cannot set up test JS plugin: %s' % e
)
# Step 3: query plugins to see if our plugins exist
response = self.send_request(method='GET')
assert_equal(response.status, 200)
response_data = response.read().decode(encoding='utf-8')
assert_not_equal(response_data, empty_response)
for (d, f) in self.jsplugins:
if f:
assert "'" + d + "'" in response_data
else:
assert "'" + d + "'" not in response_data
# Step 4: remove fake plugins
try:
for (d, f) in self.jsplugins:
dir = os.path.join(paths.IPA_JS_PLUGINS_DIR, d)
file = os.path.join(dir, f)
if f and os.path.exists(file):
os.unlink(file)
if os.path.exists(dir):
os.rmdir(dir)
except OSError:
pass
# Step 5: make sure default response has no additional plugins
response = self.send_request(method='GET')
assert_equal(response.status, 200)
response_data = response.read().decode(encoding='utf-8')
assert_equal(response_data, empty_response)

23
ipatests/test_ipaserver/test_ldap.py
View File

@@ -27,6 +27,7 @@
from __future__ import absolute_import
from datetime import datetime, timedelta
import os
import sys
@@ -140,6 +141,28 @@ class test_ldap:
cert = entry_attrs.get('usercertificate')[0]
assert cert.serial_number is not None
def test_generalized_time(self):
"""
Test that LDAP generalized time is converted to/from datetime
"""
self.conn = ldap2(api)
try:
self.conn.connect(autobind=True)
except errors.ACIError:
pytest.skip("Only executed as root")
if not api.Backend.rpcclient.isconnected():
api.Backend.rpcclient.connect()
newdate = datetime.now() + timedelta(days=365)
lastdate = api.Backend.ldap2.encode(newdate).decode('utf-8')
api.Command["user_mod"](
"admin",
**dict(setattr=("krbprincipalexpiration=%s" % lastdate))
)
result = api.Command["user_find"](
**dict(krbprincipalexpiration=lastdate)
)
assert result['count'] == 1
@pytest.mark.tier0
@pytest.mark.needs_ipaapi

88
ipatests/test_ipaserver/test_login_password.py Normal file
View File

@@ -0,0 +1,88 @@
# Copyright (C) 2023 Red Hat
# see file 'COPYING' for use and warranty information
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
import pytest
import uuid
from ipatests.test_ipaserver.httptest import Unauthorized_HTTP_test
from ipatests.test_xmlrpc.xmlrpc_test import XMLRPC_test
from ipatests.util import assert_equal
from ipalib import api, errors
from ipapython.ipautil import run
testuser = u'tuser'
password = u'password'
@pytest.mark.tier1
class test_login_password(XMLRPC_test, Unauthorized_HTTP_test):
app_uri = '/ipa/session/login_password'
@pytest.fixture(autouse=True)
def login_setup(self, request):
ccache = os.path.join('/tmp', str(uuid.uuid4()))
try:
api.Command['user_add'](uid=testuser, givenname=u'Test', sn=u'User')
api.Command['passwd'](testuser, password=password)
run(['kinit', testuser], stdin='{0}\n{0}\n{0}\n'.format(password),
env={"KRB5CCNAME": ccache})
except errors.ExecutionError as e:
pytest.skip(
'Cannot set up test user: %s' % e
)
def fin():
try:
api.Command['user_del']([testuser])
except errors.NotFound:
pass
os.unlink(ccache)
request.addfinalizer(fin)
def _login(self, user, password, host=None):
return self.send_request(params={'user': str(user),
'password' : str(password)},
host=host)
def test_bad_options(self):
for params in (
None, # no params
{"user": "foo"}, # missing options
{"user": "foo", "password": ""}, # empty option
):
response = self.send_request(params=params)
assert_equal(response.status, 400)
assert_equal(response.reason, 'Bad Request')
def test_invalid_auth(self):
response = self._login(testuser, 'wrongpassword')
assert_equal(response.status, 401)
assert_equal(response.getheader('X-IPA-Rejection-Reason'),
'invalid-password')
def test_invalid_referer(self):
response = self._login(testuser, password, 'attacker.test')
assert_equal(response.status, 400)
def test_success(self):
response = self._login(testuser, password)
assert_equal(response.status, 200)
assert response.getheader('X-IPA-Rejection-Reason') is None

136
ipatests/test_ipaserver/test_referer.py Normal file
View File

@@ -0,0 +1,136 @@
# Copyright (C) 2023 Red Hat
# see file 'COPYING' for use and warranty information
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
import pytest
import uuid
from ipatests.test_ipaserver.httptest import Unauthorized_HTTP_test
from ipatests.test_xmlrpc.xmlrpc_test import XMLRPC_test
from ipatests.util import assert_equal
from ipalib import api, errors
from ipapython.ipautil import run
testuser = u'tuser'
password = u'password'
@pytest.mark.tier1
class test_referer(XMLRPC_test, Unauthorized_HTTP_test):
@pytest.fixture(autouse=True)
def login_setup(self, request):
ccache = os.path.join('/tmp', str(uuid.uuid4()))
tokenid = None
try:
api.Command['user_add'](uid=testuser, givenname=u'Test', sn=u'User')
api.Command['passwd'](testuser, password=password)
run(['kinit', testuser], stdin='{0}\n{0}\n{0}\n'.format(password),
env={"KRB5CCNAME": ccache})
result = api.Command["otptoken_add"](
type='HOTP', description='testotp',
ipatokenotpalgorithm='sha512', ipatokenowner=testuser,
ipatokenotpdigits='6')
tokenid = result['result']['ipatokenuniqueid'][0]
except errors.ExecutionError as e:
pytest.skip(
'Cannot set up test user: %s' % e
)
def fin():
try:
api.Command['user_del']([testuser])
api.Command['otptoken_del']([tokenid])
except errors.NotFound:
pass
os.unlink(ccache)
request.addfinalizer(fin)
def _request(self, params={}, host=None):
# implicit is that self.app_uri is set to the appropriate value
return self.send_request(params=params, host=host)
def test_login_password_valid(self):
"""Valid authentication of a user"""
self.app_uri = "/ipa/session/login_password"
response = self._request(
params={'user': 'tuser', 'password': password})
assert_equal(response.status, 200, self.app_uri)
def test_change_password_valid(self):
"""This actually changes the user password"""
self.app_uri = "/ipa/session/change_password"
response = self._request(
params={'user': 'tuser',
'old_password': password,
'new_password': 'new_password'}
)
assert_equal(response.status, 200, self.app_uri)
def test_sync_token_valid(self):
"""We aren't testing that sync works, just that we can get there"""
self.app_uri = "/ipa/session/sync_token"
response = self._request(
params={'user': 'tuser',
'first_code': '1234',
'second_code': '5678',
'password': 'password'})
assert_equal(response.status, 200, self.app_uri)
def test_i18n_messages_valid(self):
# i18n_messages requires a valid JSON request and we send
# nothing. If we get a 500 error then it got past the
# referer check.
self.app_uri = "/ipa/i18n_messages"
response = self._request()
assert_equal(response.status, 500, self.app_uri)
# /ipa/session/login_x509 is not tested yet as it requires
# significant additional setup.
# This can be manually verified by adding
# Satisfy Any and Require all granted to the configuration
# section and comment out all Auth directives. The request
# will fail and log that there is no KRB5CCNAME which comes
# after the referer check.
def test_endpoints_auth_required(self):
"""Test endpoints that require pre-authorization which will
fail before we even get to the Referer check
"""
self.endpoints = {
"/ipa/xml",
"/ipa/session/login_kerberos",
"/ipa/session/json",
"/ipa/session/xml"
}
for self.app_uri in self.endpoints:
response = self._request(host="attacker.test")
# referer is checked after auth
assert_equal(response.status, 401, self.app_uri)
def notest_endpoints_invalid(self):
"""Pass in a bad Referer, expect a 400 Bad Request"""
self.endpoints = {
"/ipa/session/login_password",
"/ipa/session/change_password",
"/ipa/session/sync_token",
}
for self.app_uri in self.endpoints:
response = self._request(host="attacker.test")
assert_equal(response.status, 400, self.app_uri)

233
ipatests/test_ipaserver/test_secure_ajp_connector.py Normal file
View File

@@ -0,0 +1,233 @@
# Copyright (C) 2021 FreeIPA Project Contributors - see LICENSE file
from collections import namedtuple
from io import BytesIO
from lxml.etree import parse as myparse
import pytest
import textwrap
from unittest.mock import mock_open, patch
from ipaplatform.constants import constants
from ipaserver.install import dogtaginstance
class MyDogtagInstance(dogtaginstance.DogtagInstance):
"""Purpose is to avoid reading configuration files.
The real DogtagInstance will open up the system store and
try to determine the actual version of tomcat installed.
Fake it instead.
"""
def __init__(self, is_newer):
self.service_user = constants.PKI_USER
self.ajp_secret = None
self.is_newer = is_newer
def _is_newer_tomcat_version(self, default=None):
return self.is_newer
def mock_etree_parse(data):
"""Convert a string into a file-like object to pass in the XML"""
f = BytesIO(data.strip().encode('utf-8'))
return myparse(f)
def mock_pkiuser_entity():
"""Return struct_passwd for mocked pkiuser"""
StructPasswd = namedtuple(
"StructPasswd",
[
"pw_name",
"pw_passwd",
"pw_uid",
"pw_gid",
"pw_gecos",
"pw_dir",
"pw_shell",
]
)
pkiuser_entity = StructPasswd(
constants.PKI_USER,
pw_passwd="x",
pw_uid=-1,
pw_gid=-1,
pw_gecos="",
pw_dir="/dev/null",
pw_shell="/sbin/nologin",
)
return pkiuser_entity
# Format of test_data is:
# (
# is_newer_tomcat (boolean),
# XML input,
# expected secret attribute(s),
# expected password(s),
# rewrite of XML file expected (boolean),
# )
test_data = (
(
# Case 1: Upgrade requiredSecret to secret
True,
textwrap.dedent("""
<?xml version="1.0" encoding="UTF-8"?>
<Server port="1234" shutdown="SHUTDOWN">
<Service name="Catalina">
<Connector port="9000" protocol="AJP/1.3" redirectPort="443"
address="localhost" requiredSecret="testing_ajp_secret" />
</Service>
</Server>
"""),
('secret',),
('testing_ajp_secret',),
('requiredSecret',),
True,
),
(
# Case 2: One connector with secret, no update is needed
True,
textwrap.dedent("""
<?xml version="1.0" encoding="UTF-8"?>
<Server port="1234" shutdown="SHUTDOWN">
<Service name="Catalina">
<Connector port="9000" protocol="AJP/1.3" redirectPort="443"
address="localhost" secret="testing_ajp_secret" />
</Service>
</Server>
"""),
('secret',),
('testing_ajp_secret',),
('requiredSecret',),
False,
),
(
# Case 3: Two connectors, old secret attribute, different secrets
True,
textwrap.dedent("""
<?xml version="1.0" encoding="UTF-8"?>
<Server port="1234" shutdown="SHUTDOWN">
<Service name="Catalina">
<Connector port="9000" protocol="AJP/1.3" redirectPort="443"
address="localhost4" requiredSecret="testing_ajp_secret" />
<Connector port="9000" protocol="AJP/1.3" redirectPort="443"
address="localhost6" requiredSecret="other_secret" />
</Service>
</Server>
"""),
('secret', 'secret'),
('testing_ajp_secret', 'testing_ajp_secret'),
('requiredSecret', 'requiredSecret'),
True,
),
(
# Case 4: Two connectors, new secret attribute, same secrets
True,
textwrap.dedent("""
<?xml version="1.0" encoding="UTF-8"?>
<Server port="1234" shutdown="SHUTDOWN">
<Service name="Catalina">
<Connector port="9000" protocol="AJP/1.3" redirectPort="443"
address="localhost4" secret="testing_ajp_secret" />
<Connector port="9000" protocol="AJP/1.3" redirectPort="443"
address="localhost6" secret="testing_ajp_secret" />
</Service>
</Server>
"""),
('secret', 'secret'),
('testing_ajp_secret', 'testing_ajp_secret'),
('requiredSecret', 'requiredSecret'),
False,
),
(
# Case 5: Two connectors, no secrets
True,
textwrap.dedent("""
<?xml version="1.0" encoding="UTF-8"?>
<Server port="1234" shutdown="SHUTDOWN">
<Service name="Catalina">
<Connector port="9000" protocol="AJP/1.3" redirectPort="443"
address="localhost4" />
<Connector port="9000" protocol="AJP/1.3" redirectPort="443"
address="localhost6" />
</Service>
</Server>
"""),
('secret', 'secret'),
('RANDOM', 'RANDOM'),
('requiredSecret', 'requiredSecret'),
True,
),
(
# Case 6: Older tomcat, no update needed for requiredSecret
False,
textwrap.dedent("""
<?xml version="1.0" encoding="UTF-8"?>
<Server port="1234" shutdown="SHUTDOWN">
<Service name="Catalina">
<Connector port="9000" protocol="AJP/1.3" redirectPort="443"
address="localhost" requiredSecret="testing_ajp_secret" />
</Service>
</Server>
"""),
('requiredSecret',),
('testing_ajp_secret',),
('secret',),
False,
),
(
# Case 7: Older tomcat, both secrets are present, one s/b removed
False,
textwrap.dedent("""
<?xml version="1.0" encoding="UTF-8"?>
<Server port="1234" shutdown="SHUTDOWN">
<Service name="Catalina">
<Connector port="9000" protocol="AJP/1.3" redirectPort="443"
address="localhost" requiredSecret="testing_ajp_secret"
secret="other_secret" />
</Service>
</Server>
"""),
('requiredSecret',),
('testing_ajp_secret',),
('secret',),
True,
),
)
class TestAJPSecretUpgrade:
@patch("ipaplatform.base.constants.pwd.getpwnam")
@patch("ipaplatform.base.constants.os.chown")
@patch("ipaserver.install.dogtaginstance.lxml.etree.parse")
@pytest.mark.parametrize("test_data", test_data)
def test_connecter(self, mock_parse, mock_chown, mock_getpwnam, test_data):
is_newer, data, secret, expect, ex_secret, rewrite = test_data
mock_chown.return_value = None
mock_parse.return_value = mock_etree_parse(data)
mock_getpwnam.return_value = mock_pkiuser_entity()
dogtag = MyDogtagInstance(is_newer)
with patch('ipaserver.install.dogtaginstance.open', mock_open()) \
as mocked_file:
dogtag.secure_ajp_connector()
if rewrite:
newdata = mocked_file().write.call_args
f = BytesIO(newdata[0][0])
server_xml = myparse(f)
doc = server_xml.getroot()
connectors = doc.xpath('//Connector[@protocol="AJP/1.3"]')
assert len(connectors) == len(secret)
i = 0
for connector in connectors:
if expect[i] == 'RANDOM':
assert connector.attrib[secret[i]]
else:
assert connector.attrib[secret[i]] == expect[i]
assert connector.attrib.get(ex_secret[i]) is None
i += 1
else:
assert mocked_file().write.call_args is None

13
ipatests/test_ipaserver/test_topology_plugin.py
View File

@@ -10,6 +10,8 @@ from ipapython.dn import DN
import pytest
REPL_PLUGIN_NAME_TEMPLATE = 'Multi%s Replication Plugin'
@pytest.mark.tier1
class TestTopologyPlugin:
"""
@@ -35,12 +37,13 @@ class TestTopologyPlugin:
@pytest.mark.skipif(os.path.isfile(pwfile) is False,
reason="You did not provide a .dmpw file with the DM password")
def test_topologyplugin(self):
supplier = REPL_PLUGIN_NAME_TEMPLATE % 'supplier'
pluginattrs = {
u'nsslapd-pluginPath': [u'libtopology'],
u'nsslapd-pluginVendor': [u'freeipa'],
u'cn': [u'IPA Topology Configuration'],
u'nsslapd-plugin-depends-on-named':
[u'Multimaster Replication Plugin', u'ldbm database'],
[supplier, u'ldbm database'],
u'nsslapd-topo-plugin-shared-replica-root': [u'dc=example,dc=com'],
u'nsslapd-pluginVersion': [u'1.0'],
u'nsslapd-topo-plugin-shared-config-base':
@@ -72,5 +75,13 @@ class TestTopologyPlugin:
bind_pw=dm_password)
entry = self.conn.get_entry(topoplugindn)
assert(set(entry.keys()) == set(pluginattrs.keys()))
# Handle different names for replication plugin
key = 'nsslapd-plugin-depends-on-named'
plugin_dependencies = entry[key]
if supplier not in plugin_dependencies:
mm = REPL_PLUGIN_NAME_TEMPLATE % 'master'
pluginattrs[key] = [mm, 'ldbm database']
for i in checkvalues:
assert(set(pluginattrs[i]) == set(entry[i]))