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

Imported Upstream version 4.7.2

Browse Source
This commit is contained in:
Mario Fetka
2021-08-09 20:54:00 +02:00
parent 3bfaa6e020
commit a791de49a2
2175 changed files with 1764288 additions and 331861 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
ipatests/test_ipaserver/__init__.py
View File

@@ -20,3 +20,7 @@
"""
Sub-package containing unit tests for `ipaserver` package.
"""
import ipatests.util
ipatests.util.check_ipaclient_unittests()

16
ipatests/test_ipaserver/data/pskc-figure7.xml
View File

@@ -8,14 +8,14 @@
<xenc11:DerivedKey>
<xenc11:KeyDerivationMethod
Algorithm="http://www.rsasecurity.com/rsalabs/pkcs/schemas/pkcs-5v2-0#pbkdf2">
<xenc11:PBKDF2-params>
<xenc11:Salt>
<xenc11:Specified>Ej7/PEpyEpw=</xenc11:Specified>
</xenc11:Salt>
<xenc11:IterationCount>1000</xenc11:IterationCount>
<xenc11:KeyLength>16</xenc11:KeyLength>
<xenc11:PRF/>
</xenc11:PBKDF2-params>
<pkcs5:PBKDF2-params>
<Salt>
<Specified>Ej7/PEpyEpw=</Specified>
</Salt>
<IterationCount>1000</IterationCount>
<KeyLength>16</KeyLength>
<PRF/>
</pkcs5:PBKDF2-params>
</xenc11:KeyDerivationMethod>
<xenc:ReferenceList>
<xenc:DataReference URI="#ED"/>

20
ipatests/test_ipaserver/httptest.py
View File

