diff options
Diffstat (limited to 'ipalib')
| -rw-r--r-- | ipalib/pkcs10.py | 116 | ||||
| -rw-r--r-- | ipalib/plugins/cert.py | 103 |
2 files changed, 177 insertions, 42 deletions
diff --git a/ipalib/pkcs10.py b/ipalib/pkcs10.py index 5958d5a21..f35e200a2 100644 --- a/ipalib/pkcs10.py +++ b/ipalib/pkcs10.py @@ -21,7 +21,7 @@ import os import sys import base64 import nss.nss as nss -from pyasn1.type import univ, namedtype, tag +from pyasn1.type import univ, char, namedtype, tag from pyasn1.codec.der import decoder from ipapython import ipautil from ipalib import api @@ -29,6 +29,10 @@ from ipalib import api PEM = 0 DER = 1 +SAN_DNSNAME = 'DNS name' +SAN_OTHERNAME_UPN = 'Other Name (OID.1.3.6.1.4.1.311.20.2.3)' +SAN_OTHERNAME_KRB5PRINCIPALNAME = 'Other Name (OID.1.3.6.1.5.2.2)' + def get_subject(csr, datatype=PEM): """ Given a CSR return the subject value. @@ -41,6 +45,89 @@ def get_subject(csr, datatype=PEM): 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) + return tuple(nss.oid_dotted_decimal(ext.oid_tag)[4:] + for ext in request.extensions) + +class _PrincipalName(univ.Sequence): + componentType = namedtype.NamedTypes( + namedtype.NamedType('name-type', univ.Integer().subtype( + explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 0)) + ), + namedtype.NamedType('name-string', univ.SequenceOf(char.GeneralString()).subtype( + explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 1)) + ), + ) + +class _KRB5PrincipalName(univ.Sequence): + componentType = namedtype.NamedTypes( + namedtype.NamedType('realm', char.GeneralString().subtype( + explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 0)) + ), + namedtype.NamedType('principalName', _PrincipalName().subtype( + explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 1)) + ), + ) + +def _decode_krb5principalname(data): + principal = decoder.decode(data, asn1Spec=_KRB5PrincipalName())[0] + realm = (str(principal['realm']).replace('\\', '\\\\') + .replace('@', '\\@')) + name = principal['principalName']['name-string'] + name = '/'.join(str(n).replace('\\', '\\\\') + .replace('/', '\\/') + .replace('@', '\\@') for n in name) + name = '%s@%s' % (name, realm) + return name + +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', univ.Choice().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() + def get_subjectaltname(csr, datatype=PEM): """ Given a CSR return the subjectaltname value, if any. @@ -48,13 +135,26 @@ def get_subjectaltname(csr, datatype=PEM): The return value is a tuple of strings or None """ request = load_certificate_request(csr, datatype) - try: - for extension in request.extensions: - if extension.oid_tag == nss.SEC_OID_X509_SUBJECT_ALT_NAME: - return nss.x509_alt_name(extension.value) - finally: - del request - return None + for extension in request.extensions: + if extension.oid_tag == nss.SEC_OID_X509_SUBJECT_ALT_NAME: + break + else: + return None + del request + + nss_names = nss.x509_alt_name(extension.value, nss.AsObject) + asn1_names = decoder.decode(extension.value.data, + asn1Spec=_SubjectAltName())[0] + names = [] + for nss_name, asn1_name in zip(nss_names, asn1_names): + name_type = nss_name.type_string + if name_type == SAN_OTHERNAME_KRB5PRINCIPALNAME: + name = _decode_krb5principalname(asn1_name['otherName']['value']) + else: + name = nss_name.name + names.append((name_type, name)) + + return tuple(names) # Unfortunately, NSS can only parse the extension request attribute, so # we have to parse friendly name ourselves (see RFC 2986) diff --git a/ipalib/plugins/cert.py b/ipalib/plugins/cert.py index b1fa32b7c..e4918a480 100644 --- a/ipalib/plugins/cert.py +++ b/ipalib/plugins/cert.py @@ -42,6 +42,7 @@ from ipalib import output from ipalib.plugins.service import validate_principal import nss.nss as nss from nss.error import NSPRError +from pyasn1.error import PyAsn1Error __doc__ = _(""" IPA certificate operations @@ -136,17 +137,6 @@ def validate_pkidate(ugettext, value): return None -def get_csr_hostname(csr): - """ - Return the value of CN in the subject of the request or None - """ - try: - subject = pkcs10.get_subject(csr) - return subject.common_name #pylint: disable=E1101 - except NSPRError, nsprerr: - raise errors.CertificateOperationError( - error=_('Failure decoding Certificate Signing Request: %s') % nsprerr) - def validate_csr(ugettext, csr): """ Ensure the CSR is base64-encoded and can be decoded by our PKCS#10 @@ -290,6 +280,14 @@ class cert_request(VirtualCommand): ), ) + _allowed_extensions = { + '2.5.29.14': None, # Subject Key Identifier + '2.5.29.15': None, # Key Usage + '2.5.29.17': 'request certificate with subjectaltname', + '2.5.29.19': None, # Basic Constraints + '2.5.29.37': None, # Extended Key Usage + } + def execute(self, csr, **kw): ldap = self.api.Backend.ldap2 principal = kw.get('principal') @@ -313,10 +311,22 @@ class cert_request(VirtualCommand): if not bind_principal.startswith('host/'): self.check_access() - # FIXME: add support for subject alt name + try: + subject = pkcs10.get_subject(csr) + extensions = pkcs10.get_extensions(csr) + subjectaltname = pkcs10.get_subjectaltname(csr) or () + except (NSPRError, PyAsn1Error), e: + raise errors.CertificateOperationError( + error=_("Failure decoding Certificate Signing Request: %s") % e) + + if not bind_principal.startswith('host/'): + for ext in extensions: + operation = self._allowed_extensions.get(ext) + if operation: + self.check_access(operation) # Ensure that the hostname in the CSR matches the principal - subject_host = get_csr_hostname(csr) + subject_host = subject.common_name #pylint: disable=E1101 if not subject_host: raise errors.ValidationError(name='csr', error=_("No hostname was found in subject of request.")) @@ -328,28 +338,40 @@ class cert_request(VirtualCommand): "does not match principal hostname '%(hostname)s'") % dict( subject_host=subject_host, hostname=hostname)) + for ext in extensions: + if ext not in self._allowed_extensions: + raise errors.ValidationError( + name='csr', error=_("extension %s is forbidden") % ext) + + for name_type, name in subjectaltname: + if name_type not in (pkcs10.SAN_DNSNAME, + pkcs10.SAN_OTHERNAME_KRB5PRINCIPALNAME, + pkcs10.SAN_OTHERNAME_UPN): + raise errors.ValidationError( + name='csr', + error=_("subject alt name type %s is forbidden") % + name_type) + dn = None service = None # See if the service exists and punt if it doesn't and we aren't # going to add it try: - if not principal.startswith('host/'): - service = api.Command['service_show'](principal, all=True)['result'] - dn = service['dn'] + if servicename != 'host': + service = api.Command['service_show'](principal, all=True) else: - hostname = get_host_from_principal(principal) - service = api.Command['host_show'](hostname, all=True)['result'] - dn = service['dn'] + service = api.Command['host_show'](hostname, all=True) except errors.NotFound, e: if not add: raise errors.NotFound(reason=_("The service principal for " "this request doesn't exist.")) try: - service = api.Command['service_add'](principal, **{'force': True})['result'] - dn = service['dn'] + service = api.Command['service_add'](principal, force=True) except errors.ACIError: raise errors.ACIError(info=_('You need to be a member of ' 'the serviceadmin role to add services')) + service = service['result'] + dn = service['dn'] # We got this far so the service entry exists, can we write it? if not ldap.can_write(dn, "usercertificate"): @@ -357,25 +379,38 @@ class cert_request(VirtualCommand): "to the 'userCertificate' attribute of entry '%s'.") % dn) # Validate the subject alt name, if any - subjectaltname = pkcs10.get_subjectaltname(csr) - if subjectaltname is not None: - for name in subjectaltname: + for name_type, name in subjectaltname: + if name_type == pkcs10.SAN_DNSNAME: name = unicode(name) try: - hostentry = api.Command['host_show'](name, all=True)['result'] - hostdn = hostentry['dn'] + if servicename == 'host': + altservice = api.Command['host_show'](name, all=True) + else: + altprincipal = '%s/%s@%s' % (servicename, name, realm) + altservice = api.Command['service_show']( + altprincipal, all=True) except errors.NotFound: # We don't want to issue any certificates referencing # machines we don't know about. Nothing is stored in this # host record related to this certificate. - raise errors.NotFound(reason=_('no host record for ' - 'subject alt name %s in certificate request') % name) - authprincipal = getattr(context, 'principal') - if authprincipal.startswith("host/"): - if not hostdn in service.get('managedby_host', []): - raise errors.ACIError(info=_( - "Insufficient privilege to create a certificate " - "with subject alt name '%s'.") % name) + raise errors.NotFound(reason=_('The service principal for ' + 'subject alt name %s in certificate request does not ' + 'exist') % name) + altdn = altservice['result']['dn'] + if not ldap.can_write(altdn, "usercertificate"): + raise errors.ACIError(info=_( + "Insufficient privilege to create a certificate with " + "subject alt name '%s'.") % name) + elif name_type in (pkcs10.SAN_OTHERNAME_KRB5PRINCIPALNAME, + pkcs10.SAN_OTHERNAME_UPN): + if name != principal: + raise errors.ACIError( + info=_("Principal '%s' in subject alt name does not " + "match requested service principal") % name) + else: + raise errors.ACIError( + info=_("Subject alt name type %s is forbidden") % + name_type) if 'usercertificate' in service: serial = x509.get_serial_number(service['usercertificate'][0], datatype=x509.DER) |