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 09064c3b7b8c57eefe3750ba80961369359d3569 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 | 95 +++++++++++++++++++++++++-------------------- ipaserver/plugins/dogtag.py | 10 +++++ 2 files changed, 63 insertions(+), 42 deletions(-) diff --git a/ipaserver/plugins/cert.py b/ipaserver/plugins/cert.py index 1a6d045..b4b46d5 100644 --- a/ipaserver/plugins/cert.py +++ b/ipaserver/plugins/cert.py @@ -162,6 +162,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 @@ -1296,18 +1301,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, **options): result = collections.OrderedDict() try: @@ -1320,11 +1314,11 @@ def _cert_search(self, all, raw, pkey_only, **options): except ValueError: return result, True, True - result[key] = self._get_cert_obj(cert, all, raw, pkey_only) + result[key] = {'certificate': base64.b64encode(cert).decode('ascii')} 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', @@ -1357,7 +1351,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'] @@ -1373,28 +1366,17 @@ def _ca_search(self, all, raw, pkey_only, exactly, **options): except KeyError: continue - if pkey_only: - obj = {'serial_number': serial_number} - else: - obj = ra_obj - if all: - obj.update(ra.get_certificate(str(serial_number))) + obj = ra_obj - if not raw: - obj['issuer'] = issuer - obj['subject'] = DN(ra_obj['subject']) - 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']) + if not pkey_only and 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')) obj['cacn'] = ca_obj['cn'][0] @@ -1402,7 +1384,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 = [] @@ -1469,7 +1451,10 @@ def _ldap_search(self, all, raw, pkey_only, no_members, **options): try: obj = result[key] except KeyError: - obj = self._get_cert_obj(cert, all, raw, pkey_only) + obj = {} + if not pkey_only and all: + obj['certificate'] = ( + base64.b64encode(cert).decode('ascii')) result[key] = obj if not pkey_only and (all or not no_members): @@ -1477,10 +1462,6 @@ def _ldap_search(self, all, raw, pkey_only, no_members, **options): 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, @@ -1537,6 +1518,36 @@ def execute(self, criteria=None, all=False, raw=False, pkey_only=False, truncated = truncated or sub_truncated complete = complete or sub_complete + ca_objs = {} + ra = self.api.Backend.ra + for key, obj in six.iteritems(result): + if pkey_only: + obj = { + k: v for k, v in six.iteritems(obj) + if k in ('serial_number', 'cacn') + } + elif all and 'cacn' in obj: + 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(key[1]))) + 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, not pkey_only and 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