@@ -20,10 +20,9 @@
Base class for HTTP request tests
"""
import urllib
import httplib
from six.moves import urllib
from ipalib import api
from ipalib import api, util
class Unauthorized_HTTP_test(object):
"""
@@ -32,7 +31,9 @@ class Unauthorized_HTTP_test(object):
"""
app_uri = ''
host = api.env.host
cacert = api.env.tls_ca_cert
content_type = 'application/x-www-form-urlencoded'
accept_language = 'en-us'
def send_request(self, method='POST', params=None):
"""
@@ -41,12 +42,17 @@ class Unauthorized_HTTP_test(object):
:param key When not None, overrides default app_uri
"""
if params is not None:
params = urllib.urlencode(params, True)
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
headers = {'Content-Type' : self.content_type,
'Referer' : url}
headers = {'Content-Type': self.content_type,
'Accept-Language': self.accept_language,
'Referer': url}
conn = httplib.HTTPSConnection(self.host)
conn = util.create_https_connection(
self.host, cafile=self.cacert)
conn.request(method, self.app_uri, params, headers)
return conn.getresponse()

58
ipatests/test_ipaserver/test_adtrust_mockup.py Normal file
View File

@@ -0,0 +1,58 @@
# Copyright (C) 2018 FreeIPA Project Contributors - see LICENSE file
from __future__ import print_function
import ipaserver.install.adtrust as adtr
from ipaserver.install.adtrust import set_and_check_netbios_name
from collections import namedtuple
from unittest import TestCase
try:
from unittest import mock
except ImportError:
import mock
from io import StringIO
class ApiMockup(object):
Backend = namedtuple('Backend', 'ldap2')
Calls = namedtuple('Callbacks', 'retrieve_netbios_name')
env = namedtuple('Environment', 'domain')
class TestNetbiosName(TestCase):
@classmethod
def setUpClass(cls):
api = ApiMockup()
ldap2 = namedtuple('LDAP', 'isconnected')
ldap2.isconnected = mock.MagicMock(return_value=True)
api.Backend.ldap2 = ldap2
api.Calls.retrieve_netbios_name = adtr.retrieve_netbios_name
adtr.retrieve_netbios_name = mock.MagicMock(return_value=None)
cls.api = api
@classmethod
def tearDownClass(cls):
adtr.retrieve_netbios_name = cls.api.Calls.retrieve_netbios_name
def test_NetbiosName(self):
"""
Test set_and_check_netbios_name() using permutation of two inputs:
- predefined and not defined NetBIOS name
- unattended and interactive run
As result, the function has to return expected NetBIOS name in
all cases. For interactive run we override input to force what
we expect.
"""
self.api.env.domain = 'example.com'
expected_nname = 'EXAMPLE'
# NetBIOS name, unattended, should set the name?
tests = ((expected_nname, True, False),
(None, True, True),
(None, False, True),
(expected_nname, False, False))
with mock.patch('sys.stdin', new_callable=StringIO) as stdin:
stdin.write(expected_nname + '\r')
for test in tests:
nname, setname = set_and_check_netbios_name(
test[0], test[1], self.api)
assert expected_nname == nname
assert setname == test[2]

24
ipatests/test_ipaserver/test_changepw.py
View File

@@ -17,38 +17,40 @@
# 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 nose
import unittest
from httptest import Unauthorized_HTTP_test
import pytest
from ipatests.test_ipaserver.httptest import Unauthorized_HTTP_test
from ipatests.test_xmlrpc.xmlrpc_test import XMLRPC_test
from ipatests.util import assert_equal, assert_not_equal
from ipatests.util import assert_equal
from ipalib import api, errors
from ipapython.dn import DN
import ldap
from ipapython.ipaldap import ldap_initialize
testuser = u'tuser'
old_password = u'old_password'
new_password = u'new_password'
@pytest.mark.tier1
class test_changepw(XMLRPC_test, Unauthorized_HTTP_test):
app_uri = '/ipa/session/change_password'
def setUp(self):
super(test_changepw, self).setUp()
def setup(self):
try:
api.Command['user_add'](uid=testuser, givenname=u'Test', sn=u'User')
api.Command['passwd'](testuser, password=u'old_password')
except errors.ExecutionError, e:
raise nose.SkipTest(
except errors.ExecutionError as e:
raise unittest.SkipTest(
'Cannot set up test user: %s' % e
)
def tearDown(self):
def teardown(self):
try:
api.Command['user_del']([testuser])
except errors.NotFound:
pass
super(test_changepw, self).tearDown()
def _changepw(self, user, old_password, new_password):
return self.send_request(params={'user': str(user),
@@ -58,7 +60,7 @@ class test_changepw(XMLRPC_test, Unauthorized_HTTP_test):
def _checkpw(self, user, password):
dn = str(DN(('uid', user), api.env.container_user, api.env.basedn))
conn = ldap.initialize(api.env.ldap_uri)
conn = ldap_initialize(api.env.ldap_uri)
try:
conn.simple_bind_s(dn, password)
finally:

41
ipatests/test_ipaserver/test_dnssec.py Normal file
View File

@@ -0,0 +1,41 @@
#
# Copyright (C) 2016 FreeIPA Contributors see COPYING for license
#
"""
Test the `ipaserver/dnssec` package.
"""
import dns.name
from ipaserver.dnssec.odsmgr import ODSZoneListReader
ZONELIST_XML = """<?xml version="1.0" encoding="UTF-8"?>
<ZoneList>
<Zone name="ipa.example">
<Policy>default</Policy>
<Adapters>
<Input>
<Adapter type="File">/var/lib/ipa/dns/zone/entryUUID/12345</Adapter>
</Input>
<Output>
<Adapter type="File">/var/lib/ipa/dns/zone/entryUUID/12345</Adapter>
</Output>
</Adapters>
</Zone>
</ZoneList>
"""
def test_ods_zonelist_reader():
uuid = '12345'
name = dns.name.from_text('ipa.example.')
reader = ODSZoneListReader("<ZoneList/>")
assert reader.mapping == {}
assert reader.names == set()
assert reader.uuids == set()
reader = ODSZoneListReader(ZONELIST_XML)
assert reader.mapping == {uuid: name}
assert reader.names == {name}
assert reader.uuids == {uuid}

120
ipatests/test_ipaserver/test_i18n_messages.py Normal file
View File

@@ -0,0 +1,120 @@
#
# Copyright (C) 2018 FreeIPA Contributors see COPYING for license
#
import json
import os
import pytest
from ipalib import api
from ipatests.test_xmlrpc.xmlrpc_test import XMLRPC_test
from ipatests.test_ipaserver.httptest import Unauthorized_HTTP_test
from ipatests.util import (assert_equal, assert_deepequal, raises,
assert_not_equal)
from ipapython.version import API_VERSION
from ipaserver.plugins.internal import i18n_messages
@pytest.mark.tier1
class test_i18n_messages(XMLRPC_test, Unauthorized_HTTP_test):
"""
Tests for i18n_messages end point
"""
app_uri = '/ipa/i18n_messages'
content_type = 'application/json'
def _prepare_data(self, method):
"""
Construct json data required for request
"""
return {"method": "{0}".format(method),
"params": [[], {"version": "{0}".format(API_VERSION)}]}
def _prepare_accept_lang(self):
try:
return os.environ['LANGUAGE']
except KeyError:
pass
try:
return os.environ['LANG'].split('.')[0].replace('_', '-')
except KeyError:
return ''
def _fetch_i18n_msgs(self):
"""
Build translations directly via command instance
"""
return i18n_messages({}).execute()
def _fetch_i18n_msgs_http(self, accept_lang):
"""
Fetch translations via http request
"""
self.accept_language = accept_lang
params = json.dumps(self._prepare_data('i18n_messages'))
response = self.send_request(params=params)
assert_equal(response.status, 200)
response_data = response.read()
jsondata = json.loads(response_data)
assert_equal(jsondata['error'], None)
assert jsondata['result']
assert jsondata['result']['texts']
return jsondata['result']
def test_only_i18n_serves(self):
"""
Test if end point doesn't fulfill other RPC commands
"""
assert api.Command.get('user_find')
params = json.dumps(self._prepare_data('user_find/1'))
response = self.send_request(params=params)
assert_equal(response.status, 403)
assert_equal(response.reason, 'Forbidden')
response_data = response.read()
assert_equal(response_data, b'Invalid RPC command')
raises(ValueError, json.loads, response_data)
def test_only_post_serves(self):
"""
Test if end point fulfills only POST method
"""
params = json.dumps(self._prepare_data('i18n_messages'))
response = self.send_request(method='GET', params=params)
assert_equal(response.status, 405)
assert_equal(response.reason, 'Method Not Allowed')
assert response.msg
assert_equal(response.msg['allow'], 'POST')
response_data = response.read()
raises(ValueError, json.loads, response_data)
def test_i18n_receive(self):
"""
Test if translations request is successful
"""
expected_msgs = self._fetch_i18n_msgs()
actual_msgs = self._fetch_i18n_msgs_http(self._prepare_accept_lang())
assert_deepequal(expected_msgs, actual_msgs)
def test_i18n_consequence_receive(self):
"""
Test if consequence translations requests for different languages are
successful. Every request's result have to contain messages in it's
locale.
"""
prev_i18n_msgs = self._fetch_i18n_msgs_http('en-us')
cur_i18n_msgs = self._fetch_i18n_msgs_http('fr-fr')
try:
assert_equal(prev_i18n_msgs['texts']['true'], u'True')
assert_equal(cur_i18n_msgs['texts']['true'], u'Vrai')
except KeyError:
assert_not_equal(prev_i18n_msgs, cur_i18n_msgs)

0
ipatests/test_ipaserver/test_install/__init__.py Normal file
View File

12
ipatests/test_ipaserver/test_install/test_adtrustinstance.py Executable file → Normal file
View File

@@ -19,13 +19,17 @@
"""
Test `adtrustinstance`
"""
import os
import nose
import pytest
import six
from ipaserver.install import adtrustinstance
class test_adtrustinstance:
if six.PY3:
unicode = str
@pytest.mark.tier0
class test_adtrustinstance(object):
"""
Test `adtrustinstance`.
"""

96
ipatests/test_ipaserver/test_install/test_bindinstance.py Normal file
View File

@@ -0,0 +1,96 @@
#
# Copyright (C) 2018 FreeIPA Contributors. See COPYING for license
#
from __future__ import absolute_import
import tempfile
import pytest
from ipaplatform.paths import paths
from ipaserver.install.server.upgrade import named_add_crypto_policy
try:
from unittest.mock import patch # pylint: disable=import-error
except ImportError:
from mock import patch # pylint: disable=import-error
TEST_CONFIG = """
options {
\tdnssec-enable yes;
\tdnssec-validation yes;
};
include "random/file";
"""
EXPECTED_CONFIG = """
options {
\tdnssec-enable yes;
\tdnssec-validation yes;
\tinclude "/etc/crypto-policies/back-ends/bind.config";
};
include "random/file";
"""
# bindinstance.named_conf_exists() looks for a section like this
IPA_DYNDB_CONFIG = """
dyndb "ipa" "/usr/lib/bind/ldap.so" {
};
"""
POLICY_FILE = "/etc/crypto-policies/back-ends/bind.config"
@pytest.fixture
def namedconf():
with tempfile.NamedTemporaryFile('w+') as f:
with patch.multiple(paths,
NAMED_CONF=f.name,
NAMED_CRYPTO_POLICY_FILE=POLICY_FILE):
yield f.name
@patch('ipaserver.install.sysupgrade.get_upgrade_state')
@patch('ipaserver.install.sysupgrade.set_upgrade_state')
def test_add_crypto_policy(m_set, m_get, namedconf):
m_get.return_value = False
with open(namedconf, 'w') as f:
f.write(TEST_CONFIG)
f.write(IPA_DYNDB_CONFIG)
result = named_add_crypto_policy()
assert result
m_get.assert_called_with('named.conf', 'add_crypto_policy')
m_set.assert_called_with('named.conf', 'add_crypto_policy', True)
with open(namedconf) as f:
content = f.read()
assert content == ''.join([EXPECTED_CONFIG, IPA_DYNDB_CONFIG])
m_get.reset_mock()
m_set.reset_mock()
m_get.return_value = True
named_add_crypto_policy()
m_get.assert_called_with('named.conf', 'add_crypto_policy')
m_set.assert_not_called()
@patch('ipaserver.install.sysupgrade.get_upgrade_state')
@patch('ipaserver.install.sysupgrade.set_upgrade_state')
def test_add_crypto_policy_no_ipa(m_set, m_get, namedconf):
# Test if the update step is skipped when named.conf doesn't contain
# IPA related settings.
m_get.return_value = False
with open(namedconf, 'w') as f:
f.write(TEST_CONFIG)
result = named_add_crypto_policy()
assert not result
m_get.assert_not_called()
m_set.assert_not_called()

125
ipatests/test_ipaserver/test_install/test_cainstance.py Normal file
View File

@@ -0,0 +1,125 @@
#
# Copyright (C) 2017 FreeIPA Contributors see COPYING for license
#
from binascii import hexlify
import pickle
# pylint: disable=import-error
from six.moves.configparser import RawConfigParser
# pylint: enable=import-error
from six import StringIO
import pytest
from ipaserver.install import cainstance
pytestmark = pytest.mark.tier0
class test_ExternalCAProfile(object):
def test_MSCSTemplateV1_good(self):
o = cainstance.MSCSTemplateV1("MySubCA")
assert hexlify(o.get_ext_data()) == b'1e0e004d007900530075006200430041'
def test_MSCSTemplateV1_bad(self):
with pytest.raises(ValueError):
cainstance.MSCSTemplateV1("MySubCA:1")
def test_MSCSTemplateV1_pickle_roundtrip(self):
o = cainstance.MSCSTemplateV1("MySubCA")
s = pickle.dumps(o)
assert o.get_ext_data() == pickle.loads(s).get_ext_data()
def test_MSCSTemplateV2_too_few_parts(self):
with pytest.raises(ValueError):
cainstance.MSCSTemplateV2("1.2.3.4")
def test_MSCSTemplateV2_too_many_parts(self):
with pytest.raises(ValueError):
cainstance.MSCSTemplateV2("1.2.3.4:100:200:300")
def test_MSCSTemplateV2_bad_oid(self):
with pytest.raises(ValueError):
cainstance.MSCSTemplateV2("not_an_oid:1")
def test_MSCSTemplateV2_non_numeric_major_version(self):
with pytest.raises(ValueError):
cainstance.MSCSTemplateV2("1.2.3.4:major:200")
def test_MSCSTemplateV2_non_numeric_minor_version(self):
with pytest.raises(ValueError):
cainstance.MSCSTemplateV2("1.2.3.4:100:minor")
def test_MSCSTemplateV2_major_version_lt_zero(self):
with pytest.raises(ValueError):
cainstance.MSCSTemplateV2("1.2.3.4:-1:200")
def test_MSCSTemplateV2_minor_version_lt_zero(self):
with pytest.raises(ValueError):
cainstance.MSCSTemplateV2("1.2.3.4:100:-1")
def test_MSCSTemplateV2_major_version_gt_max(self):
with pytest.raises(ValueError):
cainstance.MSCSTemplateV2("1.2.3.4:4294967296:200")
def test_MSCSTemplateV2_minor_version_gt_max(self):
with pytest.raises(ValueError):
cainstance.MSCSTemplateV2("1.2.3.4:100:4294967296")
def test_MSCSTemplateV2_good_major(self):
o = cainstance.MSCSTemplateV2("1.2.3.4:4294967295")
assert hexlify(o.get_ext_data()) == b'300c06032a0304020500ffffffff'
def test_MSCSTemplateV2_good_major_minor(self):
o = cainstance.MSCSTemplateV2("1.2.3.4:4294967295:0")
assert hexlify(o.get_ext_data()) \
== b'300f06032a0304020500ffffffff020100'
def test_MSCSTemplateV2_pickle_roundtrip(self):
o = cainstance.MSCSTemplateV2("1.2.3.4:4294967295:0")
s = pickle.dumps(o)
assert o.get_ext_data() == pickle.loads(s).get_ext_data()
def test_ExternalCAProfile_dispatch(self):
"""
Test that constructing ExternalCAProfile actually returns an
instance of the appropriate subclass.
"""
assert isinstance(
cainstance.ExternalCAProfile("MySubCA"),
cainstance.MSCSTemplateV1)
assert isinstance(
cainstance.ExternalCAProfile("1.2.3.4:100"),
cainstance.MSCSTemplateV2)
def test_write_pkispawn_config_file_MSCSTemplateV1(self):
template = cainstance.MSCSTemplateV1(u"SubCA")
expected = (
'[CA]\n'
'pki_req_ext_oid = 1.3.6.1.4.1.311.20.2\n'
'pki_req_ext_data = 1e0a00530075006200430041\n\n'
)
self._test_write_pkispawn_config_file(template, expected)
def test_write_pkispawn_config_file_MSCSTemplateV2(self):
template = cainstance.MSCSTemplateV2(u"1.2.3.4:4294967295")
expected = (
'[CA]\n'
'pki_req_ext_oid = 1.3.6.1.4.1.311.21.7\n'
'pki_req_ext_data = 300c06032a0304020500ffffffff\n\n'
)
self._test_write_pkispawn_config_file(template, expected)
def _test_write_pkispawn_config_file(self, template, expected):
"""
Test that the values we read from an ExternalCAProfile
object can be used to produce a reasonable-looking pkispawn
configuration.
"""
config = RawConfigParser()
config.optionxform = str
config.add_section("CA")
config.set("CA", "pki_req_ext_oid", template.ext_oid)
config.set("CA", "pki_req_ext_data",
hexlify(template.get_ext_data()).decode('ascii'))
out = StringIO()
config.write(out)
assert out.getvalue() == expected

147
ipatests/test_ipaserver/test_install/test_installer.py Normal file
View File

@@ -0,0 +1,147 @@
#
# Copyright (C) 2018 FreeIPA Contributors. See COPYING for license
#
from __future__ import absolute_import
import six
from abc import ABCMeta, abstractproperty
from collections import namedtuple
import itertools
from ipatests.util import assert_equal
from ipaserver.install.ipa_replica_install import ReplicaInstall
Keyval = namedtuple('Keyval', ['option', 'value'])
class InstallerTestBase(six.with_metaclass(ABCMeta, object)):
OPTS_DICT = {}
# don't allow creating classes with tested_cls unspecified
@abstractproperty
def tested_cls(self):
return None
def setup_class(self):
"""Initializes the tested class so that it can be used later on
"""
self.tested_cls.make_parser()
assert \
getattr(self.tested_cls, 'option_parser', False), \
("Unable to generate option parser for {}"
.format(self.tested_cls.__name__))
self._populate_opts_dict()
@classmethod
def _populate_opts_dict(cls):
"""Populate the class-owned OPTS_DICT with available options
"""
if not getattr(cls.tested_cls, 'option_parser', False):
raise RuntimeError("You need to create the parser of the tested "
"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
# add options outside groups
for opt in cls.tested_cls.option_parser.option_list:
cls.OPTS_DICT[opt.dest] = opt._short_opts + opt._long_opts
def parse_cli_args(self, args):
"""Parses the CLI-like arguments and returns them in python objects
:param args: A string representing the CLI arguments
:returns: dictionary and a list of parsed options and arguments
"""
return self.tested_cls.option_parser.parse_args(args.split())
def get_installer_instance(self, args):
"""Get instance of the configuring class
"""
parsed_opts = self.parse_cli_args(args)
cls = self.tested_cls.get_command_class(*parsed_opts)
command_instance = cls(*parsed_opts)
return command_instance.init_configurator()
def combine_options(self, *args):
return ' '.join(args)
def all_option_permutations(self, *key_val):
"""Gets all short-option/long-option permutations
:param key_val: Keyval tuples specifying the options to grab from
OPTS_DICT along with the values to assign to them
:returns: list or list of lists of all permutations
"""
if len(key_val) == 0:
return []
elif len(key_val) == 1:
val = key_val[0].value
return ['{opt} {val}'.format(opt=opt, val=val)
for opt in self.OPTS_DICT[key_val[0].option]]
permutation_lists = [self.OPTS_DICT[k.option] for k in key_val]
permutation = itertools.product(*permutation_lists)
ret = []
for p in permutation:
opt_vals = []
for i, kv in enumerate(key_val):
opt_vals.append(
'{opt} {val}'.format(
opt=p[i], val=kv.value))
ret.append(opt_vals)
return ret
class TestReplicaInstaller(InstallerTestBase):
tested_cls = ReplicaInstall
PASSWORD = Keyval("auto_password",
"c3ca2246bcf309d1b636581ce429da3522a8aec4")
ADMIN_PASSWORD = Keyval("admin_password", "milan_je_buh123")
PRINCIPAL = Keyval("principal", "ubercool_guy")
def test_password_option_DL1(self):
# OTP enrollment
for passwd_opt in self.all_option_permutations(self.PASSWORD):
ic = self.get_installer_instance(passwd_opt)
assert_equal(ic.password, self.PASSWORD.value)
# admin principal enrollment
for adm_password_opt in (
self.all_option_permutations(self.ADMIN_PASSWORD)
):
ic = self.get_installer_instance(adm_password_opt)
assert_equal(ic.password, None)
assert_equal(ic.admin_password, self.ADMIN_PASSWORD.value)
# if principal is set, we interpret --password as that principal's
for passwd_opt, principal_opt in (
self.all_option_permutations(self.PASSWORD, self.PRINCIPAL)
):
ic = self.get_installer_instance(
self.combine_options(passwd_opt, principal_opt))
assert_equal(ic.password, None)
assert_equal(ic.principal, self.PRINCIPAL.value)
assert_equal(ic.admin_password, self.PASSWORD.value)
# if principal is set, we interpret --password as that principal's
# unless admin-password is also specified, in which case it's once
# again an OTP
for adm_password_opt, passwd_opt, principal_opt in (
self.all_option_permutations(
self.ADMIN_PASSWORD, self.PASSWORD, self.PRINCIPAL)
):
ic = self.get_installer_instance(
self.combine_options(
adm_password_opt, passwd_opt, principal_opt))
assert_equal(ic.password, self.PASSWORD.value)
assert_equal(ic.principal, self.PRINCIPAL.value)
assert_equal(ic.admin_password, self.ADMIN_PASSWORD.value)

152
ipatests/test_ipaserver/test_install/test_installutils.py Normal file
View File

@@ -0,0 +1,152 @@
#
# Copyright (C) 2017 FreeIPA Contributors. See COPYING for license
#
from __future__ import absolute_import
import binascii
import os
import re
import shutil
import subprocess
import tempfile
import textwrap
import pytest
from ipaplatform.paths import paths
from ipapython import ipautil
from ipaserver.install import installutils
from ipaserver.install import ipa_backup
from ipaserver.install import ipa_restore
@pytest.fixture
def tempdir(request):
tempdir = tempfile.mkdtemp()
def fin():
shutil.rmtree(tempdir)
request.addfinalizer(fin)
return tempdir
GPG_GENKEY = textwrap.dedent("""
%echo Generating a standard key
Key-Type: RSA
Key-Length: 2048
Name-Real: IPA Backup
Name-Comment: IPA Backup
Name-Email: root@example.com
Expire-Date: 0
Passphrase: {passphrase}
%commit
%echo done
""")
@pytest.fixture
def gpgkey(request, tempdir):
passphrase = "Secret123"
gnupghome = os.path.join(tempdir, "gnupg")
os.makedirs(gnupghome, 0o700)
# provide clean env for gpg test
env = os.environ.copy()
orig_gnupghome = env.get('GNUPGHOME')
env['GNUPGHOME'] = gnupghome
env['LC_ALL'] = 'C.UTF-8'
env['LANGUAGE'] = 'C'
devnull = open(os.devnull, 'w')
# allow passing passphrases to agent
with open(os.path.join(gnupghome, "gpg-agent.conf"), 'w') as f:
f.write("verbose\n")
f.write("allow-preset-passphrase\n")
# run agent in background
agent = subprocess.Popen(
[paths.GPG_AGENT, '--batch', '--daemon'],
env=env, stdout=devnull, stderr=devnull
)
def fin():
if orig_gnupghome is not None:
os.environ['GNUPGHOME'] = orig_gnupghome
else:
os.environ.pop('GNUPGHOME', None)
agent.kill()
agent.wait()
request.addfinalizer(fin)
# create public / private key pair
keygen = os.path.join(gnupghome, 'keygen')
with open(keygen, 'w') as f:
f.write(GPG_GENKEY.format(passphrase=passphrase))
subprocess.check_call(
[paths.GPG2, '--batch', '--gen-key', keygen],
env=env, stdout=devnull, stderr=devnull
)
# get keygrip of private key
out = subprocess.check_output(
[paths.GPG2, "--list-secret-keys", "--with-keygrip"],
env=env, stderr=subprocess.STDOUT
)
mo = re.search("Keygrip = ([A-Z0-9]{32,})", out.decode('utf-8'))
if mo is None:
raise ValueError(out.decode('utf-8'))
keygrip = mo.group(1)
# unlock private key
cmd = "PRESET_PASSPHRASE {} -1 {}".format(
keygrip,
binascii.hexlify(passphrase.encode('utf-8')).decode('utf-8')
)
subprocess.check_call(
[paths.GPG_CONNECT_AGENT, cmd, "/bye"],
env=env, stdout=devnull, stderr=devnull
)
# set env for the rest of the progress
os.environ['GNUPGHOME'] = gnupghome
def test_gpg_encrypt(tempdir):
src = os.path.join(tempdir, "data.txt")
encrypted = os.path.join(tempdir, "data.gpg")
decrypted = os.path.join(tempdir, "data.out")
passwd = 'Secret123'
payload = 'Dummy text\n'
with open(src, 'w') as f:
f.write(payload)
installutils.encrypt_file(src, encrypted, password=passwd)
assert os.path.isfile(encrypted)
installutils.decrypt_file(encrypted, decrypted, password=passwd)
assert os.path.isfile(decrypted)
with open(decrypted) as f:
assert f.read() == payload
with pytest.raises(ipautil.CalledProcessError):
installutils.decrypt_file(encrypted, decrypted, password='invalid')
def test_gpg_asymmetric(tempdir, gpgkey):
src = os.path.join(tempdir, "asymmetric.txt")
encrypted = src + ".gpg"
payload = 'Dummy text\n'
with open(src, 'w') as f:
f.write(payload)
ipa_backup.encrypt_file(src, remove_original=True)
assert os.path.isfile(encrypted)
assert not os.path.exists(src)
ipa_restore.decrypt_file(tempdir, encrypted)
assert os.path.isfile(src)
with open(src) as f:
assert f.read() == payload

2
ipatests/test_ipaserver/test_install/test_service.py
View File

@@ -22,8 +22,10 @@ Tests for the `ipaserver.service` module.
"""
from ipaserver.install import service
import pytest
@pytest.mark.tier0
def test_format_seconds():
assert service.format_seconds(0) == '0 seconds'
assert service.format_seconds(1) == '1 second'

