From 23051783b79241d712fe3d957c75c78f7ab05d9c Mon Sep 17 00:00:00 2001 From: William Brown Date: Tue, 22 Aug 2017 09:11:56 +1000 Subject: [PATCH] Ticket 95 - extend tls certmap test Bug Description: Extend the TLS certmap test to account for more complex certmap configurations. Fix Description: Add a variety of tests for certmap, based on dncomps, filtercomps, cmap attr, basedn. Relies on https://pagure.io/389-ds-base/issue/49218 https://pagure.io/lib389/issue/95 Author: wibrown Review by: ??? --- lib389/__init__.py | 26 ++++---- lib389/_mapped_object.py | 34 ++++++----- lib389/idm/user.py | 15 ++++- lib389/instance/setup.py | 8 ++- lib389/nss_ssl.py | 20 +++++- lib389/plugins_v4.py | 62 +++++++++++++++++++ lib389/tests/tls_external_test.py | 125 +++++++++++++++++++++++++++++++------- 7 files changed, 237 insertions(+), 53 deletions(-) create mode 100644 lib389/plugins_v4.py diff --git a/lib389/__init__.py b/lib389/__init__.py index 0b0a1fa..fb5e7fe 100644 --- a/lib389/__init__.py +++ b/lib389/__init__.py @@ -520,7 +520,7 @@ class DirSrv(SimpleLDAPObject, object): (self.sslport or self.port))) - def openConnection(self, *args, **kwargs): + def clone(self, args_instance={}): """ Open a new connection to our LDAP server *IMPORTANT* @@ -536,7 +536,6 @@ class DirSrv(SimpleLDAPObject, object): args_instance[SER_SERVERID_PROP] = self.serverid args_standalone = args_instance.copy() server.allocate(args_standalone) - server.open(*args, **kwargs) return server @@ -1027,6 +1026,9 @@ class DirSrv(SimpleLDAPObject, object): # There are cases (especially CACERT/USERCERTS) where when one connection # is open set_option SILENTLY fails!!!! # + # PYTHON LDAP SHARES ldap and per conn options as GLOBAL STATE for cert + # related components. You CAN NOT make this work the way you want. + # # You MAY need to set post_open=False in your DirSrv start/restart instance! ################## @@ -1036,6 +1038,15 @@ class DirSrv(SimpleLDAPObject, object): certdir = self.get_cert_dir() log.debug("Using dirsrv ca certificate %s" % certdir) + if certdir is not None: + """ + We have a certificate directory, so lets start up TLS negotiations + """ + # Note this sets LDAP.OPT not SELF. Because once self has opened + # it can NOT change opts AT ALL. + ldap.set_option(ldap.OPT_X_TLS_CACERTDIR, ensure_str(certdir)) + log.debug("Using external ca certificate %s" % certdir) + if userkey is not None: # Note this sets LDAP.OPT not SELF. Because once self has opened # it can NOT change opts AT ALL. @@ -1047,15 +1058,6 @@ class DirSrv(SimpleLDAPObject, object): ldap.set_option(ldap.OPT_X_TLS_CERTFILE, ensure_str(usercert)) log.debug("Using user certificate %s" % usercert) - if certdir is not None: - """ - We have a certificate directory, so lets start up TLS negotiations - """ - # Note this sets LDAP.OPT not SELF. Because once self has opened - # it can NOT change opts AT ALL. - ldap.set_option(ldap.OPT_X_TLS_CACERTDIR, ensure_str(certdir)) - log.debug("Using external ca certificate %s" % certdir) - if certdir or starttls: try: # Note this sets LDAP.OPT not SELF. Because once self has opened @@ -1067,7 +1069,7 @@ class DirSrv(SimpleLDAPObject, object): log.fatal('TLS negotiation failed: %s' % str(e)) raise e - ## NOW INIT THIS. This MUST be after all the ldap.OPT set above, + ## NOW INIT THIS. This MUST be before all the ldap.OPT set below, # so that we inherit the settings correctly!!!! if self.verbose: self.log.info('open(): Connecting to uri %s' % uri) diff --git a/lib389/_mapped_object.py b/lib389/_mapped_object.py index 6b50f83..9e27a68 100644 --- a/lib389/_mapped_object.py +++ b/lib389/_mapped_object.py @@ -13,7 +13,7 @@ import logging from functools import partial from lib389._entry import Entry -from lib389._constants import DIRSRV_STATE_ONLINE +from lib389._constants import DIRSRV_STATE_ONLINE, SER_ROOT_DN, SER_ROOT_PW from lib389.utils import ( ensure_bytes, ensure_str, ensure_int, ensure_list_bytes, ensure_list_str, ensure_list_int @@ -377,9 +377,9 @@ class DSLdapObject(DSLogging): # If the account can be bound to, this will attempt to do so. We don't check # for exceptions, just pass them back! def bind(self, password=None, *args, **kwargs): - conn = self._instance.openConnection(*args, **kwargs) - conn.simple_bind_s(self.dn, password) - return conn + inst_clone = self._instance.clone({SER_ROOT_DN: self.dn, SER_ROOT_PW: password}) + inst_clone.open(*args, **kwargs) + return inst_clone def delete(self): """ @@ -435,7 +435,9 @@ class DSLdapObject(DSLogging): if basedn is None: raise ldap.UNWILLING_TO_PERFORM('Invalid request to create. basedn cannot be None') - if properties.get(self._rdn_attribute, None) is not None: + if rdn is not None: + tdn = ensure_str('%s,%s' % (rdn, basedn)) + elif properties.get(self._rdn_attribute, None) is not None: # Favour the value in the properties dictionary v = properties.get(self._rdn_attribute) rdn = ensure_str(v[0]) @@ -603,17 +605,17 @@ class DSLdapObjects(DSLogging): raise ldap.UNWILLING_TO_PERFORM("properties must be a dictionary") # Get the rdn out of the properties if it's unset??? - if rdn is None and self._rdn_attribute in properties: - # First see if we can get it from the properties. - trdn = properties.get(self._rdn_attribute) - if type(trdn) == str: - rdn = trdn - elif type(trdn) == list and len(trdn) != 1: - raise ldap.UNWILLING_TO_PERFORM("Cannot determine rdn %s from properties. Too many choices" % (self._rdn_attribute)) - elif type(trdn) == list: - rdn = trdn[0] - else: - raise ldap.UNWILLING_TO_PERFORM("Cannot determine rdn %s from properties, Invalid type" % type(trdn)) + # if rdn is None and self._rdn_attribute in properties: + # # First see if we can get it from the properties. + # trdn = properties.get(self._rdn_attribute) + # if type(trdn) == str: + # rdn = "%s=%s" % (self._rdn_attribute, trdn) + # elif type(trdn) == list and len(trdn) != 1: + # raise ldap.UNWILLING_TO_PERFORM("Cannot determine rdn %s from properties. Too many choices" % (self._rdn_attribute)) + # elif type(trdn) == list: + # rdn = "%s=%s" % (self._rdn_attribute, trdn[0]) + # else: + # raise ldap.UNWILLING_TO_PERFORM("Cannot determine rdn %s from properties, Invalid type" % type(trdn)) return (rdn, properties) diff --git a/lib389/idm/user.py b/lib389/idm/user.py index ee4e08d..e4228c9 100644 --- a/lib389/idm/user.py +++ b/lib389/idm/user.py @@ -52,6 +52,7 @@ class UserAccount(Account): self._create_objectclasses.append('inetUser') else: self._create_objectclasses.append('nsMemberOf') + self._create_objectclasses.append('nsAccount') user_compare_exclude = [ 'nsUniqueId', 'modifyTimestamp', @@ -67,6 +68,15 @@ class UserAccount(Account): return super(UserAccount, self)._validate(rdn, properties, basedn) + def enroll_certificate(self, der_path): + if ds_is_older('1.3.7'): + raise Exception("This version of DS does not support nsAccount") + # Given a cert path, add this to the object as a userCertificate + crt = None + with open(der_path, 'rb') as f: + crt = f.read() + self.add('usercertificate;binary', crt) + # Add a set password function.... # Can't I actually just set, and it will hash? @@ -81,5 +91,8 @@ class UserAccounts(DSLdapObjects): ] self._filterattrs = [RDN] self._childobject = UserAccount - self._basedn = '{},{}'.format(rdn, basedn) + if rdn is None: + self._basedn = basedn + else: + self._basedn = '{},{}'.format(rdn, basedn) diff --git a/lib389/instance/setup.py b/lib389/instance/setup.py index cc94b48..424d847 100644 --- a/lib389/instance/setup.py +++ b/lib389/instance/setup.py @@ -353,7 +353,13 @@ class SetupDs(object): srcfile = os.path.join(slapd['sysconf_dir'], 'dirsrv/config/slapd-collations.conf') dstfile = os.path.join(slapd['config_dir'], 'slapd-collations.conf') shutil.copy2(srcfile, dstfile) - os.chown(slapd['schema_dir'], slapd['user_uid'], slapd['group_gid']) + os.chown(dstfile, slapd['user_uid'], slapd['group_gid']) + + # Copy in the certmap configuration + srcfile = os.path.join(slapd['sysconf_dir'], 'dirsrv/config/certmap.conf') + dstfile = os.path.join(slapd['config_dir'], 'certmap.conf') + shutil.copy2(srcfile, dstfile) + os.chown(dstfile, slapd['user_uid'], slapd['group_gid']) # If we are on the correct platform settings, systemd if general['systemd'] and not self.containerised: diff --git a/lib389/nss_ssl.py b/lib389/nss_ssl.py index 90e78f9..cad32ca 100644 --- a/lib389/nss_ssl.py +++ b/lib389/nss_ssl.py @@ -15,6 +15,7 @@ import random import string import re import socket +import time # from nss import nss from subprocess import check_call, check_output from lib389.passwd import password_generate @@ -102,6 +103,8 @@ class NssSsl(object): Create a self signed CA. """ + # Wait a second to avoid an NSS bug with serial ids based on time. + time.sleep(1) # Create noise. self._generate_noise('%s/noise.txt' % self.dirsrv.get_cert_dir()) # Now run the command. Can we do this with NSS native? @@ -258,6 +261,8 @@ class NssSsl(object): if self.dirsrv.host not in alt_names: alt_names.append(self.dirsrv.host) + # Wait a second to avoid an NSS bug with serial ids based on time. + time.sleep(1) # Create noise. self._generate_noise('%s/noise.txt' % self.dirsrv.get_cert_dir()) cmd = [ @@ -295,6 +300,8 @@ class NssSsl(object): Name is the uid of the account, and will become the CN of the cert. """ + # Wait a second to avoid an NSS bug with serial ids based on time. + time.sleep(1) cmd = [ '/usr/bin/certutil', '-S', @@ -358,6 +365,16 @@ class NssSsl(object): '-clcerts', '-nodes' ]) + # Convert the cert for userCertificate attr + check_call([ + 'openssl', + 'x509', + '-inform', 'PEM', + '-outform', 'DER', + '-in', '%s/%s%s.crt' % (self.dirsrv.get_cert_dir(), USER_PREFIX, name), + '-out', '%s/%s%s.der' % (self.dirsrv.get_cert_dir(), USER_PREFIX, name), + ]) + return True def get_rsa_user(self, name): @@ -367,5 +384,6 @@ class NssSsl(object): ca_path = '%s/ca.crt' % self.dirsrv.get_cert_dir() key_path = '%s/%s%s.key' % (self.dirsrv.get_cert_dir(), USER_PREFIX, name) crt_path = '%s/%s%s.crt' % (self.dirsrv.get_cert_dir(), USER_PREFIX, name) - return {'ca': ca_path, 'key': key_path, 'crt': crt_path} + crt_der_path = '%s/%s%s.der' % (self.dirsrv.get_cert_dir(), USER_PREFIX, name) + return {'ca': ca_path, 'key': key_path, 'crt': crt_path, 'crt_der_path': crt_der_path} diff --git a/lib389/plugins_v4.py b/lib389/plugins_v4.py new file mode 100644 index 0000000..f42182d --- /dev/null +++ b/lib389/plugins_v4.py @@ -0,0 +1,62 @@ +# --- BEGIN COPYRIGHT BLOCK --- +# Copyright (C) 2017 Red Hat, Inc. +# All rights reserved. +# +# License: GPL (version 3 or any later version). +# See LICENSE for details. +# --- END COPYRIGHT BLOCK --- + +import copy + +from lib389._mapped_object import DSLdapObjects, DSLdapObject + +class DylibPlugin4(DSLdapObject): + _plugin_properties = { + 'nsslapd-pluginEnabled' : 'off', + 'nsslapd-pluginInitfunc' : 'NONE', + 'nsslapd-pluginPath': 'NONE', + } + + def __init__(self, instance, dn=None, batch=False): + super(DylibPlugin4, self).__init__(instance, dn, batch) + self._rdn_attribute = 'cn' + self._must_attributes = [ + 'cn', + 'nsslapd-pluginPath', + 'nsslapd-pluginInitfunc', + 'nsslapd-pluginEnabled', + ] + self._create_objectclasses = ['top', 'nsDylibPlugin4'] + # We'll mark this protected, and people can just disable the plugins. + self._protected = True + + def enable(self): + self.set('nsslapd-pluginEnabled', 'on') + + def disable(self): + self.set('nsslapd-pluginEnabled', 'off') + + def status(self): + return self.get_attr_val_utf8('nsslapd-pluginEnabled') == 'on' + + def create(self, rdn=None, properties=None, basedn=None): + # When we create plugins, we don't want people to have to consider all + # the little details. Plus, the server during creation needs to be able + # to create these from nothing. + # As a result, all the named plugins carry a default properties + # dictionary that can be used. + + # Copy the plugin internal properties. + internal_properties = copy.deepcopy(self._plugin_properties) + if properties is not None: + internal_properties.update(properties) + return super(DylibPlugin4, self).create(rdn, internal_properties, basedn) + +class CertmapPlugin(DylibPlugin4): + def __init__(self, instance, dn="cn=certmap plugin,cn=plugins,cn=config", batch=False): + super(CertmapPlugin, self).__init__(instance, dn, batch) + + + + + diff --git a/lib389/tests/tls_external_test.py b/lib389/tests/tls_external_test.py index 68aaa9d..47ebaf8 100644 --- a/lib389/tests/tls_external_test.py +++ b/lib389/tests/tls_external_test.py @@ -8,16 +8,52 @@ # import ldap +import pytest +from lib389 import DirSrv +from lib389.utils import ds_is_older from lib389.topologies import topology_st from lib389.utils import logging from lib389.idm.user import UserAccounts -from lib389._constants import DEFAULT_SUFFIX, SECUREPORT_STANDALONE1 +from lib389.plugins_v4 import CertmapPlugin +from lib389.backend import Backends, BACKEND_SAMPLE_ENTRIES +from lib389._constants import DEFAULT_SUFFIX, SECUREPORT_STANDALONE1, INSTALL_LATEST_CONFIG from lib389.config import CertmapLegacy +CERT_SUFFIX = 'O=testing,L=lib389,ST=Queensland,C=AU' + log = logging.getLogger(__name__) + +def _check_bind(inst, dn, tls_locs): + #server.allocate(args_standalone) + inst_clone = inst.clone() + inst_clone.open(saslmethod='EXTERNAL', + connOnly=True, + certdir=inst.get_cert_dir(), + userkey=tls_locs['key'], + usercert=tls_locs['crt']) + assert(inst_clone.whoami_s().lower() == "dn: %s" % dn.lower()) + inst_clone.close() + log.debug("PASS: bound as %s" % dn) + +@pytest.mark.skipif(ds_is_older('1.3.7'), reason="Not implemented in 1.3.7") +def test_certmap_plugin_upgrade(topology_st): + standalone = topology_st.standalone + + certmap = CertmapPlugin(standalone) + # Now, reach in and allow deletion! + certmap._protected = False + certmap.delete() + + # Restart the server + standalone.restart() + + # Now assert that the item exists again. + certmap_reload = CertmapPlugin(standalone) + assert certmap_reload.status() == True + def test_tls_external(topology_st): standalone = topology_st.standalone @@ -28,23 +64,41 @@ def test_tls_external(topology_st): assert(standalone.nss_ssl.create_rsa_ca() is True) assert(standalone.nss_ssl.create_rsa_key_and_cert() is True) # Create a user - assert(standalone.nss_ssl.create_rsa_user('testuser') is True) + assert(standalone.nss_ssl.create_rsa_user('testuser_a') is True) # Now get the details of where the key and crt are. - tls_locs = standalone.nss_ssl.get_rsa_user('testuser') - # {'ca': ca_path, 'key': key_path, 'crt': crt_path} + testuser_a_tls_locs = standalone.nss_ssl.get_rsa_user('testuser_a') # Start again standalone.start() - users = UserAccounts(standalone, DEFAULT_SUFFIX) - user = users.create(properties={ - 'uid': 'testuser', - 'cn' : 'testuser', + # This user is added to a different backend to test "without" dncomps + # IE we try to bind to the DN in the cert. + backends = Backends(standalone) + backends.create(properties={ + 'cn': 'certRoot', + 'nsslapd-suffix': CERT_SUFFIX, + BACKEND_SAMPLE_ENTRIES: INSTALL_LATEST_CONFIG + }) + + c_users = UserAccounts(standalone, CERT_SUFFIX, rdn=None) + testuser_b = c_users.create(rdn='cn=testuser_a', properties={ + 'uid': 'testuser_b', + 'cn' : 'testuser_b', + 'o': 'testing', 'sn' : 'user', - 'uidNumber' : '1000', - 'gidNumber' : '2000', - 'homeDirectory' : '/home/testuser' + 'uidNumber' : '1001', + 'gidNumber' : '2001', + 'homeDirectory' : '/home/testuser_b', + 'nsCertSubjectDn': 'CN=testuser_a,O=testing,L=lib389,ST=Queensland,C=AU' }) + testuser_b.enroll_certificate(testuser_a_tls_locs['crt_der_path']) + + ## REMEMBER the bind logic is in this order: + # If dncomps == None: bind to subject DN in cert + # If dncomps == None, cmapattr set, search object with cmap attr. nsCertSubjectDN + # if dncomps == '', use filter comps ot search an entyr below basedn + # if dncomps == 'attr', use these to construct a DN + standalone.rsa.create() # Set the secure port and nsslapd-security @@ -54,7 +108,8 @@ def test_tls_external(topology_st): # Now turn on the certmap. cm = CertmapLegacy(standalone) certmaps = cm.list() - certmaps['default']['DNComps'] = '' + # This means to use filter comps instead of dn comps + certmaps['default']['DNComps'] = None certmaps['default']['FilterComps'] = ['cn'] certmaps['default']['VerifyCert'] = 'off' cm.set(certmaps) @@ -66,17 +121,43 @@ def test_tls_external(topology_st): standalone.restart(post_open=False) # Now attempt a bind with TLS external - conn = standalone.openConnection(saslmethod='EXTERNAL', connOnly=True, certdir=standalone.get_cert_dir(), userkey=tls_locs['key'], usercert=tls_locs['crt']) + _check_bind(standalone, 'cn=testuser_a,o=testing,l=lib389,st=Queensland,c=AU', testuser_a_tls_locs) + + # Change the certmap and try and verify the certificate + certmaps['default']['CmapLdapAttr'] = 'nsCertSubjectDN' + cm.set(certmaps) + standalone.restart(post_open=False) + _check_bind(standalone, 'cn=testuser_a,o=testing,l=lib389,st=Queensland,c=AU', testuser_a_tls_locs) + + # Change the basedn and check it works. + certmaps['default']['basedn'] = 'o=testing,l=lib389,st=Queensland,c=AU' + cm.set(certmaps) + standalone.restart(post_open=False) + _check_bind(standalone, 'cn=testuser_a,o=testing,l=lib389,st=Queensland,c=AU', testuser_a_tls_locs) + + # Change the certmap and try and verify the certificate + certmaps['default']['VerifyCert'] = 'on' + cm.set(certmaps) + standalone.restart(post_open=False) + _check_bind(standalone, 'cn=testuser_a,o=testing,l=lib389,st=Queensland,c=AU', testuser_a_tls_locs) - assert(conn.whoami_s() == "dn: uid=testuser,ou=People,dc=example,dc=com") + # Check filter comps + certmaps['default']['DNComps'] = '' + certmaps['default']['CmapLdapAttr'] = None + cm.set(certmaps) + standalone.restart(post_open=False) + _check_bind(standalone, 'cn=testuser_a,o=testing,l=lib389,st=Queensland,c=AU', testuser_a_tls_locs) - # Backup version of the code: - # ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER) - # ldap.set_option(ldap.OPT_X_TLS_CACERTFILE, tls_locs['ca']) - # ldap.set_option(ldap.OPT_X_TLS_KEYFILE, tls_locs['key']) - # ldap.set_option(ldap.OPT_X_TLS_CERTFILE, tls_locs['crt']) - # conn = ldap.initialize(standalone.toLDAPURL()) + # Change the filter comps to check o and cn + certmaps['default']['FilterComps'] = ['cn', 'o'] + cm.set(certmaps) + standalone.restart(post_open=False) + _check_bind(standalone, 'cn=testuser_a,o=testing,l=lib389,st=Queensland,c=AU', testuser_a_tls_locs) - # sasl_auth = ldap.sasl.external() - # conn.sasl_interactive_bind_s("", sasl_auth) + # Finally, check the dncomps + # The order of these matter as they are extract and appended in *this* order. + certmaps['default']['DNComps'] = ['cn', 'o', 'l', 'st', 'c'] + cm.set(certmaps) + standalone.restart(post_open=False) + _check_bind(standalone, 'cn=testuser_a,o=testing,l=lib389,st=Queensland,c=AU', testuser_a_tls_locs) -- 1.8.3.1