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

Reviewed-By: Jan Cholasta <jcholast@redhat.com>
Reviewed-By: Florence Blanc-Renaud <frenaud@redhat.com>
This commit is contained in:
Fraser Tweedale
2016-11-10 10:21:47 +01:00
committed by David Kupka
parent 9522970bfa
commit 66637f766d
4 changed files with 151 additions and 168 deletions
+17 -79
View File
@@ -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.
"""
if datatype == PEM:
csr = strip_header(csr)
csr = base64.b64decode(csr)
# 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()
def load_certificate_request(data):
"""
Load a PEM or base64-encoded PKCS #10 certificate request.
:return: a python-cryptography ``Certificate`` object.
:raises: ``ValueError`` if unable to load the request
"""
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())
return nss.CertificateRequest(csr)
if __name__ == '__main__':
nss.nss_init_nodb()
# 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))
+39
View File
@@ -39,6 +39,7 @@ import sys
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 ipalib import errors
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
+43 -52
View File
@@ -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 @@ class cert_request(Create, BaseCertMethod, VirtualCommand):
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 @@ class cert_request(Create, BaseCertMethod, VirtualCommand):
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 @@ class cert_request(Create, BaseCertMethod, VirtualCommand):
)
# 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 @@ class cert_request(Create, BaseCertMethod, VirtualCommand):
"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 @@ class cert_request(Create, BaseCertMethod, VirtualCommand):
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 @@ class cert_request(Create, BaseCertMethod, VirtualCommand):
"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 @@ class cert_request(Create, BaseCertMethod, VirtualCommand):
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:
+52 -37
View File
@@ -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 @@ class test_update(object):
"""
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 @@ class test_update(object):
subject = request.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')
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'
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 @@ class test_update(object):
subject = request.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')
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)
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 @@ class test_update(object):
"""
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')