293
ipatests/test_ipaserver/test_ipap11helper.py Normal file
View File

@@ -0,0 +1,293 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2015 FreeIPA Contributors see COPYING for license
#
"""
Test the `ipapython/ipap11helper/p11helper.c` module.
"""
from __future__ import absolute_import
from binascii import hexlify
import os
import os.path
import logging
import subprocess
import tempfile
import pytest
from ipaplatform.paths import paths
from ipaserver import p11helper as _ipap11helper
pytestmark = pytest.mark.tier0
CONFIG_DATA = """
# SoftHSM v2 configuration file
directories.tokendir = %s/tokens
objectstore.backend = file
"""
LIBSOFTHSM = paths.LIBSOFTHSM2_SO
SOFTHSM2_UTIL = paths.SOFTHSM2_UTIL
logging.basicConfig(level=logging.INFO)
log = logging.getLogger('t')
master_key_label = u"master-ž" # random non-ascii character to test unicode
master_key_id = "m"
replica1_key_label = u"replica1"
replica1_key_id = "id1"
replica1_import_label = u"replica1-import"
replica1_import_id = "id1-import"
replica1_new_label = u"replica1-new-label-ž"
replica2_key_label = u"replica2"
replica2_key_id = "id2"
replica_non_existent_label = u"replica-nonexistent"
@pytest.fixture(scope="module")
def token_path():
token_path = tempfile.mkdtemp(prefix='pytest_', suffix='_pkcs11')
os.mkdir(os.path.join(token_path, 'tokens'))
return token_path
@pytest.fixture(scope="module")
def p11(request, token_path):
with open(os.path.join(token_path, 'softhsm2.conf'), 'w') as cfg:
cfg.write(CONFIG_DATA % token_path)
args = [
SOFTHSM2_UTIL, '--init-token', '--free',
'--label', 'test',
'--pin', '1234',
'--so-pin', '1234'
]
os.environ['SOFTHSM2_CONF'] = os.path.join(token_path, 'softhsm2.conf')
subprocess.check_call(args, cwd=token_path)
try:
p11 = _ipap11helper.P11_Helper('test', "1234", LIBSOFTHSM)
except _ipap11helper.Error:
pytest.fail('Failed to initialize the helper object.', pytrace=False)
def fin():
try:
p11.finalize()
except _ipap11helper.Error:
pytest.fail('Failed to finalize the helper object.', pytrace=False)
finally:
subprocess.call(
[SOFTHSM2_UTIL, '--delete-token', '--label', 'test'],
cwd=token_path
)
del os.environ['SOFTHSM2_CONF']
request.addfinalizer(fin)
return p11
class test_p11helper(object):
def test_generate_master_key(self, p11):
assert p11.generate_master_key(master_key_label, master_key_id,
key_length=16, cka_wrap=True,
cka_unwrap=True)
def test_search_for_master_key(self, p11):
master_key = p11.find_keys(_ipap11helper.KEY_CLASS_SECRET_KEY,
label=master_key_label, id=master_key_id)
assert len(master_key) == 1, "The master key should exist."
def test_generate_replica_key_pair(self, p11):
assert p11.generate_replica_key_pair(replica1_key_label,
replica1_key_id,
pub_cka_wrap=True,
priv_cka_unwrap=True)
def test_find_key(self, p11):
rep1_pub = p11.find_keys(_ipap11helper.KEY_CLASS_PUBLIC_KEY,
label=replica1_key_label, cka_wrap=True)
assert len(rep1_pub) == 1, ("replica key pair has to contain "
"1 pub key instead of %s" % len(rep1_pub))
rep1_priv = p11.find_keys(_ipap11helper.KEY_CLASS_PRIVATE_KEY,
label=replica1_key_label, cka_unwrap=True)
assert len(rep1_priv) == 1, ("replica key pair has to contain 1 "
"private key instead of %s" %
len(rep1_priv))
def test_find_key_by_uri(self, p11):
rep1_pub = p11.find_keys(uri="pkcs11:object=replica1;objecttype=public")
assert len(rep1_pub) == 1, ("replica key pair has to contain 1 pub "
"key instead of %s" % len(rep1_pub))
def test_get_attribute_from_object(self, p11):
rep1_pub = p11.find_keys(_ipap11helper.KEY_CLASS_PUBLIC_KEY,
label=replica1_key_label, cka_wrap=True)[0]
iswrap = p11.get_attribute(rep1_pub, _ipap11helper.CKA_WRAP)
assert iswrap is True, "replica public key has to have CKA_WRAP = TRUE"
def test_generate_replica_keypair_with_extractable_private_key(self, p11):
assert p11.generate_replica_key_pair(replica2_key_label,
replica2_key_id,
pub_cka_wrap=True,
priv_cka_unwrap=True,
priv_cka_extractable=True)
def test_find_key_on_nonexistent_key_pair(self, p11):
test_list = p11.find_keys(_ipap11helper.KEY_CLASS_PUBLIC_KEY,
label=replica_non_existent_label)
assert len(test_list) == 0, ("list should be empty because label "
"'%s' should not exist" %
replica_non_existent_label)
def test_export_import_of_public_key(self, p11, token_path):
rep1_pub = p11.find_keys(_ipap11helper.KEY_CLASS_PUBLIC_KEY,
label=replica1_key_label, cka_wrap=True)[0]
pub = p11.export_public_key(rep1_pub)
log.debug("Exported public key %s", hexlify(pub))
pubfile = os.path.join(token_path, "public_key.asn1.der")
with open(pubfile, "wb") as f:
f.write(pub)
rep1_pub_import = p11.import_public_key(replica1_import_label,
replica1_import_id,
pub,
cka_wrap=True)
log.debug('imported replica 1 public key: %s', rep1_pub_import)
# test public key import
rep1_modulus_orig = p11.get_attribute(rep1_pub,
_ipap11helper.CKA_MODULUS)
rep1_modulus_import = p11.get_attribute(rep1_pub_import,
_ipap11helper.CKA_MODULUS)
log.debug('rep1_modulus_orig = 0x%s', hexlify(rep1_modulus_orig))
log.debug('rep1_modulus_import = 0x%s', hexlify(rep1_modulus_import))
assert rep1_modulus_import == rep1_modulus_orig
rep1_pub_exp_orig = p11.get_attribute(
rep1_pub, _ipap11helper.CKA_PUBLIC_EXPONENT)
rep1_pub_exp_import = p11.get_attribute(
rep1_pub_import, _ipap11helper.CKA_PUBLIC_EXPONENT)
log.debug('rep1_pub_exp_orig = 0x%s', hexlify(rep1_pub_exp_orig))
log.debug('rep1_pub_exp_import = 0x%s', hexlify(rep1_pub_exp_import))
assert rep1_pub_exp_import == rep1_pub_exp_orig
def test_wrap_unwrap_key_by_master_key_with_AES(self, p11, token_path):
master_key = p11.find_keys(_ipap11helper.KEY_CLASS_SECRET_KEY,
label=master_key_label, id=master_key_id)[0]
rep2_priv = p11.find_keys(_ipap11helper.KEY_CLASS_PRIVATE_KEY,
label=replica2_key_label, cka_unwrap=True)[0]
log.debug("wrapping dnssec priv key by master key")
wrapped_priv = p11.export_wrapped_key(
rep2_priv, master_key, _ipap11helper.MECH_AES_KEY_WRAP_PAD
)
assert wrapped_priv
log.debug("wrapped_dnssec priv key: %s", hexlify(wrapped_priv))
privfile = os.path.join(token_path, "wrapped_priv.der")
with open(privfile, "wb") as f:
f.write(wrapped_priv)
assert p11.import_wrapped_private_key(
u'test_import_wrapped_priv',
'1',
wrapped_priv,
master_key,
_ipap11helper.MECH_AES_KEY_WRAP_PAD,
_ipap11helper.KEY_TYPE_RSA
)
def test_wrap_unwrap_key_by_master_key_with_RSA_PKCS(self, p11):
master_key = p11.find_keys(_ipap11helper.KEY_CLASS_SECRET_KEY,
label=master_key_label, id=master_key_id)[0]
rep2_pub = p11.find_keys(_ipap11helper.KEY_CLASS_PUBLIC_KEY,
label=replica2_key_label, cka_wrap=True)[0]
rep2_priv = p11.find_keys(_ipap11helper.KEY_CLASS_PRIVATE_KEY,
label=replica2_key_label, cka_unwrap=True)[0]
wrapped = p11.export_wrapped_key(master_key,
rep2_pub,
_ipap11helper.MECH_RSA_PKCS)
assert wrapped
log.debug("wrapped key MECH_RSA_PKCS (secret master wrapped by pub "
"key): %s", hexlify(wrapped))
assert p11.import_wrapped_secret_key(u'test_import_wrapped',
'2',
wrapped,
rep2_priv,
_ipap11helper.MECH_RSA_PKCS,
_ipap11helper.KEY_TYPE_AES)
def test_wrap_unwrap_by_master_key_with_RSA_PKCS_OAEP(self, p11):
master_key = p11.find_keys(_ipap11helper.KEY_CLASS_SECRET_KEY,
label=master_key_label, id=master_key_id)[0]
rep2_pub = p11.find_keys(_ipap11helper.KEY_CLASS_PUBLIC_KEY,
label=replica2_key_label, cka_wrap=True)[0]
rep2_priv = p11.find_keys(_ipap11helper.KEY_CLASS_PRIVATE_KEY,
label=replica2_key_label, cka_unwrap=True)[0]
wrapped = p11.export_wrapped_key(master_key,
rep2_pub,
_ipap11helper.MECH_RSA_PKCS_OAEP)
assert wrapped
log.debug("wrapped key MECH_RSA_PKCS_OAEP (secret master wrapped by "
"pub key): %s", hexlify(wrapped))
assert p11.import_wrapped_secret_key(u'test_import_wrapped',
'3',
wrapped,
rep2_priv,
_ipap11helper.MECH_RSA_PKCS_OAEP,
_ipap11helper.KEY_TYPE_AES)
def test_set_attribute_on_object(self, p11):
rep1_pub = p11.find_keys(_ipap11helper.KEY_CLASS_PUBLIC_KEY,
label=replica1_key_label, cka_wrap=True)[0]
test_label = replica1_new_label
p11.set_attribute(rep1_pub, _ipap11helper.CKA_LABEL, test_label)
assert p11.get_attribute(rep1_pub, _ipap11helper.CKA_LABEL) \
== test_label, "The labels do not match."
def test_do_not_generate_identical_master_keys(self, p11):
with pytest.raises(_ipap11helper.DuplicationError):
p11.generate_master_key(master_key_label, master_key_id,
key_length=16)
master_key = p11.find_keys(_ipap11helper.KEY_CLASS_SECRET_KEY,
label=master_key_label)
assert len(master_key) == 1, ("There shouldn't be multiple keys "
"with the same label.")
def test_delete_key(self, p11):
master_key = p11.find_keys(_ipap11helper.KEY_CLASS_SECRET_KEY,
label=master_key_label, id=master_key_id)[0]
rep1_pub = p11.find_keys(_ipap11helper.KEY_CLASS_PUBLIC_KEY,
label=replica1_new_label, cka_wrap=True)[0]
rep2_priv = p11.find_keys(_ipap11helper.KEY_CLASS_PRIVATE_KEY,
label=replica2_key_label, cka_unwrap=True)[0]
for key in (rep1_pub, rep2_priv, master_key):
p11.delete_key(key)
master_key = p11.find_keys(_ipap11helper.KEY_CLASS_SECRET_KEY,
label=master_key_label, id=master_key_id)
assert len(master_key) == 0, "The master key should be deleted."
rep1_pub = p11.find_keys(_ipap11helper.KEY_CLASS_PUBLIC_KEY,
label=replica1_new_label, cka_wrap=True)
assert len(rep1_pub) == 0, ("The public key of replica1 pair should "
"be deleted.")
rep2_priv = p11.find_keys(_ipap11helper.KEY_CLASS_PRIVATE_KEY,
label=replica2_key_label, cka_unwrap=True)
assert len(rep2_priv) == 0, ("The private key of replica2 pair should"
" be deleted.")

