From f7a88f8550550057ca3e37b76a2c67e480d3d529 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 | 14 +++++- lib389/nss_ssl.py | 20 +++++++- lib389/tests/tls_external_test.py | 99 ++++++++++++++++++++++++++++++--------- 5 files changed, 141 insertions(+), 52 deletions(-) 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..03cdac9 100644 --- a/lib389/idm/user.py +++ b/lib389/idm/user.py @@ -40,6 +40,8 @@ class UserAccount(Account): self._create_objectclasses = [ 'top', 'account', + 'nsAccount', + 'nsMemberOf', 'posixaccount', # inetOrgPerson allows userCertificate 'inetOrgPerson', @@ -67,6 +69,13 @@ class UserAccount(Account): return super(UserAccount, self)._validate(rdn, properties, basedn) + def enroll_certificate(self, der_path): + # 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 +90,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/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/tests/tls_external_test.py b/lib389/tests/tls_external_test.py index 68aaa9d..4014838 100644 --- a/lib389/tests/tls_external_test.py +++ b/lib389/tests/tls_external_test.py @@ -9,15 +9,32 @@ import ldap +from lib389 import DirSrv 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.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) + + def test_tls_external(topology_st): standalone = topology_st.standalone @@ -28,23 +45,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 +89,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 +102,36 @@ 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) - assert(conn.whoami_s() == "dn: uid=testuser,ou=People,dc=example,dc=com") + # 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) - # 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 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) + + # 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) + + # 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) + + # Change the basedn and check it works. + standalone.config.set('nsslapd-certmap-basedn', 'o=testing,l=lib389,st=Queensland,c=AU') + 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) -- 1.8.3.1