From 34f175dc045799bd541e4bfa8f442eb2bd0a252e Mon Sep 17 00:00:00 2001 From: William Brown Date: Wed, 4 Oct 2017 10:49:15 +1000 Subject: [PATCH 1/2] 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/tests/tls_external_test.py | 125 +++++++++++++++++++++++++++++++------- 5 files changed, 156 insertions(+), 52 deletions(-) diff --git a/lib389/__init__.py b/lib389/__init__.py index 39e504f..a13a1ec 100644 --- a/lib389/__init__.py +++ b/lib389/__init__.py @@ -521,7 +521,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* @@ -537,7 +537,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 @@ -1028,6 +1027,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! ################## @@ -1037,6 +1039,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. @@ -1048,15 +1059,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 @@ -1068,7 +1070,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 f73cb25..5b34fec 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 @@ -534,9 +534,9 @@ class DSLdapObject(DSLogging): :returns: Connection with a binding as the entry """ - 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): """Deletes the object defined by self._dn. @@ -592,7 +592,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]) @@ -804,17 +806,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/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