126
ipatests/test_ipaserver/test_kadmin.py Normal file
View File

@@ -0,0 +1,126 @@
#
# Copyright (C) 2016 FreeIPA Contributors see COPYING for license
#
"""
Test suite for creating principals via kadmin.local and modifying their keys
"""
import os
import pytest
import tempfile
from ipalib import api
from ipaserver.install import installutils
from ipatests.test_util import yield_fixture
@yield_fixture()
def keytab():
fd, keytab_path = tempfile.mkstemp(suffix='.keytab')
os.close(fd)
try:
yield keytab_path
finally:
try:
os.remove(keytab_path)
except OSError:
pass
@pytest.fixture()
def service_in_kerberos_subtree(request):
princ = u'svc1/{0.host}@{0.realm}'.format(api.env)
installutils.kadmin_addprinc(princ)
def fin():
try:
installutils.kadmin(
'delprinc -force {}'.format(princ))
except Exception:
pass
request.addfinalizer(fin)
return princ
@pytest.fixture()
def service_in_service_subtree(request):
princ = u'svc2/{0.host}@{0.realm}'.format(api.env)
rpcclient = api.Backend.rpcclient
was_connected = rpcclient.isconnected()
if not was_connected:
rpcclient.connect()
api.Command.service_add(princ)
def fin():
try:
api.Command.service_del(princ)
except Exception:
pass
try:
if not was_connected:
rpcclient.disconnect()
except Exception:
pass
request.addfinalizer(fin)
return princ
@pytest.fixture(params=[service_in_kerberos_subtree,
service_in_service_subtree])
def service(request):
return request.param(request)
@pytest.mark.skipif(
os.getuid() != 0, reason="kadmin.local is accesible only to root")
class TestKadmin(object):
def assert_success(self, command, *args):
"""
Since kadmin.local returns 0 also when internal errors occur, we have
to catch the command's stderr and check that it is empty
"""
result = command(*args)
assert not result.error_output
def test_create_keytab(self, service, keytab):
"""
tests that ktadd command works for both types of services
"""
self.assert_success(
installutils.create_keytab,
keytab,
service)
def test_change_key(self, service, keytab):
"""
tests that both types of service can have passwords changed using
kadmin
"""
self.assert_success(
installutils.create_keytab,
keytab,
service)
self.assert_success(
installutils.kadmin,
'change_password -randkey {}'.format(service))
def test_append_key(self, service, keytab):
"""
Tests that we can create a new keytab for both service types and then
append new keys to it
"""
self.assert_success(
installutils.create_keytab,
keytab,
service)
self.assert_success(
installutils.create_keytab,
keytab,
service)

166
ipatests/test_ipaserver/test_ldap.py
View File

