URL: https://github.com/freeipa/freeipa/pull/217 Author: frasertweedale Title: #217: change certificate processing code to use python-cryptography Action: opened
PR body: """ This commit changes certificate processing code to use python-cryptography instead of NSS. Part of the refactoring effort, certificates sub-effort. Reviewed at dkupka/freeipa:pull/1 """ To pull the PR as Git branch: git remote add ghfreeipa https://github.com/freeipa/freeipa git fetch ghfreeipa pull/217/head:pr217 git checkout pr217
From 48d4aa5be45ff73d7384d2aa004550f595ec6570 Mon Sep 17 00:00:00 2001 From: Fraser Tweedale <ftwee...@redhat.com> Date: Mon, 10 Oct 2016 16:08:52 +1000 Subject: [PATCH 1/7] dn: support conversion from python-cryptography Name The upcoming change to using python-cryptography for certificate process will require a way to convert ``cryptography.x509.name.Name`` values to ``ipapython.dn.DN``. Update the ``DN`` constructor to accept a ``Name``. Part of: https://fedorahosted.org/freeipa/ticket/6398 --- ipapython/dn.py | 43 ++++++++++++++++++++++++++++++++++++-- ipatests/test_ipapython/test_dn.py | 23 ++++++++++++++++++-- 2 files changed, 62 insertions(+), 4 deletions(-) diff --git a/ipapython/dn.py b/ipapython/dn.py index 3ee35c6..682e0ca 100644 --- a/ipapython/dn.py +++ b/ipapython/dn.py @@ -422,6 +422,7 @@ import sys import functools +import cryptography.x509 from ldap.dn import str2dn, dn2str from ldap import DECODING_ERROR import six @@ -976,6 +977,8 @@ class DN(object): to yield one or more RDN's which will be appended in order to the DN. The parsing recognizes the DN syntax escaping rules. + * A single ``cryptography.x509.name.Name`` object. + * A RDN object, the RDN will copied respecting the constructors keyword configuration parameters and appended in order. @@ -1125,9 +1128,17 @@ def _rdns_from_value(self, value): rdns = [[ava]] elif isinstance(value, RDN): rdns = [value.to_openldap()] + elif isinstance(value, cryptography.x509.name.Name): + rdns = list(reversed([ + [get_ava( + _ATTR_NAME_BY_OID.get(ava.oid, ava.oid.dotted_string), + ava.value)] + for ava in value + ])) else: - raise TypeError("must be str, unicode, tuple, or RDN or DN, got %s instead" % - type(value)) + raise TypeError( + "must be str, unicode, tuple, Name, RDN or DN, got %s instead" + % type(value)) return rdns def _rdns_from_sequence(self, seq): @@ -1407,3 +1418,31 @@ def rindex(self, pattern, start=None, end=None): if i == -1: raise ValueError("pattern not found") return i + + +_ATTR_NAME_BY_OID = { + cryptography.x509.oid.NameOID.COMMON_NAME: 'CN', + cryptography.x509.oid.NameOID.COUNTRY_NAME: 'C', + cryptography.x509.oid.NameOID.LOCALITY_NAME: 'L', + cryptography.x509.oid.NameOID.STATE_OR_PROVINCE_NAME: 'ST', + cryptography.x509.oid.NameOID.ORGANIZATION_NAME: 'O', + cryptography.x509.oid.NameOID.ORGANIZATIONAL_UNIT_NAME: 'OU', + cryptography.x509.oid.NameOID.SERIAL_NUMBER: 'serialNumber', + cryptography.x509.oid.NameOID.SURNAME: 'SN', + cryptography.x509.oid.NameOID.GIVEN_NAME: 'givenName', + cryptography.x509.oid.NameOID.TITLE: 'title', + cryptography.x509.oid.NameOID.GENERATION_QUALIFIER: 'generationQualifier', + cryptography.x509.oid.NameOID.DN_QUALIFIER: 'dnQualifier', + cryptography.x509.oid.NameOID.PSEUDONYM: 'pseudonym', + cryptography.x509.oid.NameOID.DOMAIN_COMPONENT: 'DC', + cryptography.x509.oid.NameOID.EMAIL_ADDRESS: 'E', + cryptography.x509.oid.NameOID.JURISDICTION_COUNTRY_NAME: + 'incorporationCountry', + cryptography.x509.oid.NameOID.JURISDICTION_LOCALITY_NAME: + 'incorporationLocality', + cryptography.x509.oid.NameOID.JURISDICTION_STATE_OR_PROVINCE_NAME: + 'incorporationState', + cryptography.x509.oid.NameOID.BUSINESS_CATEGORY: 'businessCategory', + cryptography.x509.ObjectIdentifier('2.5.4.9'): 'STREET', + cryptography.x509.ObjectIdentifier('0.9.2342.19200300.100.1.1'): 'UID', +} diff --git a/ipatests/test_ipapython/test_dn.py b/ipatests/test_ipapython/test_dn.py index a96bd33..3ca3b57 100644 --- a/ipatests/test_ipapython/test_dn.py +++ b/ipatests/test_ipapython/test_dn.py @@ -2,6 +2,7 @@ import unittest import pytest +from cryptography import x509 import six from ipapython.dn import DN, RDN, AVA @@ -621,7 +622,7 @@ class TestDN(unittest.TestCase): def setUp(self): # ava1 must sort before ava2 self.attr1 = 'cn' - self.value1 = 'Bob' + self.value1 = u'Bob' self.str_ava1 = '%s=%s' % (self.attr1, self.value1) self.ava1 = AVA(self.attr1, self.value1) @@ -629,7 +630,7 @@ def setUp(self): self.rdn1 = RDN((self.attr1, self.value1)) self.attr2 = 'ou' - self.value2 = 'people' + self.value2 = u'people' self.str_ava2 = '%s=%s' % (self.attr2, self.value2) self.ava2 = AVA(self.attr2, self.value2) @@ -656,6 +657,11 @@ def setUp(self): self.base_container_dn = DN((self.attr1, self.value1), self.container_dn, self.base_dn) + self.x500name = x509.Name([ + x509.NameAttribute( + x509.NameOID.ORGANIZATIONAL_UNIT_NAME, self.value2), + x509.NameAttribute(x509.NameOID.COMMON_NAME, self.value1), + ]) def assertExpectedClass(self, klass, obj, component): self.assertIs(obj.__class__, expected_class(klass, component)) @@ -794,6 +800,19 @@ def test_create(self): self.assertEqual(dn1[0], self.rdn1) self.assertEqual(dn1[1], self.rdn2) + # Create with a python-cryptography 'Name' + dn1 = DN(self.x500name) + self.assertEqual(len(dn1), 2) + self.assertExpectedClass(DN, dn1, 'self') + for i in range(0, len(dn1)): + self.assertExpectedClass(DN, dn1[i], 'RDN') + for j in range(0, len(dn1[i])): + self.assertExpectedClass(DN, dn1[i][j], 'AVA') + self.assertIsInstance(dn1[i].attr, unicode) + self.assertIsInstance(dn1[i].value, unicode) + self.assertEqual(dn1[0], self.rdn1) + self.assertEqual(dn1[1], self.rdn2) + # Create with RDN, and 2 DN's (e.g. attr + container + base) dn1 = DN((self.attr1, self.value1), self.container_dn, self.base_dn) self.assertEqual(len(dn1), 5) From 0ffa9af31395907925ef6125953002d0380bc286 Mon Sep 17 00:00:00 2001 From: Fraser Tweedale <ftwee...@redhat.com> Date: Mon, 10 Oct 2016 14:31:49 +1000 Subject: [PATCH 2/7] pkcs10: use python-cryptography for CSR processing Update ``ipalib.pkcs10`` module to use python-cryptography for CSR processing instead of NSS. Part of: https://fedorahosted.org/freeipa/ticket/6398 --- ipalib/pkcs10.py | 94 +++++++----------------------------- ipalib/x509.py | 39 +++++++++++++++ ipaserver/plugins/cert.py | 95 +++++++++++++++++-------------------- ipatests/test_pkcs10/test_pkcs10.py | 93 +++++++++++++++++++++--------------- 4 files changed, 152 insertions(+), 169 deletions(-) diff --git a/ipalib/pkcs10.py b/ipalib/pkcs10.py index 158ebb3..44a529d 100644 --- a/ipalib/pkcs10.py +++ b/ipalib/pkcs10.py @@ -19,71 +19,12 @@ from __future__ import print_function +import binascii import sys -import base64 -import nss.nss as nss +from cryptography.hazmat.backends import default_backend +import cryptography.x509 from pyasn1.type import univ, namedtype, tag from pyasn1.codec.der import decoder -import six -from ipalib import x509 - -if six.PY3: - unicode = str - -PEM = 0 -DER = 1 - -def get_subject(csr, datatype=PEM): - """ - Given a CSR return the subject value. - - This returns an nss.DN object. - """ - request = load_certificate_request(csr, datatype) - try: - return request.subject - finally: - del request - -def get_extensions(csr, datatype=PEM): - """ - Given a CSR return OIDs of certificate extensions. - - The return value is a tuple of strings - """ - request = load_certificate_request(csr, datatype) - - # Work around a bug in python-nss where nss.oid_dotted_decimal - # errors on unrecognised OIDs - # - # https://bugzilla.redhat.com/show_bug.cgi?id=1246729 - # - def get_prefixed_oid_str(ext): - """Returns a string like 'OID.1.2...'.""" - if ext.oid_tag == 0: - return repr(ext) - else: - return nss.oid_dotted_decimal(ext.oid) - - return tuple(get_prefixed_oid_str(ext)[4:] - for ext in request.extensions) - - -def get_subjectaltname(csr, datatype=PEM): - """ - Given a CSR return the subjectaltname value, if any. - - The return value is a tuple of strings or None - """ - request = load_certificate_request(csr, datatype) - for extension in request.extensions: - if extension.oid_tag == nss.SEC_OID_X509_SUBJECT_ALT_NAME: - break - else: - return None - del request - - return x509.decode_generalnames(extension.value) # Unfortunately, NSS can only parse the extension request attribute, so @@ -148,31 +89,28 @@ def strip_header(csr): return csr -def load_certificate_request(csr, datatype=PEM): - """ - Given a base64-encoded certificate request, with or without the - header/footer, return a request object. + +def load_certificate_request(data): """ - if datatype == PEM: - csr = strip_header(csr) - csr = base64.b64decode(csr) + Load a PEM or base64-encoded PKCS #10 certificate request. - # A fail-safe so we can always read a CSR. python-nss/NSS will segfault - # otherwise - if not nss.nss_is_initialized(): - nss.nss_init_nodb() + :return: a python-cryptography ``Certificate`` object. + :raises: ``ValueError`` if unable to load the request - return nss.CertificateRequest(csr) + """ + data = strip_header(data) + try: + data = binascii.a2b_base64(data) + except binascii.Error as e: + raise ValueError(e) + return cryptography.x509.load_der_x509_csr(data, default_backend()) -if __name__ == '__main__': - nss.nss_init_nodb() +if __name__ == '__main__': # Read PEM request from stdin and print out its components csrlines = sys.stdin.readlines() csr = ''.join(csrlines) print(load_certificate_request(csr)) - print(get_subject(csr)) - print(get_subjectaltname(csr)) print(get_friendlyname(csr)) diff --git a/ipalib/x509.py b/ipalib/x509.py index e986a97..e67aab6 100644 --- a/ipalib/x509.py +++ b/ipalib/x509.py @@ -39,6 +39,7 @@ import base64 import re +import cryptography.x509 import nss.nss as nss from nss.error import NSPRError from pyasn1.type import univ, char, namedtype, tag @@ -52,6 +53,9 @@ from ipaplatform.paths import paths from ipapython.dn import DN +if six.PY3: + unicode = str + PEM = 0 DER = 1 @@ -513,6 +517,41 @@ def decode_generalnames(secitem): return names +class KRB5PrincipalName(cryptography.x509.general_name.OtherName): + def __init__(self, type_id, value): + super(KRB5PrincipalName, self).__init__(type_id, value) + self.name = _decode_krb5principalname(value) + + +class UPN(cryptography.x509.general_name.OtherName): + def __init__(self, type_id, value): + super(UPN, self).__init__(type_id, value) + self.name = unicode( + decoder.decode(value, asn1Spec=char.UTF8String())[0]) + + +OTHERNAME_CLASS_MAP = { + SAN_KRB5PRINCIPALNAME: KRB5PrincipalName, + SAN_UPN: UPN, +} + + +def process_othernames(gns): + """ + Process python-cryptography GeneralName values, yielding + OtherName values of more specific type if type is known. + + """ + for gn in gns: + if isinstance(gn, cryptography.x509.general_name.OtherName): + cls = OTHERNAME_CLASS_MAP.get( + gn.type_id.dotted_string, + cryptography.x509.general_name.OtherName) + yield cls(gn.type_id, gn.value) + else: + yield gn + + if __name__ == '__main__': # this can be run with: # python ipalib/x509.py < /etc/ipa/ca.crt diff --git a/ipaserver/plugins/cert.py b/ipaserver/plugins/cert.py index a66cc38..5e85942 100644 --- a/ipaserver/plugins/cert.py +++ b/ipaserver/plugins/cert.py @@ -20,14 +20,13 @@ # along with this program. If not, see <http://www.gnu.org/licenses/>. import base64 -import binascii import collections import datetime import os +import cryptography.x509 from nss import nss from nss.error import NSPRError -from pyasn1.error import PyAsn1Error import six from ipalib import Command, Str, Int, Flag @@ -174,31 +173,9 @@ def validate_csr(ugettext, csr): return try: pkcs10.load_certificate_request(csr) - except (TypeError, binascii.Error) as e: - raise errors.Base64DecodeError(reason=str(e)) - except Exception as e: + except (TypeError, ValueError) as e: raise errors.CertificateOperationError(error=_('Failure decoding Certificate Signing Request: %s') % e) -def normalize_csr(csr): - """ - Strip any leading and trailing cruft around the BEGIN/END block - """ - end_len = 37 - s = csr.find('-----BEGIN NEW CERTIFICATE REQUEST-----') - if s == -1: - s = csr.find('-----BEGIN CERTIFICATE REQUEST-----') - e = csr.find('-----END NEW CERTIFICATE REQUEST-----') - if e == -1: - e = csr.find('-----END CERTIFICATE REQUEST-----') - if e != -1: - end_len = 33 - - if s > -1 and e > -1: - # We're normalizing here, not validating - csr = csr[s:e+end_len] - - return csr - def normalize_serial_number(num): """ @@ -515,7 +492,6 @@ class cert_request(Create, BaseCertMethod, VirtualCommand): 'csr', validate_csr, label=_('CSR'), cli_name='csr_file', - normalizer=normalize_csr, noextrawhitespace=False, ), ) @@ -607,17 +583,21 @@ def execute(self, csr, all=False, raw=False, **kw): caacl_check(principal_type, principal, ca, profile_id) try: - subject = pkcs10.get_subject(csr) - extensions = pkcs10.get_extensions(csr) - subjectaltname = pkcs10.get_subjectaltname(csr) or () - except (NSPRError, PyAsn1Error, ValueError) as e: + csr_obj = pkcs10.load_certificate_request(csr) + except ValueError as e: raise errors.CertificateOperationError( error=_("Failure decoding Certificate Signing Request: %s") % e) + try: + ext_san = csr_obj.extensions.get_extension_for_oid( + cryptography.x509.oid.ExtensionOID.SUBJECT_ALTERNATIVE_NAME) + except cryptography.x509.extensions.ExtensionNotFound: + ext_san = None + # self-service and host principals may bypass SAN permission check if (bind_principal_string != principal_string and bind_principal_type != HOST): - if '2.5.29.17' in extensions: + if ext_san is not None: self.check_access('request certificate with subjectaltname') dn = None @@ -650,10 +630,14 @@ def execute(self, csr, all=False, raw=False, **kw): dn = principal_obj['dn'] # Ensure that the DN in the CSR matches the principal - cn = subject.common_name #pylint: disable=E1101 - if not cn: + # + # We only look at the "most specific" CN value + cns = csr_obj.subject.get_attributes_for_oid( + cryptography.x509.oid.NameOID.COMMON_NAME) + if len(cns) == 0: raise errors.ValidationError(name='csr', error=_("No Common Name was found in subject of request.")) + cn = cns[-1].value # "most specific" is end of list if principal_type in (SERVICE, HOST): if cn.lower() != principal.hostname.lower(): @@ -670,8 +654,11 @@ def execute(self, csr, all=False, raw=False, **kw): ) # check email address - mail = subject.email_address #pylint: disable=E1101 - if mail is not None and mail not in principal_obj.get('mail', []): + # + # fail if any email addr from DN does not appear in ldap entry + email_addrs = csr_obj.subject.get_attributes_for_oid( + cryptography.x509.oid.NameOID.EMAIL_ADDRESS) + if len(set(email_addrs) - set(principal_obj.get('mail', []))) > 0: raise errors.ValidationError( name='csr', error=_( @@ -685,9 +672,12 @@ def execute(self, csr, all=False, raw=False, **kw): "to the 'userCertificate' attribute of entry '%s'.") % dn) # Validate the subject alt name, if any - for name_type, desc, name, _der_name in subjectaltname: - if name_type == nss.certDNSName: - name = unicode(name) + generalnames = [] + if ext_san is not None: + generalnames = x509.process_othernames(ext_san.value) + for gn in generalnames: + if isinstance(gn, cryptography.x509.general_name.DNSName): + name = gn.value alt_principal = None alt_principal_obj = None try: @@ -703,8 +693,9 @@ def execute(self, csr, all=False, raw=False, **kw): elif principal_type == USER: raise errors.ValidationError( name='csr', - error=_("subject alt name type %s is forbidden " - "for user principals") % desc + error=_( + "subject alt name type %s is forbidden " + "for user principals") % "DNSName" ) except errors.NotFound: # We don't want to issue any certificates referencing @@ -721,17 +712,15 @@ def execute(self, csr, all=False, raw=False, **kw): "with subject alt name '%s'.") % name) if alt_principal is not None and not bypass_caacl: caacl_check(principal_type, alt_principal, ca, profile_id) - elif name_type in [ - (nss.certOtherName, x509.SAN_UPN), - (nss.certOtherName, x509.SAN_KRB5PRINCIPALNAME), - ]: - if name != principal_string: + elif isinstance(gn, (x509.KRB5PrincipalName, x509.UPN)): + if gn.name != principal_string: raise errors.ACIError( - info=_("Principal '%s' in subject alt name does not " - "match requested principal") % name) - elif name_type == nss.certRFC822Name: + info=_( + "Principal '%s' in subject alt name does not " + "match requested principal") % gn.name) + elif isinstance(gn, cryptography.x509.general_name.RFC822Name): if principal_type == USER: - if name not in principal_obj.get('mail', []): + if gn.value not in principal_obj.get('mail', []): raise errors.ValidationError( name='csr', error=_( @@ -741,12 +730,14 @@ def execute(self, csr, all=False, raw=False, **kw): else: raise errors.ValidationError( name='csr', - error=_("subject alt name type %s is forbidden " - "for non-user principals") % desc + error=_( + "subject alt name type %s is forbidden " + "for non-user principals") % "RFC822Name" ) else: raise errors.ACIError( - info=_("Subject alt name type %s is forbidden") % desc) + info=_("Subject alt name type %s is forbidden") + % type(gn).__name__) # Request the certificate try: diff --git a/ipatests/test_pkcs10/test_pkcs10.py b/ipatests/test_pkcs10/test_pkcs10.py index 152d8e7..850ed91 100644 --- a/ipatests/test_pkcs10/test_pkcs10.py +++ b/ipatests/test_pkcs10/test_pkcs10.py @@ -20,18 +20,12 @@ Test the `pkcs10.py` module. """ -# FIXME: Pylint errors -# pylint: disable=no-member - -import binascii - import nose from ipalib import pkcs10 from ipapython import ipautil -import nss.nss as nss -from nss.error import NSPRError import pytest import os +import cryptography.x509 @pytest.mark.tier0 @@ -41,7 +35,6 @@ class test_update(object): """ def setup(self): - nss.nss_init_nodb() self.testdir = os.path.abspath(os.path.dirname(__file__)) if not ipautil.file_exists(os.path.join(self.testdir, "test0.csr")): @@ -57,13 +50,19 @@ def test_0(self): """ Test simple CSR with no attributes """ - csr = self.read_file("test0.csr") + csr = pkcs10.load_certificate_request(self.read_file("test0.csr")) - subject = pkcs10.get_subject(csr) + subject = csr.subject - assert(subject.common_name == 'test.example.com') - assert(subject.state_name == 'California') - assert(subject.country_name == 'US') + cn = subject.get_attributes_for_oid( + cryptography.x509.NameOID.COMMON_NAME)[-1].value + assert(cn == 'test.example.com') + st = subject.get_attributes_for_oid( + cryptography.x509.NameOID.STATE_OR_PROVINCE_NAME)[-1].value + assert(st == 'California') + c = subject.get_attributes_for_oid( + cryptography.x509.NameOID.COUNTRY_NAME)[-1].value + assert(c == 'US') def test_1(self): """ @@ -74,13 +73,20 @@ def test_1(self): subject = request.subject - assert(subject.common_name == 'test.example.com') - assert(subject.state_name == 'California') - assert(subject.country_name == 'US') - - for extension in request.extensions: - if extension.oid_tag == nss.SEC_OID_X509_SUBJECT_ALT_NAME: - assert nss.x509_alt_name(extension.value)[0] == 'testlow.example.com' + cn = subject.get_attributes_for_oid( + cryptography.x509.NameOID.COMMON_NAME)[-1].value + assert(cn == 'test.example.com') + st = subject.get_attributes_for_oid( + cryptography.x509.NameOID.STATE_OR_PROVINCE_NAME)[-1].value + assert(st == 'California') + c = subject.get_attributes_for_oid( + cryptography.x509.NameOID.COUNTRY_NAME)[-1].value + assert(c == 'US') + + san = request.extensions.get_extension_for_oid( + cryptography.x509.ExtensionOID.SUBJECT_ALTERNATIVE_NAME).value + dns = san.get_values_for_type(cryptography.x509.DNSName) + assert dns[0] == 'testlow.example.com' def test_2(self): """ @@ -91,18 +97,32 @@ def test_2(self): subject = request.subject - assert(subject.common_name == 'test.example.com') - assert(subject.state_name == 'California') - assert(subject.country_name == 'US') - - for extension in request.extensions: - if extension.oid_tag == nss.SEC_OID_X509_SUBJECT_ALT_NAME: - assert nss.x509_alt_name(extension.value)[0] == 'testlow.example.com' - if extension.oid_tag == nss.SEC_OID_X509_CRL_DIST_POINTS: - pts = nss.CRLDistributionPts(extension.value) - urls = pts[0].get_general_names() - assert('http://ca.example.com/my.crl' in urls) - assert('http://other.example.com/my.crl' in urls) + cn = subject.get_attributes_for_oid( + cryptography.x509.NameOID.COMMON_NAME)[-1].value + assert(cn == 'test.example.com') + st = subject.get_attributes_for_oid( + cryptography.x509.NameOID.STATE_OR_PROVINCE_NAME)[-1].value + assert(st == 'California') + c = subject.get_attributes_for_oid( + cryptography.x509.NameOID.COUNTRY_NAME)[-1].value + assert(c == 'US') + + san = request.extensions.get_extension_for_oid( + cryptography.x509.ExtensionOID.SUBJECT_ALTERNATIVE_NAME).value + dns = san.get_values_for_type(cryptography.x509.DNSName) + assert dns[0] == 'testlow.example.com' + + crldps = request.extensions.get_extension_for_oid( + cryptography.x509.ExtensionOID.CRL_DISTRIBUTION_POINTS).value + gns = [] + for crldp in crldps: + gns.extend(crldp.full_name) + uris = [ + u'http://ca.example.com/my.crl', + u'http://other.example.com/my.crl', + ] + for uri in uris: + assert cryptography.x509.UniformResourceIdentifier(uri) in gns def test_3(self): """ @@ -110,18 +130,13 @@ def test_3(self): """ csr = self.read_file("test3.csr") - try: + with pytest.raises(ValueError): pkcs10.load_certificate_request(csr) - except NSPRError as nsprerr: - # (SEC_ERROR_BAD_DER) security library: improperly formatted DER-encoded message. - assert(nsprerr. errno== -8183) def test_4(self): """ Test CSR with badly formatted base64-encoded data """ csr = self.read_file("test4.csr") - try: + with pytest.raises(ValueError): pkcs10.load_certificate_request(csr) - except (TypeError, binascii.Error) as typeerr: - assert(str(typeerr) == 'Incorrect padding') From 065f5dbfb762c2a7a2e9b764413b56c7739053fc Mon Sep 17 00:00:00 2001 From: Fraser Tweedale <ftwee...@redhat.com> Date: Wed, 12 Oct 2016 11:03:56 +1000 Subject: [PATCH 3/7] pkcs10: remove pyasn1 PKCS #10 spec In the dogtag-ipa-ca-renew-agent-submit certmonger renewal helper, we currently use our hand-rolled PKCS #10 pyasn1 specification to parse the friendlyName out of CSRs generated by certmonger (it contains the NSSDB nickname of the cert). Use other information from the renewal helper process environment to determine the nickname and remove our PKCS #10 pyasn1 spec. Part of: https://fedorahosted.org/freeipa/ticket/6398 --- .../certmonger/dogtag-ipa-ca-renew-agent-submit | 54 +++++++++++++++------- ipalib/pkcs10.py | 50 -------------------- 2 files changed, 38 insertions(+), 66 deletions(-) diff --git a/install/certmonger/dogtag-ipa-ca-renew-agent-submit b/install/certmonger/dogtag-ipa-ca-renew-agent-submit index f2b15cf..916ac4d 100755 --- a/install/certmonger/dogtag-ipa-ca-renew-agent-submit +++ b/install/certmonger/dogtag-ipa-ca-renew-agent-submit @@ -39,10 +39,10 @@ import six from ipapython import ipautil from ipapython.dn import DN -from ipalib import api, errors, pkcs10, x509 +from ipalib import api, errors, x509 from ipaplatform.paths import paths from ipaserver.plugins.ldap2 import ldap2 -from ipaserver.install import cainstance, certs +from ipaserver.install import cainstance, dsinstance, certs # This is a certmonger CA helper script for IPA CA subsystem cert renewal. See # https://git.fedorahosted.org/cgit/certmonger.git/tree/doc/submit.txt for more @@ -65,8 +65,38 @@ if six.PY3: IPA_CA_NICKNAME = 'caSigningCert cert-pki-ca' def get_nickname(): - csr = os.environ.get('CERTMONGER_CSR') - return pkcs10.get_friendlyname(csr) if csr else None + subject = os.environ.get('CERTMONGER_REQ_SUBJECT') + if not subject: + return None + + subject_base = dsinstance.DsInstance().find_subject_base() + if not subject_base: + return None + + nickname_by_subject_dn = { + DN('CN=Certificate Authority', subject_base): + 'caSigningCert cert-pki-ca', + DN('CN=CA Audit', subject_base): 'auditSigningCert cert-pki-ca', + DN('CN=OCSP Subsystem', subject_base): 'ocspSigningCert cert-pki-ca', + DN('CN=CA Subsystem', subject_base): 'subsystemCert cert-pki-ca', + DN('CN=KRA Audit', subject_base): 'auditSigningCert cert-pki-kra', + DN('CN=KRA Transport Certificate', subject_base): + 'transportCert cert-pki-kra', + DN('CN=KRA Storage Certificate', subject_base): + 'storageCert cert-pki-kra', + DN('CN=IPA RA', subject_base): 'ipaCert', + } + + try: + return nickname_by_subject_dn[DN(subject)] + except KeyError: + if not api.Backend.ldap2.isconnected(): + api.Backend.ldap2.connect() + cas = api.Command.ca_find(ipacasubjectdn=DN(subject))['result'] + if len(cas) == 0: + return None + return 'caSigningCert cert-pki-ca {}'.format(cas[0]['ipacaid'][0]) + def is_lightweight_ca(): nickname = get_nickname() or '' @@ -216,13 +246,9 @@ def store_cert(): else: return (OPERATION_NOT_SUPPORTED_BY_HELPER,) - csr = os.environ.get('CERTMONGER_CSR') - if not csr: - return (UNCONFIGURED, "Certificate request not provided") - - nickname = pkcs10.get_friendlyname(csr) + nickname = get_nickname() if not nickname: - return (REJECTED, "No friendly name in the certificate request") + return (REJECTED, "Nickname could not be determined") cert = os.environ.get('CERTMONGER_CERTIFICATE') if not cert: @@ -325,13 +351,9 @@ def retrieve_or_reuse_cert(): Retrieve certificate from LDAP. If the certificate is not available, reuse the old certificate. """ - csr = os.environ.get('CERTMONGER_CSR') - if not csr: - return (UNCONFIGURED, "Certificate request not provided") - - nickname = pkcs10.get_friendlyname(csr) + nickname = get_nickname() if not nickname: - return (REJECTED, "No friendly name in the certificate request") + return (REJECTED, "Nickname could not be determined") cert = os.environ.get('CERTMONGER_CERTIFICATE') if not cert: diff --git a/ipalib/pkcs10.py b/ipalib/pkcs10.py index 44a529d..7145c13 100644 --- a/ipalib/pkcs10.py +++ b/ipalib/pkcs10.py @@ -23,57 +23,8 @@ import sys from cryptography.hazmat.backends import default_backend import cryptography.x509 -from pyasn1.type import univ, namedtype, tag -from pyasn1.codec.der import decoder -# Unfortunately, NSS can only parse the extension request attribute, so -# we have to parse friendly name ourselves (see RFC 2986) -class _Attribute(univ.Sequence): - componentType = namedtype.NamedTypes( - namedtype.NamedType('type', univ.ObjectIdentifier()), - namedtype.NamedType('values', univ.Set()), - ) - -class _Attributes(univ.SetOf): - componentType = _Attribute() - -class _CertificationRequestInfo(univ.Sequence): - componentType = namedtype.NamedTypes( - namedtype.NamedType('version', univ.Integer()), - namedtype.NamedType('subject', univ.Sequence()), - namedtype.NamedType('subjectPublicKeyInfo', univ.Sequence()), - namedtype.OptionalNamedType('attributes', _Attributes().subtype( - implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 0))) - ) - -class _CertificationRequest(univ.Sequence): - componentType = namedtype.NamedTypes( - namedtype.NamedType('certificationRequestInfo', - _CertificationRequestInfo()), - namedtype.NamedType('signatureAlgorithm', univ.Sequence()), - namedtype.NamedType('signatureValue', univ.BitString()), - ) - -_FRIENDLYNAME = univ.ObjectIdentifier('1.2.840.113549.1.9.20') - -def get_friendlyname(csr, datatype=PEM): - """ - Given a CSR return the value of the friendlyname attribute, if any. - - The return value is a string. - """ - if datatype == PEM: - csr = strip_header(csr) - csr = base64.b64decode(csr) - - csr = decoder.decode(csr, asn1Spec=_CertificationRequest())[0] - for attribute in csr['certificationRequestInfo']['attributes']: - if attribute['type'] == _FRIENDLYNAME: - return unicode(attribute['values'][0]) - - return None - def strip_header(csr): """ Remove the header and footer from a CSR. @@ -113,4 +64,3 @@ def load_certificate_request(data): csr = ''.join(csrlines) print(load_certificate_request(csr)) - print(get_friendlyname(csr)) From 8b73daa18e6946c0f1a8672b41494ee6d746238f Mon Sep 17 00:00:00 2001 From: Fraser Tweedale <ftwee...@redhat.com> Date: Tue, 11 Oct 2016 12:43:22 +1000 Subject: [PATCH 4/7] x509: avoid use of nss.data_to_hex Avoid use of the nss.data_to_hex function for formatting certificate fingerprints. Add our own helper functions to format the fingerprints as hex (with colons). Part of: https://fedorahosted.org/freeipa/ticket/6398 --- ipalib/x509.py | 23 +++++++++++++++++++++++ ipaserver/plugins/cert.py | 8 ++++---- ipaserver/plugins/service.py | 6 ++++-- 3 files changed, 31 insertions(+), 6 deletions(-) diff --git a/ipalib/x509.py b/ipalib/x509.py index e67aab6..cac5e9c 100644 --- a/ipalib/x509.py +++ b/ipalib/x509.py @@ -33,6 +33,7 @@ from __future__ import print_function +import binascii import collections import os import sys @@ -552,6 +553,28 @@ def process_othernames(gns): yield gn +def chunk(size, s): + """Yield chunks of the specified size from the given string. + + The input must be a multiple of the chunk size (otherwise + trailing characters are dropped). + + Works on character strings only. + + """ + return (u''.join(span) for span in six.moves.zip(*[iter(s)] * size)) + + +def add_colons(s): + """Add colons between each nibble pair in a hex string.""" + return u':'.join(chunk(2, s)) + + +def to_hex_with_colons(bs): + """Convert bytes to a hex string with colons.""" + return add_colons(binascii.hexlify(bs).decode('utf-8')) + + if __name__ == '__main__': # this can be run with: # python ipalib/x509.py < /etc/ipa/ca.crt diff --git a/ipaserver/plugins/cert.py b/ipaserver/plugins/cert.py index 5e85942..a534c4d 100644 --- a/ipaserver/plugins/cert.py +++ b/ipaserver/plugins/cert.py @@ -379,10 +379,10 @@ def _parse(self, obj, full=True): obj['valid_not_before'] = unicode(cert.valid_not_before_str) obj['valid_not_after'] = unicode(cert.valid_not_after_str) if full: - obj['md5_fingerprint'] = unicode( - nss.data_to_hex(nss.md5_digest(cert.der_data), 64)[0]) - obj['sha1_fingerprint'] = unicode( - nss.data_to_hex(nss.sha1_digest(cert.der_data), 64)[0]) + obj['md5_fingerprint'] = x509.to_hex_with_colons( + nss.md5_digest(cert.der_data)) + obj['sha1_fingerprint'] = x509.to_hex_with_colons( + nss.sha1_digest(cert.der_data)) try: ext_san = cert.get_extension(nss.SEC_OID_X509_SUBJECT_ALT_NAME) diff --git a/ipaserver/plugins/service.py b/ipaserver/plugins/service.py index e57ca52..a39ba32 100644 --- a/ipaserver/plugins/service.py +++ b/ipaserver/plugins/service.py @@ -274,8 +274,10 @@ def set_certificate_attrs(entry_attrs): entry_attrs['issuer'] = unicode(cert.issuer) entry_attrs['valid_not_before'] = unicode(cert.valid_not_before_str) entry_attrs['valid_not_after'] = unicode(cert.valid_not_after_str) - entry_attrs['md5_fingerprint'] = unicode(nss.data_to_hex(nss.md5_digest(cert.der_data), 64)[0]) - entry_attrs['sha1_fingerprint'] = unicode(nss.data_to_hex(nss.sha1_digest(cert.der_data), 64)[0]) + entry_attrs['md5_fingerprint'] = x509.to_hex_with_colons( + nss.md5_digest(cert.der_data)) + entry_attrs['sha1_fingerprint'] = x509.to_hex_with_colons( + nss.sha1_digest(cert.der_data)) def check_required_principal(ldap, principal): """ From a762a38e5ad949c8b2a8a68874502af726defc43 Mon Sep 17 00:00:00 2001 From: Fraser Tweedale <ftwee...@redhat.com> Date: Tue, 11 Oct 2016 17:02:58 +1000 Subject: [PATCH 5/7] x509: use pyasn1-modules X.509 specs Remove our hand-rolled pyasn1 specifications for X.509 in favour of those provided by the pyasn1-modules library. This also avoids a bug in our _Extension spec wherein parsing fails if the 'critical' field is absent. Part of: https://fedorahosted.org/freeipa/ticket/6398 --- ipalib/x509.py | 103 ++++----------------------------------------------------- 1 file changed, 7 insertions(+), 96 deletions(-) diff --git a/ipalib/x509.py b/ipalib/x509.py index cac5e9c..a807d12 100644 --- a/ipalib/x509.py +++ b/ipalib/x509.py @@ -45,6 +45,7 @@ from nss.error import NSPRError from pyasn1.type import univ, char, namedtype, tag from pyasn1.codec.der import decoder, encoder +from pyasn1_modules import rfc2459 import six from ipalib import api @@ -200,49 +201,11 @@ def is_self_signed(certificate, datatype=PEM, dbdir=None): del nsscert return self_signed -class _Name(univ.Choice): - componentType = namedtype.NamedTypes( - namedtype.NamedType('rdnSequence', - univ.SequenceOf()), - ) - -class _TBSCertificate(univ.Sequence): - componentType = namedtype.NamedTypes( - namedtype.NamedType( - 'version', - univ.Integer().subtype(explicitTag=tag.Tag( - tag.tagClassContext, tag.tagFormatSimple, 0))), - namedtype.NamedType('serialNumber', univ.Integer()), - namedtype.NamedType('signature', univ.Sequence()), - namedtype.NamedType('issuer', _Name()), - namedtype.NamedType('validity', univ.Sequence()), - namedtype.NamedType('subject', _Name()), - namedtype.NamedType('subjectPublicKeyInfo', univ.Sequence()), - namedtype.OptionalNamedType( - 'issuerUniquedID', - univ.BitString().subtype(implicitTag=tag.Tag( - tag.tagClassContext, tag.tagFormatSimple, 1))), - namedtype.OptionalNamedType( - 'subjectUniquedID', - univ.BitString().subtype(implicitTag=tag.Tag( - tag.tagClassContext, tag.tagFormatSimple, 2))), - namedtype.OptionalNamedType( - 'extensions', - univ.Sequence().subtype(explicitTag=tag.Tag( - tag.tagClassContext, tag.tagFormatSimple, 3))), - ) - -class _Certificate(univ.Sequence): - componentType = namedtype.NamedTypes( - namedtype.NamedType('tbsCertificate', _TBSCertificate()), - namedtype.NamedType('signatureAlgorithm', univ.Sequence()), - namedtype.NamedType('signature', univ.BitString()), - ) def _get_der_field(cert, datatype, dbdir, field): cert = load_certificate(cert, datatype, dbdir) cert = cert.der_data - cert = decoder.decode(cert, _Certificate())[0] + cert = decoder.decode(cert, rfc2459.Certificate())[0] field = cert['tbsCertificate'][field] field = encoder.encode(field) return field @@ -364,77 +327,24 @@ def write_certificate_list(rawcerts, filename): except (IOError, OSError) as e: raise errors.FileError(reason=str(e)) -class _Extension(univ.Sequence): - componentType = namedtype.NamedTypes( - namedtype.NamedType('extnID', univ.ObjectIdentifier()), - namedtype.NamedType('critical', univ.Boolean()), - namedtype.NamedType('extnValue', univ.OctetString()), - ) def _encode_extension(oid, critical, value): - ext = _Extension() + ext = rfc2459.Extension() ext['extnID'] = univ.ObjectIdentifier(oid) ext['critical'] = univ.Boolean(critical) - ext['extnValue'] = univ.OctetString(value) + ext['extnValue'] = univ.Any(encoder.encode(univ.OctetString(value))) ext = encoder.encode(ext) return ext -class _ExtKeyUsageSyntax(univ.SequenceOf): - componentType = univ.ObjectIdentifier() def encode_ext_key_usage(ext_key_usage): - eku = _ExtKeyUsageSyntax() + eku = rfc2459.ExtKeyUsageSyntax() for i, oid in enumerate(ext_key_usage): eku[i] = univ.ObjectIdentifier(oid) eku = encoder.encode(eku) return _encode_extension('2.5.29.37', EKU_ANY not in ext_key_usage, eku) -class _AnotherName(univ.Sequence): - componentType = namedtype.NamedTypes( - namedtype.NamedType('type-id', univ.ObjectIdentifier()), - namedtype.NamedType('value', univ.Any().subtype( - explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 0)) - ), - ) - - -class _GeneralName(univ.Choice): - componentType = namedtype.NamedTypes( - namedtype.NamedType('otherName', _AnotherName().subtype( - implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 0)) - ), - namedtype.NamedType('rfc822Name', char.IA5String().subtype( - implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 1)) - ), - namedtype.NamedType('dNSName', char.IA5String().subtype( - implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 2)) - ), - namedtype.NamedType('x400Address', univ.Sequence().subtype( - implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 3)) - ), - namedtype.NamedType('directoryName', _Name().subtype( - implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 4)) - ), - namedtype.NamedType('ediPartyName', univ.Sequence().subtype( - implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 5)) - ), - namedtype.NamedType('uniformResourceIdentifier', char.IA5String().subtype( - implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 6)) - ), - namedtype.NamedType('iPAddress', univ.OctetString().subtype( - implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 7)) - ), - namedtype.NamedType('registeredID', univ.ObjectIdentifier().subtype( - implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 8)) - ), - ) - - -class _SubjectAltName(univ.SequenceOf): - componentType = _GeneralName() - - class _PrincipalName(univ.Sequence): componentType = namedtype.NamedTypes( namedtype.NamedType('name-type', univ.Integer().subtype( @@ -488,7 +398,8 @@ def decode_generalnames(secitem): """ nss_names = nss.x509_alt_name(secitem, repr_kind=nss.AsObject) - asn1_names = decoder.decode(secitem.data, asn1Spec=_SubjectAltName())[0] + asn1_names = decoder.decode( + secitem.data, asn1Spec=rfc2459.SubjectAltName())[0] names = [] for nss_name, asn1_name in zip(nss_names, asn1_names): # NOTE: we use the NSS enum to identify the name type. From 888303e36a73602974b97e43cda5674214276e88 Mon Sep 17 00:00:00 2001 From: Fraser Tweedale <ftwee...@redhat.com> Date: Thu, 13 Oct 2016 17:12:31 +1000 Subject: [PATCH 6/7] x509: use python-cryptography to process certs Update x509.load_certificate and related functions to return python-cryptography ``Certificate`` objects. Update the call sites accordingly, including removal of NSS initialisation code. Also update GeneralName parsing code to return python-cryptography GeneralName values, for consistency with other code that processes GeneralNames. The new function, `get_san_general_names`, and associated helper functions, can be removed when python-cryptography provides a way to deal with unrecognised critical extensions. Part of: https://fedorahosted.org/freeipa/ticket/6398 --- client/ipa-client-install | 27 +- install/tools/ipa-replica-conncheck | 15 +- ipalib/certstore.py | 12 +- ipalib/x509.py | 327 ++++++++++++------------ ipapython/certdb.py | 17 +- ipaserver/install/ca.py | 2 +- ipaserver/install/cainstance.py | 25 +- ipaserver/install/certs.py | 9 +- ipaserver/install/installutils.py | 14 +- ipaserver/install/ipa_cacert_manage.py | 103 ++++---- ipaserver/install/krainstance.py | 2 +- ipaserver/plugins/cert.py | 115 +++++---- ipaserver/plugins/service.py | 20 +- ipatests/test_ipalib/test_x509.py | 66 ++--- ipatests/test_ipaserver/test_ldap.py | 8 +- ipatests/test_ipaserver/test_otptoken_import.py | 4 - 16 files changed, 369 insertions(+), 397 deletions(-) diff --git a/client/ipa-client-install b/client/ipa-client-install index 639810b..c228ea3 100755 --- a/client/ipa-client-install +++ b/client/ipa-client-install @@ -35,9 +35,9 @@ try: import gssapi import netifaces - import nss.nss as nss import SSSDConfig from six.moves.urllib.parse import urlparse, urlunparse + from cryptography.hazmat.primitives import serialization from ipapython.ipa_log_manager import standard_logging_setup, root_logger from ipaclient import ipadiscovery @@ -92,15 +92,10 @@ def parse_options(): if not os.path.isabs(value): raise OptionValueError("%s option '%s' is not an absolute file path" % (opt, value)) - initialized = nss.nss_is_initialized() try: - cert = x509.load_certificate_from_file(value) + x509.load_certificate_from_file(value) except Exception: raise OptionValueError("%s option '%s' is not a valid certificate file" % (opt, value)) - else: - del(cert) - if not initialized: - nss.nss_shutdown() parser.values.ca_cert_file = value @@ -300,10 +295,10 @@ def cert_summary(msg, certs, indent=' '): else: s = '' for cert in certs: - s += '%sSubject: %s\n' % (indent, cert.subject) - s += '%sIssuer: %s\n' % (indent, cert.issuer) - s += '%sValid From: %s\n' % (indent, cert.valid_not_before_str) - s += '%sValid Until: %s\n' % (indent, cert.valid_not_after_str) + s += '%sSubject: %s\n' % (indent, DN(cert.subject)) + s += '%sIssuer: %s\n' % (indent, DN(cert.issuer)) + s += '%sValid From: %s\n' % (indent, cert.not_valid_before) + s += '%sValid Until: %s\n' % (indent, cert.not_valid_after) s += '\n' s = s[:-1] @@ -2148,7 +2143,10 @@ def get_ca_certs(fstore, options, server, basedn, realm): if ca_certs is not None: try: - ca_certs = [cert.der_data for cert in ca_certs] + ca_certs = [ + cert.public_bytes(serialization.Encoding.DER) + for cert in ca_certs + ] x509.write_certificate_list(ca_certs, ca_file) except Exception as e: if os.path.exists(ca_file): @@ -2815,7 +2813,10 @@ def install(options, env, fstore, statestore): # Add CA certs to a temporary NSS database ca_certs = x509.load_certificate_list_from_file(CACERT) - ca_certs = [cert.der_data for cert in ca_certs] + ca_certs = [ + cert.public_bytes(serialization.Encoding.DER) + for cert in ca_certs + ] try: pwd_file = ipautil.write_tmp_file(ipautil.ipa_generate_password()) tmp_db.create_db(pwd_file.name) diff --git a/install/tools/ipa-replica-conncheck b/install/tools/ipa-replica-conncheck index 067afb7..4045e41 100755 --- a/install/tools/ipa-replica-conncheck +++ b/install/tools/ipa-replica-conncheck @@ -21,6 +21,7 @@ from __future__ import print_function from ipapython.config import IPAOptionParser +from ipapython.dn import DN from ipapython import version from ipapython import ipautil, certdb from ipalib import api, errors, x509 @@ -40,7 +41,7 @@ from socket import SOCK_STREAM, SOCK_DGRAM import distutils.spawn from ipaplatform.paths import paths import gssapi -from nss import nss +from cryptography.hazmat.primitives import serialization CONNECT_TIMEOUT = 5 RESPONDERS = [ ] @@ -121,16 +122,12 @@ def parse_options(): raise OptionValueError( "%s option '%s' is not an absolute file path" % (opt, value)) - initialized = nss.nss_is_initialized() try: x509.load_certificate_list_from_file(value) except Exception: raise OptionValueError( "%s option '%s' is not a valid certificate file" % (opt, value)) - finally: - if not initialized: - nss.nss_shutdown() parser.values.ca_cert_file = value @@ -472,12 +469,12 @@ def main(): nss_db.create_db(password_file.name) ca_certs = x509.load_certificate_list_from_file( - options.ca_cert_file, dbdir=nss_db.secdir) + options.ca_cert_file) for ca_cert in ca_certs: + data = ca_cert.public_bytes( + serialization.Encoding.DER) nss_db.add_cert( - ca_cert.der_data, str(ca_cert.subject), 'C,,') - del ca_cert - del ca_certs + data, str(DN(ca_cert.subject)), 'C,,') else: nss_dir = None diff --git a/ipalib/certstore.py b/ipalib/certstore.py index d17cb2b..70ae942 100644 --- a/ipalib/certstore.py +++ b/ipalib/certstore.py @@ -22,7 +22,6 @@ LDAP shared certificate store. """ -from nss.error import NSPRError from pyasn1.error import PyAsn1Error from ipapython.dn import DN @@ -31,11 +30,12 @@ def _parse_cert(dercert): try: - subject = x509.get_subject(dercert, x509.DER) - issuer = x509.get_issuer(dercert, x509.DER) - serial_number = x509.get_serial_number(dercert, x509.DER) + cert = x509.load_certificate(dercert, x509.DER) + subject = DN(cert.subject) + issuer = DN(cert.issuer) + serial_number = cert.serial public_key_info = x509.get_der_public_key_info(dercert, x509.DER) - except (NSPRError, PyAsn1Error) as e: + except (ValueError, PyAsn1Error) as e: raise ValueError("failed to decode certificate: %s" % e) subject = str(subject).replace('\\;', '\\3b') @@ -54,7 +54,7 @@ def init_ca_entry(entry, dercert, nickname, trusted, ext_key_usage): if ext_key_usage is not None: try: cert_eku = x509.get_ext_key_usage(dercert, x509.DER) - except NSPRError as e: + except ValueError as e: raise ValueError("failed to decode certificate: %s" % e) if cert_eku is not None: cert_eku -= {x509.EKU_SERVER_AUTH, x509.EKU_CLIENT_AUTH, diff --git a/ipalib/x509.py b/ipalib/x509.py index a807d12..7f7a89c 100644 --- a/ipalib/x509.py +++ b/ipalib/x509.py @@ -28,31 +28,27 @@ # # cert: the certificate is a PEM-encoded certificate # dercert: the certificate is DER-encoded -# nsscert: the certificate is an NSS Certificate object # rawcert: the cert is in an unknown format from __future__ import print_function import binascii -import collections -import os +import datetime +import ipaddress import sys import base64 import re +from cryptography.hazmat.backends import default_backend import cryptography.x509 -import nss.nss as nss -from nss.error import NSPRError from pyasn1.type import univ, char, namedtype, tag from pyasn1.codec.der import decoder, encoder from pyasn1_modules import rfc2459 import six from ipalib import api -from ipalib import _ from ipalib import util from ipalib import errors -from ipaplatform.paths import paths from ipapython.dn import DN if six.PY3: @@ -95,32 +91,16 @@ def strip_header(pem): return pem -def initialize_nss_database(dbdir=None): - """ - Initializes NSS database, if not initialized yet. Uses a proper database - directory (.ipa/alias or HTTPD_ALIAS_DIR), depending on the value of - api.env.in_tree. - """ - if not nss.nss_is_initialized(): - if dbdir is None: - if 'in_tree' in api.env: - if api.env.in_tree: - dbdir = api.env.dot_ipa + os.sep + 'alias' - else: - dbdir = paths.HTTPD_ALIAS_DIR - nss.nss_init(dbdir) - else: - nss.nss_init_nodb() - else: - nss.nss_init(dbdir) - -def load_certificate(data, datatype=PEM, dbdir=None): +def load_certificate(data, datatype=PEM): """ - Given a base64-encoded certificate, with or without the - header/footer, return a request object. + Load an X.509 certificate. + + :param datatype: PEM for base64-encoded data (with or without header), + or DER + :return: a python-cryptography ``CertificateSigningRequest`` object. + :raises: ``ValueError`` if unable to load the certificate. - Returns a nss.Certificate type """ if type(data) in (tuple, list): data = data[0] @@ -129,82 +109,50 @@ def load_certificate(data, datatype=PEM, dbdir=None): data = strip_header(data) data = base64.b64decode(data) - initialize_nss_database(dbdir=dbdir) + return cryptography.x509.load_der_x509_certificate(data, default_backend()) - if six.PY2: - return nss.Certificate(buffer(data)) # pylint: disable=buffer-builtin - else: - # In python 3 , `bytes` has the buffer interface - return nss.Certificate(data) def load_certificate_from_file(filename, dbdir=None): """ Load a certificate from a PEM file. - Returns a nss.Certificate type - """ - fd = open(filename, 'r') - data = fd.read() - fd.close() - - return load_certificate(data, PEM, dbdir) - -def load_certificate_list(data, dbdir=None): - certs = PEM_REGEX.findall(data) - certs = [load_certificate(cert, PEM, dbdir) for cert in certs] - return certs + Returns a python-cryptography ``Certificate`` object. -def load_certificate_list_from_file(filename, dbdir=None): """ - Load a certificate list from a PEM file. + with open(filename, mode='rb') as f: + return load_certificate(f.read(), PEM) + - Returns a list of nss.Certificate objects. +def load_certificate_list(data): """ - fd = open(filename, 'r') - data = fd.read() - fd.close() + Load a certificate list from a sequence of concatenated PEMs. - return load_certificate_list(data, dbdir) + Return a list of python-cryptography ``Certificate`` objects. -def get_subject(certificate, datatype=PEM, dbdir=None): - """ - Load an X509.3 certificate and get the subject. """ + certs = PEM_REGEX.findall(data) + certs = [load_certificate(cert, PEM) for cert in certs] + return certs - nsscert = load_certificate(certificate, datatype, dbdir) - subject = nsscert.subject - del(nsscert) - return subject -def get_issuer(certificate, datatype=PEM, dbdir=None): - """ - Load an X509.3 certificate and get the issuer. +def load_certificate_list_from_file(filename): """ + Load a certificate list from a PEM file. - nsscert = load_certificate(certificate, datatype, dbdir) - issuer = nsscert.issuer - del(nsscert) - return issuer + Return a list of python-cryptography ``Certificate`` objects. -def get_serial_number(certificate, datatype=PEM, dbdir=None): """ - Return the decimal value of the serial number. - """ - nsscert = load_certificate(certificate, datatype, dbdir) - serial_number = nsscert.serial_number - del(nsscert) - return serial_number + with open(filename) as f: + return load_certificate_list(f.read()) + -def is_self_signed(certificate, datatype=PEM, dbdir=None): - nsscert = load_certificate(certificate, datatype, dbdir) - self_signed = (nsscert.issuer == nsscert.subject) - del nsscert - return self_signed +def is_self_signed(certificate, datatype=PEM): + cert = load_certificate(certificate, datatype) + return cert.issuer == cert.subject def _get_der_field(cert, datatype, dbdir, field): - cert = load_certificate(cert, datatype, dbdir) - cert = cert.der_data + cert = normalize_certificate(cert) cert = decoder.decode(cert, rfc2459.Certificate())[0] field = cert['tbsCertificate'][field] field = encoder.encode(field) @@ -222,20 +170,17 @@ def get_der_serial_number(cert, datatype=PEM, dbdir=None): def get_der_public_key_info(cert, datatype=PEM, dbdir=None): return _get_der_field(cert, datatype, dbdir, 'subjectPublicKeyInfo') -def get_ext_key_usage(certificate, datatype=PEM, dbdir=None): - nsscert = load_certificate(certificate, datatype, dbdir) - if not nsscert.extensions: - return None - for ext in nsscert.extensions: - if ext.oid_tag == nss.SEC_OID_X509_EXT_KEY_USAGE: - break - else: +def get_ext_key_usage(certificate, datatype=PEM): + cert = load_certificate(certificate, datatype) + try: + eku = cert.extensions.get_extension_for_oid( + cryptography.x509.oid.ExtensionOID.EXTENDED_KEY_USAGE).value + except cryptography.x509.ExtensionNotFound: return None - eku = nss.x509_ext_key_usage(ext.value, nss.AsDottedDecimal) - eku = set(o[4:] for o in eku) - return eku + return set(oid.dotted_string for oid in eku) + def make_pem(data): """ @@ -270,27 +215,21 @@ def normalize_certificate(rawcert): else: dercert = rawcert - # At this point we should have a certificate, either because the data - # was base64-encoded and now its not or it came in as DER format. - # Let's decode it and see. Fetching the serial number will pass the - # certificate through the NSS DER parser. + # At this point we should have a DER certificate. + # Attempt to decode it. validate_certificate(dercert, datatype=DER) return dercert -def validate_certificate(cert, datatype=PEM, dbdir=None): +def validate_certificate(cert, datatype=PEM): """ - Perform certificate validation by trying to load it into NSS database + Perform cert validation by trying to load it via python-cryptography. """ try: - load_certificate(cert, datatype=datatype, dbdir=dbdir) - except NSPRError as nsprerr: - if nsprerr.errno == -8183: # SEC_ERROR_BAD_DER - raise errors.CertificateFormatError( - error=_('improperly formatted DER-encoded certificate')) - else: - raise errors.CertificateFormatError(error=str(nsprerr)) + load_certificate(cert, datatype=datatype) + except ValueError as e: + raise errors.CertificateFormatError(error=str(e)) def write_certificate(rawcert, filename): @@ -379,56 +318,6 @@ def _decode_krb5principalname(data): return name -GeneralNameInfo = collections.namedtuple( - 'GeneralNameInfo', ('type', 'desc', 'value', 'der_value')) - - -def decode_generalnames(secitem): - """ - Decode a GeneralNames object (this the data for the Subject - Alt Name and Issuer Alt Name extensions, among others). - - ``secitem`` - The input is the DER-encoded extension data, without the - OCTET STRING header, as an nss SecItem object. - - Return a list of ``GeneralNameInfo`` namedtuples. The - ``der_value`` field is set for otherNames, otherwise it is - ``None``. - - """ - nss_names = nss.x509_alt_name(secitem, repr_kind=nss.AsObject) - asn1_names = decoder.decode( - secitem.data, asn1Spec=rfc2459.SubjectAltName())[0] - names = [] - for nss_name, asn1_name in zip(nss_names, asn1_names): - # NOTE: we use the NSS enum to identify the name type. - # (For otherName we also tuple it up with the type-id OID). - # The enum does not correspond exactly to the ASN.1 tags. - # If we ever want to switch to using the true tag numbers, - # the expression to get the tag is: - # - # asn1_name.getComponent().getTagSet()[0].asTuple()[2] - # - if nss_name.type_enum == nss.certOtherName: - oid = str(asn1_name['otherName']['type-id']) - nametype = (nss_name.type_enum, oid) - der_value = asn1_name['otherName']['value'].asOctets() - else: - nametype = nss_name.type_enum - der_value = None - - if nametype == (nss.certOtherName, SAN_KRB5PRINCIPALNAME): - name = _decode_krb5principalname(asn1_name['otherName']['value']) - else: - name = nss_name.name - - gni = GeneralNameInfo(nametype, nss_name.type_string, name, der_value) - names.append(gni) - - return names - - class KRB5PrincipalName(cryptography.x509.general_name.OtherName): def __init__(self, type_id, value): super(KRB5PrincipalName, self).__init__(type_id, value) @@ -464,6 +353,100 @@ def process_othernames(gns): yield gn +def get_san_general_names(cert): + """ + Return SAN general names from a python-cryptography + certificate object. If the SAN extension is not present, + return an empty sequence. + + Because python-cryptography does not yet provide a way to + handle unrecognised critical extensions (which may occur), + we must parse the certificate and extract the General Names. + For uniformity with other code, we manually construct values + of python-crytography GeneralName subtypes. + + python-cryptography does not yet provide types for + ediPartyName or x400Address, so we drop these name types. + + otherNames are NOT instantiated to more specific types where + the type is known. Use ``process_othernames`` to do that. + + When python-cryptography can handle certs with unrecognised + critical extensions and implements ediPartyName and + x400Address, this function (and helpers) will be redundant + and should go away. + + """ + tbs = decoder.decode( + cert.tbs_certificate_bytes, + asn1Spec=rfc2459.TBSCertificate() + )[0] + OID_SAN = univ.ObjectIdentifier('2.5.29.17') + gns = [] + for ext in tbs['extensions']: + if ext['extnID'] == OID_SAN: + der = decoder.decode( + ext['extnValue'], asn1Spec=univ.OctetString())[0] + gns = decoder.decode(der, asn1Spec=rfc2459.SubjectAltName())[0] + break + + GENERAL_NAME_CONSTRUCTORS = { + 'rfc822Name': lambda x: cryptography.x509.RFC822Name(unicode(x)), + 'dNSName': lambda x: cryptography.x509.DNSName(unicode(x)), + 'directoryName': _pyasn1_to_cryptography_directoryname, + 'registeredID': _pyasn1_to_cryptography_registeredid, + 'iPAddress': _pyasn1_to_cryptography_ipaddress, + 'uniformResourceIdentifier': + lambda x: cryptography.x509.UniformResourceIdentifier(unicode(x)), + 'otherName': _pyasn1_to_cryptography_othername, + } + + result = [] + + for gn in gns: + gn_type = gn.getName() + if gn_type in GENERAL_NAME_CONSTRUCTORS: + result.append( + GENERAL_NAME_CONSTRUCTORS[gn_type](gn.getComponent())) + + return result + + +def _pyasn1_to_cryptography_directoryname(dn): + attrs = [] + + # Name is CHOICE { RDNSequence } (only one possibility) + for rdn in dn.getComponent(): + for ava in rdn: + attr = cryptography.x509.NameAttribute( + _pyasn1_to_cryptography_oid(ava['type']), + unicode(decoder.decode(ava['value'])[0]) + ) + attrs.append(attr) + + return cryptography.x509.DirectoryName(cryptography.x509.Name(attrs)) + + +def _pyasn1_to_cryptography_registeredid(oid): + return cryptography.x509.RegisteredID(_pyasn1_to_cryptography_oid(oid)) + + +def _pyasn1_to_cryptography_ipaddress(octet_string): + return cryptography.x509.IPAddress( + ipaddress.ip_address(bytes(octet_string))) + + +def _pyasn1_to_cryptography_othername(on): + return cryptography.x509.OtherName( + _pyasn1_to_cryptography_oid(on['type-id']), + bytes(on['value']) + ) + + +def _pyasn1_to_cryptography_oid(oid): + return cryptography.x509.ObjectIdentifier(str(oid)) + + def chunk(size, s): """Yield chunks of the specified size from the given string. @@ -486,20 +469,34 @@ def to_hex_with_colons(bs): return add_colons(binascii.hexlify(bs).decode('utf-8')) +class UTC(datetime.tzinfo): + ZERO = datetime.timedelta(0) + + def tzname(self, dt): + return "UTC" + + def utcoffset(self, dt): + return self.ZERO + + def dst(self, dt): + return self.ZERO + + +def format_datetime(t): + if t.tzinfo is None: + t = t.replace(tzinfo=UTC()) + return unicode(t.strftime("%a %b %d %H:%M:%S %Y %Z")) + + if __name__ == '__main__': # this can be run with: # python ipalib/x509.py < /etc/ipa/ca.crt - api.bootstrap() - api.finalize() - - nss.nss_init_nodb() - - # Read PEM certs from stdin and print out its components + # Read PEM cert from stdin and print out its components certlines = sys.stdin.readlines() cert = ''.join(certlines) - nsscert = load_certificate(cert) + cert = load_certificate(cert) - print(nsscert) + print(cert) diff --git a/ipapython/certdb.py b/ipapython/certdb.py index 06666c0..c2fe599 100644 --- a/ipapython/certdb.py +++ b/ipapython/certdb.py @@ -22,10 +22,12 @@ import tempfile import shutil import base64 +from cryptography.hazmat.primitives import serialization from nss import nss from nss.error import NSPRError from ipaplatform.paths import paths +from ipapython.dn import DN from ipapython.ipa_log_manager import root_logger from ipapython import ipautil from ipalib import x509 @@ -258,7 +260,7 @@ def import_files(self, files, db_password_filename, import_keys=False, 'X.509 CERTIFICATE'): try: x509.load_certificate(match.group(2)) - except NSPRError as e: + except ValueError as e: if label != 'CERTIFICATE': root_logger.warning( "Skipping certificate in %s at line %s: %s", @@ -334,7 +336,7 @@ def import_files(self, files, db_password_filename, import_keys=False, # Try to load the file as DER certificate try: x509.load_certificate(data, x509.DER) - except NSPRError: + except ValueError: pass else: data = x509.make_pem(base64.b64encode(data)) @@ -379,12 +381,11 @@ def import_files(self, files, db_password_filename, import_keys=False, raise RuntimeError( "No server certificates found in %s" % (', '.join(files))) - nss_certs = x509.load_certificate_list(extracted_certs) - nss_cert = None - for nss_cert in nss_certs: - nickname = str(nss_cert.subject) - self.add_cert(nss_cert.der_data, nickname, ',,') - del nss_certs, nss_cert + certs = x509.load_certificate_list(extracted_certs) + for cert in certs: + nickname = str(DN(cert.subject)) + data = cert.public_bytes(serialization.Encoding.DER) + self.add_cert(data, nickname, ',,') if extracted_key: in_file = ipautil.write_tmp_file(extracted_certs + extracted_key) diff --git a/ipaserver/install/ca.py b/ipaserver/install/ca.py index 88ec627..921e494 100644 --- a/ipaserver/install/ca.py +++ b/ipaserver/install/ca.py @@ -102,7 +102,7 @@ def install_check(standalone, replica_config, options): cert = db.get_cert_from_db(nickname) if not cert: continue - subject = DN(str(x509.get_subject(cert))) + subject = DN(x509.load_certificate(cert).subject) if subject in (DN('CN=Certificate Authority', subject_base), DN('CN=IPA RA', subject_base)): raise ScriptError( diff --git a/ipaserver/install/cainstance.py b/ipaserver/install/cainstance.py index 1b7acef..7b26e74 100644 --- a/ipaserver/install/cainstance.py +++ b/ipaserver/install/cainstance.py @@ -554,9 +554,9 @@ def __spawn_instance(self): config.set("CA", "pki_req_ext_data", "1E0A00530075006200430041") elif self.external == 2: - cert = x509.load_certificate_from_file(self.cert_file) cert_file = tempfile.NamedTemporaryFile() - x509.write_certificate(cert.der_data, cert_file.name) + with open(self.cert_file) as f: + x509.write_certificate(f.read(), cert_file.name) cert_file.flush() result = ipautil.run( @@ -778,7 +778,7 @@ def __create_ca_agent(self): userstate=["1"], userCertificate=[cert_data], description=['2;%s;%s;%s' % ( - cert.serial_number, + cert.serial, DN(('CN', 'Certificate Authority'), self.subject_base), DN(('CN', 'IPA RA'), self.subject_base))]) conn.add_entry(entry) @@ -1674,8 +1674,9 @@ def update_people_entry(dercert): is needed when a certificate is renewed. """ def make_filter(dercert): - subject = x509.get_subject(dercert, datatype=x509.DER) - issuer = x509.get_issuer(dercert, datatype=x509.DER) + cert = x509.load_certificate(dercert, datatype=x509.DER) + subject = DN(cert.subject) + issuer = DN(cert.issuer) return ldap2.ldap2.combine_filters( [ ldap2.ldap2.make_filter({'objectClass': 'inetOrgPerson'}), @@ -1686,9 +1687,10 @@ def make_filter(dercert): ldap2.ldap2.MATCH_ALL) def make_entry(dercert, entry): - serial_number = x509.get_serial_number(dercert, datatype=x509.DER) - subject = x509.get_subject(dercert, datatype=x509.DER) - issuer = x509.get_issuer(dercert, datatype=x509.DER) + cert = x509.load_certificate(dercert, datatype=x509.DER) + serial_number = cert.serial + subject = DN(cert.subject) + issuer = DN(cert.issuer) entry['usercertificate'].append(dercert) entry['description'] = '2;%d;%s;%s' % (serial_number, issuer, subject) return entry @@ -1702,15 +1704,16 @@ def update_authority_entry(dercert): serial number to match the given cert. """ def make_filter(dercert): - subject = x509.get_subject(dercert, datatype=x509.DER) + cert = x509.load_certificate(dercert, datatype=x509.DER) + subject = str(DN(cert.subject)) return ldap2.ldap2.make_filter( dict(objectclass='authority', authoritydn=subject), rules=ldap2.ldap2.MATCH_ALL, ) def make_entry(dercert, entry): - serial_number = x509.get_serial_number(dercert, datatype=x509.DER) - entry['authoritySerial'] = serial_number + cert = x509.load_certificate(dercert, datatype=x509.DER) + entry['authoritySerial'] = cert.serial return entry return __update_entry_from_cert(make_filter, make_entry, dercert) diff --git a/ipaserver/install/certs.py b/ipaserver/install/certs.py index 31fd36c..a730250 100644 --- a/ipaserver/install/certs.py +++ b/ipaserver/install/certs.py @@ -60,9 +60,8 @@ def get_cert_nickname(cert): representation of the first RDN in the subject and subject_dn is a DN object. """ - nsscert = x509.load_certificate(cert) - subject = str(nsscert.subject) - dn = DN(subject) + cert_obj = x509.load_certificate(cert) + dn = DN(cert_obj.subject) return (str(dn[0]), dn) @@ -304,8 +303,8 @@ def track_server_cert(self, nickname, principal, password_file=None, command=Non return cert = self.get_cert_from_db(nickname) - nsscert = x509.load_certificate(cert, dbdir=self.secdir) - subject = str(nsscert.subject) + cert_obj = x509.load_certificate(cert) + subject = str(DN(cert_obj.subject)) certmonger.add_principal(request_id, principal) certmonger.add_subject(request_id, subject) diff --git a/ipaserver/install/installutils.py b/ipaserver/install/installutils.py index fb9579a..bee501a 100644 --- a/ipaserver/install/installutils.py +++ b/ipaserver/install/installutils.py @@ -921,10 +921,9 @@ def load_pkcs12(cert_files, key_password, key_nickname, ca_cert_files, if ca_cert is None: ca_cert = cert - nss_cert = x509.load_certificate(cert, x509.DER) - subject = DN(str(nss_cert.subject)) - issuer = DN(str(nss_cert.issuer)) - del nss_cert + cert_obj = x509.load_certificate(cert, x509.DER) + subject = DN(cert_obj.subject) + issuer = DN(cert_obj.issuer) if subject == issuer: break @@ -1046,10 +1045,9 @@ def load_external_cert(files, subject_base): for nickname, _trust_flags in nssdb.list_certs(): cert = nssdb.get_cert(nickname, pem=True) - nss_cert = x509.load_certificate(cert) - subject = DN(str(nss_cert.subject)) - issuer = DN(str(nss_cert.issuer)) - del nss_cert + cert_obj = x509.load_certificate(cert) + subject = DN(cert_obj.subject) + issuer = DN(cert_obj.issuer) cache[nickname] = (cert, subject, issuer) if subject == ca_subject: diff --git a/ipaserver/install/ipa_cacert_manage.py b/ipaserver/install/ipa_cacert_manage.py index af08ba6..0dcb70f 100644 --- a/ipaserver/install/ipa_cacert_manage.py +++ b/ipaserver/install/ipa_cacert_manage.py @@ -21,8 +21,7 @@ import os from optparse import OptionGroup -from nss import nss -from nss.error import NSPRError +from cryptography.hazmat.primitives import serialization import gssapi from ipapython import admintool, certmonger, ipautil @@ -187,7 +186,7 @@ def renew_external_step_1(self, ca): "--external-cert-file=/path/to/signed_certificate " "--external-cert-file=/path/to/external_ca_certificate") - def renew_external_step_2(self, ca, old_cert): + def renew_external_step_2(self, ca, old_cert_der): print("Importing the renewed CA certificate, please wait") options = self.options @@ -195,55 +194,54 @@ def renew_external_step_2(self, ca, old_cert): cert_file, ca_file = installutils.load_external_cert( options.external_cert_files, x509.subject_base()) - nss_cert = None - nss.nss_init(paths.PKI_TOMCAT_ALIAS_DIR) - try: - nss_cert = x509.load_certificate(old_cert, x509.DER) - subject = nss_cert.subject - der_subject = x509.get_der_subject(old_cert, x509.DER) - #pylint: disable=E1101 - pkinfo = nss_cert.subject_public_key_info.format() - #pylint: enable=E1101 - - nss_cert = x509.load_certificate_from_file(cert_file.name) - cert = nss_cert.der_data - if nss_cert.subject != subject: - raise admintool.ScriptError( - "Subject name mismatch (visit " - "http://www.freeipa.org/page/Troubleshooting for " - "troubleshooting guide)") - if x509.get_der_subject(cert, x509.DER) != der_subject: - raise admintool.ScriptError( - "Subject name encoding mismatch (visit " - "http://www.freeipa.org/page/Troubleshooting for " - "troubleshooting guide)") - #pylint: disable=E1101 - if nss_cert.subject_public_key_info.format() != pkinfo: - raise admintool.ScriptError( - "Subject public key info mismatch (visit " - "http://www.freeipa.org/page/Troubleshooting for " - "troubleshooting guide)") - #pylint: enable=E1101 - finally: - del nss_cert - nss.nss_shutdown() + old_cert_obj = x509.load_certificate(old_cert_der, x509.DER) + old_der_subject = x509.get_der_subject(old_cert_der, x509.DER) + old_spki = old_cert_obj.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo + ) + + with open(cert_file.name) as f: + new_cert_data = f.read() + new_cert_der = x509.normalize_certificate(new_cert_data) + new_cert_obj = x509.load_certificate(new_cert_der, x509.DER) + new_der_subject = x509.get_der_subject(new_cert_der, x509.DER) + new_spki = new_cert_obj.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo + ) + + if new_cert_obj.subject != old_cert_obj.subject: + raise admintool.ScriptError( + "Subject name mismatch (visit " + "http://www.freeipa.org/page/Troubleshooting for " + "troubleshooting guide)") + if new_der_subject != old_der_subject: + raise admintool.ScriptError( + "Subject name encoding mismatch (visit " + "http://www.freeipa.org/page/Troubleshooting for " + "troubleshooting guide)") + if new_spki != old_spki: + raise admintool.ScriptError( + "Subject public key info mismatch (visit " + "http://www.freeipa.org/page/Troubleshooting for " + "troubleshooting guide)") with certs.NSSDatabase() as tmpdb: pw = ipautil.write_tmp_file(ipautil.ipa_generate_password()) tmpdb.create_db(pw.name) - tmpdb.add_cert(old_cert, 'IPA CA', 'C,,') + tmpdb.add_cert(old_cert_der, 'IPA CA', 'C,,') try: - tmpdb.add_cert(cert, 'IPA CA', 'C,,') + tmpdb.add_cert(new_cert_der, 'IPA CA', 'C,,') except ipautil.CalledProcessError as e: raise admintool.ScriptError( "Not compatible with the current CA certificate: %s" % e) ca_certs = x509.load_certificate_list_from_file(ca_file.name) for ca_cert in ca_certs: - tmpdb.add_cert(ca_cert.der_data, str(ca_cert.subject), 'C,,') - del ca_certs - del ca_cert + data = ca_cert.public_bytes(serialization.Encoding.DER) + tmpdb.add_cert(data, str(DN(ca_cert.subject)), 'C,,') try: tmpdb.verify_ca_cert_validity('IPA CA') @@ -266,14 +264,14 @@ def renew_external_step_2(self, ca, old_cert): ('cn', 'ipa'), ('cn', 'etc'), api.env.basedn) try: entry = conn.get_entry(dn, ['usercertificate']) - entry['usercertificate'] = [cert] + entry['usercertificate'] = [new_cert_der] conn.update_entry(entry) except errors.NotFound: entry = conn.make_entry( dn, objectclass=['top', 'pkiuser', 'nscontainer'], cn=[self.cert_nickname], - usercertificate=[cert]) + usercertificate=[new_cert_der]) conn.add_entry(entry) except errors.EmptyModlist: pass @@ -313,21 +311,16 @@ def install(self): options = self.options cert_filename = self.args[1] - nss_cert = None try: - try: - nss_cert = x509.load_certificate_from_file(cert_filename) - except IOError as e: - raise admintool.ScriptError( - "Can't open \"%s\": %s" % (cert_filename, e)) - except (TypeError, NSPRError, ValueError) as e: - raise admintool.ScriptError("Not a valid certificate: %s" % e) - subject = nss_cert.subject - cert = nss_cert.der_data - finally: - del nss_cert + cert_obj = x509.load_certificate_from_file(cert_filename) + except IOError as e: + raise admintool.ScriptError( + "Can't open \"%s\": %s" % (cert_filename, e)) + except (TypeError, ValueError) as e: + raise admintool.ScriptError("Not a valid certificate: %s" % e) + cert = cert_obj.public_bytes(serialization.Encoding.DER) - nickname = options.nickname or str(subject) + nickname = options.nickname or str(DN(cert_obj.subject)) ca_certs = certstore.get_ca_certs_nss(api.Backend.ldap2, api.env.basedn, diff --git a/ipaserver/install/krainstance.py b/ipaserver/install/krainstance.py index 3150578..77f23c1 100644 --- a/ipaserver/install/krainstance.py +++ b/ipaserver/install/krainstance.py @@ -297,7 +297,7 @@ def __create_kra_agent(self): usertype=["undefined"], userCertificate=[cert_data], description=['2;%s;%s;%s' % ( - cert.serial_number, + cert.serial, DN(('CN', 'Certificate Authority'), self.subject_base), DN(('CN', 'IPA RA'), self.subject_base))]) conn.add_entry(entry) diff --git a/ipaserver/plugins/cert.py b/ipaserver/plugins/cert.py index a534c4d..4362d82 100644 --- a/ipaserver/plugins/cert.py +++ b/ipaserver/plugins/cert.py @@ -22,11 +22,11 @@ import base64 import collections import datetime +from operator import attrgetter import os import cryptography.x509 -from nss import nss -from nss.error import NSPRError +from cryptography.hazmat.primitives import hashes import six from ipalib import Command, Str, Int, Flag @@ -224,7 +224,7 @@ def bind_principal_can_manage_cert(cert): """Check that the bind principal can manage the given cert. ``cert`` - An NSS certificate object. + A python-cryptography ``Certificate`` object. """ bind_principal = kerberos.Principal(getattr(context, 'principal')) @@ -233,9 +233,14 @@ def bind_principal_can_manage_cert(cert): hostname = bind_principal.hostname - # If we have a hostname we want to verify that the subject - # of the certificate matches it. - return hostname == cert.subject.common_name #pylint: disable=E1101 + # Verify that hostname matches subject of cert. + # We check the "most-specific" CN value. + cns = cert.subject.get_attributes_for_oid( + cryptography.x509.oid.NameOID.COMMON_NAME) + if len(cns) == 0: + return False # no CN in subject + else: + return hostname == cns[-1].value class BaseCertObject(Object): @@ -370,30 +375,27 @@ def _parse(self, obj, full=True): attribute. """ - cert = obj.get('certificate') - if cert is not None: - cert = x509.load_certificate(cert) - obj['subject'] = DN(unicode(cert.subject)) - obj['issuer'] = DN(unicode(cert.issuer)) - obj['serial_number'] = cert.serial_number - obj['valid_not_before'] = unicode(cert.valid_not_before_str) - obj['valid_not_after'] = unicode(cert.valid_not_after_str) + if 'certificate' in obj: + cert = x509.load_certificate(obj['certificate']) + obj['subject'] = DN(cert.subject) + obj['issuer'] = DN(cert.issuer) + obj['serial_number'] = cert.serial + obj['valid_not_before'] = x509.format_datetime( + cert.not_valid_before) + obj['valid_not_after'] = x509.format_datetime( + cert.not_valid_after) if full: obj['md5_fingerprint'] = x509.to_hex_with_colons( - nss.md5_digest(cert.der_data)) + cert.fingerprint(hashes.MD5())) obj['sha1_fingerprint'] = x509.to_hex_with_colons( - nss.sha1_digest(cert.der_data)) + cert.fingerprint(hashes.SHA1())) - try: - ext_san = cert.get_extension(nss.SEC_OID_X509_SUBJECT_ALT_NAME) - general_names = x509.decode_generalnames(ext_san.value) - except KeyError: - general_names = [] + general_names = x509.process_othernames( + x509.get_san_general_names(cert)) - for name_type, _desc, name, der_name in general_names: + for gn in general_names: try: - self._add_san_attribute( - obj, full, name_type, name, der_name) + self._add_san_attribute(obj, full, gn) except Exception: # Invalid GeneralName (i.e. not a valid X.509 cert); # don't fail but log something about it @@ -404,45 +406,52 @@ def _parse(self, obj, full=True): if serial_number is not None: obj['serial_number_hex'] = u'0x%X' % serial_number - - def _add_san_attribute( - self, obj, full, name_type, name, der_name): + def _add_san_attribute(self, obj, full, gn): name_type_map = { - nss.certRFC822Name: 'san_rfc822name', - nss.certDNSName: 'san_dnsname', - nss.certX400Address: 'san_x400address', - nss.certDirectoryName: 'san_directoryname', - nss.certEDIPartyName: 'san_edipartyname', - nss.certURI: 'san_uri', - nss.certIPAddress: 'san_ipaddress', - nss.certRegisterID: 'san_oid', - (nss.certOtherName, x509.SAN_UPN): 'san_other_upn', - (nss.certOtherName, x509.SAN_KRB5PRINCIPALNAME): 'san_other_kpn', + cryptography.x509.RFC822Name: + ('san_rfc822name', attrgetter('value')), + cryptography.x509.DNSName: ('san_dnsname', attrgetter('value')), + # cryptography.x509.???: 'san_x400address', + cryptography.x509.DirectoryName: + ('san_directoryname', lambda x: DN(x.value)), + # cryptography.x509.???: 'san_edipartyname', + cryptography.x509.UniformResourceIdentifier: + ('san_uri', attrgetter('value')), + cryptography.x509.IPAddress: + ('san_ipaddress', attrgetter('value')), + cryptography.x509.RegisteredID: + ('san_oid', attrgetter('value.dotted_string')), + cryptography.x509.OtherName: ('san_other', _format_othername), + x509.UPN: ('san_other_upn', attrgetter('name')), + x509.KRB5PrincipalName: ('san_other_kpn', attrgetter('name')), } default_attrs = { 'san_rfc822name', 'san_dnsname', 'san_other_upn', 'san_other_kpn', } - attr_name = name_type_map.get(name_type, 'san_other') + if type(gn) not in name_type_map: + return + + attr_name, format_name = name_type_map[type(gn)] if full or attr_name in default_attrs: - if attr_name != 'san_other': - name_formatted = name - else: - # display as "OID : b64(DER)" - name_formatted = u'{}:{}'.format( - name_type[1], base64.b64encode(der_name)) - attr_value = self.params[attr_name].type(name_formatted) + attr_value = self.params[attr_name].type(format_name(gn)) obj.setdefault(attr_name, []).append(attr_value) if full and attr_name.startswith('san_other_'): # also include known otherName in generic otherName attribute - name_formatted = u'{}:{}'.format( - name_type[1], base64.b64encode(der_name)) - attr_value = self.params['san_other'].type(name_formatted) + attr_value = self.params['san_other'].type(_format_othername(gn)) obj.setdefault('san_other', []).append(attr_value) +def _format_othername(on): + """Format a python-cryptography OtherName for display.""" + return u'{}:{}'.format( + on.type_id.dotted_string, + base64.b64encode(on.value) + ) + + class BaseCertMethod(Method): def get_options(self): yield self.obj.params['cacn'].clone(query=True) @@ -909,7 +918,7 @@ def execute(self, serial_number, all=False, raw=False, no_members=False, raise acierr # pylint: disable=E0702 ca_obj = api.Command.ca_show(options['cacn'])['result'] - if DN(unicode(cert.issuer)) != DN(ca_obj['ipacasubjectdn'][0]): + if DN(cert.issuer) != DN(ca_obj['ipacasubjectdn'][0]): # DN of cert differs from what we requested raise errors.NotFound( reason=_("Certificate with serial number %(serial)s " @@ -1132,16 +1141,16 @@ def get_options(self): def _get_cert_key(self, cert): try: - nss_cert = x509.load_certificate(cert, x509.DER) - except NSPRError as e: + cert_obj = x509.load_certificate(cert, x509.DER) + except ValueError as e: message = messages.SearchResultTruncated( reason=_("failed to load certificate: %s") % e, ) self.add_message(message) - raise ValueError("failed to load certificate") + raise - return (DN(unicode(nss_cert.issuer)), nss_cert.serial_number) + return (DN(cert_obj.issuer), cert_obj.serial) def _get_cert_obj(self, cert, all, raw, pkey_only): obj = {'certificate': unicode(base64.b64encode(cert))} diff --git a/ipaserver/plugins/service.py b/ipaserver/plugins/service.py index a39ba32..ddae37f 100644 --- a/ipaserver/plugins/service.py +++ b/ipaserver/plugins/service.py @@ -19,6 +19,7 @@ # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. +from cryptography.hazmat.primitives import hashes import six from ipalib import api, errors, messages @@ -49,8 +50,6 @@ from ipapython import kerberos from ipapython.dn import DN -import nss.nss as nss - if six.PY3: unicode = str @@ -268,16 +267,17 @@ def set_certificate_attrs(entry_attrs): cert = entry_attrs['usercertificate'] cert = x509.normalize_certificate(cert) cert = x509.load_certificate(cert, datatype=x509.DER) - entry_attrs['subject'] = unicode(cert.subject) - entry_attrs['serial_number'] = unicode(cert.serial_number) - entry_attrs['serial_number_hex'] = u'0x%X' % cert.serial_number - entry_attrs['issuer'] = unicode(cert.issuer) - entry_attrs['valid_not_before'] = unicode(cert.valid_not_before_str) - entry_attrs['valid_not_after'] = unicode(cert.valid_not_after_str) + entry_attrs['subject'] = unicode(DN(cert.subject)) + entry_attrs['serial_number'] = unicode(cert.serial) + entry_attrs['serial_number_hex'] = u'0x%X' % cert.serial + entry_attrs['issuer'] = unicode(DN(cert.issuer)) + entry_attrs['valid_not_before'] = x509.format_datetime( + cert.not_valid_before) + entry_attrs['valid_not_after'] = x509.format_datetime(cert.not_valid_after) entry_attrs['md5_fingerprint'] = x509.to_hex_with_colons( - nss.md5_digest(cert.der_data)) + cert.fingerprint(hashes.MD5())) entry_attrs['sha1_fingerprint'] = x509.to_hex_with_colons( - nss.sha1_digest(cert.der_data)) + cert.fingerprint(hashes.SHA1())) def check_required_principal(ldap, principal): """ diff --git a/ipatests/test_ipalib/test_x509.py b/ipatests/test_ipalib/test_x509.py index f765bc9..750e086 100644 --- a/ipatests/test_ipalib/test_x509.py +++ b/ipatests/test_ipalib/test_x509.py @@ -22,9 +22,9 @@ """ import base64 +import datetime import pytest -from nss.error import NSPRError from ipalib import x509 from ipapython.dn import DN @@ -57,17 +57,25 @@ def test_1_load_base64_cert(self): # Load a good cert x509.load_certificate(goodcert) + # Should handle list/tuple + x509.load_certificate((goodcert,)) + x509.load_certificate([goodcert]) + # Load a good cert with headers newcert = '-----BEGIN CERTIFICATE-----' + goodcert + '-----END CERTIFICATE-----' x509.load_certificate(newcert) + # Should handle list/tuple + x509.load_certificate((newcert,)) + x509.load_certificate([newcert]) + # Load a good cert with bad headers newcert = '-----BEGIN CERTIFICATE-----' + goodcert with pytest.raises((TypeError, ValueError)): x509.load_certificate(newcert) # Load a bad cert - with pytest.raises(NSPRError): + with pytest.raises(ValueError): x509.load_certificate(badcert) def test_1_load_der_cert(self): @@ -80,53 +88,23 @@ def test_1_load_der_cert(self): # Load a good cert x509.load_certificate(der, x509.DER) - def test_2_get_subject(self): - """ - Test retrieving the subject - """ - subject = x509.get_subject(goodcert) - assert DN(str(subject)) == DN(('CN','ipa.example.com'),('O','IPA')) - - der = base64.b64decode(goodcert) - subject = x509.get_subject(der, x509.DER) - assert DN(str(subject)) == DN(('CN','ipa.example.com'),('O','IPA')) - - # We should be able to pass in a tuple/list of certs too - subject = x509.get_subject((goodcert)) - assert DN(str(subject)) == DN(('CN','ipa.example.com'),('O','IPA')) - - subject = x509.get_subject([goodcert]) - assert DN(str(subject)) == DN(('CN','ipa.example.com'),('O','IPA')) - - def test_2_get_serial_number(self): - """ - Test retrieving the serial number - """ - serial = x509.get_serial_number(goodcert) - assert serial == 1093 - - der = base64.b64decode(goodcert) - serial = x509.get_serial_number(der, x509.DER) - assert serial == 1093 - - # We should be able to pass in a tuple/list of certs too - serial = x509.get_serial_number((goodcert)) - assert serial == 1093 - - serial = x509.get_serial_number([goodcert]) - assert serial == 1093 + # Should handle list/tuple + x509.load_certificate((der,), x509.DER) + x509.load_certificate([der], x509.DER) def test_3_cert_contents(self): """ Test the contents of a certificate """ - # Verify certificate contents. This exercises python-nss more than - # anything but confirms our usage of it. + # Verify certificate contents. This exercises python-cryptography + # more than anything but confirms our usage of it. + not_before = datetime.datetime(2010, 6, 25, 13, 0, 42) + not_after = datetime.datetime(2015, 6, 25, 13, 0, 42) cert = x509.load_certificate(goodcert) - assert DN(str(cert.subject)) == DN(('CN','ipa.example.com'),('O','IPA')) - assert DN(str(cert.issuer)) == DN(('CN','IPA Test Certificate Authority')) - assert cert.serial_number == 1093 - assert cert.valid_not_before_str == 'Fri Jun 25 13:00:42 2010 UTC' - assert cert.valid_not_after_str == 'Thu Jun 25 13:00:42 2015 UTC' + assert DN(cert.subject) == DN(('CN', 'ipa.example.com'), ('O', 'IPA')) + assert DN(cert.issuer) == DN(('CN', 'IPA Test Certificate Authority')) + assert cert.serial == 1093 + assert cert.not_valid_before == not_before + assert cert.not_valid_after == not_after diff --git a/ipatests/test_ipaserver/test_ldap.py b/ipatests/test_ipaserver/test_ldap.py index 904c841..1ea9959 100644 --- a/ipatests/test_ipaserver/test_ldap.py +++ b/ipatests/test_ipaserver/test_ldap.py @@ -80,7 +80,7 @@ def test_GSSAPI(self): entry_attrs = self.conn.get_entry(self.dn, ['usercertificate']) cert = entry_attrs.get('usercertificate') cert = cert[0] - serial = unicode(x509.get_serial_number(cert, x509.DER)) + serial = x509.load_certificate(cert, x509.DER).serial assert serial is not None def test_simple(self): @@ -99,7 +99,7 @@ def test_simple(self): entry_attrs = self.conn.get_entry(self.dn, ['usercertificate']) cert = entry_attrs.get('usercertificate') cert = cert[0] - serial = unicode(x509.get_serial_number(cert, x509.DER)) + serial = x509.load_certificate(cert, x509.DER).serial assert serial is not None def test_Backend(self): @@ -127,7 +127,7 @@ def test_Backend(self): entry_attrs = result['result'] cert = entry_attrs.get('usercertificate') cert = cert[0] - serial = unicode(x509.get_serial_number(cert, x509.DER)) + serial = x509.load_certificate(cert, x509.DER).serial assert serial is not None def test_autobind(self): @@ -143,7 +143,7 @@ def test_autobind(self): entry_attrs = self.conn.get_entry(self.dn, ['usercertificate']) cert = entry_attrs.get('usercertificate') cert = cert[0] - serial = unicode(x509.get_serial_number(cert, x509.DER)) + serial = x509.load_certificate(cert, x509.DER).serial assert serial is not None diff --git a/ipatests/test_ipaserver/test_otptoken_import.py b/ipatests/test_ipaserver/test_otptoken_import.py index f1b4331..b885cef 100644 --- a/ipatests/test_ipaserver/test_otptoken_import.py +++ b/ipatests/test_ipaserver/test_otptoken_import.py @@ -20,7 +20,6 @@ import os import pytest from nss import nss -from ipalib.x509 import initialize_nss_database from ipaserver.install.ipa_otptoken_import import PSKCDocument, ValidationError @@ -30,9 +29,6 @@ @pytest.mark.tier1 class test_otptoken_import(object): - def teardown(self): - initialize_nss_database() - def test_figure3(self): doc = PSKCDocument(os.path.join(basename, "pskc-figure3.xml")) assert doc.keyname is None From 396149ac019eea28c127cdb65efc2119f62b65e0 Mon Sep 17 00:00:00 2001 From: Fraser Tweedale <ftwee...@redhat.com> Date: Fri, 4 Nov 2016 17:07:41 +1000 Subject: [PATCH 7/7] Remove __main__ code from ipalib.x509 and ipalib.pkcs10 This code was presumably once used for testing, but has been subsumed by the actual test suite. Part of: https://fedorahosted.org/freeipa/ticket/6398 --- ipalib/pkcs10.py | 10 ---------- ipalib/x509.py | 15 --------------- 2 files changed, 25 deletions(-) diff --git a/ipalib/pkcs10.py b/ipalib/pkcs10.py index 7145c13..a993519 100644 --- a/ipalib/pkcs10.py +++ b/ipalib/pkcs10.py @@ -20,7 +20,6 @@ from __future__ import print_function import binascii -import sys from cryptography.hazmat.backends import default_backend import cryptography.x509 @@ -55,12 +54,3 @@ def load_certificate_request(data): except binascii.Error as e: raise ValueError(e) return cryptography.x509.load_der_x509_csr(data, default_backend()) - - -if __name__ == '__main__': - # Read PEM request from stdin and print out its components - - csrlines = sys.stdin.readlines() - csr = ''.join(csrlines) - - print(load_certificate_request(csr)) diff --git a/ipalib/x509.py b/ipalib/x509.py index 7f7a89c..e1c3867 100644 --- a/ipalib/x509.py +++ b/ipalib/x509.py @@ -35,7 +35,6 @@ import binascii import datetime import ipaddress -import sys import base64 import re @@ -486,17 +485,3 @@ def format_datetime(t): if t.tzinfo is None: t = t.replace(tzinfo=UTC()) return unicode(t.strftime("%a %b %d %H:%M:%S %Y %Z")) - - -if __name__ == '__main__': - # this can be run with: - # python ipalib/x509.py < /etc/ipa/ca.crt - - # Read PEM cert from stdin and print out its components - - certlines = sys.stdin.readlines() - cert = ''.join(certlines) - - cert = load_certificate(cert) - - print(cert)
-- 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
