URL: https://github.com/freeipa/freeipa/pull/677 Author: HonzaCholasta Title: #677: cert: defer cert-find result post-processing Action: synchronized
To pull the PR as Git branch: git remote add ghfreeipa https://github.com/freeipa/freeipa git fetch ghfreeipa pull/677/head:pr677 git checkout pr677
From 2a3a05a076590b7d668d7c56a52d23529029cc19 Mon Sep 17 00:00:00 2001 From: Jan Cholasta <jchol...@redhat.com> Date: Thu, 30 Mar 2017 08:33:30 +0000 Subject: [PATCH] cert: defer cert-find result post-processing Rather than post-processing the results of each internal search, post-process the combined result. This avoids expensive per-certificate searches on certificates which won't even be included in the combined result when cert-find is executed with the --all option. https://pagure.io/freeipa/issue/6808 --- ipaserver/plugins/cert.py | 93 +++++++++++++++++++++++++++------------------ ipaserver/plugins/dogtag.py | 10 +++++ 2 files changed, 66 insertions(+), 37 deletions(-) diff --git a/ipaserver/plugins/cert.py b/ipaserver/plugins/cert.py index 5590913..1a425de 100644 --- a/ipaserver/plugins/cert.py +++ b/ipaserver/plugins/cert.py @@ -250,6 +250,11 @@ def normalize_pkidate(value): return datetime.datetime.strptime(value, PKIDATE_FORMAT) +def convert_pkidatetime(value): + value = datetime.datetime.fromtimestamp(int(value) // 1000) + return x509.format_datetime(value) + + def validate_csr(ugettext, csr): """ Ensure the CSR is base64-encoded and can be decoded by our PKCS#10 @@ -1384,18 +1389,7 @@ def _get_cert_key(self, cert): return (DN(cert_obj.issuer), cert_obj.serial_number) - def _get_cert_obj(self, cert, all, raw, pkey_only): - obj = {'certificate': base64.b64encode(cert).decode('ascii')} - - full = not pkey_only and all - if not raw: - self.obj._parse(obj, full) - if not full: - del obj['certificate'] - - return obj - - def _cert_search(self, all, raw, pkey_only, **options): + def _cert_search(self, pkey_only, **options): result = collections.OrderedDict() try: @@ -1404,15 +1398,19 @@ def _cert_search(self, all, raw, pkey_only, **options): return result, False, False try: - key = self._get_cert_key(cert) + issuer, serial_number = self._get_cert_key(cert) except ValueError: return result, True, True - result[key] = self._get_cert_obj(cert, all, raw, pkey_only) + obj = {'serial_number': serial_number} + if not pkey_only: + obj['certificate'] = base64.b64encode(cert).decode('ascii') + + result[issuer, serial_number] = obj return result, False, True - def _ca_search(self, all, raw, pkey_only, exactly, **options): + def _ca_search(self, raw, pkey_only, exactly, **options): ra_options = {} for name in ('revocation_reason', 'issuer', @@ -1445,7 +1443,6 @@ def _ca_search(self, all, raw, pkey_only, exactly, **options): return result, False, complete ca_objs = self.api.Command.ca_find( - all=all, timelimit=0, sizelimit=0, )['result'] @@ -1465,24 +1462,16 @@ def _ca_search(self, all, raw, pkey_only, exactly, **options): obj = {'serial_number': serial_number} else: obj = ra_obj - if all: - obj.update(ra.get_certificate(str(serial_number))) if not raw: obj['issuer'] = issuer obj['subject'] = DN(ra_obj['subject']) + obj['valid_not_before'] = ( + convert_pkidatetime(obj['valid_not_before'])) + obj['valid_not_after'] = ( + convert_pkidatetime(obj['valid_not_after'])) obj['revoked'] = ( ra_obj['status'] in (u'REVOKED', u'REVOKED_EXPIRED')) - if all: - obj['certificate'] = ( - obj['certificate'].replace('\r\n', '')) - self.obj._parse(obj) - - if 'certificate_chain' in ca_obj: - cert = x509.load_certificate(obj['certificate']) - cert_der = cert.public_bytes(serialization.Encoding.DER) - obj['certificate_chain'] = ( - [cert_der] + ca_obj['certificate_chain']) obj['cacn'] = ca_obj['cn'][0] @@ -1490,7 +1479,7 @@ def _ca_search(self, all, raw, pkey_only, exactly, **options): return result, False, complete - def _ldap_search(self, all, raw, pkey_only, no_members, **options): + def _ldap_search(self, all, pkey_only, no_members, **options): ldap = self.api.Backend.ldap2 filters = [] @@ -1549,26 +1538,25 @@ def _ldap_search(self, all, raw, pkey_only, no_members, **options): for attr in ('usercertificate', 'usercertificate;binary'): for cert in entry.get(attr, []): try: - key = self._get_cert_key(cert) + issuer, serial_number = self._get_cert_key(cert) except ValueError: truncated = True continue try: - obj = result[key] + obj = result[issuer, serial_number] except KeyError: - obj = self._get_cert_obj(cert, all, raw, pkey_only) - result[key] = obj + obj = {'serial_number': serial_number} + if not pkey_only and all: + obj['certificate'] = ( + base64.b64encode(cert).decode('ascii')) + result[issuer, serial_number] = obj if not pkey_only and (all or not no_members): owners = obj.setdefault('owner', []) if entry.dn not in owners: owners.append(entry.dn) - if not raw: - for obj in six.itervalues(result): - self.obj._fill_owners(obj) - return result, truncated, complete def execute(self, criteria=None, all=False, raw=False, pkey_only=False, @@ -1625,6 +1613,37 @@ def execute(self, criteria=None, all=False, raw=False, pkey_only=False, truncated = truncated or sub_truncated complete = complete or sub_complete + if not pkey_only: + ca_objs = {} + ra = self.api.Backend.ra + + for key, obj in six.iteritems(result): + if all and 'cacn' in obj: + _issuer, serial_number = key + cacn = obj['cacn'] + + try: + ca_obj = ca_objs[cacn] + except KeyError: + ca_obj = ca_objs[cacn] = ( + self.api.Command.ca_show(cacn, all=True)['result']) + + obj.update(ra.get_certificate(str(serial_number))) + if not raw: + obj['certificate'] = ( + obj['certificate'].replace('\r\n', '')) + + if 'certificate_chain' in ca_obj: + cert = x509.load_certificate(obj['certificate']) + cert_der = ( + cert.public_bytes(serialization.Encoding.DER)) + obj['certificate_chain'] = ( + [cert_der] + ca_obj['certificate_chain']) + + if not raw: + self.obj._parse(obj, all) + self.obj._fill_owners(obj) + result = list(six.itervalues(result)) if sizelimit > 0 and len(result) > sizelimit: if not truncated: diff --git a/ipaserver/plugins/dogtag.py b/ipaserver/plugins/dogtag.py index d1dd707..3997531 100644 --- a/ipaserver/plugins/dogtag.py +++ b/ipaserver/plugins/dogtag.py @@ -1945,6 +1945,16 @@ def convert_time(value): if len(issuer_dn) == 1: response_request['issuer'] = unicode(issuer_dn[0].text) + not_valid_before = cert.xpath('NotValidBefore') + if len(not_valid_before) == 1: + response_request['valid_not_before'] = ( + unicode(not_valid_before[0].text)) + + not_valid_after = cert.xpath('NotValidAfter') + if len(not_valid_after) == 1: + response_request['valid_not_after'] = ( + unicode(not_valid_after[0].text)) + status = cert.xpath('Status') if len(status) == 1: response_request['status'] = unicode(status[0].text)
-- Manage your subscription for the Freeipa-devel mailing list: https://www.redhat.com/mailman/listinfo/freeipa-devel Contribute to FreeIPA: http://www.freeipa.org/page/Contribute/Code