@@ -25,34 +25,38 @@
# The DM password needs to be set in ~/.ipa/.dmpw
from __future__ import absolute_import
import os
import sys
import unittest
import nose
from nose.tools import assert_raises # pylint: disable=E0611
import nss.nss as nss
import pytest
import six
from ipaserver.plugins.ldap2 import ldap2
from ipalib.plugins.service import service, service_show
from ipalib.plugins.host import host
from ipalib import api, x509, create_api, errors
from ipapython import ipautil
from ipaplatform.paths import paths
from ipaserver.plugins.ldap2 import ldap2, AUTOBIND_DISABLED
from ipalib import api, create_api, errors
from ipapython.dn import DN
if six.PY3:
unicode = str
@pytest.mark.tier0
@pytest.mark.needs_ipaapi
class test_ldap(object):
"""
Test various LDAP client bind methods.
"""
def setUp(self):
def setup(self):
self.conn = None
self.ldapuri = 'ldap://%s' % ipautil.format_netloc(api.env.host)
self.ccache = paths.TMP_KRB5CC % os.getuid()
nss.nss_init_nodb()
self.ldapuri = api.env.ldap_uri
self.dn = DN(('krbprincipalname','ldap/%s@%s' % (api.env.host, api.env.realm)),
('cn','services'),('cn','accounts'),api.env.basedn)
def tearDown(self):
def teardown(self):
if self.conn and self.conn.isconnected():
self.conn.disconnect()
@@ -60,8 +64,8 @@ class test_ldap(object):
"""
Test an anonymous LDAP bind using ldap2
"""
self.conn = ldap2(shared_instance=False, ldap_uri=self.ldapuri)
self.conn.connect()
self.conn = ldap2(api)
self.conn.connect(autobind=AUTOBIND_DISABLED)
dn = api.env.basedn
entry_attrs = self.conn.get_entry(dn, ['associateddomain'])
domain = entry_attrs.single_value['associateddomain']
@@ -71,34 +75,29 @@ class test_ldap(object):
"""
Test a GSSAPI LDAP bind using ldap2
"""
if not ipautil.file_exists(self.ccache):
raise nose.SkipTest('Missing ccache %s' % self.ccache)
self.conn = ldap2(shared_instance=False, ldap_uri=self.ldapuri)
self.conn.connect(ccache='FILE:%s' % self.ccache)
self.conn = ldap2(api)
self.conn.connect(autobind=AUTOBIND_DISABLED)
entry_attrs = self.conn.get_entry(self.dn, ['usercertificate'])
cert = entry_attrs.get('usercertificate')
cert = cert[0]
serial = unicode(x509.get_serial_number(cert, x509.DER))
assert serial is not None
cert = entry_attrs.get('usercertificate')[0]
assert cert.serial_number is not None
def test_simple(self):
"""
Test a simple LDAP bind using ldap2
"""
pwfile = api.env.dot_ipa + os.sep + ".dmpw"
if ipautil.file_exists(pwfile):
fp = open(pwfile, "r")
dm_password = fp.read().rstrip()
fp.close()
if os.path.isfile(pwfile):
with open(pwfile, "r") as fp:
dm_password = fp.read().rstrip()
else:
raise nose.SkipTest("No directory manager password in %s" % pwfile)
self.conn = ldap2(shared_instance=False, ldap_uri=self.ldapuri)
raise unittest.SkipTest(
"No directory manager password in %s" % pwfile
)
self.conn = ldap2(api)
self.conn.connect(bind_dn=DN(('cn', 'directory manager')), bind_pw=dm_password)
entry_attrs = self.conn.get_entry(self.dn, ['usercertificate'])
cert = entry_attrs.get('usercertificate')
cert = cert[0]
serial = unicode(x509.get_serial_number(cert, x509.DER))
assert serial is not None
cert = entry_attrs.get('usercertificate')[0]
assert cert.serial_number is not None
def test_Backend(self):
"""
@@ -109,46 +108,40 @@ class test_ldap(object):
# a client-only api. Then we register in the commands and objects
# we need for the test.
myapi = create_api(mode=None)
myapi.bootstrap(context='cli', in_server=True, in_tree=True)
myapi.register(ldap2)
myapi.register(host)
myapi.register(service)
myapi.register(service_show)
myapi.bootstrap(context='cli', in_server=True, confdir=paths.ETC_IPA)
myapi.finalize()
pwfile = api.env.dot_ipa + os.sep + ".dmpw"
if ipautil.file_exists(pwfile):
fp = open(pwfile, "r")
dm_password = fp.read().rstrip()
fp.close()
if os.path.isfile(pwfile):
with open(pwfile, "r") as fp:
dm_password = fp.read().rstrip()
else:
raise nose.SkipTest("No directory manager password in %s" % pwfile)
raise unittest.SkipTest(
"No directory manager password in %s" % pwfile
)
myapi.Backend.ldap2.connect(bind_dn=DN(('cn', 'Directory Manager')), bind_pw=dm_password)
result = myapi.Command['service_show']('ldap/%s@%s' % (api.env.host, api.env.realm,))
entry_attrs = result['result']
cert = entry_attrs.get('usercertificate')
cert = cert[0]
serial = unicode(x509.get_serial_number(cert, x509.DER))
assert serial is not None
cert = entry_attrs.get('usercertificate')[0]
assert cert.serial_number is not None
def test_autobind(self):
"""
Test an autobind LDAP bind using ldap2
"""
ldapuri = 'ldapi://%%2fvar%%2frun%%2fslapd-%s.socket' % api.env.realm.replace('.','-')
self.conn = ldap2(shared_instance=False, ldap_uri=ldapuri)
self.conn = ldap2(api)
try:
self.conn.connect(autobind=True)
except errors.ACIError:
raise nose.SkipTest("Only executed as root")
raise unittest.SkipTest("Only executed as root")
entry_attrs = self.conn.get_entry(self.dn, ['usercertificate'])
cert = entry_attrs.get('usercertificate')
cert = cert[0]
serial = unicode(x509.get_serial_number(cert, x509.DER))
assert serial is not None
cert = entry_attrs.get('usercertificate')[0]
assert cert.serial_number is not None
@pytest.mark.tier0
@pytest.mark.needs_ipaapi
class test_LDAPEntry(object):
"""
Test the LDAPEntry class
@@ -158,14 +151,14 @@ class test_LDAPEntry(object):
dn1 = DN(('cn', cn1[0]))
dn2 = DN(('cn', cn2[0]))
def setUp(self):
self.ldapuri = 'ldap://%s' % ipautil.format_netloc(api.env.host)
self.conn = ldap2(shared_instance=False, ldap_uri=self.ldapuri)
self.conn.connect()
def setup(self):
self.ldapuri = api.env.ldap_uri
self.conn = ldap2(api)
self.conn.connect(autobind=AUTOBIND_DISABLED)
self.entry = self.conn.make_entry(self.dn1, cn=self.cn1)
def tearDown(self):
def teardown(self):
if self.conn and self.conn.isconnected():
self.conn.disconnect()
@@ -175,9 +168,15 @@ class test_LDAPEntry(object):
assert u'cn' in e
assert u'cn' in e.keys()
assert 'CN' in e
assert 'CN' not in e.keys()
if six.PY2:
assert 'CN' not in e.keys()
else:
assert 'CN' in e.keys()
assert 'commonName' in e
assert 'commonName' not in e.keys()
if six.PY2:
assert 'commonName' not in e.keys()
else:
assert 'commonName' in e.keys()
assert e['CN'] is self.cn1
assert e['CN'] is e[u'cn']
@@ -190,9 +189,15 @@ class test_LDAPEntry(object):
assert u'cn' in e
assert u'cn' in e.keys()
assert 'CN' in e
assert 'CN' not in e.keys()
if six.PY2:
assert 'CN' not in e.keys()
else:
assert 'CN' in e.keys()
assert 'commonName' in e
assert 'commonName' not in e.keys()
if six.PY2:
assert 'commonName' not in e.keys()
else:
assert 'commonName' in e.keys()
assert e['CN'] is self.cn2
assert e['CN'] is e[u'cn']
@@ -209,7 +214,7 @@ class test_LDAPEntry(object):
def test_popitem(self):
e = self.entry
assert e.popitem() == ('cn', self.cn1)
e.keys() == []
assert list(e) == []
def test_setdefault(self):
e = self.entry
@@ -229,7 +234,7 @@ class test_LDAPEntry(object):
assert e.pop('cn') == self.cn1
assert 'cn' not in e
assert e.pop('cn', 'default') is 'default'
with assert_raises(KeyError):
with pytest.raises(KeyError):
e.pop('cn')
def test_clear(self):
@@ -238,12 +243,19 @@ class test_LDAPEntry(object):
assert not e
assert 'cn' not in e
@pytest.mark.skipif(sys.version_info >= (3, 0), reason="Python 2 only")
def test_has_key(self):
e = self.entry
assert not e.has_key('xyz')
assert e.has_key('cn')
assert e.has_key('COMMONNAME')
def test_in(self):
e = self.entry
assert 'xyz' not in e
assert 'cn' in e
assert 'COMMONNAME' in e
def test_get(self):
e = self.entry
assert e.get('cn') == self.cn1
@@ -265,35 +277,35 @@ class test_LDAPEntry(object):
assert e['test'] is nice
raw = e.raw['test']
assert raw == ['1', '2', '3']
assert raw == [b'1', b'2', b'3']
nice.remove(1)
assert e.raw['test'] is raw
assert raw == ['2', '3']
assert raw == [b'2', b'3']
raw.append('4')
raw.append(b'4')
assert e['test'] is nice
assert nice == [2, 3, u'4']
nice.remove(2)
raw.append('5')
raw.append(b'5')
assert nice == [3, u'4']
assert raw == ['2', '3', '4', '5']
assert raw == [b'2', b'3', b'4', b'5']
assert e['test'] is nice
assert e.raw['test'] is raw
assert nice == [3, u'4', u'5']
assert raw == ['3', '4', '5']
assert raw == [b'3', b'4', b'5']
nice.insert(0, 2)
raw.remove('4')
raw.remove(b'4')
assert nice == [2, 3, u'4', u'5']
assert raw == ['3', '5']
assert raw == [b'3', b'5']
assert e.raw['test'] is raw
assert e['test'] is nice
assert nice == [2, 3, u'5']
assert raw == ['3', '5', '2']
assert raw == [b'3', b'5', b'2']
raw = ['a', 'b']
raw = [b'a', b'b']
e.raw['test'] = raw
assert e['test'] is not nice
assert e['test'] == [u'a', u'b']
@@ -301,7 +313,7 @@ class test_LDAPEntry(object):
nice = 'not list'
e['test'] = nice
assert e['test'] is nice
assert e.raw['test'] == ['not list']
assert e.raw['test'] == [b'not list']
e.raw['test'].append('second')
e.raw['test'].append(b'second')
assert e['test'] == ['not list', u'second']

92
ipatests/test_ipaserver/test_migratepw.py Normal file
View File

@@ -0,0 +1,92 @@
#
# Copyright (C) 2018 FreeIPA Contributors see COPYING for license
#
import pytest
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
testuser = u'tuser'
password = u'password'
@pytest.mark.tier1
class test_migratepw(XMLRPC_test, Unauthorized_HTTP_test):
"""
Test password migrate end point
"""
app_uri = '/ipa/migration/migration.py'
def setup(self):
"""
Prepare for tests
"""
api.Command['user_add'](uid=testuser, givenname=u'Test', sn=u'User')
api.Command['passwd'](testuser, password=password)
def teardown(self):
"""
Clean up
"""
try:
api.Command['user_del']([testuser])
except errors.NotFound:
pass
def _migratepw(self, user, password, method='POST'):
"""
Make password migrate request to server
"""
return self.send_request(method, params={'username': str(user),
'password': str(password)},
)
def test_bad_params(self):
"""
Test against bad (missing, empty) params
"""
for params in (None, # no params
{'username': 'foo'}, # missing password
{'password': 'bar'}, # missing username
{'username': '',
'password': ''}, # empty options
{'username': '',
'password': 'bar'}, # empty username
{'username': 'foo',
'password': ''}, # empty password
):
response = self.send_request(params=params)
assert_equal(response.status, 400)
assert_equal(response.reason, 'Bad Request')
def test_not_post_method(self):
"""
Test redirection of non POST request
"""
response = self._migratepw(testuser, password, method='GET')
assert_equal(response.status, 302)
assert response.msg
assert_equal(response.msg['Location'], 'index.html')
def test_invalid_password(self):
"""
Test invalid password
"""
response = self._migratepw(testuser, 'wrongpassword')
assert_equal(response.status, 200)
assert_equal(response.getheader('X-IPA-Migrate-Result'),
'invalid-password')
def test_migration_success(self):
"""
Test successful migration scenario
"""
response = self._migratepw(testuser, password)
assert_equal(response.status, 200)
assert_equal(response.getheader('X-IPA-Migrate-Result'), 'ok')

131
ipatests/test_ipaserver/test_otptoken_import.py
View File

