Imported Debian patch 4.7.2-3
This commit is contained in:
Timo Aaltonen
committed by
Mario Fetka
Mario Fetka
parent
27edeba051
commit
8bc559c5a1
277
ipatests/test_cmdline/test_ipagetkeytab.py
Executable file → Normal file
277
ipatests/test_cmdline/test_ipagetkeytab.py
Executable file → Normal file
@@ -29,38 +29,31 @@ import tempfile
|
||||
import gssapi
|
||||
import pytest
|
||||
|
||||
from ipapython.ipautil import private_ccache
|
||||
from ipalib import api, errors
|
||||
from ipalib.request import context
|
||||
from ipalib import api
|
||||
from ipaplatform.paths import paths
|
||||
from ipapython import ipautil, ipaldap
|
||||
from ipaserver.plugins.ldap2 import ldap2
|
||||
from ipatests.test_cmdline.cmdline import cmdline_test
|
||||
from ipatests.test_xmlrpc.tracker import host_plugin, service_plugin
|
||||
from ipatests.test_xmlrpc.xmlrpc_test import fuzzy_digits, add_oc
|
||||
from contextlib import contextmanager
|
||||
|
||||
|
||||
@contextmanager
|
||||
def use_keytab(principal, keytab):
|
||||
with private_ccache() as ccache_file:
|
||||
try:
|
||||
old_principal = getattr(context, 'principal', None)
|
||||
name = gssapi.Name(principal, gssapi.NameType.kerberos_principal)
|
||||
store = {'ccache': ccache_file,
|
||||
'client_keytab': keytab}
|
||||
gssapi.Credentials(name=name, usage='initiate', store=store)
|
||||
conn = ldap2(api)
|
||||
conn.connect(ccache=ccache_file,
|
||||
autobind=ipaldap.AUTOBIND_DISABLED)
|
||||
yield conn
|
||||
conn.disconnect()
|
||||
except gssapi.exceptions.GSSError as e:
|
||||
raise Exception('Unable to bind to LDAP. Error initializing '
|
||||
'principal %s in %s: %s' % (principal, keytab,
|
||||
str(e)))
|
||||
finally:
|
||||
setattr(context, 'principal', old_principal)
|
||||
try:
|
||||
tmpdir = tempfile.mkdtemp(prefix = "tmp-")
|
||||
ccache_file = 'FILE:%s/ccache' % tmpdir
|
||||
name = gssapi.Name(principal, gssapi.NameType.kerberos_principal)
|
||||
store = {'ccache': ccache_file,
|
||||
'client_keytab': keytab}
|
||||
os.environ['KRB5CCNAME'] = ccache_file
|
||||
gssapi.Credentials(name=name, usage='initiate', store=store)
|
||||
conn = ldap2(api)
|
||||
conn.connect(autobind=ipaldap.AUTOBIND_DISABLED)
|
||||
conn.disconnect()
|
||||
except gssapi.exceptions.GSSError as e:
|
||||
raise Exception('Unable to bind to LDAP. Error initializing principal %s in %s: %s' % (principal, keytab, str(e)))
|
||||
finally:
|
||||
os.environ.pop('KRB5CCNAME', None)
|
||||
if tmpdir:
|
||||
shutil.rmtree(tmpdir)
|
||||
|
||||
|
||||
@pytest.fixture(scope='class')
|
||||
@@ -70,7 +63,7 @@ def test_host(request):
|
||||
|
||||
|
||||
@pytest.fixture(scope='class')
|
||||
def test_service(request, test_host, keytab_retrieval_setup):
|
||||
def test_service(request, test_host):
|
||||
service_tracker = service_plugin.ServiceTracker(u'srv', test_host.name)
|
||||
test_host.ensure_exists()
|
||||
return service_tracker.make_fixture(request)
|
||||
@@ -84,9 +77,10 @@ class KeytabRetrievalTest(cmdline_test):
|
||||
command = "ipa-getkeytab"
|
||||
keytabname = None
|
||||
|
||||
@pytest.fixture(autouse=True, scope="class")
|
||||
def keytab_retrieval_setup(self, request, cmdline_setup):
|
||||
cls = request.cls
|
||||
@classmethod
|
||||
def setup_class(cls):
|
||||
super(KeytabRetrievalTest, cls).setup_class()
|
||||
|
||||
keytabfd, keytabname = tempfile.mkstemp()
|
||||
|
||||
os.close(keytabfd)
|
||||
@@ -94,16 +88,17 @@ class KeytabRetrievalTest(cmdline_test):
|
||||
|
||||
cls.keytabname = keytabname
|
||||
|
||||
def fin():
|
||||
try:
|
||||
os.unlink(cls.keytabname)
|
||||
except OSError:
|
||||
pass
|
||||
@classmethod
|
||||
def teardown_class(cls):
|
||||
super(KeytabRetrievalTest, cls).teardown_class()
|
||||
|
||||
request.addfinalizer(fin)
|
||||
try:
|
||||
os.unlink(cls.keytabname)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def run_ipagetkeytab(self, service_principal, args=tuple(),
|
||||
raiseonerr=False, stdin=None):
|
||||
raiseonerr=False):
|
||||
new_args = [self.command,
|
||||
"-p", service_principal,
|
||||
"-k", self.keytabname]
|
||||
@@ -115,7 +110,7 @@ class KeytabRetrievalTest(cmdline_test):
|
||||
|
||||
return ipautil.run(
|
||||
new_args,
|
||||
stdin=stdin,
|
||||
stdin=None,
|
||||
raiseonerr=raiseonerr,
|
||||
capture_error=True)
|
||||
|
||||
@@ -167,9 +162,7 @@ class test_ipagetkeytab(KeytabRetrievalTest):
|
||||
"""
|
||||
Try to use the service keytab.
|
||||
"""
|
||||
with use_keytab(test_service.name, self.keytabname) as conn:
|
||||
assert conn.can_read(test_service.dn, 'objectclass') is True
|
||||
assert getattr(context, 'principal') == test_service.name
|
||||
use_keytab(test_service.name, self.keytabname)
|
||||
|
||||
def test_4_disable(self, test_service):
|
||||
"""
|
||||
@@ -193,77 +186,10 @@ class test_ipagetkeytab(KeytabRetrievalTest):
|
||||
Try to use the disabled keytab
|
||||
"""
|
||||
try:
|
||||
with use_keytab(test_service.name, self.keytabname) as conn:
|
||||
assert conn.can_read(test_service.dn, 'objectclass') is True
|
||||
assert getattr(context, 'principal') == test_service.name
|
||||
use_keytab(test_service.name, self.keytabname)
|
||||
except Exception as errmsg:
|
||||
assert('Unable to bind to LDAP. Error initializing principal' in str(errmsg))
|
||||
|
||||
def test_6_quiet_mode(self, test_service):
|
||||
"""
|
||||
Try to use quiet mode
|
||||
"""
|
||||
test_service.ensure_exists()
|
||||
# getkeytab without quiet mode option enabled
|
||||
result = self.run_ipagetkeytab(test_service.name)
|
||||
err = result.error_output.split("\n")[0]
|
||||
assert err == f"Keytab successfully retrieved and stored in:" \
|
||||
f" {self.keytabname}"
|
||||
assert result.returncode == 0
|
||||
|
||||
# getkeytab with quiet mode option enabled
|
||||
result1 = self.run_ipagetkeytab(test_service.name, args=tuple("-q"))
|
||||
assert result1.returncode == 0
|
||||
|
||||
def test_7_server_name_check(self, test_service):
|
||||
"""
|
||||
Try to use -s for server name
|
||||
"""
|
||||
test_service.ensure_exists()
|
||||
self.assert_success(test_service.name, args=["-s", api.env.host])
|
||||
|
||||
def test_8_keytab_encryption_check(self, test_service):
|
||||
"""
|
||||
Try to use -e for different types of encryption check
|
||||
"""
|
||||
encryptes_list = [
|
||||
"aes256-cts-hmac-sha1-96",
|
||||
"aes128-cts-hmac-sha256-128",
|
||||
]
|
||||
self.assert_success(
|
||||
test_service.name, args=["-e", ",".join(encryptes_list)]
|
||||
)
|
||||
|
||||
def test_dangling_symlink(self, test_service):
|
||||
# see https://pagure.io/freeipa/issue/4607
|
||||
test_service.ensure_exists()
|
||||
|
||||
fd, symlink_target = tempfile.mkstemp()
|
||||
os.close(fd)
|
||||
os.unlink(symlink_target)
|
||||
# create dangling symlink
|
||||
os.symlink(self.keytabname, symlink_target)
|
||||
|
||||
try:
|
||||
self.assert_success(test_service.name, raiseonerr=True)
|
||||
assert os.path.isfile(symlink_target)
|
||||
assert os.path.samefile(self.keytabname, symlink_target)
|
||||
finally:
|
||||
os.unlink(symlink_target)
|
||||
|
||||
|
||||
def retrieve_dm_password():
|
||||
dmpw_file = os.path.join(api.env.dot_ipa, '.dmpw')
|
||||
|
||||
if not os.path.isfile(dmpw_file):
|
||||
raise errors.NotFound(reason='{} file required '
|
||||
'for this test'.format(dmpw_file))
|
||||
|
||||
with open(dmpw_file, 'r') as f:
|
||||
dm_password = f.read().strip()
|
||||
|
||||
return dm_password
|
||||
|
||||
|
||||
class TestBindMethods(KeytabRetrievalTest):
|
||||
"""
|
||||
@@ -273,13 +199,17 @@ class TestBindMethods(KeytabRetrievalTest):
|
||||
dm_password = None
|
||||
ca_cert = None
|
||||
|
||||
@pytest.fixture(autouse=True, scope="class")
|
||||
def bindmethods_setup(self, request, keytab_retrieval_setup):
|
||||
cls = request.cls
|
||||
try:
|
||||
cls.dm_password = retrieve_dm_password()
|
||||
except errors.NotFound as e:
|
||||
pytest.skip(e.args)
|
||||
@classmethod
|
||||
def setup_class(cls):
|
||||
super(TestBindMethods, cls).setup_class()
|
||||
|
||||
dmpw_file = os.path.join(api.env.dot_ipa, '.dmpw')
|
||||
|
||||
if not os.path.isfile(dmpw_file):
|
||||
pytest.skip('{} file required for this test'.format(dmpw_file))
|
||||
|
||||
with open(dmpw_file, 'r') as f:
|
||||
cls.dm_password = f.read().strip()
|
||||
|
||||
tempfd, temp_ca_cert = tempfile.mkstemp()
|
||||
|
||||
@@ -289,12 +219,14 @@ class TestBindMethods(KeytabRetrievalTest):
|
||||
|
||||
cls.ca_cert = temp_ca_cert
|
||||
|
||||
def fin():
|
||||
try:
|
||||
os.unlink(cls.ca_cert)
|
||||
except OSError:
|
||||
pass
|
||||
request.addfinalizer(fin)
|
||||
@classmethod
|
||||
def teardown_class(cls):
|
||||
super(TestBindMethods, cls).teardown_class()
|
||||
|
||||
try:
|
||||
os.unlink(cls.ca_cert)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def check_ldapi(self):
|
||||
if not api.env.ldap_uri.startswith('ldapi://'):
|
||||
@@ -393,106 +325,3 @@ class TestBindMethods(KeytabRetrievalTest):
|
||||
'-H', 'ldaps://{}'.format(api.env.host),
|
||||
'-Y', 'EXTERNAL'],
|
||||
raiseonerr=False)
|
||||
|
||||
|
||||
class SMBServiceTracker(service_plugin.ServiceTracker):
|
||||
def __init__(self, name, host_fqdn, options=None):
|
||||
super(SMBServiceTracker, self).__init__(name, host_fqdn,
|
||||
options=options)
|
||||
# Create SMB service principal that has POSIX attributes to allow
|
||||
# generating SID and adding proper objectclasses
|
||||
self.create_keys |= {u'uidnumber', u'gidnumber'}
|
||||
self.options[u'addattr'] = [
|
||||
u'objectclass=ipaIDObject', u'uidNumber=-1', u'gidNumber=-1']
|
||||
|
||||
def track_create(self, **options):
|
||||
super(SMBServiceTracker, self).track_create(**options)
|
||||
self.attrs[u'uidnumber'] = [fuzzy_digits]
|
||||
self.attrs[u'gidnumber'] = [fuzzy_digits]
|
||||
self.attrs[u'objectclass'].append(u'ipaIDObject')
|
||||
|
||||
|
||||
@pytest.fixture(scope='class')
|
||||
def test_smb_svc(request, test_host, smb_service_setup):
|
||||
service_tracker = SMBServiceTracker(u'cifs', test_host.name)
|
||||
test_host.ensure_exists()
|
||||
return service_tracker.make_fixture(request)
|
||||
|
||||
|
||||
@pytest.mark.tier0
|
||||
@pytest.mark.skipif(u'ipantuserattrs' not in add_oc([], u'ipantuserattrs'),
|
||||
reason="Must have trust support enabled for this test")
|
||||
class test_smb_service(KeytabRetrievalTest):
|
||||
"""
|
||||
Test `ipa-getkeytab` for retrieving explicit enctypes
|
||||
"""
|
||||
command = "ipa-getkeytab"
|
||||
dm_password = None
|
||||
keytabname = None
|
||||
|
||||
@pytest.fixture(autouse=True, scope="class")
|
||||
def smb_service_setup(self, request, keytab_retrieval_setup):
|
||||
cls = request.cls
|
||||
try:
|
||||
cls.dm_password = retrieve_dm_password()
|
||||
except errors.NotFound as e:
|
||||
pytest.skip(e.args)
|
||||
|
||||
def test_create(self, test_smb_svc):
|
||||
"""
|
||||
Create a keytab with `ipa-getkeytab` for an existing service.
|
||||
"""
|
||||
test_smb_svc.ensure_exists()
|
||||
|
||||
# Request a keytab with explicit encryption types
|
||||
enctypes = ['aes128-cts-hmac-sha1-96',
|
||||
'aes256-cts-hmac-sha1-96', 'arcfour-hmac']
|
||||
args = ['-e', ','.join(enctypes), '-s', api.env.host]
|
||||
self.assert_success(test_smb_svc.name, args=args, raiseonerr=True)
|
||||
|
||||
def test_use(self, test_smb_svc):
|
||||
"""
|
||||
Try to use the service keytab to regenerate ipaNTHash value
|
||||
"""
|
||||
# Step 1. Extend objectclass to allow ipaNTHash attribute
|
||||
# We cannot verify write access to objectclass
|
||||
with use_keytab(test_smb_svc.name, self.keytabname) as conn:
|
||||
entry = conn.get_entry(test_smb_svc.dn, ['objectclass'])
|
||||
entry['objectclass'].extend(['ipaNTUserAttrs'])
|
||||
try:
|
||||
conn.update_entry(entry)
|
||||
except errors.ACIError:
|
||||
assert ('No correct ACI to the allow ipaNTUserAttrs '
|
||||
'for SMB service' in "failure")
|
||||
|
||||
# Step 2. With ipaNTUserAttrs in place, we can ask to regenerate
|
||||
# ipaNTHash value. We can also verify it is possible to write to
|
||||
# ipaNTHash attribute while being an SMB service
|
||||
with use_keytab(test_smb_svc.name, self.keytabname) as conn:
|
||||
assert conn.can_write(test_smb_svc.dn, 'ipaNTHash') is True
|
||||
entry = conn.get_entry(test_smb_svc.dn, ['ipaNTHash'])
|
||||
entry['ipanthash'] = b'MagicRegen'
|
||||
try:
|
||||
conn.update_entry(entry)
|
||||
except errors.ACIError:
|
||||
assert ("No correct ACI to the ipaNTHash for SMB service"
|
||||
in "failure")
|
||||
except errors.EmptyResult:
|
||||
assert "No arcfour-hmac in Kerberos keys" in "failure"
|
||||
except errors.DatabaseError:
|
||||
# Most likely ipaNTHash already existed -- we either get
|
||||
# OPERATIONS_ERROR or UNWILLING_TO_PERFORM, both map to
|
||||
# the same DatabaseError class.
|
||||
assert "LDAP Entry corruption after generation" in "failure"
|
||||
|
||||
# Update succeeded, now we have either MagicRegen (broken) or
|
||||
# a real NT hash in the entry. However, we can only retrieve it as
|
||||
# a cn=Directory Manager. When bind_dn is None, ldap2.connect() wil
|
||||
# default to cn=Directory Manager.
|
||||
conn = ldap2(api)
|
||||
conn.connect(bind_dn=None, bind_pw=self.dm_password,
|
||||
autobind=ipaldap.AUTOBIND_DISABLED)
|
||||
entry = conn.retrieve(test_smb_svc.dn, ['ipaNTHash'])
|
||||
ipanthash = entry.single_value.get('ipanthash')
|
||||
conn.disconnect()
|
||||
assert ipanthash != b'MagicRegen', 'LDBM backend entry corruption'
|
||||
|
||||
Reference in New Issue
Block a user