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

Imported Upstream version 4.0.5

Browse Source
This commit is contained in:
Mario Fetka
2021-07-25 07:50:50 +02:00
parent 8ff3be4216
commit 3bfaa6e020
2049 changed files with 317193 additions and 1632423 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
ipatests/test_ipaserver/__init__.py
View File

@@ -20,7 +20,3 @@
"""
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">
<pkcs5:PBKDF2-params>
<Salt>
<Specified>Ej7/PEpyEpw=</Specified>
</Salt>
<IterationCount>1000</IterationCount>
<KeyLength>16</KeyLength>
<PRF/>
</pkcs5:PBKDF2-params>
<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>
</xenc11:KeyDerivationMethod>
<xenc:ReferenceList>
<xenc:DataReference URI="#ED"/>

13
ipatests/test_ipaserver/httptest.py
View File

@@ -20,9 +20,10 @@
Base class for HTTP request tests
"""
from six.moves import urllib
import urllib
import httplib
from ipalib import api, util
from ipalib import api
class Unauthorized_HTTP_test(object):
"""
@@ -31,7 +32,6 @@ class Unauthorized_HTTP_test(object):
"""
app_uri = ''
host = api.env.host
cacert = api.env.tls_ca_cert
content_type = 'application/x-www-form-urlencoded'
def send_request(self, method='POST', params=None):
@@ -41,15 +41,12 @@ class Unauthorized_HTTP_test(object):
:param key When not None, overrides default app_uri
"""
if params is not None:
# urlencode *can* take two arguments
# pylint: disable=too-many-function-args
params = urllib.parse.urlencode(params, True)
params = urllib.urlencode(params, True)
url = 'https://' + self.host + self.app_uri
headers = {'Content-Type' : self.content_type,
'Referer' : url}
conn = util.create_https_connection(
self.host, cafile=self.cacert)
conn = httplib.HTTPSConnection(self.host)
conn.request(method, self.app_uri, params, headers)
return conn.getresponse()

17
ipatests/test_ipaserver/test_changepw.py
View File

@@ -18,38 +18,37 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import nose
import ldap
import pytest
from ipatests.test_ipaserver.httptest import Unauthorized_HTTP_test
from httptest import Unauthorized_HTTP_test
from ipatests.test_xmlrpc.xmlrpc_test import XMLRPC_test
from ipatests.util import assert_equal
from ipatests.util import assert_equal, assert_not_equal
from ipalib import api, errors
from ipapython.dn import DN
import ldap
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):
def setUp(self):
super(test_changepw, self).setUp()
try:
api.Command['user_add'](uid=testuser, givenname=u'Test', sn=u'User')
api.Command['passwd'](testuser, password=u'old_password')
except errors.ExecutionError as e:
except errors.ExecutionError, e:
raise nose.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),

41
ipatests/test_ipaserver/test_dnssec.py
View File

@@ -1,41 +0,0 @@
#
# 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}

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

12
ipatests/test_ipaserver/test_install/test_adtrustinstance.py
View File

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

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

@@ -1,125 +0,0 @@
#
# 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

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

@@ -22,10 +22,8 @@ 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'

279
ipatests/test_ipaserver/test_ipap11helper.py
View File

@@ -1,279 +0,0 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2015 FreeIPA Contributors see COPYING for license
#
"""
Test the `ipapython/ipap11helper/p11helper.c` module.
"""
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 p11(request):
token_path = tempfile.mkdtemp(prefix='pytest_', suffix='_pkcs11')
os.chdir(token_path)
os.mkdir('tokens')
with open('softhsm2.conf', 'w') as cfg:
cfg.write(CONFIG_DATA % token_path)
os.environ['SOFTHSM2_CONF'] = os.path.join(token_path, 'softhsm2.conf')
subprocess.check_call([SOFTHSM2_UTIL, '--init-token', '--free',
'--label', 'test', '--pin', '1234', '--so-pin',
'1234'])
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'])
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):
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))
with open("public_key.asn1.der", "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):
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))
with open("wrapped_priv.der", "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
View File

@@ -1,126 +0,0 @@
#
# 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)

149
ipatests/test_ipaserver/test_ldap.py
View File

@@ -26,36 +26,33 @@
# The DM password needs to be set in ~/.ipa/.dmpw
import os
import sys
import pytest
import nose
from nose.tools import assert_raises # pylint: disable=E0611
import six
import nss.nss as nss
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 = api.env.ldap_uri
self.ldapuri = 'ldap://%s' % ipautil.format_netloc(api.env.host)
self.ccache = paths.TMP_KRB5CC % os.getuid()
nss.nss_init_nodb()
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()
@@ -63,8 +60,8 @@ class test_ldap(object):
"""
Test an anonymous LDAP bind using ldap2
"""
self.conn = ldap2(api)
self.conn.connect(autobind=AUTOBIND_DISABLED)
self.conn = ldap2(shared_instance=False, ldap_uri=self.ldapuri)
self.conn.connect()
dn = api.env.basedn
entry_attrs = self.conn.get_entry(dn, ['associateddomain'])
domain = entry_attrs.single_value['associateddomain']
@@ -74,27 +71,34 @@ class test_ldap(object):
"""
Test a GSSAPI LDAP bind using ldap2
"""
self.conn = ldap2(api)
self.conn.connect(autobind=AUTOBIND_DISABLED)
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)
entry_attrs = self.conn.get_entry(self.dn, ['usercertificate'])
cert = entry_attrs.get('usercertificate')[0]
assert cert.serial_number is not None
cert = entry_attrs.get('usercertificate')
cert = cert[0]
serial = unicode(x509.get_serial_number(cert, x509.DER))
assert serial is not None
def test_simple(self):
"""
Test a simple LDAP bind using ldap2
"""
pwfile = api.env.dot_ipa + os.sep + ".dmpw"
if os.path.isfile(pwfile):
with open(pwfile, "r") as fp:
dm_password = fp.read().rstrip()
if ipautil.file_exists(pwfile):
fp = open(pwfile, "r")
dm_password = fp.read().rstrip()
fp.close()
else:
raise nose.SkipTest("No directory manager password in %s" % pwfile)
self.conn = ldap2(api)
self.conn = ldap2(shared_instance=False, ldap_uri=self.ldapuri)
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')[0]
assert cert.serial_number is not None
cert = entry_attrs.get('usercertificate')
cert = cert[0]
serial = unicode(x509.get_serial_number(cert, x509.DER))
assert serial is not None
def test_Backend(self):
"""
@@ -105,38 +109,46 @@ 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, confdir=paths.ETC_IPA)
myapi.bootstrap(context='cli', in_server=True, in_tree=True)
myapi.register(ldap2)
myapi.register(host)
myapi.register(service)
myapi.register(service_show)
myapi.finalize()
pwfile = api.env.dot_ipa + os.sep + ".dmpw"
if os.path.isfile(pwfile):
with open(pwfile, "r") as fp:
dm_password = fp.read().rstrip()
if ipautil.file_exists(pwfile):
fp = open(pwfile, "r")
dm_password = fp.read().rstrip()
fp.close()
else:
raise nose.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')[0]
assert cert.serial_number is not None
cert = entry_attrs.get('usercertificate')
cert = cert[0]
serial = unicode(x509.get_serial_number(cert, x509.DER))
assert serial is not None
def test_autobind(self):
"""
Test an autobind LDAP bind using ldap2
"""
self.conn = ldap2(api)
ldapuri = 'ldapi://%%2fvar%%2frun%%2fslapd-%s.socket' % api.env.realm.replace('.','-')
self.conn = ldap2(shared_instance=False, ldap_uri=ldapuri)
try:
self.conn.connect(autobind=True)
except errors.ACIError:
raise nose.SkipTest("Only executed as root")
entry_attrs = self.conn.get_entry(self.dn, ['usercertificate'])
cert = entry_attrs.get('usercertificate')[0]
assert cert.serial_number is not None
cert = entry_attrs.get('usercertificate')
cert = cert[0]
serial = unicode(x509.get_serial_number(cert, x509.DER))
assert serial is not None
@pytest.mark.tier0
@pytest.mark.needs_ipaapi
class test_LDAPEntry(object):
"""
Test the LDAPEntry class
@@ -146,14 +158,14 @@ class test_LDAPEntry(object):
dn1 = DN(('cn', cn1[0]))
dn2 = DN(('cn', cn2[0]))
def setup(self):
self.ldapuri = api.env.ldap_uri
self.conn = ldap2(api)
self.conn.connect(autobind=AUTOBIND_DISABLED)
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()
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()
@@ -163,15 +175,9 @@ class test_LDAPEntry(object):
assert u'cn' in e
assert u'cn' in e.keys()
assert 'CN' in e
if six.PY2:
assert 'CN' not in e.keys()
else:
assert 'CN' in e.keys()
assert 'CN' not in e.keys()
assert 'commonName' in e
if six.PY2:
assert 'commonName' not in e.keys()
else:
assert 'commonName' in e.keys()
assert 'commonName' not in e.keys()
assert e['CN'] is self.cn1
assert e['CN'] is e[u'cn']
@@ -184,15 +190,9 @@ class test_LDAPEntry(object):
assert u'cn' in e
assert u'cn' in e.keys()
assert 'CN' in e
if six.PY2:
assert 'CN' not in e.keys()
else:
assert 'CN' in e.keys()
assert 'CN' not in e.keys()
assert 'commonName' in e
if six.PY2:
assert 'commonName' not in e.keys()
else:
assert 'commonName' in e.keys()
assert 'commonName' not in e.keys()
assert e['CN'] is self.cn2
assert e['CN'] is e[u'cn']
@@ -209,7 +209,7 @@ class test_LDAPEntry(object):
def test_popitem(self):
e = self.entry
assert e.popitem() == ('cn', self.cn1)
assert list(e) == []
e.keys() == []
def test_setdefault(self):
e = self.entry
@@ -238,19 +238,12 @@ 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
@@ -272,35 +265,35 @@ class test_LDAPEntry(object):
assert e['test'] is nice
raw = e.raw['test']
assert raw == [b'1', b'2', b'3']
assert raw == ['1', '2', '3']
nice.remove(1)
assert e.raw['test'] is raw
assert raw == [b'2', b'3']
assert raw == ['2', '3']
raw.append(b'4')
raw.append('4')
assert e['test'] is nice
assert nice == [2, 3, u'4']
nice.remove(2)
raw.append(b'5')
raw.append('5')
assert nice == [3, u'4']
assert raw == [b'2', b'3', b'4', b'5']
assert raw == ['2', '3', '4', '5']
assert e['test'] is nice
assert e.raw['test'] is raw
assert nice == [3, u'4', u'5']
assert raw == [b'3', b'4', b'5']
assert raw == ['3', '4', '5']
nice.insert(0, 2)
raw.remove(b'4')
raw.remove('4')
assert nice == [2, 3, u'4', u'5']
assert raw == [b'3', b'5']
assert raw == ['3', '5']
assert e.raw['test'] is raw
assert e['test'] is nice
assert nice == [2, 3, u'5']
assert raw == [b'3', b'5', b'2']
assert raw == ['3', '5', '2']
raw = [b'a', b'b']
raw = ['a', 'b']
e.raw['test'] = raw
assert e['test'] is not nice
assert e['test'] == [u'a', u'b']
@@ -308,7 +301,7 @@ class test_LDAPEntry(object):
nice = 'not list'
e['test'] = nice
assert e['test'] is nice
assert e.raw['test'] == [b'not list']
assert e.raw['test'] == ['not list']
e.raw['test'].append(b'second')
e.raw['test'].append('second')
assert e['test'] == ['not list', u'second']

131
ipatests/test_ipaserver/test_otptoken_import.py
View File

@@ -17,16 +17,15 @@
# 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 pytest
import sys
import nose
from nss import nss
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"))
@@ -36,7 +35,7 @@ class test_otptoken_import(object):
'ipatokenotpkey': u'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ',
'ipatokenvendor': u'Manufacturer',
'ipatokenserial': u'987654321',
'ipatokenhotpcounter': 0,
'ipatokenhotpcounter': 0L,
'ipatokenotpdigits': 8,
'type': u'hotp',
})]
@@ -62,89 +61,91 @@ class test_otptoken_import(object):
assert False
def test_figure6(self):
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'})]
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()
def test_figure7(self):
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'})]
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()
def test_figure8(self):
nss.nss_init_nodb()
try:
PSKCDocument(os.path.join(basename, "pskc-figure8.xml"))
doc = 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:
PSKCDocument(os.path.join(basename, "pskc-invalid.xml"))
doc = 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"))
for t in doc.getKeyPackages():
t._PSKCKeyPackage__process()
[(t.id, t.options) for t in doc.getKeyPackages()]
except ValidationError: # Unsupported token type.
pass
else:
assert False
finally:
nss.nss_shutdown()
def test_full(self):
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'
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()

58
ipatests/test_ipaserver/test_rpcserver.py
View File

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

55
ipatests/test_ipaserver/test_secrets.py
View File

@@ -1,55 +0,0 @@
# 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 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):
try:
shutil.rmtree('test-ipa-sec-store')
except Exception: # pylint: disable=broad-except
pass
testdir = 'test-ipa-sec-store'
pwfile = os.path.join(testdir, 'pwfile')
os.mkdir(testdir)
with open(pwfile, 'w') as f:
f.write('testpw')
cls.certdb = os.path.join(testdir, 'certdb')
os.mkdir(cls.certdb)
cls.cert2db = os.path.join(testdir, 'cert2db')
os.mkdir(cls.cert2db)
seedfile = os.path.join(testdir, 'seedfile')
with open(seedfile, 'wb') as f:
seed = os.urandom(1024)
f.write(seed)
subprocess.call(['certutil', '-d', cls.certdb, '-N', '-f', pwfile])
subprocess.call(['certutil', '-d', cls.cert2db, '-N', '-f', pwfile])
subprocess.call(['certutil', '-d', cls.certdb, '-S', '-f', pwfile,
'-s', 'CN=testCA', '-n', 'testCACert', '-x',
'-t', 'CT,C,C', '-m', '1', '-z', seedfile])
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)

743
ipatests/test_ipaserver/test_serverroles.py
View File

@@ -1,743 +0,0 @@
#
# Copyright (C) 2016 FreeIPA Contributors see COPYING for license
#
"""
Tests for the serverroles backend
"""
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
View File

@@ -1,74 +0,0 @@
#
# 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]))

51
ipatests/test_ipaserver/test_version_comparison.py
View File

@@ -1,51 +0,0 @@
#
# Copyright (C) 2015 FreeIPA Contributors see COPYING for license
#
"""
tests for correct RPM version comparison
"""
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)
)