@@ -17,15 +17,16 @@
# 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 codecs
import os
import sys
import nose
from nss import nss
import pytest
from ipaserver.install.ipa_otptoken_import import PSKCDocument, ValidationError
from ipaserver.install.ipa_otptoken_import import convertHashName
basename = os.path.join(os.path.dirname(__file__), "data")
@pytest.mark.tier1
class test_otptoken_import(object):
def test_figure3(self):
doc = PSKCDocument(os.path.join(basename, "pskc-figure3.xml"))
@@ -35,7 +36,7 @@ class test_otptoken_import(object):
'ipatokenotpkey': u'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ',
'ipatokenvendor': u'Manufacturer',
'ipatokenserial': u'987654321',
'ipatokenhotpcounter': 0L,
'ipatokenhotpcounter': 0,
'ipatokenotpdigits': 8,
'type': u'hotp',
})]
@@ -61,91 +62,89 @@ class test_otptoken_import(object):
assert False
def test_figure6(self):
nss.nss_init_nodb()
try:
doc = PSKCDocument(os.path.join(basename, "pskc-figure6.xml"))
assert doc.keyname == 'Pre-shared-key'
doc.setKey('12345678901234567890123456789012'.decode('hex'))
assert [(t.id, t.options) for t in doc.getKeyPackages()] == \
[(u'12345678', {
'ipatokenotpkey': u'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ',
'ipatokenvendor': u'Manufacturer',
'ipatokenserial': u'987654321',
'ipatokenhotpcounter': 0L,
'ipatokenotpdigits': 8,
'type': u'hotp'})]
finally:
nss.nss_shutdown()
doc = PSKCDocument(os.path.join(basename, "pskc-figure6.xml"))
assert doc.keyname == 'Pre-shared-key'
doc.setKey(codecs.decode('12345678901234567890123456789012', 'hex'))
assert [(t.id, t.options) for t in doc.getKeyPackages()] == \
[(u'12345678', {
'ipatokenotpkey': u'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ',
'ipatokenvendor': u'Manufacturer',
'ipatokenserial': u'987654321',
'ipatokenhotpcounter': 0,
'ipatokenotpdigits': 8,
'type': u'hotp'})]
def test_figure7(self):
nss.nss_init_nodb()
try:
doc = PSKCDocument(os.path.join(basename, "pskc-figure7.xml"))
assert doc.keyname == 'My Password 1'
doc.setKey('qwerty')
assert [(t.id, t.options) for t in doc.getKeyPackages()] == \
[(u'123456', {
'ipatokenotpkey': u'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ',
'ipatokenvendor': u'TokenVendorAcme',
'ipatokenserial': u'987654321',
'ipatokenotpdigits': 8,
'type': u'hotp'})]
finally:
nss.nss_shutdown()
doc = PSKCDocument(os.path.join(basename, "pskc-figure7.xml"))
assert doc.keyname == 'My Password 1'
doc.setKey(b'qwerty')
assert [(t.id, t.options) for t in doc.getKeyPackages()] == \
[(u'123456', {
'ipatokenotpkey': u'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ',
'ipatokenvendor': u'TokenVendorAcme',
'ipatokenserial': u'987654321',
'ipatokenotpdigits': 8,
'type': u'hotp'})]
def test_figure8(self):
nss.nss_init_nodb()
try:
doc = PSKCDocument(os.path.join(basename, "pskc-figure8.xml"))
PSKCDocument(os.path.join(basename, "pskc-figure8.xml"))
except NotImplementedError: # X.509 is not supported.
pass
else:
assert False
finally:
nss.nss_shutdown()
def test_invalid(self):
nss.nss_init_nodb()
try:
doc = PSKCDocument(os.path.join(basename, "pskc-invalid.xml"))
PSKCDocument(os.path.join(basename, "pskc-invalid.xml"))
except ValueError: # File is invalid.
pass
else:
assert False
finally:
nss.nss_shutdown()
def test_mini(self):
nss.nss_init_nodb()
try:
doc = PSKCDocument(os.path.join(basename, "pskc-mini.xml"))
[(t.id, t.options) for t in doc.getKeyPackages()]
for t in doc.getKeyPackages():
t._PSKCKeyPackage__process()
except ValidationError: # Unsupported token type.
pass
else:
assert False
finally:
nss.nss_shutdown()
def test_full(self):
nss.nss_init_nodb()
try:
doc = PSKCDocument(os.path.join(basename, "full.xml"))
assert [(t.id, t.options) for t in doc.getKeyPackages()] == \
[(u'KID1', {
'ipatokenotpkey': u'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ',
'ipatokennotafter': u'20060531000000Z',
'ipatokennotbefore': u'20060501000000Z',
'ipatokenserial': u'SerialNo-IssueNo',
'ipatokentotpclockoffset': 60000,
'ipatokenotpalgorithm': u'sha1',
'ipatokenvendor': u'iana.dummy',
'description': u'FriendlyName',
'ipatokentotptimestep': 200,
'ipatokenhotpcounter': 0L,
'ipatokenmodel': u'Model',
'ipatokenotpdigits': 8,
'type': u'hotp',
})]
finally:
nss.nss_shutdown()
doc = PSKCDocument(os.path.join(basename, "full.xml"))
assert [(t.id, t.options) for t in doc.getKeyPackages()] == \
[(u'KID1', {
'ipatokenotpkey': u'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ',
'ipatokennotafter': u'20060531000000Z',
'ipatokennotbefore': u'20060501000000Z',
'ipatokenserial': u'SerialNo-IssueNo',
'ipatokentotpclockoffset': 60000,
'ipatokenotpalgorithm': u'sha1',
'ipatokenvendor': u'iana.dummy',
'description': u'FriendlyName',
'ipatokentotptimestep': 200,
'ipatokenhotpcounter': 0,
'ipatokenmodel': u'Model',
'ipatokenotpdigits': 8,
'type': u'hotp',
})]
def test_valid_tokens(self):
assert convertHashName('sha1') == u'sha1'
assert convertHashName('hmac-sha1') == u'sha1'
assert convertHashName('sha224') == u'sha224'
assert convertHashName('hmac-sha224') == u'sha224'
assert convertHashName('sha256') == u'sha256'
assert convertHashName('hmac-sha256') == u'sha256'
assert convertHashName('sha384') == u'sha384'
assert convertHashName('hmac-sha384') == u'sha384'
assert convertHashName('sha512') == u'sha512'
assert convertHashName('hmac-sha512') == u'sha512'
def test_invalid_tokens(self):
"""The conversion defaults to sha1 on unknown hashing"""
assert convertHashName('something-sha256') == u'sha1'
assert convertHashName('') == u'sha1'
assert convertHashName(None) == u'sha1'

58
ipatests/test_ipaserver/test_rpcserver.py
View File

@@ -22,12 +22,18 @@ Test the `ipaserver.rpc` module.
"""
import json
import pytest
from ipatests.util import create_test_api, assert_equal, raises, PluginTester
from ipatests.data import unicode_str
from ipalib import errors, Command
import six
from ipatests.util import assert_equal, raises, PluginTester
from ipalib import errors
from ipaserver import rpcserver
if six.PY3:
unicode = str
pytestmark = pytest.mark.tier0
class StartResponse(object):
def __init__(self):
@@ -47,7 +53,8 @@ class StartResponse(object):
def test_not_found():
f = rpcserver.HTTP_Status()
api = 'the api instance'
f = rpcserver.HTTP_Status(api)
t = rpcserver._not_found_template
s = StartResponse()
@@ -55,7 +62,7 @@ def test_not_found():
url = '/ipa/foo/stuff'
assert_equal(
f.not_found(None, s, url, None),
[t % dict(url='/ipa/foo/stuff')]
[(t % dict(url='/ipa/foo/stuff')).encode('utf-8')]
)
assert s.status == '404 Not Found'
assert s.headers == [('Content-Type', 'text/html; charset=utf-8')]
@@ -65,46 +72,51 @@ def test_not_found():
url ='&nbsp;' + '<script>do_bad_stuff();</script>'
assert_equal(
f.not_found(None, s, url, None),
[t % dict(url='&amp;nbsp;&lt;script&gt;do_bad_stuff();&lt;/script&gt;')]
[(t % dict(
url='&amp;nbsp;&lt;script&gt;do_bad_stuff();&lt;/script&gt;')
).encode('utf-8')]
)
assert s.status == '404 Not Found'
assert s.headers == [('Content-Type', 'text/html; charset=utf-8')]
def test_bad_request():
f = rpcserver.HTTP_Status()
api = 'the api instance'
f = rpcserver.HTTP_Status(api)
t = rpcserver._bad_request_template
s = StartResponse()
assert_equal(
f.bad_request(None, s, 'illegal request'),
[t % dict(message='illegal request')]
[(t % dict(message='illegal request')).encode('utf-8')]
)
assert s.status == '400 Bad Request'
assert s.headers == [('Content-Type', 'text/html; charset=utf-8')]
def test_internal_error():
f = rpcserver.HTTP_Status()
api = 'the api instance'
f = rpcserver.HTTP_Status(api)
t = rpcserver._internal_error_template
s = StartResponse()
assert_equal(
f.internal_error(None, s, 'request failed'),
[t % dict(message='request failed')]
[(t % dict(message='request failed')).encode('utf-8')]
)
assert s.status == '500 Internal Server Error'
assert s.headers == [('Content-Type', 'text/html; charset=utf-8')]
def test_unauthorized_error():
f = rpcserver.HTTP_Status()
api = 'the api instance'
f = rpcserver.HTTP_Status(api)
t = rpcserver._unauthorized_template
s = StartResponse()
assert_equal(
f.unauthorized(None, s, 'unauthorized', 'password-expired'),
[t % dict(message='unauthorized')]
[(t % dict(message='unauthorized')).encode('utf-8')]
)
assert s.status == '401 Unauthorized'
assert s.headers == [('Content-Type', 'text/html; charset=utf-8'),
@@ -139,7 +151,8 @@ class test_session(object):
[environ[k] for k in ('SCRIPT_NAME', 'PATH_INFO')]
)
inst = self.klass()
api = 'the api instance'
inst = self.klass(api)
inst.mount(app1, '/foo/stuff')
inst.mount(app2, '/bar')
@@ -157,13 +170,14 @@ class test_session(object):
pass
# Test that mount works:
inst = self.klass()
api = 'the api instance'
inst = self.klass(api)
inst.mount(app1, 'foo')
assert inst['foo'] is app1
assert list(inst) == ['foo']
# Test that StandardError is raise if trying override a mount:
e = raises(StandardError, inst.mount, app2, 'foo')
# Test that Exception is raise if trying override a mount:
e = raises(Exception, inst.mount, app2, 'foo')
assert str(e) == '%s.mount(): cannot replace %r with %r at %r' % (
'wsgi_dispatch', app1, app2, 'foo'
)
@@ -182,7 +196,7 @@ class test_xmlserver(PluginTester):
_plugin = rpcserver.xmlserver
def test_marshaled_dispatch(self): # FIXME
(o, api, home) = self.instance('Backend', in_server=True)
self.instance('Backend', in_server=True)
class test_jsonserver(PluginTester):
@@ -196,12 +210,14 @@ class test_jsonserver(PluginTester):
"""
Test the `ipaserver.rpcserver.jsonserver.unmarshal` method.
"""
(o, api, home) = self.instance('Backend', in_server=True)
o, _api, _home = self.instance('Backend', in_server=True)
# Test with invalid JSON-data:
e = raises(errors.JSONError, o.unmarshal, 'this wont work')
assert isinstance(e.error, ValueError)
assert unicode(e.error) == 'No JSON object could be decoded'
if six.PY2:
assert unicode(e.error) == 'No JSON object could be decoded'
else:
assert str(e.error).startswith('Expecting value: ')
# Test with non-dict type:
e = raises(errors.JSONError, o.unmarshal, json.dumps([1, 2, 3]))
@@ -241,7 +257,7 @@ class test_jsonserver(PluginTester):
assert unicode(e.error) == 'params[1] (aka options) must be a dict'
# Test with valid values:
args = (u'jdoe', )
args = [u'jdoe']
options = dict(givenname=u'John', sn='Doe')
d = dict(method=u'user_add', params=(args, options), id=18)
assert o.unmarshal(json.dumps(d)) == (u'user_add', args, options, 18)

64
ipatests/test_ipaserver/test_secrets.py Normal file
View File

@@ -0,0 +1,64 @@
# Copyright (C) 2015 FreeIPA Project Contributors - see LICENSE file
from __future__ import print_function
from ipaserver.secrets.store import iSecStore, NAME_DB_MAP, NSSCertDB
import os
import shutil
import subprocess
import tempfile
import unittest
def _test_password_callback():
with open('test-ipa-sec-store/pwfile') as f:
password = f.read()
return password
class TestiSecStore(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.testdir = tempfile.mkdtemp(suffix='ipa-sec-store')
pwfile = os.path.join(cls.testdir, 'pwfile')
with open(pwfile, 'w') as f:
f.write('testpw')
cls.certdb = os.path.join(cls.testdir, 'certdb')
os.mkdir(cls.certdb)
cls.cert2db = os.path.join(cls.testdir, 'cert2db')
os.mkdir(cls.cert2db)
seedfile = os.path.join(cls.testdir, 'seedfile')
with open(seedfile, 'wb') as f:
seed = os.urandom(1024)
f.write(seed)
subprocess.call(
['certutil', '-d', cls.certdb, '-N', '-f', pwfile],
cwd=cls.testdir
)
subprocess.call(
['certutil', '-d', cls.cert2db, '-N', '-f', pwfile],
cwd=cls.testdir
)
subprocess.call(
['certutil', '-d', cls.certdb, '-S', '-f', pwfile,
'-s', 'CN=testCA', '-n', 'testCACert', '-x',
'-t', 'CT,C,C', '-m', '1', '-z', seedfile],
cwd=cls.testdir
)
@classmethod
def tearDownClass(cls):
shutil.rmtree(cls.testdir)
def test_iSecStore(self):
iss = iSecStore({})
NAME_DB_MAP['test'] = {
'type': 'NSSDB',
'path': self.certdb,
'handler': NSSCertDB,
'pwcallback': _test_password_callback,
}
value = iss.get('keys/test/testCACert')
NAME_DB_MAP['test']['path'] = self.cert2db
iss.set('keys/test/testCACert', value)

745
ipatests/test_ipaserver/test_serverroles.py Normal file
View File

@@ -0,0 +1,745 @@
#
# Copyright (C) 2016 FreeIPA Contributors see COPYING for license
#
"""
Tests for the serverroles backend
"""
from __future__ import absolute_import
from collections import namedtuple
import ldap
import pytest
from ipaplatform.paths import paths
from ipalib import api, create_api, errors
from ipapython.dn import DN
pytestmark = pytest.mark.needs_ipaapi
def _make_service_entry(ldap_backend, dn, enabled=True, other_config=None):
mods = {
'objectClass': ['top', 'nsContainer', 'ipaConfigObject'],
}
if enabled:
mods.update({'ipaConfigString': ['enabledService']})
if other_config is not None:
mods.setdefault('ipaConfigString', [])
mods['ipaConfigString'].extend(other_config)
return ldap_backend.make_entry(dn, **mods)
def _make_master_entry(ldap_backend, dn, ca=False):
mods = {
'objectClass': [
'top',
'nsContainer',
'ipaReplTopoManagedServer',
'ipaSupportedDomainLevelConfig',
'ipaConfigObject',
],
'ipaMaxDomainLevel': ['1'],
'ipaMinDomainLevel': ['0'],
'ipaReplTopoManagedsuffix': [str(api.env.basedn)]
}
if ca:
mods['ipaReplTopoManagedsuffix'].append('o=ipaca')
return ldap_backend.make_entry(dn, **mods)
_adtrust_agents = DN(
('cn', 'adtrust agents'),
('cn', 'sysaccounts'),
('cn', 'etc'),
api.env.basedn
)
master_data = {
'ca-dns-dnssec-keymaster-pkinit-server': {
'services': {
'CA': {
'enabled': True,
},
'DNS': {
'enabled': True,
},
'DNSKeySync': {
'enabled': True,
},
'DNSSEC': {
'enabled': True,
'config': ['DNSSecKeyMaster']
},
'KDC': {
'enabled': True,
'config': ['pkinitEnabled']
}
},
'expected_roles': {
'enabled': ['IPA master', 'CA server', 'DNS server']
},
'expected_attributes': {'DNS server': 'dnssec_key_master_server',
'IPA master': 'pkinit_server_server'}
},
'ca-kra-renewal-master-pkinit-server': {
'services': {
'CA': {
'enabled': True,
'config': ['caRenewalMaster']
},
'KRA': {
'enabled': True,
},
'KDC': {
'enabled': True,
'config': ['pkinitEnabled']
},
},
'expected_roles': {
'enabled': ['IPA master', 'CA server', 'KRA server']
},
'expected_attributes': {'CA server': 'ca_renewal_master_server',
'IPA master': 'pkinit_server_server'}
},
'dns-trust-agent': {
'services': {
'DNS': {
'enabled': True,
},
'DNSKeySync': {
'enabled': True,
}
},
'attributes': {
_adtrust_agents: {
'member': ['host']
}
},
'expected_roles': {
'enabled': ['IPA master', 'DNS server', 'AD trust agent']
}
},
'trust-agent': {
'attributes': {
_adtrust_agents: {
'member': ['host']
}
},
'expected_roles': {
'enabled': ['IPA master', 'AD trust agent']
}
},
'trust-controller-dns': {
'services': {
'ADTRUST': {
'enabled': True,
},
'DNS': {
'enabled': True,
},
'DNSKeySync': {
'enabled': True,
}
},
'attributes': {
_adtrust_agents: {
'member': ['host', 'cifs']
}
},
'expected_roles': {
'enabled': ['IPA master', 'AD trust agent', 'AD trust controller',
'DNS server']
}
},
'trust-controller-ca': {
'services': {
'ADTRUST': {
'enabled': True,
},
'CA': {
'enabled': True,
},
},
'attributes': {
_adtrust_agents: {
'member': ['host', 'cifs']
}
},
'expected_roles': {
'enabled': ['IPA master', 'AD trust agent', 'AD trust controller',
'CA server']
}
},
'configured-ca': {
'services': {
'CA': {
'enabled': False,
},
},
'expected_roles': {
'enabled': ['IPA master'],
'configured': ['CA server']
}
},
'configured-dns': {
'services': {
'DNS': {
'enabled': False,
},
'DNSKeySync': {
'enabled': False,
}
},
'expected_roles': {
'enabled': ['IPA master'],
'configured': ['DNS server']
}
},
'mixed-state-dns': {
'services': {
'DNS': {
'enabled': False
},
'DNSKeySync': {
'enabled': True
}
},
'expected_roles': {
'enabled': ['IPA master'],
'configured': ['DNS server']
}
},
}
class MockMasterTopology(object):
"""
object that will set up and tear down entries in LDAP backend to mimic
a presence of real IPA masters with services running on them.
"""
ipamaster_services = [u'KDC', u'HTTP', u'KPASSWD']
def __init__(self, api_instance, domain_data):
self.api = api_instance
self.domain = self.api.env.domain
self.domain_data = domain_data
self.masters_base = DN(
self.api.env.container_masters, self.api.env.basedn)
self.test_master_dn = DN(
('cn', self.api.env.host), self.api.env.container_masters,
self.api.env.basedn)
self.ldap = self.api.Backend.ldap2
self.existing_masters = {
m['cn'][0] for m in self.api.Command.server_find(
u'', sizelimit=0,
pkey_only=True,
no_members=True,
raw=True)['result']}
self.original_dns_configs = self._remove_test_host_attrs()
def iter_domain_data(self):
MasterData = namedtuple('MasterData',
['dn', 'fqdn', 'services', 'attrs'])
for name in self.domain_data:
fqdn = self.get_fqdn(name)
master_dn = self.get_master_dn(name)
master_services = self.domain_data[name].get('services', {})
master_attributes = self.domain_data[name].get('attributes', {})
yield MasterData(
dn=master_dn,
fqdn=fqdn,
services=master_services,
attrs=master_attributes
)
def get_fqdn(self, name):
return '.'.join([name, self.domain])
def get_master_dn(self, name):
return DN(('cn', self.get_fqdn(name)), self.masters_base)
def get_service_dn(self, name, master_dn):
return DN(('cn', name), master_dn)
def _add_host_entry(self, fqdn):
self.api.Command.host_add(fqdn, force=True)
self.api.Command.hostgroup_add_member(u'ipaservers', host=fqdn)
def _del_host_entry(self, fqdn):
try:
self.api.Command.host_del(fqdn)
except errors.NotFound:
pass
def _add_service_entry(self, service, fqdn):
return self.api.Command.service_add(
'/'.join([service, fqdn]),
force=True
)
def _del_service_entry(self, service, fqdn):
try:
self.api.Command.service_del(
'/'.join([service, fqdn]),
)
except errors.NotFound:
pass
def _add_svc_entries(self, master_dn, svc_desc):
for name in svc_desc:
svc_dn = self.get_service_dn(name, master_dn)
svc_mods = svc_desc[name]
self.ldap.add_entry(
_make_service_entry(
self.ldap,
svc_dn,
enabled=svc_mods['enabled'],
other_config=svc_mods.get('config', None)))
self._add_ipamaster_services(master_dn)
def _remove_svc_master_entries(self, master_dn):
try:
entries = self.ldap.get_entries(
master_dn, ldap.SCOPE_SUBTREE
)
except errors.NotFound:
return
if entries:
entries.sort(key=lambda x: len(x.dn), reverse=True)
for entry in entries:
self.ldap.delete_entry(entry)
def _add_ipamaster_services(self, master_dn):
"""
add all the service entries which are part of the IPA Master role
"""
for svc_name in self.ipamaster_services:
svc_dn = self.get_service_dn(svc_name, master_dn)
try:
self.ldap.get_entry(svc_dn)
except errors.NotFound:
self.ldap.add_entry(_make_service_entry(self.ldap, svc_dn))
def _add_members(self, dn, fqdn, member_attrs):
entry_attrs = self.ldap.get_entry(dn)
value = entry_attrs.get('member', [])
for a in member_attrs:
if a == 'host':
value.append(
str(self.api.Object.host.get_dn(fqdn)))
else:
result = self._add_service_entry(a, fqdn)['result']
value.append(str(result['dn']))
entry_attrs['member'] = value
self.ldap.update_entry(entry_attrs)
def _remove_members(self, dn, fqdn, member_attrs):
entry_attrs = self.ldap.get_entry(dn)
value = set(entry_attrs.get('member', []))
if not value:
return
for a in member_attrs:
if a == 'host':
try:
value.remove(
str(self.api.Object.host.get_dn(fqdn)))
except KeyError:
pass
else:
try:
value.remove(
str(self.api.Object.service.get_dn(
'/'.join([a, fqdn]))))
except KeyError:
pass
self._del_service_entry(a, fqdn)
entry_attrs['member'] = list(value)
try:
self.ldap.update_entry(entry_attrs)
except (errors.NotFound, errors.EmptyModlist):
pass
def _remove_test_host_attrs(self):
original_dns_configs = []
for attr_name in (
'caRenewalMaster', 'dnssecKeyMaster', 'pkinitEnabled'):
try:
svc_entry = self.ldap.find_entry_by_attr(
'ipaConfigString', attr_name, 'ipaConfigObject',
base_dn=self.test_master_dn)
except errors.NotFound:
continue
else:
original_dns_configs.append(
(svc_entry.dn, list(svc_entry.get('ipaConfigString', [])))
)
svc_entry[u'ipaConfigString'].remove(attr_name)
self.ldap.update_entry(svc_entry)
return original_dns_configs
def _restore_test_host_attrs(self):
for dn, config in self.original_dns_configs:
try:
svc_entry = self.api.Backend.ldap2.get_entry(dn)
svc_entry['ipaConfigString'] = config
self.ldap.update_entry(svc_entry)
except (errors.NotFound, errors.EmptyModlist):
continue
def setup_data(self):
for master_data in self.iter_domain_data():
# create host
self._add_host_entry(master_data.fqdn)
# create master
self.ldap.add_entry(
_make_master_entry(
self.ldap,
master_data.dn,
ca='CA' in master_data.services))
# now add service entries
self._add_svc_entries(master_data.dn, master_data.services)
# optionally add some attributes required e.g. by AD trust roles
for entry_dn, attrs in master_data.attrs.items():
if 'member' in attrs:
self._add_members(
entry_dn,
master_data.fqdn,
attrs['member']
)
def teardown_data(self):
for master_data in self.iter_domain_data():
# first remove the master entries and service containers
self._remove_svc_master_entries(master_data.dn)
# optionally clean up leftover attributes
for entry_dn, attrs in master_data.attrs.items():
if 'member' in attrs:
self._remove_members(
entry_dn,
master_data.fqdn,
attrs['member'],
)
# finally remove host entry
self._del_host_entry(master_data.fqdn)
self._restore_test_host_attrs()
@pytest.fixture(scope='module')
def mock_api(request):
test_api = create_api(mode=None)
test_api.bootstrap(in_server=True,
ldap_uri=api.env.ldap_uri,
confdir=paths.ETC_IPA)
test_api.finalize()
if not test_api.Backend.ldap2.isconnected():
test_api.Backend.ldap2.connect()
def finalize():
test_api.Backend.ldap2.disconnect()
request.addfinalizer(finalize)
return test_api
@pytest.fixture(scope='module')
def mock_masters(request, mock_api):
"""
Populate the LDAP backend with test data
"""
if not api.Backend.rpcclient.isconnected():
api.Backend.rpcclient.connect()
master_topo = MockMasterTopology(mock_api, master_data)
def finalize():
master_topo.teardown_data()
if api.Backend.rpcclient.isconnected():
api.Backend.rpcclient.disconnect()
request.addfinalizer(finalize)
master_topo.setup_data()
return master_topo
def enabled_role_iter(master_data):
for m, data in master_data.items():
for role in data['expected_roles']['enabled']:
yield m, role
def provided_role_iter(master_data):
for m, data in master_data.items():
yield m, data['expected_roles']['enabled']
def configured_role_iter(master_data):
for m, data in master_data.items():
if 'configured' in data['expected_roles']:
for role in data['expected_roles']['configured']:
yield m, role
def role_provider_iter(master_data):
result = {}
for m, data in master_data.items():
for role in data['expected_roles']['enabled']:
if role not in result:
result[role] = []
result[role].append(m)
for role_name, masters in result.items():
yield role_name, masters
def attribute_masters_iter(master_data):
for m, data in master_data.items():
if 'expected_attributes' in data:
for assoc_role, attr in data['expected_attributes'].items():
yield m, assoc_role, attr
def dns_servers_iter(master_data):
for m, data in master_data.items():
if "DNS server" in data['expected_roles']['enabled']:
yield m
@pytest.fixture(params=list(enabled_role_iter(master_data)),
ids=['role: {}, master: {}, enabled'.format(role, m)
for m, role in enabled_role_iter(master_data)])
def enabled_role(request):
return request.param
@pytest.fixture(params=list(provided_role_iter(master_data)),
ids=["{}: {}".format(m, ', '.join(roles)) for m, roles in
provided_role_iter(master_data)])
def provided_roles(request):
return request.param
@pytest.fixture(params=list(configured_role_iter(master_data)),
ids=['role: {}, master: {}, configured'.format(role, m)
for m, role in configured_role_iter(master_data)])
def configured_role(request):
return request.param
@pytest.fixture(params=list(role_provider_iter(master_data)),
ids=['{} providers'.format(role_name)
for role_name, _m in
role_provider_iter(master_data)])
def role_providers(request):
return request.param
@pytest.fixture(params=list(attribute_masters_iter(master_data)),
ids=['{} of {}: {}'.format(attr, role, m) for m, role, attr in
attribute_masters_iter(master_data)])
def attribute_providers(request):
return request.param
@pytest.fixture(params=list(dns_servers_iter(master_data)),
ids=list(dns_servers_iter(master_data)))
def dns_server(request):
return request.param
class TestServerRoleStatusRetrieval(object):
def retrieve_role(self, master, role, mock_api, mock_masters):
fqdn = mock_masters.get_fqdn(master)
return mock_api.Backend.serverroles.server_role_retrieve(
server_server=fqdn, role_servrole=role)
def find_role(self, role_name, mock_api, mock_masters, master=None):
if master is not None:
hostname = mock_masters.get_fqdn(master)
else:
hostname = None
result = mock_api.Backend.serverroles.server_role_search(
server_server=hostname,
role_servrole=role_name)
return [
r for r in result if r[u'server_server'] not in
mock_masters.existing_masters]
def get_enabled_roles_on_master(self, master, mock_api, mock_masters):
fqdn = mock_masters.get_fqdn(master)
result = mock_api.Backend.serverroles.server_role_search(
server_server=fqdn, role_servrole=None, status=u'enabled'
)
return sorted(set(r[u'role_servrole'] for r in result))
def get_masters_with_enabled_role(self, role_name, mock_api, mock_masters):
result = mock_api.Backend.serverroles.server_role_search(
server_server=None, role_servrole=role_name)
return sorted(
r[u'server_server'] for r in result if
r[u'status'] == u'enabled' and r[u'server_server'] not in
mock_masters.existing_masters)
def test_listing_of_enabled_role(
self, mock_api, mock_masters, enabled_role):
master, role_name = enabled_role
result = self.retrieve_role(master, role_name, mock_api, mock_masters)
assert result[0][u'status'] == u'enabled'
def test_listing_of_configured_role(
self, mock_api, mock_masters, configured_role):
master, role_name = configured_role
result = self.retrieve_role(master, role_name, mock_api, mock_masters)
assert result[0][u'status'] == u'configured'
def test_role_providers(
self, mock_api, mock_masters, role_providers):
role_name, providers = role_providers
expected_masters = sorted(mock_masters.get_fqdn(m) for m in providers)
actual_masters = self.get_masters_with_enabled_role(
role_name, mock_api, mock_masters)
assert expected_masters == actual_masters
def test_provided_roles_on_master(
self, mock_api, mock_masters, provided_roles):
master, expected_roles = provided_roles
expected_roles.sort()
actual_roles = self.get_enabled_roles_on_master(
master, mock_api, mock_masters)
assert expected_roles == actual_roles
def test_unknown_role_status_raises_notfound(self, mock_api, mock_masters):
unknown_role = 'IAP maestr'
fqdn = mock_masters.get_fqdn('ca-dns-dnssec-keymaster-pkinit-server')
with pytest.raises(errors.NotFound):
mock_api.Backend.serverroles.server_role_retrieve(
fqdn, unknown_role)
def test_no_servrole_queries_all_roles_on_server(self, mock_api,
mock_masters):
master_name = 'ca-dns-dnssec-keymaster-pkinit-server'
enabled_roles = master_data[master_name]['expected_roles']['enabled']
result = self.find_role(None, mock_api, mock_masters,
master=master_name)
for r in result:
if r[u'role_servrole'] in enabled_roles:
assert r[u'status'] == u'enabled'
else:
assert r[u'status'] == u'absent'
def test_invalid_substring_search_returns_nothing(self, mock_api,
mock_masters):
invalid_substr = 'fwfgbb'
assert (not self.find_role(invalid_substr, mock_api, mock_masters,
'ca-dns-dnssec-keymaster-pkinit-server'))
class TestServerAttributes(object):
def config_retrieve(self, assoc_role_name, mock_api):
return mock_api.Backend.serverroles.config_retrieve(
assoc_role_name)
def config_update(self, mock_api, **attrs_values):
return mock_api.Backend.serverroles.config_update(**attrs_values)
def test_attribute_master(self, mock_api, mock_masters,
attribute_providers):
master, assoc_role, attr_name = attribute_providers
fqdn = mock_masters.get_fqdn(master)
actual_attr_masters = self.config_retrieve(
assoc_role, mock_api)[attr_name]
assert fqdn in actual_attr_masters
def test_set_attribute_on_the_same_provider_raises_emptymodlist(
self, mock_api, mock_masters):
attr_name = "ca_renewal_master_server"
role_name = "CA server"
existing_renewal_master = self.config_retrieve(
role_name, mock_api)[attr_name]
with pytest.raises(errors.EmptyModlist):
self.config_update(
mock_api, **{attr_name: existing_renewal_master})
def test_set_attribute_on_master_without_assoc_role_raises_validationerror(
self, mock_api, mock_masters):
attr_name = "ca_renewal_master_server"
non_ca_fqdn = mock_masters.get_fqdn('trust-controller-dns')
with pytest.raises(errors.ValidationError):
self.config_update(mock_api, **{attr_name: non_ca_fqdn})
def test_set_unknown_attribute_on_master_raises_notfound(
self, mock_api, mock_masters):
attr_name = "ca_renuwal_maztah"
fqdn = mock_masters.get_fqdn('trust-controller-ca')
with pytest.raises(errors.NotFound):
self.config_update(mock_api, **{attr_name: [fqdn]})
def test_set_ca_renewal_master_on_other_ca_and_back(self, mock_api,
mock_masters):
attr_name = "ca_renewal_master_server"
role_name = "CA server"
original_renewal_master = self.config_retrieve(
role_name, mock_api)[attr_name]
other_ca_server = mock_masters.get_fqdn('trust-controller-ca')
for host in (other_ca_server, original_renewal_master):
self.config_update(mock_api, **{attr_name: host})
assert (
self.config_retrieve(role_name, mock_api)[attr_name] == host)

74
ipatests/test_ipaserver/test_topology_plugin.py Normal file
View File

@@ -0,0 +1,74 @@
#
# Copyright (C) 2015 FreeIPA Contributors see COPYING for license
#
import io
import os
from ipaserver.plugins.ldap2 import ldap2
from ipalib import api
from ipapython.dn import DN
import pytest
@pytest.mark.tier1
class TestTopologyPlugin(object):
"""
Test Topology plugin from the DS point of view
Testcase: http://www.freeipa.org/page/V4/Manage_replication_topology/
Test_plan#Test_case:
_Replication_Topology_is_listed_among_directory_server_plugins
"""
pwfile = os.path.join(api.env.dot_ipa, ".dmpw")
def setup(self):
"""
setup for test
"""
self.conn = None
def teardown(self):
if self.conn and self.conn.isconnected():
self.conn.disconnect()
@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):
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'],
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':
[u'cn=ipa,cn=etc,dc=example,dc=com'],
u'nsslapd-pluginDescription': [u'ipa-topology-plugin'],
u'nsslapd-pluginEnabled': [u'on'],
u'nsslapd-pluginId': [u'ipa-topology-plugin'],
u'objectClass': [u'top', u'nsSlapdPlugin', u'extensibleObject'],
u'nsslapd-topo-plugin-startup-delay': [u'20'],
u'nsslapd-topo-plugin-shared-binddngroup':
[u'cn=replication managers,cn=sysaccounts,cn=etc,dc=example,dc=com'],
u'nsslapd-pluginType': [u'object'],
u'nsslapd-pluginInitfunc': [u'ipa_topo_init']
}
variable_attrs = {u'nsslapd-topo-plugin-shared-replica-root',
u'nsslapd-topo-plugin-shared-config-base',
u'nsslapd-topo-plugin-shared-binddngroup'}
# Now eliminate keys that have domain-dependent values.
checkvalues = set(pluginattrs.keys()) - variable_attrs
topoplugindn = DN(('cn', 'IPA Topology Configuration'),
('cn', 'plugins'),
('cn', 'config'))
pwfile = os.path.join(api.env.dot_ipa, ".dmpw")
with io.open(pwfile, "r") as f:
dm_password = f.read().rstrip()
self.conn = ldap2(api)
self.conn.connect(bind_dn=DN(('cn', 'directory manager')),
bind_pw=dm_password)
entry = self.conn.get_entry(topoplugindn)
assert(set(entry.keys()) == set(pluginattrs.keys()))
for i in checkvalues:
assert(set(pluginattrs[i]) == set(entry[i]))

53
ipatests/test_ipaserver/test_version_comparison.py Normal file
View File

@@ -0,0 +1,53 @@
#
# Copyright (C) 2015 FreeIPA Contributors see COPYING for license
#
"""
tests for correct RPM version comparison
"""
from __future__ import absolute_import
from ipaplatform.tasks import tasks
import pytest
version_strings = [
("3.0.0-1.el6", "3.0.0-2.el6", "older"),
("3.0.0-1.el6_8", "3.0.0-1.el6_8.1", "older"),
("3.0.0-42.el6", "3.0.0-1.el6", "newer"),
("3.0.0-1.el6", "3.0.0-42.el6", "older"),
("3.0.0-42.el6", "3.3.3-1.fc20", "older"),
("4.2.0-15.el7", "4.2.0-15.el7_2.3", "older"),
("4.2.0-15.el7_2", "4.2.0-15.el7_2.3", "older"),
("4.2.0-15.el7_2.3", "4.2.0-15.el7_2.3", "equal"),
("4.2.0-15.el7_2.3", "4.2.0-15.el7_2.2", "newer"),
("4.2.0-1.fc23", "4.2.1-1.fc23", "older"),
("4.2.3-alpha1.fc23", "4.2.3-2.fc23", "older"), # numeric version elements
# have precedence over
# non-numeric ones
("4.3.90.201601080923GIT55aeea7-0.fc23", "4.3.0-1.fc23", "newer")
]
@pytest.fixture(params=version_strings)
def versions(request):
return request.param
class TestVersionComparsion(object):
def test_versions(self, versions):
version_string1, version_string2, expected_comparison = versions
ver1 = tasks.parse_ipa_version(version_string1)
ver2 = tasks.parse_ipa_version(version_string2)
if expected_comparison == "newer":
assert ver1 > ver2
elif expected_comparison == "older":
assert ver1 < ver2
elif expected_comparison == "equal":
assert ver1 == ver2
else:
raise TypeError(
"Unexpected comparison string: {}".format(expected_comparison)
)