URL: https://github.com/freeipa/freeipa/pull/878
Author: stlaz
Title: #878: [experimental] Make certificate an object
Action: opened
PR body:
"""
This is a WIP patchset (you can tell it's WIP by the lack of the commit
messages) which implements certificates as objects. This should
very much simplify the way certificates are processed in the whole
project and is a first part of this effort. This makes us not care about
the certificate format (=> no encoding/decoding) until we really need
to write it somewhere.
In the second part of this effort, there should be a Certificate parameter
to the framework which should further improve the format-agnostic
behavior.
This is a part of Python 3 effort which showed that it's necessary to
improve how certificates are worked with.
"""
To pull the PR as Git branch:
git remote add ghfreeipa https://github.com/freeipa/freeipa
git fetch ghfreeipa pull/878/head:pr878
git checkout pr878
From eca199275315008579f577e448ffddd6b4d5ba1b Mon Sep 17 00:00:00 2001
From: Stanislav Laznicka <slazn...@redhat.com>
Date: Fri, 16 Jun 2017 09:36:26 +0200
Subject: [PATCH 1/4] Split x509.load_certificate() into PEM/DER functions
Splitting the load_certificate() function into two separate helps
us word the requirements for the input explicitly. It also makes
our backend similar to the one of python-cryptography so eventually
we can swap python-cryptography for IPA x509 module.
TICKET TBD
---
ipaclient/install/client.py | 2 +-
ipaclient/plugins/cert.py | 6 ++--
ipaclient/plugins/vault.py | 4 +--
ipalib/install/certstore.py | 2 +-
ipalib/x509.py | 57 +++++++++++++++++++++-------------
ipapython/certdb.py | 12 ++++----
ipaserver/install/ca.py | 2 +-
ipaserver/install/cainstance.py | 10 +++---
ipaserver/install/certs.py | 4 +--
ipaserver/install/installutils.py | 4 +--
ipaserver/plugins/baseuser.py | 2 +-
ipaserver/plugins/cert.py | 18 ++++++-----
ipaserver/plugins/host.py | 6 ++--
ipaserver/plugins/service.py | 12 ++++----
ipatests/test_ipalib/test_x509.py | 65 +++++++++++++++++++--------------------
15 files changed, 110 insertions(+), 96 deletions(-)
diff --git a/ipaclient/install/client.py b/ipaclient/install/client.py
index 1aa50372a4..30cecc882e 100644
--- a/ipaclient/install/client.py
+++ b/ipaclient/install/client.py
@@ -1646,7 +1646,7 @@ def get_ca_certs_from_ldap(server, basedn, realm):
root_logger.debug("get_ca_certs_from_ldap() error: %s", e)
raise
- certs = [x509.load_certificate(c[0], x509.DER) for c in certs
+ certs = [x509.load_der_x509_certificate(c[0]) for c in certs
if c[2] is not False]
return certs
diff --git a/ipaclient/plugins/cert.py b/ipaclient/plugins/cert.py
index 541b82a5ed..245b3c57aa 100644
--- a/ipaclient/plugins/cert.py
+++ b/ipaclient/plugins/cert.py
@@ -66,9 +66,9 @@ def forward(self, *args, **options):
certs = result['result']['certificate_chain']
else:
certs = [result['result']['certificate']]
- certs = (x509.normalize_certificate(cert) for cert in certs)
- certs = (x509.make_pem(base64.b64encode(cert)) for cert in certs)
- with open(certificate_out, 'w') as f:
+ certs = (x509.ensure_der_format(cert) for cert in certs)
+ certs = (x509.load_der_x509_certificate(cert) for cert in certs)
+ with open(certificate_out, 'wb') as f:
f.write('\n'.join(certs))
return result
diff --git a/ipaclient/plugins/vault.py b/ipaclient/plugins/vault.py
index 34988daf3c..75e8da4fe5 100644
--- a/ipaclient/plugins/vault.py
+++ b/ipaclient/plugins/vault.py
@@ -683,8 +683,8 @@ def internal(self, algo, *args, **options):
# retrieve transport certificate (cached by vaultconfig_show)
response = self.api.Command.vaultconfig_show()
- transport_cert = x509.load_certificate(
- response['result']['transport_cert'], x509.DER)
+ transport_cert = x509.load_der_x509_certificate(
+ response['result']['transport_cert'])
# call with the retrieved transport certificate
return self._do_internal(algo, transport_cert, True,
*args, **options)
diff --git a/ipalib/install/certstore.py b/ipalib/install/certstore.py
index bc2079fb12..0d0902fc1a 100644
--- a/ipalib/install/certstore.py
+++ b/ipalib/install/certstore.py
@@ -30,7 +30,7 @@
def _parse_cert(dercert):
try:
- cert = x509.load_certificate(dercert, x509.DER)
+ cert = x509.load_der_x509_certificate(dercert)
subject = DN(cert.subject)
issuer = DN(cert.issuer)
serial_number = cert.serial_number
diff --git a/ipalib/x509.py b/ipalib/x509.py
index 4d866a64c5..aabe0eb5ce 100644
--- a/ipalib/x509.py
+++ b/ipalib/x509.py
@@ -19,7 +19,7 @@
# Certificates should be stored internally DER-encoded. We can be passed
# a certificate several ways: read if from LDAP, read it from a 3rd party
-# app (dogtag, candlepin, etc) or as user input. The normalize_certificate()
+# app (dogtag, candlepin, etc) or as user input. The ensure_der_format()
# function will convert an incoming certificate to DER-encoding.
# Conventions
@@ -101,24 +101,29 @@ def strip_header(pem):
return pem
-def load_certificate(data, datatype=PEM):
"""
- Load an X.509 certificate.
+def load_pem_x509_certificate(data):
+ """
+ Load an X.509 certificate in PEM format.
- :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.
-
"""
- if type(data) in (tuple, list):
- data = data[0]
+ return _IPACertificate(
+ crypto_x509.load_pem_x509_certificate(data, backend=default_backend())
+ )
- if (datatype == PEM):
- data = strip_header(data)
- data = base64.b64decode(data)
- return cryptography.x509.load_der_x509_certificate(data, default_backend())
+def load_der_x509_certificate(data):
+ """
+ Load an X.509 certificate in PEM format.
+
+ :return: a python-cryptography ``CertificateSigningRequest`` object.
+ :raises: ``ValueError`` if unable to load the certificate.
+ """
+ return _IPACertificate(
+ crypto_x509.load_der_x509_certificate(data, backend=default_backend())
+ )
def load_certificate_from_file(filename, dbdir=None):
@@ -126,10 +131,9 @@ def load_certificate_from_file(filename, dbdir=None):
Load a certificate from a PEM file.
Returns a python-cryptography ``Certificate`` object.
-
"""
with open(filename, mode='rb') as f:
- return load_certificate(f.read(), PEM)
+ return load_pem_x509_certificate(f.read())
def load_certificate_list(data):
@@ -140,8 +144,7 @@ def load_certificate_list(data):
"""
certs = PEM_REGEX.findall(data)
- certs = [load_certificate(cert, PEM) for cert in certs]
- return certs
+ return [load_pem_x509_certificate(cert) for cert in certs]
def load_certificate_list_from_file(filename):
@@ -151,7 +154,7 @@ def load_certificate_list_from_file(filename):
Return a list of python-cryptography ``Certificate`` objects.
"""
- with open(filename) as f:
+ with open(filename, 'rb') as f:
return load_certificate_list(f.read())
@@ -242,7 +245,8 @@ def make_pem(data):
pemcert + \
'\n-----END CERTIFICATE-----'
-def normalize_certificate(rawcert):
+
+def ensure_der_format(rawcert):
"""
Incoming certificates should be DER-encoded. If not it is converted to
DER-format.
@@ -274,17 +278,26 @@ def normalize_certificate(rawcert):
# At this point we should have a DER certificate.
# Attempt to decode it.
- validate_certificate(dercert, datatype=DER)
-
+ validate_der_x509_certificate(dercert)
return dercert
-def validate_certificate(cert, datatype=PEM):
+def validate_pem_x509_certificate(cert):
+ """
+ Perform cert validation by trying to load it via python-cryptography.
+ """
+ try:
+ load_pem_x509_certificate(cert)
+ except ValueError as e:
+ raise errors.CertificateFormatError(error=str(e))
+
+
+def validate_der_x509_certificate(cert):
"""
Perform cert validation by trying to load it via python-cryptography.
"""
try:
- load_certificate(cert, datatype=datatype)
+ load_der_x509_certificate(cert)
except ValueError as e:
raise errors.CertificateFormatError(error=str(e))
diff --git a/ipapython/certdb.py b/ipapython/certdb.py
index 8c53821916..b9e364cdca 100644
--- a/ipapython/certdb.py
+++ b/ipapython/certdb.py
@@ -454,7 +454,7 @@ def import_files(self, files, import_keys=False, key_password=None,
if label in ('CERTIFICATE', 'X509 CERTIFICATE',
'X.509 CERTIFICATE'):
try:
- x509.load_certificate(match.group(2))
+ x509.load_pem_x509_certificate(match.group(2))
except ValueError as e:
if label != 'CERTIFICATE':
root_logger.warning(
@@ -525,7 +525,7 @@ def import_files(self, files, import_keys=False, key_password=None,
# Try to load the file as DER certificate
try:
- x509.load_certificate(data, x509.DER)
+ x509.load_der_x509_certificate(data)
except ValueError:
pass
else:
@@ -571,7 +571,7 @@ def import_files(self, files, import_keys=False, key_password=None,
"No server certificates found in %s" % (', '.join(files)))
for cert_pem in extracted_certs:
- cert = x509.load_certificate(cert_pem)
+ cert = x509.load_pem_x509_certificate(cert_pem)
nickname = str(DN(cert.subject))
data = cert.public_bytes(serialization.Encoding.DER)
self.add_cert(data, nickname, EMPTY_TRUST_FLAGS)
@@ -682,7 +682,7 @@ def verify_server_cert_validity(self, nickname, hostname):
Raises a ValueError if the certificate is invalid.
"""
cert = self.get_cert(nickname)
- cert = x509.load_certificate(cert, x509.DER)
+ cert = x509.load_der_x509_certificate(cert)
try:
self.run_certutil(['-V', '-n', nickname, '-u', 'V'],
@@ -702,7 +702,7 @@ def verify_server_cert_validity(self, nickname, hostname):
def verify_ca_cert_validity(self, nickname):
cert = self.get_cert(nickname)
- cert = x509.load_certificate(cert, x509.DER)
+ cert = x509.load_der_x509_certificate(cert)
if not cert.subject:
raise ValueError("has empty subject")
@@ -736,6 +736,6 @@ def verify_ca_cert_validity(self, nickname):
def verify_kdc_cert_validity(self, nickname, realm):
nicknames = self.get_trust_chain(nickname)
certs = [self.get_cert(nickname) for nickname in nicknames]
- certs = [x509.load_certificate(cert, x509.DER) for cert in certs]
+ certs = [x509.load_der_x509_certificate(cert) for cert in certs]
verify_kdc_cert_validity(certs[-1], certs[:-1], realm)
diff --git a/ipaserver/install/ca.py b/ipaserver/install/ca.py
index 52cb20f1cb..a88bee7e89 100644
--- a/ipaserver/install/ca.py
+++ b/ipaserver/install/ca.py
@@ -194,7 +194,7 @@ def install_check(standalone, replica_config, options):
cert = db.get_cert_from_db(nickname)
if not cert:
continue
- subject = DN(x509.load_certificate(cert).subject)
+ subject = DN(x509.load_pem_x509_certificate(cert).subject)
if subject == DN(options._ca_subject):
raise ScriptError(
"Certificate with subject %s is present in %s, "
diff --git a/ipaserver/install/cainstance.py b/ipaserver/install/cainstance.py
index 28a702dba5..531f28f658 100644
--- a/ipaserver/install/cainstance.py
+++ b/ipaserver/install/cainstance.py
@@ -1415,7 +1415,7 @@ def update_people_entry(dercert):
is needed when a certificate is renewed.
"""
def make_filter(dercert):
- cert = x509.load_certificate(dercert, datatype=x509.DER)
+ cert = x509.load_der_x509_certificate(dercert)
subject = DN(cert.subject)
issuer = DN(cert.issuer)
return ldap2.ldap2.combine_filters(
@@ -1428,8 +1428,8 @@ def make_filter(dercert):
ldap2.ldap2.MATCH_ALL)
def make_entry(dercert, entry):
- cert = x509.load_certificate(dercert, datatype=x509.DER)
- serial_number = cert.serial
+ cert = x509.load_der_x509_certificate(dercert)
+ serial_number = cert.serial_number
subject = DN(cert.subject)
issuer = DN(cert.issuer)
entry['usercertificate'].append(dercert)
@@ -1445,7 +1445,7 @@ def update_authority_entry(dercert):
serial number to match the given cert.
"""
def make_filter(dercert):
- cert = x509.load_certificate(dercert, datatype=x509.DER)
+ cert = x509.load_der_x509_certificate(dercert)
subject = str(DN(cert.subject))
return ldap2.ldap2.make_filter(
dict(objectclass='authority', authoritydn=subject),
@@ -1453,7 +1453,7 @@ def make_filter(dercert):
)
def make_entry(dercert, entry):
- cert = x509.load_certificate(dercert, datatype=x509.DER)
+ cert = x509.load_der_x509_certificate(dercert)
entry['authoritySerial'] = cert.serial_number
return entry
diff --git a/ipaserver/install/certs.py b/ipaserver/install/certs.py
index 02c479d925..5b1b6ec22a 100644
--- a/ipaserver/install/certs.py
+++ b/ipaserver/install/certs.py
@@ -56,7 +56,7 @@ def get_cert_nickname(cert):
representation of the first RDN in the subject and subject_dn
is a DN object.
"""
- cert_obj = x509.load_certificate(cert)
+ cert_obj = x509.load_pem_x509_certificate(cert)
dn = DN(cert_obj.subject)
return (str(dn[0]), dn)
@@ -359,7 +359,7 @@ def track_server_cert(self, nickname, principal, password_file=None, command=Non
return
cert = self.get_cert_from_db(nickname)
- cert_obj = x509.load_certificate(cert)
+ cert_obj = x509.load_pem_x509_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 04bf4fcd20..e9cc2a78a9 100644
--- a/ipaserver/install/installutils.py
+++ b/ipaserver/install/installutils.py
@@ -1051,7 +1051,7 @@ def load_pkcs12(cert_files, key_password, key_nickname, ca_cert_files,
if ca_cert is None:
ca_cert = cert
- cert_obj = x509.load_certificate(cert, x509.DER)
+ cert_obj = x509.load_der_x509_certificate(cert)
subject = DN(cert_obj.subject)
issuer = DN(cert_obj.issuer)
@@ -1182,7 +1182,7 @@ def load_external_cert(files, ca_subject):
for nickname, _trust_flags in nssdb.list_certs():
cert = nssdb.get_cert(nickname, pem=True)
- cert_obj = x509.load_certificate(cert)
+ cert_obj = x509.load_pem_x509_certificate(cert)
subject = DN(cert_obj.subject)
issuer = DN(cert_obj.issuer)
diff --git a/ipaserver/plugins/baseuser.py b/ipaserver/plugins/baseuser.py
index 8fbec733a6..8b902aa4e8 100644
--- a/ipaserver/plugins/baseuser.py
+++ b/ipaserver/plugins/baseuser.py
@@ -799,7 +799,7 @@ def _convert_options_to_certmap(cls, entry_attrs, issuer=None,
data.append(cls._build_mapdata(subject, issuer))
for dercert in certificates:
- cert = x509.load_certificate(dercert, x509.DER)
+ cert = x509.load_der_x509_certificate(dercert)
issuer = DN(cert.issuer)
subject = DN(cert.subject)
if not subject:
diff --git a/ipaserver/plugins/cert.py b/ipaserver/plugins/cert.py
index b62f82541d..aa0ea250ca 100644
--- a/ipaserver/plugins/cert.py
+++ b/ipaserver/plugins/cert.py
@@ -328,7 +328,7 @@ def ca_kdc_check(api_instance, hostname):
def validate_certificate(value):
- return x509.validate_certificate(value, x509.DER)
+ return x509.validate_der_x509_certificate(value)
def bind_principal_can_manage_cert(cert):
@@ -369,7 +369,7 @@ class BaseCertObject(Object):
'certificate', validate_certificate,
label=_("Certificate"),
doc=_("Base-64 encoded certificate."),
- normalizer=x509.normalize_certificate,
+ normalizer=x509.ensure_der_format,
flags={'no_create', 'no_update', 'no_search'},
),
Bytes(
@@ -493,7 +493,7 @@ def _parse(self, obj, full=True):
"""
if 'certificate' in obj:
- cert = x509.load_certificate(obj['certificate'])
+ cert = x509.load_pem_x509_certificate(obj['certificate'])
obj['subject'] = DN(cert.subject)
obj['issuer'] = DN(cert.issuer)
obj['serial_number'] = cert.serial_number
@@ -945,7 +945,7 @@ def execute(self, csr, all=False, raw=False, chain=False, **kw):
"used for krbtgt certificates")
if 'certificate_chain' in ca_obj:
- cert = x509.load_certificate(result['certificate'])
+ cert = x509.load_pem_x509_certificate(result['certificate'])
cert = cert.public_bytes(serialization.Encoding.DER)
result['certificate_chain'] = [cert] + ca_obj['certificate_chain']
@@ -1164,7 +1164,7 @@ def execute(self, serial_number, all=False, raw=False, no_members=False,
# we don't tell Dogtag the issuer (but we check the cert after).
#
result = self.Backend.ra.get_certificate(str(serial_number))
- cert = x509.load_certificate(result['certificate'])
+ cert = x509.load_pem_x509_certificate(result['certificate'])
try:
self.check_access()
@@ -1241,7 +1241,8 @@ def execute(self, serial_number, **kw):
except errors.ACIError as acierr:
self.debug("Not granted by ACI to revoke certificate, looking at principal")
try:
- cert = x509.load_certificate(resp['result']['certificate'])
+ cert = x509.load_pem_x509_certificate(
+ resp['result']['certificate'])
if not bind_principal_can_manage_cert(cert):
raise acierr
except errors.NotImplementedError:
@@ -1406,7 +1407,7 @@ def get_options(self):
def _get_cert_key(self, cert):
try:
- cert_obj = x509.load_certificate(cert, x509.DER)
+ cert_obj = x509.load_der_x509_certificate(cert)
except ValueError as e:
message = messages.SearchResultTruncated(
reason=_("failed to load certificate: %s") % e,
@@ -1662,7 +1663,8 @@ def execute(self, criteria=None, all=False, raw=False, pkey_only=False,
obj['certificate'].replace('\r\n', ''))
if 'certificate_chain' in ca_obj:
- cert = x509.load_certificate(obj['certificate'])
+ cert = x509.load_der_x509_certificate(
+ obj['certificate'])
cert_der = (
cert.public_bytes(serialization.Encoding.DER))
obj['certificate_chain'] = (
diff --git a/ipaserver/plugins/host.py b/ipaserver/plugins/host.py
index 1e1f9d82df..5a14c3bf87 100644
--- a/ipaserver/plugins/host.py
+++ b/ipaserver/plugins/host.py
@@ -687,7 +687,7 @@ def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
# save the password so it can be displayed in post_callback
setattr(context, 'randompassword', entry_attrs['userpassword'])
certs = options.get('usercertificate', [])
- certs_der = [x509.normalize_certificate(c) for c in certs]
+ certs_der = [x509.ensure_der_format(c) for c in certs]
entry_attrs['usercertificate'] = certs_der
entry_attrs['managedby'] = dn
entry_attrs['objectclass'].append('ieee802device')
@@ -891,7 +891,7 @@ def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
# verify certificates
certs = entry_attrs.get('usercertificate') or []
- certs_der = [x509.normalize_certificate(c) for c in certs]
+ certs_der = [x509.ensure_der_format(c) for c in certs]
# revoke removed certificates
ca_is_enabled = self.api.Command.ca_is_enabled()['result']
@@ -901,7 +901,7 @@ def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
except errors.NotFound:
self.obj.handle_not_found(*keys)
old_certs = entry_attrs_old.get('usercertificate', [])
- old_certs_der = [x509.normalize_certificate(c) for c in old_certs]
+ old_certs_der = [x509.ensure_der_format(c) for c in old_certs]
removed_certs_der = set(old_certs_der) - set(certs_der)
for der in removed_certs_der:
rm_certs = api.Command.cert_find(certificate=der)['result']
diff --git a/ipaserver/plugins/service.py b/ipaserver/plugins/service.py
index 03271d6858..4950227abf 100644
--- a/ipaserver/plugins/service.py
+++ b/ipaserver/plugins/service.py
@@ -216,7 +216,7 @@ def validate_certificate(ugettext, cert):
Check whether the certificate is properly encoded to DER
"""
if api.env.in_server:
- x509.validate_certificate(cert, datatype=x509.DER)
+ x509.validate_der_x509_certificate(cert)
def revoke_certs(certs):
@@ -265,8 +265,8 @@ def set_certificate_attrs(entry_attrs):
cert = entry_attrs['usercertificate'][0]
else:
cert = entry_attrs['usercertificate']
- cert = x509.normalize_certificate(cert)
- cert = x509.load_certificate(cert, datatype=x509.DER)
+ cert = x509.ensure_der_format(cert)
+ cert = x509.load_der_x509_certificate(cert)
entry_attrs['subject'] = unicode(DN(cert.subject))
entry_attrs['serial_number'] = unicode(cert.serial_number)
entry_attrs['serial_number_hex'] = u'0x%X' % cert.serial_number
@@ -629,7 +629,7 @@ def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
self.obj.validate_ipakrbauthzdata(entry_attrs)
certs = options.get('usercertificate', [])
- certs_der = [x509.normalize_certificate(c) for c in certs]
+ certs_der = [x509.ensure_der_format(c) for c in certs]
entry_attrs['usercertificate'] = certs_der
if not options.get('force', False):
@@ -701,7 +701,7 @@ def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
# verify certificates
certs = entry_attrs.get('usercertificate') or []
- certs_der = [x509.normalize_certificate(c) for c in certs]
+ certs_der = [x509.ensure_der_format(c) for c in certs]
# revoke removed certificates
ca_is_enabled = self.api.Command.ca_is_enabled()['result']
if 'usercertificate' in options and ca_is_enabled:
@@ -710,7 +710,7 @@ def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
except errors.NotFound:
self.obj.handle_not_found(*keys)
old_certs = entry_attrs_old.get('usercertificate', [])
- old_certs_der = [x509.normalize_certificate(c) for c in old_certs]
+ old_certs_der = [x509.ensure_der_format(c) for c in old_certs]
removed_certs_der = set(old_certs_der) - set(certs_der)
for der in removed_certs_der:
rm_certs = api.Command.cert_find(certificate=der)['result']
diff --git a/ipatests/test_ipalib/test_x509.py b/ipatests/test_ipalib/test_x509.py
index 1ccf99d3e5..cd845f786f 100644
--- a/ipatests/test_ipalib/test_x509.py
+++ b/ipatests/test_ipalib/test_x509.py
@@ -34,10 +34,31 @@
# certutil -
# certificate for CN=ipa.example.com,O=IPA
-goodcert = 'MIICAjCCAWugAwIBAgICBEUwDQYJKoZIhvcNAQEFBQAwKTEnMCUGA1UEAxMeSVBBIFRlc3QgQ2VydGlmaWNhdGUgQXV0aG9yaXR5MB4XDTEwMDYyNTEzMDA0MloXDTE1MDYyNTEzMDA0MlowKDEMMAoGA1UEChMDSVBBMRgwFgYDVQQDEw9pcGEuZXhhbXBsZS5jb20wgZ8wDQYJKoZIhvcNAQEBBQADgY0AMIGJAoGBAJcZ+H6+cQaN/BlzR8OYkVeJgaU5tCaV9FF1m7Ws/ftPtTJUaSL1ncp6603rjA4tH1aa/B8i8xdC46+ZbY2au8b9ryGcOsx2uaRpNLEQ2Fy//q1kQC8oM+iD8Nd6osF0a2wnugsgnJHPuJzhViaWxYgzk5DRdP81debokF3f3FX/AgMBAAGjOjA4MBEGCWCGSAGG+EIBAQQEAwIGQDATBgNVHSUEDDAKBggrBgEFBQcDATAOBgNVHQ8BAf8EBAMCBPAwDQYJKoZIhvcNAQEFBQADgYEALD6X9V9w381AzzQPcHsjIjiX3B/AF9RCGocKZUDXkdDhsD9NZ3PLPEf1AMjkraKG963HPB8scyiBbbSuSh6m7TCp0eDgRpo77zNuvd3U4Qpm0Qk+KEjtHQDjNNG6N4ZnCQPmjFPScElvc/GgW7XMbywJy2euF+3/Uip8cnPgSH4='
-
+goodcert = (
+ 'MIICAjCCAWugAwIBAgICBEUwDQYJKoZIhvcNAQEFBQAwKTEnMCUGA1UEAxMeSVBB'
+ 'IFRlc3QgQ2VydGlmaWNhdGUgQXV0aG9yaXR5MB4XDTEwMDYyNTEzMDA0MloXDTE1'
+ 'MDYyNTEzMDA0MlowKDEMMAoGA1UEChMDSVBBMRgwFgYDVQQDEw9pcGEuZXhhbXBs'
+ 'ZS5jb20wgZ8wDQYJKoZIhvcNAQEBBQADgY0AMIGJAoGBAJcZ+H6+cQaN/BlzR8OY'
+ 'kVeJgaU5tCaV9FF1m7Ws/ftPtTJUaSL1ncp6603rjA4tH1aa/B8i8xdC46+ZbY2a'
+ 'u8b9ryGcOsx2uaRpNLEQ2Fy//q1kQC8oM+iD8Nd6osF0a2wnugsgnJHPuJzhViaW'
+ 'xYgzk5DRdP81debokF3f3FX/AgMBAAGjOjA4MBEGCWCGSAGG+EIBAQQEAwIGQDAT'
+ 'BgNVHSUEDDAKBggrBgEFBQcDATAOBgNVHQ8BAf8EBAMCBPAwDQYJKoZIhvcNAQEF'
+ 'BQADgYEALD6X9V9w381AzzQPcHsjIjiX3B/AF9RCGocKZUDXkdDhsD9NZ3PLPEf1'
+ 'AMjkraKG963HPB8scyiBbbSuSh6m7TCp0eDgRpo77zNuvd3U4Qpm0Qk+KEjtHQDj'
+ 'NNG6N4ZnCQPmjFPScElvc/GgW7XMbywJy2euF+3/Uip8cnPgSH4='
+)
+
+goodcert_headers = (
+ '-----BEGIN CERTIFICATE-----\n{cert}\n-----END CERTIFICATE-----'
+ .format(cert=goodcert)
+)
# The base64-encoded string 'bad cert'
-badcert = 'YmFkIGNlcnQ='
+badcert = (
+ '-----BEGIN CERTIFICATE-----\n'
+ 'YmFkIGNlcnQ=\r\n'
+ '-----END CERTIFICATE-----'
+)
+
class test_x509(object):
"""
@@ -55,39 +76,21 @@ 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])
+ x509.load_pem_x509_certificate(goodcert_headers)
# Load a good cert with headers and leading text
newcert = (
- 'leading text\n-----BEGIN CERTIFICATE-----' +
- goodcert +
- '-----END CERTIFICATE-----')
- x509.load_certificate(newcert)
-
- # Should handle list/tuple
- x509.load_certificate((newcert,))
- x509.load_certificate([newcert])
+ 'leading text\n' + goodcert_headers)
+ x509.load_pem_x509_certificate(newcert)
# Load a good cert with bad headers
- newcert = '-----BEGIN CERTIFICATE-----' + goodcert
+ newcert = '-----BEGIN CERTIFICATE-----' + goodcert_headers
with pytest.raises((TypeError, ValueError)):
- x509.load_certificate(newcert)
+ x509.load_pem_x509_certificate(newcert)
# Load a bad cert
with pytest.raises(ValueError):
- x509.load_certificate(badcert)
+ x509.load_pem_x509_certificate(badcert)
def test_1_load_der_cert(self):
"""
@@ -97,11 +100,7 @@ def test_1_load_der_cert(self):
der = base64.b64decode(goodcert)
# Load a good cert
- x509.load_certificate(der, x509.DER)
-
- # Should handle list/tuple
- x509.load_certificate((der,), x509.DER)
- x509.load_certificate([der], x509.DER)
+ x509.load_der_x509_certificate(der)
def test_3_cert_contents(self):
"""
@@ -112,7 +111,7 @@ def test_3_cert_contents(self):
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)
+ cert = x509.load_pem_x509_certificate(goodcert_headers)
assert DN(cert.subject) == DN(('CN', 'ipa.example.com'), ('O', 'IPA'))
assert DN(cert.issuer) == DN(('CN', 'IPA Test Certificate Authority'))
From fd53e844bf16411f9b797c851f349120881670cc Mon Sep 17 00:00:00 2001
From: Stanislav Laznicka <slazn...@redhat.com>
Date: Fri, 16 Jun 2017 10:18:07 +0200
Subject: [PATCH 2/4] WIP: make certificates represented as objects
---
ipalib/x509.py | 236 +++++++++++++++++++++++++++++++++------------------------
1 file changed, 137 insertions(+), 99 deletions(-)
diff --git a/ipalib/x509.py b/ipalib/x509.py
index aabe0eb5ce..8319fc1b45 100644
--- a/ipalib/x509.py
+++ b/ipalib/x509.py
@@ -39,10 +39,14 @@
import base64
import re
+from cryptography import x509 as crypto_x509
from cryptography.hazmat.backends import default_backend
-import cryptography.x509
+from cryptography.hazmat.primitives.serialization import (
+ Encoding, PublicFormat
+)
from pyasn1.type import univ, char, namedtype, tag
from pyasn1.codec.der import decoder, encoder
+# from pyasn1.codec.native import decoder, encoder
from pyasn1_modules import rfc2315, rfc2459
import six
@@ -101,7 +105,119 @@ def strip_header(pem):
return pem
- """
+class _IPACertificate(object):
+ """
+ A proxy class wrapping a python-cryptography certificate representation for
+ FreeIPA purposes
+ """
+ def __init__(self, cert, backend=None):
+ """
+ :param cert: A python-cryptography Certificate object
+ :param backend: A python-cryptography Backend object
+ """
+ self._cert = cert
+ self.backend = default_backend() if backend is None else backend()
+
+ # initialize the certificate fields
+ # we have to do it this way so that some systems don't explode since
+ # some field types encode-decoding is not strongly defined
+ self._subject = self.__get_der_field('subject')
+ self._issuer = self.__get_der_field('issuer')
+
+ def __str__(self):
+ return self._cert.public_bytes(encoding=Encoding.PEM)
+
+ def __bytes__(self):
+ return self._cert.public_bytes(encoding=Encoding.DER)
+
+ def __encode_extension(oid, critical, value):
+ ext = rfc2459.Extension()
+ # TODO: nemusi se psat tak hrozne, ale v pyasn1 jsou kodeky, ktere
+ # se o toto postaraji a prevadeji do python-native typu
+ ext['extnID'] = univ.ObjectIdentifier(oid)
+ ext['critical'] = univ.Boolean(critical)
+ ext['extnValue'] = univ.Any(encoder.encode(univ.OctetString(value)))
+ ext = encoder.encode(ext)
+ return ext
+
+ def __get_der_field(self, field):
+ cert_bytes = self._cert.tbs_certificate_bytes
+ cert = decoder.decode(cert_bytes, rfc2459.Certificate())[0]
+ field = cert['tbsCertificate'][field]
+ field = encoder.encode(field)
+ return field
+
+ def is_self_signed(self):
+ """
+ :returns: True if this certificate is self-signed, False otherwise
+ """
+ return self._cert.issuer == self._cert.subject
+
+ @property
+ def subject(self):
+ return self._cert.subject
+
+ @property
+ def subject_bytes(self):
+ return self._subject
+
+ @property
+ def issuer(self):
+ return self._cert.issuer
+
+ @property
+ def issuer_bytes(self):
+ return self._issuer
+
+ @property
+ def serial(self):
+ return self.serial
+
+ @property
+ def public_key_info_bytes(self):
+ return self._cert.public_key().public_bytes(
+ encoding=Encoding.DER, format=PublicFormat.SubjectPublicKeyInfo)
+
+ @property
+ def extended_key_usage(self):
+ """
+ :raises: ValueError: if extended key usage is not present in the cert
+ """
+ ext_key_usage = self._cert.extensions.get_extension_for_oid(
+ crypto_x509.oid.ExtensionOID.EXTENDED_KEY_USAGE).value
+
+ return set(oid.dotted_string for oid in ext_key_usage)
+
+ @property
+ def extended_key_usage_bytes(self):
+ ekurfc = rfc2459.ExtKeyUsageSyntax()
+ eku = self.extended_key_usage
+ for i, oid in enumerate():
+ ekurfc[i] = univ.ObjectIdentifier(oid)
+ ekurfc = encoder.encode(ekurfc)
+ return self.__encode_extension('2.5.29.37', EKU_ANY not in eku, ekurfc)
+
+ def match_hostname(self, hostname):
+ match_cert = {}
+
+ match_cert['subject'] = match_subject = []
+ for rdn in self._cert.subject.rdns:
+ match_rdn = []
+ for ava in rdn:
+ if ava.oid == crypto_x509.oid.NameOID.COMMON_NAME:
+ match_rdn.append(('commonName', ava.value))
+ match_subject.append(match_rdn)
+
+ # TODO: toto by se asi taky dalo nacpat jako metoda teto tridy
+ values = get_san_a_label_dns_names(self._cert)
+ if values:
+ match_cert['subjectAltName'] = match_san = []
+ for value in values:
+ match_san.append(('DNS', value))
+
+ ssl.match_hostname(match_cert, DNSName(hostname).ToASCII())
+
+
def load_pem_x509_certificate(data):
"""
Load an X.509 certificate in PEM format.
@@ -141,7 +257,6 @@ def load_certificate_list(data):
Load a certificate list from a sequence of concatenated PEMs.
Return a list of python-cryptography ``Certificate`` objects.
-
"""
certs = PEM_REGEX.findall(data)
return [load_pem_x509_certificate(cert) for cert in certs]
@@ -197,42 +312,6 @@ def pkcs7_to_pems(data, datatype=PEM):
return result
-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 = normalize_certificate(cert)
- cert = decoder.decode(cert, rfc2459.Certificate())[0]
- field = cert['tbsCertificate'][field]
- field = encoder.encode(field)
- return field
-
-def get_der_subject(cert, datatype=PEM, dbdir=None):
- return _get_der_field(cert, datatype, dbdir, 'subject')
-
-def get_der_issuer(cert, datatype=PEM, dbdir=None):
- return _get_der_field(cert, datatype, dbdir, 'issuer')
-
-def get_der_serial_number(cert, datatype=PEM, dbdir=None):
- return _get_der_field(cert, datatype, dbdir, 'serialNumber')
-
-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):
- 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
-
- return set(oid.dotted_string for oid in eku)
-
-
def make_pem(data):
"""
Convert a raw base64-encoded blob into something that looks like a PE
@@ -302,58 +381,37 @@ def validate_der_x509_certificate(cert):
raise errors.CertificateFormatError(error=str(e))
-def write_certificate(rawcert, filename):
+def write_certificate(cert, filename):
"""
Write the certificate to a file in PEM format.
The cert value can be either DER or PEM-encoded, it will be normalized
to DER regardless, then back out to PEM.
"""
- dercert = normalize_certificate(rawcert)
try:
- fp = open(filename, 'w')
- fp.write(make_pem(base64.b64encode(dercert)))
- fp.close()
+ with open(filename, 'wb') as fp:
+ fp.write(cert)
except (IOError, OSError) as e:
raise errors.FileError(reason=str(e))
-def write_certificate_list(rawcerts, filename):
+
+def write_certificate_list(certs, filename):
"""
Write a list of certificates to a file in PEM format.
The cert values can be either DER or PEM-encoded, they will be normalized
to DER regardless, then back out to PEM.
"""
- dercerts = [normalize_certificate(rawcert) for rawcert in rawcerts]
try:
- with open(filename, 'w') as f:
- for cert in dercerts:
- cert = base64.b64encode(cert)
- cert = make_pem(cert)
- f.write(cert + '\n')
+ with open(filename, 'wb') as f:
+ for cert in certs:
+ f.write(cert)
except (IOError, OSError) as e:
raise errors.FileError(reason=str(e))
-def _encode_extension(oid, critical, value):
- ext = rfc2459.Extension()
- ext['extnID'] = univ.ObjectIdentifier(oid)
- ext['critical'] = univ.Boolean(critical)
- ext['extnValue'] = univ.Any(encoder.encode(univ.OctetString(value)))
- ext = encoder.encode(ext)
- return ext
-
-
-def encode_ext_key_usage(ext_key_usage):
- 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 _PrincipalName(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.NamedType('name-type', univ.Integer().subtype(
@@ -414,10 +472,10 @@ def process_othernames(gns):
"""
for gn in gns:
- if isinstance(gn, cryptography.x509.general_name.OtherName):
+ if isinstance(gn, crypto_x509.general_name.OtherName):
cls = OTHERNAME_CLASS_MAP.get(
gn.type_id.dotted_string,
- cryptography.x509.general_name.OtherName)
+ crypto_x509.general_name.OtherName)
yield cls(gn.type_id, gn.value)
else:
yield gn
@@ -471,13 +529,13 @@ def get_san_general_names(cert):
gns = _pyasn1_get_san_general_names(cert)
GENERAL_NAME_CONSTRUCTORS = {
- 'rfc822Name': lambda x: cryptography.x509.RFC822Name(unicode(x)),
- 'dNSName': lambda x: cryptography.x509.DNSName(unicode(x)),
+ 'rfc822Name': lambda x: crypto_x509.RFC822Name(unicode(x)),
+ 'dNSName': lambda x: crypto_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)),
+ lambda x: crypto_x509.UniformResourceIdentifier(unicode(x)),
'otherName': _pyasn1_to_cryptography_othername,
}
@@ -498,33 +556,33 @@ def _pyasn1_to_cryptography_directoryname(dn):
# Name is CHOICE { RDNSequence } (only one possibility)
for rdn in dn.getComponent():
for ava in rdn:
- attr = cryptography.x509.NameAttribute(
+ attr = crypto_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))
+ return crypto_x509.DirectoryName(crypto_x509.Name(attrs))
def _pyasn1_to_cryptography_registeredid(oid):
- return cryptography.x509.RegisteredID(_pyasn1_to_cryptography_oid(oid))
+ return crypto_x509.RegisteredID(_pyasn1_to_cryptography_oid(oid))
def _pyasn1_to_cryptography_ipaddress(octet_string):
- return cryptography.x509.IPAddress(
+ return crypto_x509.IPAddress(
ipaddress.ip_address(bytes(octet_string)))
def _pyasn1_to_cryptography_othername(on):
- return cryptography.x509.OtherName(
+ return crypto_x509.OtherName(
_pyasn1_to_cryptography_oid(on['type-id']),
bytes(on['value'])
)
def _pyasn1_to_cryptography_oid(oid):
- return cryptography.x509.ObjectIdentifier(str(oid))
+ return crypto_x509.ObjectIdentifier(str(oid))
def get_san_a_label_dns_names(cert):
@@ -577,23 +635,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"))
-
-
-def match_hostname(cert, hostname):
- match_cert = {}
-
- match_cert['subject'] = match_subject = []
- for rdn in cert.subject.rdns:
- match_rdn = []
- for ava in rdn:
- if ava.oid == cryptography.x509.oid.NameOID.COMMON_NAME:
- match_rdn.append(('commonName', ava.value))
- match_subject.append(match_rdn)
-
- values = get_san_a_label_dns_names(cert)
- if values:
- match_cert['subjectAltName'] = match_san = []
- for value in values:
- match_san.append(('DNS', value))
-
- ssl.match_hostname(match_cert, DNSName(hostname).ToASCII())
From 51ae03c067f20989795f7964b8903974605fc8a5 Mon Sep 17 00:00:00 2001
From: Stanislav Laznicka <slazn...@redhat.com>
Date: Fri, 16 Jun 2017 09:48:30 +0200
Subject: [PATCH 3/4] WIP
---
.../certmonger/dogtag-ipa-ca-renew-agent-submit | 9 ++---
install/restart_scripts/renew_ca_cert | 2 +-
ipaclient/plugins/vault.py | 6 ++--
ipalib/install/certstore.py | 3 +-
ipalib/x509.py | 41 ++++++++++++----------
ipaplatform/redhat/tasks.py | 17 +++++----
ipaserver/install/cainstance.py | 8 ++---
ipaserver/install/ipa_cacert_manage.py | 22 +++++-------
ipaserver/plugins/ca.py | 4 +--
ipatests/test_ipaserver/test_ldap.py | 20 +++++------
10 files changed, 62 insertions(+), 70 deletions(-)
diff --git a/install/certmonger/dogtag-ipa-ca-renew-agent-submit b/install/certmonger/dogtag-ipa-ca-renew-agent-submit
index 3d3e791449..8e21333a8b 100755
--- a/install/certmonger/dogtag-ipa-ca-renew-agent-submit
+++ b/install/certmonger/dogtag-ipa-ca-renew-agent-submit
@@ -267,7 +267,7 @@ def store_cert(**kwargs):
if not cert:
return (REJECTED, "New certificate requests not supported")
- dercert = x509.normalize_certificate(cert)
+ dercert = x509.ensure_der_format(cert)
dn = DN(('cn', nickname), ('cn', 'ca_renewal'),
('cn', 'ipa'), ('cn', 'etc'), api.env.basedn)
@@ -392,7 +392,7 @@ def retrieve_cert_continuous(reuse_existing, **kwargs):
"""
old_cert = os.environ.get('CERTMONGER_CERTIFICATE')
if old_cert:
- old_cert = x509.normalize_certificate(old_cert)
+ old_cert = x509.ensure_der_format(old_cert)
result = call_handler(retrieve_or_reuse_cert,
reuse_existing=reuse_existing,
@@ -400,7 +400,7 @@ def retrieve_cert_continuous(reuse_existing, **kwargs):
if result[0] != ISSUED or reuse_existing:
return result
- new_cert = x509.normalize_certificate(result[1])
+ new_cert = x509.ensure_der_format(result[1])
if new_cert == old_cert:
syslog.syslog(syslog.LOG_INFO, "Updated certificate not available")
# No cert available yet, tell certmonger to wait another 8 hours
@@ -431,7 +431,8 @@ def renew_ca_cert(reuse_existing, **kwargs):
cert = os.environ.get('CERTMONGER_CERTIFICATE')
if not cert:
return (REJECTED, "New certificate requests not supported")
- is_self_signed = x509.is_self_signed(cert)
+ cert = x509.load_pem_x509_certificate(cert)
+ is_self_signed = cert.is_self_signed()
operation = os.environ.get('CERTMONGER_OPERATION')
if operation == 'SUBMIT':
diff --git a/install/restart_scripts/renew_ca_cert b/install/restart_scripts/renew_ca_cert
index bb31defc0e..235024bd98 100644
--- a/install/restart_scripts/renew_ca_cert
+++ b/install/restart_scripts/renew_ca_cert
@@ -106,7 +106,7 @@ def _main():
cfg_path, 'subsystem.select', '=')
if config == 'New':
syslog.syslog(syslog.LOG_NOTICE, "Updating CS.cfg")
- if x509.is_self_signed(cert, x509.DER):
+ if cert.is_self_signed():
installutils.set_directive(
cfg_path, 'hierarchy.select', 'Root',
quotes=False, separator='=')
diff --git a/ipaclient/plugins/vault.py b/ipaclient/plugins/vault.py
index 75e8da4fe5..4881abee96 100644
--- a/ipaclient/plugins/vault.py
+++ b/ipaclient/plugins/vault.py
@@ -624,15 +624,15 @@ def forward(self, *args, **options):
response = super(vaultconfig_show, self).forward(*args, **options)
# cache transport certificate
- transport_cert = x509.load_certificate(
- response['result']['transport_cert'], x509.DER)
+ transport_cert = x509.load_der_x509_certificate(
+ response['result']['transport_cert'])
_transport_cert_cache.store_cert(
self.api.env.domain, transport_cert
)
if file:
- with open(file, 'w') as f:
+ with open(file, 'wb') as f:
f.write(response['result']['transport_cert'])
return response
diff --git a/ipalib/install/certstore.py b/ipalib/install/certstore.py
index 0d0902fc1a..7447b2d34f 100644
--- a/ipalib/install/certstore.py
+++ b/ipalib/install/certstore.py
@@ -28,13 +28,14 @@
from ipapython.certdb import get_ca_nickname, TrustFlags
from ipalib import errors, x509
+
def _parse_cert(dercert):
try:
cert = x509.load_der_x509_certificate(dercert)
subject = DN(cert.subject)
issuer = DN(cert.issuer)
serial_number = cert.serial_number
- public_key_info = x509.get_der_public_key_info(dercert, x509.DER)
+ public_key_info = cert.public_key_info_bytes
except (ValueError, PyAsn1Error) as e:
raise ValueError("failed to decode certificate: %s" % e)
diff --git a/ipalib/x509.py b/ipalib/x509.py
index 8319fc1b45..622a6b30f6 100644
--- a/ipalib/x509.py
+++ b/ipalib/x509.py
@@ -124,13 +124,11 @@ def __init__(self, cert, backend=None):
self._subject = self.__get_der_field('subject')
self._issuer = self.__get_der_field('issuer')
- def __str__(self):
- return self._cert.public_bytes(encoding=Encoding.PEM)
+ # assign the methods of the wrapped object
+ self.public_bytes = self._cert.public_bytes
+ self.fingerprint = self._cert.fingerprint
- def __bytes__(self):
- return self._cert.public_bytes(encoding=Encoding.DER)
-
- def __encode_extension(oid, critical, value):
+ def __encode_extension(self, oid, critical, value):
ext = rfc2459.Extension()
# TODO: nemusi se psat tak hrozne, ale v pyasn1 jsou kodeky, ktere
# se o toto postaraji a prevadeji do python-native typu
@@ -142,8 +140,8 @@ def __encode_extension(oid, critical, value):
def __get_der_field(self, field):
cert_bytes = self._cert.tbs_certificate_bytes
- cert = decoder.decode(cert_bytes, rfc2459.Certificate())[0]
- field = cert['tbsCertificate'][field]
+ cert = decoder.decode(cert_bytes, rfc2459.TBSCertificate())[0]
+ field = cert[field]
field = encoder.encode(field)
return field
@@ -170,8 +168,16 @@ def issuer_bytes(self):
return self._issuer
@property
- def serial(self):
- return self.serial
+ def serial_number(self):
+ return self._cert.serial_number
+
+ @property
+ def not_valid_before(self):
+ return self._cert.not_valid_before
+
+ @property
+ def not_valid_after(self):
+ return self._cert.not_valid_after
@property
def public_key_info_bytes(self):
@@ -273,11 +279,11 @@ def load_certificate_list_from_file(filename):
return load_certificate_list(f.read())
-def pkcs7_to_pems(data, datatype=PEM):
+def pkcs7_to_certs(data, datatype=PEM):
"""
Extract certificates from a PKCS #7 object.
- Return a ``list`` of X.509 PEM strings.
+ :returns: a ``list`` of ``_IPACertificate`` objects.
"""
if datatype == PEM:
match = re.match(
@@ -305,8 +311,7 @@ def pkcs7_to_pems(data, datatype=PEM):
for certificate in signed_data['certificates']:
certificate = encoder.encode(certificate)
- certificate = base64.b64encode(certificate)
- certificate = make_pem(certificate)
+ certificate = load_der_x509_certificate(certificate)
result.append(certificate)
return result
@@ -336,8 +341,6 @@ def ensure_der_format(rawcert):
if not rawcert:
return None
- rawcert = strip_header(rawcert)
-
try:
if isinstance(rawcert, bytes):
# base64 must work with utf-8, otherwise it is raw bin certificate
@@ -407,7 +410,7 @@ def write_certificate_list(certs, filename):
try:
with open(filename, 'wb') as f:
for cert in certs:
- f.write(cert)
+ f.write(cert.public_bytes(Encoding.PEM))
except (IOError, OSError) as e:
raise errors.FileError(reason=str(e))
@@ -446,13 +449,13 @@ def _decode_krb5principalname(data):
return name
-class KRB5PrincipalName(cryptography.x509.general_name.OtherName):
+class KRB5PrincipalName(crypto_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):
+class UPN(crypto_x509.general_name.OtherName):
def __init__(self, type_id, value):
super(UPN, self).__init__(type_id, value)
self.name = unicode(
diff --git a/ipaplatform/redhat/tasks.py b/ipaplatform/redhat/tasks.py
index 07efebab97..1a87b15360 100644
--- a/ipaplatform/redhat/tasks.py
+++ b/ipaplatform/redhat/tasks.py
@@ -29,7 +29,6 @@
import pwd
import shutil
import socket
-import base64
import traceback
import errno
@@ -266,10 +265,10 @@ def insert_ca_certs_into_systemwide_ca_store(self, ca_certs):
has_eku = set()
for cert, nickname, trusted, ext_key_usage in ca_certs:
try:
- subject = x509.get_der_subject(cert, x509.DER)
- issuer = x509.get_der_issuer(cert, x509.DER)
- serial_number = x509.get_der_serial_number(cert, x509.DER)
- public_key_info = x509.get_der_public_key_info(cert, x509.DER)
+ subject = cert.subject_bytes
+ issuer = cert.issuer_bytes
+ serial_number = cert.serial_number
+ public_key_info = cert.public_key_info_bytes
except (PyAsn1Error, ValueError, CertificateError) as e:
root_logger.warning(
"Failed to decode certificate \"%s\": %s", nickname, e)
@@ -281,8 +280,8 @@ def insert_ca_certs_into_systemwide_ca_store(self, ca_certs):
serial_number = urllib.parse.quote(serial_number)
public_key_info = urllib.parse.quote(public_key_info)
- cert = base64.b64encode(cert)
- cert = x509.make_pem(cert)
+ # TODO: simple way to get PEM/DER
+ cert = x509.load_der_x509_certificate(cert)
obj = ("[p11-kit-object-v1]\n"
"class: certificate\n"
@@ -302,14 +301,14 @@ def insert_ca_certs_into_systemwide_ca_store(self, ca_certs):
obj += "trusted: true\n"
elif trusted is False:
obj += "x-distrusted: true\n"
- obj += "%s\n\n" % cert
+ obj += "%s\n\n" % cert.public_bytes(x509.Encoding.PEM)
f.write(obj)
if ext_key_usage is not None and public_key_info not in has_eku:
if not ext_key_usage:
ext_key_usage = {x509.EKU_PLACEHOLDER}
try:
- ext_key_usage = x509.encode_ext_key_usage(ext_key_usage)
+ ext_key_usage = cert.extended_key_usage_bytes
except PyAsn1Error as e:
root_logger.warning(
"Failed to encode extended key usage for \"%s\": %s",
diff --git a/ipaserver/install/cainstance.py b/ipaserver/install/cainstance.py
index 531f28f658..25fae01e63 100644
--- a/ipaserver/install/cainstance.py
+++ b/ipaserver/install/cainstance.py
@@ -791,23 +791,21 @@ def __export_ca_chain(self):
data = base64.b64decode(chain)
# Get list of PEM certificates
- certlist = x509.pkcs7_to_pems(data, x509.DER)
+ certlist = x509.pkcs7_to_certs(data, x509.DER)
# We need to append the certs to the existing file, so start by
# reading the file
if ipautil.file_exists(paths.IPA_CA_CRT):
ca_certs = x509.load_certificate_list_from_file(paths.IPA_CA_CRT)
- ca_certs = [cert.public_bytes(serialization.Encoding.PEM)
- for cert in ca_certs]
certlist.extend(ca_certs)
# We have all the certificates in certlist, write them to a PEM file
for path in [paths.IPA_CA_CRT,
paths.KDC_CA_BUNDLE_PEM,
paths.CA_BUNDLE_PEM]:
- with open(path, 'w') as ipaca_pem:
+ with open(path, 'wb') as ipaca_pem:
for cert in certlist:
- ipaca_pem.write(cert)
+ ipaca_pem.write(cert.public_bytes(x509.Encoding.PEM))
ipaca_pem.write('\n')
def __request_ra_certificate(self):
diff --git a/ipaserver/install/ipa_cacert_manage.py b/ipaserver/install/ipa_cacert_manage.py
index e88e8b63ae..9327c5a2c6 100644
--- a/ipaserver/install/ipa_cacert_manage.py
+++ b/ipaserver/install/ipa_cacert_manage.py
@@ -167,7 +167,7 @@ def renew(self):
if options.self_signed is not None:
self_signed = options.self_signed
else:
- self_signed = x509.is_self_signed(cert, x509.DER)
+ self_signed = cert.is_self_signed()
if self_signed:
return self.renew_self_signed(ca)
@@ -208,25 +208,19 @@ def renew_external_step_2(self, ca, old_cert_der):
options = self.options
conn = api.Backend.ldap2
- 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
- )
+ old_cert_obj = x509.load_der_x509_certificate(old_cert_der)
+ old_der_subject = old_cert_obj.subject_bytes
+ old_spki = old_cert_obj.public_key_info_bytes
cert_file, ca_file = installutils.load_external_cert(
options.external_cert_files, DN(old_cert_obj.subject))
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
- )
+ new_cert_der = x509.ensure_der_format(new_cert_data)
+ new_cert_obj = x509.load_der_x509_certificate(new_cert_der)
+ new_der_subject = new_cert_obj.subject_bytes
+ new_spki = new_cert_obj.public_key_info_bytes
if new_cert_obj.subject != old_cert_obj.subject:
raise admintool.ScriptError(
diff --git a/ipaserver/plugins/ca.py b/ipaserver/plugins/ca.py
index 8db6ec549f..6557ce2a09 100644
--- a/ipaserver/plugins/ca.py
+++ b/ipaserver/plugins/ca.py
@@ -181,8 +181,8 @@ def set_certificate_attrs(entry, options, want_cert=True):
if want_chain or full:
pkcs7_der = ca_api.read_ca_chain(ca_id)
- pems = x509.pkcs7_to_pems(pkcs7_der, x509.DER)
- ders = [x509.normalize_certificate(pem) for pem in pems]
+ certs = x509.pkcs7_to_certs(pkcs7_der, x509.DER)
+ ders = [cert.public_bytes(x509.Encoding.DER) for cert in certs]
entry['certificate_chain'] = ders
diff --git a/ipatests/test_ipaserver/test_ldap.py b/ipatests/test_ipaserver/test_ldap.py
index d409efbe44..f2298c7abc 100644
--- a/ipatests/test_ipaserver/test_ldap.py
+++ b/ipatests/test_ipaserver/test_ldap.py
@@ -78,9 +78,8 @@ def test_GSSAPI(self):
self.conn.connect()
entry_attrs = self.conn.get_entry(self.dn, ['usercertificate'])
cert = entry_attrs.get('usercertificate')
- cert = cert[0]
- serial = x509.load_certificate(cert, x509.DER).serial_number
- assert serial is not None
+ cert = x509.load_der_x509_certificate(cert[0])
+ assert cert.serial_number is not None
def test_simple(self):
"""
@@ -96,9 +95,8 @@ def test_simple(self):
self.conn.connect(bind_dn=DN(('cn', 'directory manager')), bind_pw=dm_password)
entry_attrs = self.conn.get_entry(self.dn, ['usercertificate'])
cert = entry_attrs.get('usercertificate')
- cert = cert[0]
- serial = x509.load_certificate(cert, x509.DER).serial_number
- assert serial is not None
+ cert = x509.load_der_x509_certificate(cert[0])
+ assert cert.serial_number is not None
def test_Backend(self):
"""
@@ -123,9 +121,8 @@ def test_Backend(self):
result = myapi.Command['service_show']('ldap/%s@%s' % (api.env.host, api.env.realm,))
entry_attrs = result['result']
cert = entry_attrs.get('usercertificate')
- cert = cert[0]
- serial = x509.load_certificate(cert, x509.DER).serial_number
- assert serial is not None
+ cert = x509.load_der_x509_certificate(cert[0])
+ assert cert.serial_number is not None
def test_autobind(self):
"""
@@ -139,9 +136,8 @@ def test_autobind(self):
raise nose.SkipTest("Only executed as root")
entry_attrs = self.conn.get_entry(self.dn, ['usercertificate'])
cert = entry_attrs.get('usercertificate')
- cert = cert[0]
- serial = x509.load_certificate(cert, x509.DER).serial_number
- assert serial is not None
+ cert = x509.load_der_x509_certificate(cert[0])
+ assert cert.serial_number is not None
@pytest.mark.tier0
From 778611ff3547eb39cbf0efcdedde2ea5d70f08ef Mon Sep 17 00:00:00 2001
From: Stanislav Laznicka <slazn...@redhat.com>
Date: Fri, 16 Jun 2017 18:24:04 +0200
Subject: [PATCH 4/4] WIP2
---
install/restart_scripts/renew_ca_cert | 4 +-
ipaclient/install/client.py | 4 --
ipalib/install/certstore.py | 48 ++++++++------
ipalib/x509.py | 24 ++++---
ipaplatform/redhat/tasks.py | 7 +--
ipapython/certdb.py | 97 +++++++++++++----------------
ipaserver/install/ca.py | 2 +-
ipaserver/install/certs.py | 38 ++++-------
ipaserver/install/dogtaginstance.py | 7 ++-
ipaserver/install/dsinstance.py | 10 +--
ipaserver/install/httpinstance.py | 6 +-
ipaserver/install/installutils.py | 12 ++--
ipaserver/install/ipa_cacert_manage.py | 22 +++----
ipaserver/install/ipa_server_certinstall.py | 8 +--
ipaserver/install/plugins/upload_cacrt.py | 9 +--
ipaserver/install/server/upgrade.py | 2 +-
ipaserver/install/service.py | 7 ++-
17 files changed, 141 insertions(+), 166 deletions(-)
diff --git a/install/restart_scripts/renew_ca_cert b/install/restart_scripts/renew_ca_cert
index 235024bd98..30716531d2 100644
--- a/install/restart_scripts/renew_ca_cert
+++ b/install/restart_scripts/renew_ca_cert
@@ -29,7 +29,7 @@ import traceback
from ipalib.install import certstore
from ipapython import ipautil
-from ipalib import api, errors, x509
+from ipalib import api, errors
from ipalib.install.kinit import kinit_keytab
from ipaserver.install import certs, cainstance, installutils
from ipaserver.plugins.ldap2 import ldap2
@@ -64,7 +64,7 @@ def _main():
# Fetch the new certificate
db = certs.CertDB(api.env.realm, nssdir=paths.PKI_TOMCAT_ALIAS_DIR)
- cert = db.get_cert_from_db(nickname, pem=False)
+ cert = db.get_cert_from_db(nickname)
if not cert:
syslog.syslog(syslog.LOG_ERR, 'No certificate %s found.' % nickname)
sys.exit(1)
diff --git a/ipaclient/install/client.py b/ipaclient/install/client.py
index 30cecc882e..d65496c1d9 100644
--- a/ipaclient/install/client.py
+++ b/ipaclient/install/client.py
@@ -2677,10 +2677,6 @@ def _install(options):
# Add CA certs to a temporary NSS database
ca_certs = x509.load_certificate_list_from_file(paths.IPA_CA_CRT)
- ca_certs = [
- cert.public_bytes(serialization.Encoding.DER)
- for cert in ca_certs
- ]
try:
tmp_db.create_db()
diff --git a/ipalib/install/certstore.py b/ipalib/install/certstore.py
index 7447b2d34f..3166713220 100644
--- a/ipalib/install/certstore.py
+++ b/ipalib/install/certstore.py
@@ -29,9 +29,8 @@
from ipalib import errors, x509
-def _parse_cert(dercert):
+def _parse_cert(cert):
try:
- cert = x509.load_der_x509_certificate(dercert)
subject = DN(cert.subject)
issuer = DN(cert.issuer)
serial_number = cert.serial_number
@@ -46,15 +45,15 @@ def _parse_cert(dercert):
return subject, issuer_serial, public_key_info
-def init_ca_entry(entry, dercert, nickname, trusted, ext_key_usage):
+def init_ca_entry(entry, cert, nickname, trusted, ext_key_usage):
"""
Initialize certificate store entry for a CA certificate.
"""
- subject, issuer_serial, public_key = _parse_cert(dercert)
+ subject, issuer_serial, public_key = _parse_cert(cert)
if ext_key_usage is not None:
try:
- cert_eku = x509.get_ext_key_usage(dercert, x509.DER)
+ cert_eku = cert.extended_key_usage
except ValueError as e:
raise ValueError("failed to decode certificate: %s" % e)
if cert_eku is not None:
@@ -69,7 +68,7 @@ def init_ca_entry(entry, dercert, nickname, trusted, ext_key_usage):
entry['ipaCertSubject'] = [subject]
entry['ipaCertIssuerSerial'] = [issuer_serial]
entry['ipaPublicKey'] = [public_key]
- entry['cACertificate;binary'] = [dercert]
+ entry['cACertificate;binary'] = [cert.public_bytes(x509.Encoding.DER)]
if trusted is not None:
entry['ipaKeyTrust'] = ['trusted' if trusted else 'distrusted']
@@ -80,11 +79,12 @@ def init_ca_entry(entry, dercert, nickname, trusted, ext_key_usage):
entry['ipaKeyExtUsage'] = ext_key_usage
-def update_compat_ca(ldap, base_dn, dercert):
+def update_compat_ca(ldap, base_dn, cert):
"""
Update the CA certificate in cn=CAcert,cn=ipa,cn=etc,SUFFIX.
"""
dn = DN(('cn', 'CAcert'), ('cn', 'ipa'), ('cn', 'etc'), base_dn)
+ dercert = cert.public_bytes(x509.Encoding.DER)
try:
entry = ldap.get_entry(dn, attrs_list=['cACertificate;binary'])
entry.single_value['cACertificate;binary'] = dercert
@@ -153,12 +153,12 @@ def add_ca_cert(ldap, base_dn, dercert, nickname, trusted=None,
clean_old_config(ldap, base_dn, dn, config_ipa, config_compat)
-def update_ca_cert(ldap, base_dn, dercert, trusted=None, ext_key_usage=None,
+def update_ca_cert(ldap, base_dn, cert, trusted=None, ext_key_usage=None,
config_ipa=False, config_compat=False):
"""
Update existing entry for a CA certificate in the certificate store.
"""
- subject, issuer_serial, public_key = _parse_cert(dercert)
+ subject, issuer_serial, public_key = _parse_cert(cert)
filter = ldap.make_filter({'ipaCertSubject': subject})
result, _truncated = ldap.find_entries(
@@ -173,7 +173,8 @@ def update_ca_cert(ldap, base_dn, dercert, trusted=None, ext_key_usage=None,
for old_cert in entry['cACertificate;binary']:
# Check if we are adding a new cert
- if old_cert == dercert:
+ # TODO: comparison
+ if old_cert == cert:
break
else:
# We are adding a new cert, validate it
@@ -182,7 +183,8 @@ def update_ca_cert(ldap, base_dn, dercert, trusted=None, ext_key_usage=None,
if entry.single_value['ipaPublicKey'] != public_key:
raise ValueError("subject public key info mismatch")
entry['ipaCertIssuerSerial'].append(issuer_serial)
- entry['cACertificate;binary'].append(dercert)
+ entry['cACertificate;binary'].append(
+ cert.public_bytes(x509.Encoding.DER))
# Update key trust
if trusted is not None:
@@ -218,22 +220,24 @@ def update_ca_cert(ldap, base_dn, dercert, trusted=None, ext_key_usage=None,
entry.setdefault('ipaConfigString', []).append('compatCA')
if is_compat or config_compat:
- update_compat_ca(ldap, base_dn, dercert)
+ update_compat_ca(ldap, base_dn, cert)
ldap.update_entry(entry)
clean_old_config(ldap, base_dn, dn, config_ipa, config_compat)
-def put_ca_cert(ldap, base_dn, dercert, nickname, trusted=None,
+def put_ca_cert(ldap, base_dn, cert, nickname, trusted=None,
ext_key_usage=None, config_ipa=False, config_compat=False):
"""
Add or update entry for a CA certificate in the certificate store.
+
+ :param cert: IPACertificate
"""
try:
- update_ca_cert(ldap, base_dn, dercert, trusted, ext_key_usage,
+ update_ca_cert(ldap, base_dn, cert, trusted, ext_key_usage,
config_ipa=config_ipa, config_compat=config_compat)
except errors.NotFound:
- add_ca_cert(ldap, base_dn, dercert, nickname, trusted, ext_key_usage,
+ add_ca_cert(ldap, base_dn, cert, nickname, trusted, ext_key_usage,
config_ipa=config_ipa, config_compat=config_compat)
except errors.EmptyModlist:
pass
@@ -307,11 +311,12 @@ def get_ca_certs(ldap, base_dn, compat_realm, compat_ipa_ca,
for cert in entry.get('cACertificate;binary', []):
try:
- _parse_cert(cert)
+ cert_obj = x509.load_der_x509_certificate(cert)
+ _parse_cert(cert_obj)
except ValueError:
certs = []
break
- certs.append((cert, nickname, trusted, ext_key_usage))
+ certs.append((cert_obj, nickname, trusted, ext_key_usage))
except errors.NotFound:
try:
ldap.get_entry(container_dn, [''])
@@ -320,7 +325,8 @@ def get_ca_certs(ldap, base_dn, compat_realm, compat_ipa_ca,
dn = DN(('cn', 'CAcert'), config_dn)
entry = ldap.get_entry(dn, ['cACertificate;binary'])
- cert = entry.single_value['cACertificate;binary']
+ cert = x509.load_der_x509_certificate(
+ entry.single_value['cACertificate;binary'])
try:
subject, _issuer_serial, _public_key_info = _parse_cert(cert)
except ValueError:
@@ -355,16 +361,18 @@ def key_policy_to_trust_flags(trusted, ca, ext_key_usage):
return TrustFlags(False, trusted, ca, ext_key_usage)
-def put_ca_cert_nss(ldap, base_dn, dercert, nickname, trust_flags,
+def put_ca_cert_nss(ldap, base_dn, cert, nickname, trust_flags,
config_ipa=False, config_compat=False):
"""
Add or update entry for a CA certificate in the certificate store.
+
+ :param cert: IPACertificate
"""
trusted, ca, ext_key_usage = trust_flags_to_key_policy(trust_flags)
if ca is False:
raise ValueError("must be CA certificate")
- put_ca_cert(ldap, base_dn, dercert, nickname, trusted, ext_key_usage,
+ put_ca_cert(ldap, base_dn, cert, nickname, trusted, ext_key_usage,
config_ipa, config_compat)
diff --git a/ipalib/x509.py b/ipalib/x509.py
index 622a6b30f6..3e383172ae 100644
--- a/ipalib/x509.py
+++ b/ipalib/x509.py
@@ -105,7 +105,7 @@ def strip_header(pem):
return pem
-class _IPACertificate(object):
+class IPACertificate(object):
"""
A proxy class wrapping a python-cryptography certificate representation for
FreeIPA purposes
@@ -180,25 +180,29 @@ def not_valid_after(self):
return self._cert.not_valid_after
@property
+ def extensions(self):
+ return self._cert.extensions
+
+ @property
def public_key_info_bytes(self):
return self._cert.public_key().public_bytes(
encoding=Encoding.DER, format=PublicFormat.SubjectPublicKeyInfo)
@property
def extended_key_usage(self):
- """
- :raises: ValueError: if extended key usage is not present in the cert
- """
- ext_key_usage = self._cert.extensions.get_extension_for_oid(
- crypto_x509.oid.ExtensionOID.EXTENDED_KEY_USAGE).value
+ try:
+ ext_key_usage = self._cert.extensions.get_extension_for_oid(
+ crypto_x509.oid.ExtensionOID.EXTENDED_KEY_USAGE).value
+ except crypto_x509.ExtensionNotFound:
+ return None
return set(oid.dotted_string for oid in ext_key_usage)
@property
def extended_key_usage_bytes(self):
ekurfc = rfc2459.ExtKeyUsageSyntax()
- eku = self.extended_key_usage
- for i, oid in enumerate():
+ eku = self.extended_key_usage or {EKU_PLACEHOLDER}
+ for i, oid in enumerate(eku):
ekurfc[i] = univ.ObjectIdentifier(oid)
ekurfc = encoder.encode(ekurfc)
return self.__encode_extension('2.5.29.37', EKU_ANY not in eku, ekurfc)
@@ -231,7 +235,7 @@ def load_pem_x509_certificate(data):
:return: a python-cryptography ``CertificateSigningRequest`` object.
:raises: ``ValueError`` if unable to load the certificate.
"""
- return _IPACertificate(
+ return IPACertificate(
crypto_x509.load_pem_x509_certificate(data, backend=default_backend())
)
@@ -243,7 +247,7 @@ def load_der_x509_certificate(data):
:return: a python-cryptography ``CertificateSigningRequest`` object.
:raises: ``ValueError`` if unable to load the certificate.
"""
- return _IPACertificate(
+ return IPACertificate(
crypto_x509.load_der_x509_certificate(data, backend=default_backend())
)
diff --git a/ipaplatform/redhat/tasks.py b/ipaplatform/redhat/tasks.py
index 1a87b15360..feb2954979 100644
--- a/ipaplatform/redhat/tasks.py
+++ b/ipaplatform/redhat/tasks.py
@@ -277,12 +277,9 @@ def insert_ca_certs_into_systemwide_ca_store(self, ca_certs):
label = urllib.parse.quote(nickname)
subject = urllib.parse.quote(subject)
issuer = urllib.parse.quote(issuer)
- serial_number = urllib.parse.quote(serial_number)
+ serial_number = urllib.parse.quote(str(serial_number))
public_key_info = urllib.parse.quote(public_key_info)
- # TODO: simple way to get PEM/DER
- cert = x509.load_der_x509_certificate(cert)
-
obj = ("[p11-kit-object-v1]\n"
"class: certificate\n"
"certificate-type: x-509\n"
@@ -305,8 +302,6 @@ def insert_ca_certs_into_systemwide_ca_store(self, ca_certs):
f.write(obj)
if ext_key_usage is not None and public_key_info not in has_eku:
- if not ext_key_usage:
- ext_key_usage = {x509.EKU_PLACEHOLDER}
try:
ext_key_usage = cert.extended_key_usage_bytes
except PyAsn1Error as e:
diff --git a/ipapython/certdb.py b/ipapython/certdb.py
index b9e364cdca..a1b7c89220 100644
--- a/ipapython/certdb.py
+++ b/ipapython/certdb.py
@@ -26,9 +26,7 @@
import tempfile
from tempfile import NamedTemporaryFile
import shutil
-import base64
-from cryptography.hazmat.primitives import serialization
import cryptography.x509
from ipapython.dn import DN
@@ -91,7 +89,7 @@ def find_cert_from_txt(cert, start=0):
trailing text, pull out just the certificate part. This will return
the FIRST cert in a stream of data.
- Returns a tuple (certificate, last position in cert)
+ :returns: a tuple (IPACertificate, last position in cert)
"""
s = cert.find('-----BEGIN CERTIFICATE-----', start)
e = cert.find('-----END CERTIFICATE-----', s)
@@ -101,7 +99,7 @@ def find_cert_from_txt(cert, start=0):
if s < 0 or e < 0:
raise RuntimeError("Unable to find certificate")
- cert = cert[s:e]
+ cert = x509.load_pem_x509_certificate(cert[s:e])
return (cert, e)
@@ -182,9 +180,9 @@ def unparse_trust_flags(trust_flags):
def verify_kdc_cert_validity(kdc_cert, ca_certs, realm):
- pem_kdc_cert = kdc_cert.public_bytes(serialization.Encoding.PEM)
+ pem_kdc_cert = kdc_cert.public_bytes(x509.Encoding.PEM)
pem_ca_certs = '\n'.join(
- cert.public_bytes(serialization.Encoding.PEM) for cert in ca_certs)
+ cert.public_bytes(x509.Encoding.PEM) for cert in ca_certs)
with NamedTemporaryFile() as kdc_file, NamedTemporaryFile() as ca_file:
kdc_file.write(pem_kdc_cert)
@@ -454,7 +452,7 @@ def import_files(self, files, import_keys=False, key_password=None,
if label in ('CERTIFICATE', 'X509 CERTIFICATE',
'X.509 CERTIFICATE'):
try:
- x509.load_pem_x509_certificate(match.group(2))
+ cert = x509.load_pem_x509_certificate(body)
except ValueError as e:
if label != 'CERTIFICATE':
root_logger.warning(
@@ -462,13 +460,13 @@ def import_files(self, files, import_keys=False, key_password=None,
filename, line, e)
continue
else:
- extracted_certs.append(body)
+ extracted_certs.append(cert)
loaded = True
continue
if label in ('PKCS7', 'PKCS #7 SIGNED DATA', 'CERTIFICATE'):
try:
- certs = x509.pkcs7_to_pems(body)
+ certs = x509.pkcs7_to_certs(body)
except ipautil.CalledProcessError as e:
if label == 'CERTIFICATE':
root_logger.warning(
@@ -525,12 +523,11 @@ def import_files(self, files, import_keys=False, key_password=None,
# Try to load the file as DER certificate
try:
- x509.load_der_x509_certificate(data)
+ cert = x509.load_der_x509_certificate(data)
except ValueError:
pass
else:
- data = x509.make_pem(base64.b64encode(data))
- extracted_certs.append(data)
+ extracted_certs.append(cert)
continue
# Try to import the file as PKCS#12 file
@@ -570,34 +567,33 @@ def import_files(self, files, import_keys=False, key_password=None,
raise RuntimeError(
"No server certificates found in %s" % (', '.join(files)))
- for cert_pem in extracted_certs:
- cert = x509.load_pem_x509_certificate(cert_pem)
+ for cert in extracted_certs:
nickname = str(DN(cert.subject))
- data = cert.public_bytes(serialization.Encoding.DER)
- self.add_cert(data, nickname, EMPTY_TRUST_FLAGS)
+ self.add_cert(cert, nickname, EMPTY_TRUST_FLAGS)
if extracted_key:
- in_file = ipautil.write_tmp_file(
- '\n'.join(extracted_certs) + '\n' + extracted_key)
- out_file = tempfile.NamedTemporaryFile()
- out_password = ipautil.ipa_generate_password()
- out_pwdfile = ipautil.write_tmp_file(out_password)
- args = [
- OPENSSL, 'pkcs12',
- '-export',
- '-in', in_file.name,
- '-out', out_file.name,
- '-passin', 'file:' + self.pwd_file,
- '-passout', 'file:' + out_pwdfile.name,
- ]
- try:
- ipautil.run(args)
- except ipautil.CalledProcessError as e:
- raise RuntimeError(
- "No matching certificate found for private key from %s" %
- key_file)
+ with tempfile.NamedTemporaryFile(), tempfile.NamedTemporaryFile() \
+ as (in_file, out_file):
+ x509.write_certificate_list(extracted_certs, in_file.name)
+ in_file.write('\n{pkey}'.format(pkey=extracted_key))
+ out_password = ipautil.ipa_generate_password()
+ out_pwdfile = ipautil.write_tmp_file(out_password)
+ args = [
+ OPENSSL, 'pkcs12',
+ '-export',
+ '-in', in_file.name,
+ '-out', out_file.name,
+ '-passin', 'file:' + self.pwd_file,
+ '-passout', 'file:' + out_pwdfile.name,
+ ]
+ try:
+ ipautil.run(args)
+ except ipautil.CalledProcessError as e:
+ raise RuntimeError(
+ "No matching certificate found for private key from "
+ "%s" % key_file)
- self.import_pkcs12(out_file.name, out_password)
+ self.import_pkcs12(out_file.name, out_password)
def trust_root_cert(self, root_nickname, trust_flags):
if root_nickname[:7] == "Builtin":
@@ -613,17 +609,13 @@ def trust_root_cert(self, root_nickname, trust_flags):
raise RuntimeError(
"Setting trust on %s failed" % root_nickname)
- def get_cert(self, nickname, pem=False):
+ def get_cert(self, nickname):
args = ['-L', '-n', nickname, '-a']
try:
result = self.run_certutil(args, capture_output=True)
except ipautil.CalledProcessError:
raise RuntimeError("Failed to get %s" % nickname)
- cert = result.output
- if not pem:
- cert, _start = find_cert_from_txt(cert, start=0)
- cert = x509.strip_header(cert)
- cert = base64.b64decode(cert)
+ cert, _start = find_cert_from_txt(result.output, start=0)
return cert
def has_nickname(self, nickname):
@@ -637,9 +629,9 @@ def has_nickname(self, nickname):
def export_pem_cert(self, nickname, location):
"""Export the given cert to PEM file in the given location"""
- cert = self.get_cert(nickname, pem=True)
- with open(location, "w+") as fd:
- fd.write(cert)
+ cert = self.get_cert(nickname)
+ with open(location, "wb+") as fd:
+ fd.write(cert.public_bytes(x509.Encoding.PEM))
os.chmod(location, 0o444)
def import_pem_cert(self, nickname, flags, location):
@@ -656,7 +648,7 @@ def import_pem_cert(self, nickname, flags, location):
)
cert, st = find_cert_from_txt(certs)
- self.add_cert(cert, nickname, flags, pem=True)
+ self.add_cert(cert, nickname, flags)
try:
find_cert_from_txt(certs, st)
@@ -666,12 +658,10 @@ def import_pem_cert(self, nickname, flags, location):
raise ValueError('%s contains more than one certificate' %
location)
- def add_cert(self, cert, nick, flags, pem=False):
+ def add_cert(self, cert, nick, flags):
flags = unparse_trust_flags(flags)
- args = ["-A", "-n", nick, "-t", flags]
- if pem:
- args.append("-a")
- self.run_certutil(args, stdin=cert)
+ args = ["-A", "-n", nick, "-t", flags, '-a']
+ self.run_certutil(args, stdin=cert.public_bytes(x509.Encoding.PEM))
def delete_cert(self, nick):
self.run_certutil(["-D", "-n", nick])
@@ -682,7 +672,6 @@ def verify_server_cert_validity(self, nickname, hostname):
Raises a ValueError if the certificate is invalid.
"""
cert = self.get_cert(nickname)
- cert = x509.load_der_x509_certificate(cert)
try:
self.run_certutil(['-V', '-n', nickname, '-u', 'V'],
@@ -696,13 +685,12 @@ def verify_server_cert_validity(self, nickname, hostname):
raise ValueError(msg)
try:
- x509.match_hostname(cert, hostname)
+ cert.match_hostname(hostname)
except ValueError:
raise ValueError('invalid for server %s' % hostname)
def verify_ca_cert_validity(self, nickname):
cert = self.get_cert(nickname)
- cert = x509.load_der_x509_certificate(cert)
if not cert.subject:
raise ValueError("has empty subject")
@@ -736,6 +724,5 @@ def verify_ca_cert_validity(self, nickname):
def verify_kdc_cert_validity(self, nickname, realm):
nicknames = self.get_trust_chain(nickname)
certs = [self.get_cert(nickname) for nickname in nicknames]
- certs = [x509.load_der_x509_certificate(cert) for cert in certs]
verify_kdc_cert_validity(certs[-1], certs[:-1], realm)
diff --git a/ipaserver/install/ca.py b/ipaserver/install/ca.py
index a88bee7e89..1a2c9d15a5 100644
--- a/ipaserver/install/ca.py
+++ b/ipaserver/install/ca.py
@@ -318,7 +318,7 @@ def install_step_1(standalone, replica_config, options):
subject_base=subject_base)
dsdb = certs.CertDB(
realm_name, nssdir=dirname, subject_base=subject_base)
- cacert = cadb.get_cert_from_db('caSigningCert cert-pki-ca', pem=False)
+ cacert = cadb.get_cert_from_db('caSigningCert cert-pki-ca')
nickname = certdb.get_ca_nickname(realm_name)
trust_flags = certdb.IPA_CA_TRUST_FLAGS
dsdb.add_cert(cacert, nickname, trust_flags)
diff --git a/ipaserver/install/certs.py b/ipaserver/install/certs.py
index 5b1b6ec22a..e10fa9157a 100644
--- a/ipaserver/install/certs.py
+++ b/ipaserver/install/certs.py
@@ -56,9 +56,7 @@ def get_cert_nickname(cert):
representation of the first RDN in the subject and subject_dn
is a DN object.
"""
- cert_obj = x509.load_pem_x509_certificate(cert)
- dn = DN(cert_obj.subject)
-
+ dn = DN(cert.subject)
return (str(dn[0]), dn)
@@ -321,30 +319,20 @@ def load_cacert(self, cacert_fname, trust_flags):
nick = get_ca_nickname(self.realm)
else:
nick = str(subject_dn)
- self.nssdb.add_cert(cert, nick, trust_flags, pem=True)
+ self.nssdb.add_cert(cert, nick, trust_flags)
except RuntimeError:
break
- def get_cert_from_db(self, nickname, pem=True):
+ def get_cert_from_db(self, nickname):
"""
Retrieve a certificate from the current NSS database for nickname.
-
- pem controls whether the value returned PEM or DER-encoded. The
- default is the data straight from certutil -a.
"""
try:
args = ["-L", "-n", nickname, "-a"]
result = self.run_certutil(args, capture_output=True)
- cert = result.output
- if pem:
- return cert
- else:
- cert, _start = find_cert_from_txt(cert, start=0)
- cert = x509.strip_header(cert)
- dercert = base64.b64decode(cert)
- return dercert
+ return x509.load_pem_x509_certificate(result.output)
except ipautil.CalledProcessError:
- return ''
+ return None
def track_server_cert(self, nickname, principal, password_file=None, command=None):
"""
@@ -359,8 +347,7 @@ def track_server_cert(self, nickname, principal, password_file=None, command=Non
return
cert = self.get_cert_from_db(nickname)
- cert_obj = x509.load_pem_x509_certificate(cert)
- subject = str(DN(cert_obj.subject))
+ subject = str(DN(cert.subject))
certmonger.add_principal(request_id, principal)
certmonger.add_subject(request_id, subject)
@@ -388,8 +375,10 @@ def create_server_cert(self, nickname, hostname, subject=None):
try:
self.issue_server_cert(self.certreq_fname, self.certder_fname)
self.import_cert(self.certder_fname, nickname)
+
with open(self.certder_fname, "r") as f:
dercert = f.read()
+ return x509.load_der_x509_certificate(dercert)
finally:
for fname in (self.certreq_fname, self.certder_fname):
try:
@@ -397,8 +386,6 @@ def create_server_cert(self, nickname, hostname, subject=None):
except OSError:
pass
- return dercert
-
def request_cert(
self, subject, certtype="rsa", keysize="2048",
san_dnsnames=None):
@@ -515,8 +502,8 @@ def issue_signing_cert(self, certreq_fname, cert_fname):
with open(cert_fname, "w") as f:
f.write(cert)
- def add_cert(self, cert, nick, flags, pem=False):
- self.nssdb.add_cert(cert, nick, flags, pem)
+ def add_cert(self, cert, nick, flags):
+ self.nssdb.add_cert(cert, nick, flags)
def import_cert(self, cert_fname, nickname):
"""
@@ -589,9 +576,8 @@ def create_from_cacert(self):
newca, _st = find_cert_from_txt(newca)
cacert = self.get_cert_from_db(self.cacert_name)
- if cacert != '':
- cacert, _st = find_cert_from_txt(cacert)
+ # TODO: we may not necessary be able to compare it like this
if newca == cacert:
return
@@ -644,7 +630,7 @@ def init_from_pkcs12(self, pkcs12_fname, pkcs12_passwd,
cert, st = find_cert_from_txt(certs, st)
except RuntimeError:
break
- self.add_cert(cert, 'CA %s' % num, EMPTY_TRUST_FLAGS, pem=True)
+ self.add_cert(cert, 'CA %s' % num, EMPTY_TRUST_FLAGS)
num += 1
# We only handle one server cert
diff --git a/ipaserver/install/dogtaginstance.py b/ipaserver/install/dogtaginstance.py
index 1fdc3e50a4..194bab7020 100644
--- a/ipaserver/install/dogtaginstance.py
+++ b/ipaserver/install/dogtaginstance.py
@@ -27,7 +27,7 @@
from pki.client import PKIConnection
import pki.system
-from ipalib import api, errors
+from ipalib import api, errors, x509
from ipalib.install import certmonger
from ipaplatform import services
from ipaplatform.constants import constants
@@ -341,7 +341,7 @@ def update_cert_cs_cfg(directive, cert, cs_cfg):
needs to get the new certificate as well.
``directive`` is the directive to update in CS.cfg
- cert is a DER-encoded certificate.
+ cert is IPACertificate.
cs_cfg is the path to the CS.cfg file
"""
@@ -349,7 +349,8 @@ def update_cert_cs_cfg(directive, cert, cs_cfg):
installutils.set_directive(
cs_cfg,
directive,
- base64.b64encode(cert),
+ # this did not contain the headers previously
+ cert.public_bytes(x509.Encoding.PEM),
quotes=False,
separator='=')
diff --git a/ipaserver/install/dsinstance.py b/ipaserver/install/dsinstance.py
index 39248edb28..a66b30e18b 100644
--- a/ipaserver/install/dsinstance.py
+++ b/ipaserver/install/dsinstance.py
@@ -232,7 +232,7 @@ def __init__(self, realm_name=None, domain_name=None, fstore=None,
self.pkcs12_info = None
self.cacert_name = None
self.ca_is_configured = True
- self.dercert = None
+ self.cert = None
self.idstart = None
self.idmax = None
self.ca_subject = None
@@ -787,7 +787,7 @@ def __enable_ssl(self):
# We only handle one server cert
self.nickname = server_certs[0][0]
- self.dercert = dsdb.get_cert_from_db(self.nickname, pem=False)
+ self.cert = dsdb.get_cert_from_db(self.nickname)
if self.ca_is_configured:
dsdb.track_server_cert(
@@ -830,7 +830,7 @@ def __enable_ssl(self):
api.Backend.ldap2.disconnect()
api.Backend.ldap2.connect()
- self.dercert = dsdb.get_cert_from_db(self.nickname, pem=False)
+ self.cert = dsdb.get_cert_from_db(self.nickname)
if prev_helper is not None:
self.add_cert_to_service()
@@ -884,12 +884,12 @@ def __upload_ca_cert(self):
nicknames = dsdb.find_root_cert(self.cacert_name)[:-1]
for nickname in nicknames:
- cert = dsdb.get_cert_from_db(nickname, pem=False)
+ cert = dsdb.get_cert_from_db(nickname)
certstore.put_ca_cert_nss(conn, self.suffix, cert, nickname,
trust_flags[nickname])
nickname = self.cacert_name
- cert = dsdb.get_cert_from_db(nickname, pem=False)
+ cert = dsdb.get_cert_from_db(nickname)
cacert_flags = trust_flags[nickname]
if self.setup_pkinit:
cacert_flags = TrustFlags(
diff --git a/ipaserver/install/httpinstance.py b/ipaserver/install/httpinstance.py
index 555c82213c..f65a6388e6 100644
--- a/ipaserver/install/httpinstance.py
+++ b/ipaserver/install/httpinstance.py
@@ -133,7 +133,7 @@ def create_instance(self, realm, fqdn, domain_name, dm_password=None,
self.dm_password = dm_password
self.suffix = ipautil.realm_to_suffix(self.realm)
self.pkcs12_info = pkcs12_info
- self.dercert = None
+ self.cert = None
self.subject_base = subject_base
self.sub_dict = dict(
REALM=realm,
@@ -397,7 +397,7 @@ def __setup_ssl(self):
nickname = server_certs[0][0]
if nickname == 'ipaCert':
nickname = server_certs[1][0]
- self.dercert = db.get_cert_from_db(nickname, pem=False)
+ self.cert = db.get_cert_from_db(nickname)
if self.ca_is_configured:
db.track_server_cert(nickname, self.principal, db.passwd_fname, 'restart_httpd')
@@ -436,7 +436,7 @@ def __setup_ssl(self):
if prev_helper is not None:
certmonger.modify_ca_helper('IPA', prev_helper)
- self.dercert = db.get_cert_from_db(self.cert_nickname, pem=False)
+ self.cert = db.get_cert_from_db(self.cert_nickname)
if prev_helper is not None:
self.add_cert_to_service()
diff --git a/ipaserver/install/installutils.py b/ipaserver/install/installutils.py
index e9cc2a78a9..bb746d3be3 100644
--- a/ipaserver/install/installutils.py
+++ b/ipaserver/install/installutils.py
@@ -1180,11 +1180,9 @@ def load_external_cert(files, ca_subject):
ca_nickname = None
cache = {}
for nickname, _trust_flags in nssdb.list_certs():
- cert = nssdb.get_cert(nickname, pem=True)
-
- cert_obj = x509.load_pem_x509_certificate(cert)
- subject = DN(cert_obj.subject)
- issuer = DN(cert_obj.issuer)
+ cert = nssdb.get_cert(nickname)
+ subject = DN(cert.subject)
+ issuer = DN(cert.issuer)
cache[nickname] = (cert, subject, issuer)
if subject == ca_subject:
@@ -1217,11 +1215,11 @@ def load_external_cert(files, ca_subject):
(subject, ", ".join(files), e))
cert_file = tempfile.NamedTemporaryFile()
- cert_file.write(ca_cert_chain[0] + '\n')
+ cert_file.write(ca_cert_chain[0].public_bytes(x509.Encoding.PEM) + b'\n')
cert_file.flush()
ca_file = tempfile.NamedTemporaryFile()
- ca_file.write('\n'.join(ca_cert_chain[1:]) + '\n')
+ x509.write_certificate_list(ca_cert_chain[1:], ca_file.name)
ca_file.flush()
return cert_file, ca_file
diff --git a/ipaserver/install/ipa_cacert_manage.py b/ipaserver/install/ipa_cacert_manage.py
index 9327c5a2c6..a82897c662 100644
--- a/ipaserver/install/ipa_cacert_manage.py
+++ b/ipaserver/install/ipa_cacert_manage.py
@@ -158,7 +158,7 @@ def renew(self):
"Found certmonger request id %r", self.request_id)
db = certs.CertDB(api.env.realm, nssdir=paths.PKI_TOMCAT_ALIAS_DIR)
- cert = db.get_cert_from_db(self.cert_nickname, pem=False)
+ cert = db.get_cert_from_db(self.cert_nickname)
options = self.options
if options.external_cert_files:
@@ -202,32 +202,30 @@ 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_der):
+ def renew_external_step_2(self, ca, old_cert):
print("Importing the renewed CA certificate, please wait")
options = self.options
conn = api.Backend.ldap2
- old_cert_obj = x509.load_der_x509_certificate(old_cert_der)
- old_der_subject = old_cert_obj.subject_bytes
- old_spki = old_cert_obj.public_key_info_bytes
+ old_spki = old_cert.public_key_info_bytes
+ # TODO: this could probably be made to return normalized cert right away
cert_file, ca_file = installutils.load_external_cert(
- options.external_cert_files, DN(old_cert_obj.subject))
+ options.external_cert_files, DN(old_cert.subject))
with open(cert_file.name) as f:
new_cert_data = f.read()
new_cert_der = x509.ensure_der_format(new_cert_data)
- new_cert_obj = x509.load_der_x509_certificate(new_cert_der)
- new_der_subject = new_cert_obj.subject_bytes
- new_spki = new_cert_obj.public_key_info_bytes
+ new_cert = x509.load_der_x509_certificate(new_cert_der)
+ new_spki = new_cert.public_key_info_bytes
- if new_cert_obj.subject != old_cert_obj.subject:
+ if new_cert.subject != old_cert.subject:
raise admintool.ScriptError(
"Subject name mismatch (visit "
"http://www.freeipa.org/page/Troubleshooting for "
"troubleshooting guide)")
- if new_der_subject != old_der_subject:
+ if new_cert.subject_bytes != old_cert.subject_bytes:
raise admintool.ScriptError(
"Subject name encoding mismatch (visit "
"http://www.freeipa.org/page/Troubleshooting for "
@@ -240,7 +238,7 @@ def renew_external_step_2(self, ca, old_cert_der):
with certs.NSSDatabase() as tmpdb:
tmpdb.create_db()
- tmpdb.add_cert(old_cert_der, 'IPA CA', EXTERNAL_CA_TRUST_FLAGS)
+ tmpdb.add_cert(old_cert, 'IPA CA', EXTERNAL_CA_TRUST_FLAGS)
try:
tmpdb.add_cert(new_cert_der, 'IPA CA', EXTERNAL_CA_TRUST_FLAGS)
diff --git a/ipaserver/install/ipa_server_certinstall.py b/ipaserver/install/ipa_server_certinstall.py
index a14a84f188..a6b45f29fe 100644
--- a/ipaserver/install/ipa_server_certinstall.py
+++ b/ipaserver/install/ipa_server_certinstall.py
@@ -212,9 +212,9 @@ def install_kdc_cert(self):
# Start tracking only if the cert was issued by IPA CA
# Retrieve IPA CA
ipa_ca_cert = cdb.get_cert_from_db(
- get_ca_nickname(api.env.realm),
- pem=False)
+ get_ca_nickname(api.env.realm))
# And compare with the CA which signed this certificate
+ # TODO: perhaps this comparison does not work as well
if ca_cert == ipa_ca_cert:
certmonger.start_tracking(
(paths.KDC_CERT, paths.KDC_KEY),
@@ -282,9 +282,9 @@ def import_cert(self, dirname, pkcs12_passwd, old_cert, principal, command):
# Start tracking only if the cert was issued by IPA CA
# Retrieve IPA CA
ipa_ca_cert = cdb.get_cert_from_db(
- get_ca_nickname(api.env.realm),
- pem=False)
+ get_ca_nickname(api.env.realm))
# And compare with the CA which signed this certificate
+ # TODO: comparison
if ca_cert == ipa_ca_cert:
cdb.track_server_cert(server_cert,
principal,
diff --git a/ipaserver/install/plugins/upload_cacrt.py b/ipaserver/install/plugins/upload_cacrt.py
index a1957ca5b6..be70ffedd2 100644
--- a/ipaserver/install/plugins/upload_cacrt.py
+++ b/ipaserver/install/plugins/upload_cacrt.py
@@ -20,7 +20,7 @@
from ipalib.install import certstore
from ipaplatform.paths import paths
from ipaserver.install import certs
-from ipalib import Registry, errors
+from ipalib import Registry, errors, x509
from ipalib import Updater
from ipapython import certdb
from ipapython.dn import DN
@@ -56,7 +56,7 @@ def execute(self, **options):
continue
if nickname == ca_nickname and ca_enabled:
trust_flags = certdb.IPA_CA_TRUST_FLAGS
- cert = db.get_cert_from_db(nickname, pem=False)
+ cert = db.get_cert_from_db(nickname)
trust, _ca, eku = certstore.trust_flags_to_key_policy(trust_flags)
dn = DN(('cn', nickname), ('cn', 'certificates'), ('cn', 'ipa'),
@@ -86,6 +86,7 @@ def execute(self, **options):
pass
if ca_cert:
+ dercert = ca_cert.public_bytes(x509.Encoding.DER)
dn = DN(('cn', 'CACert'), ('cn', 'ipa'), ('cn','etc'),
self.api.env.basedn)
try:
@@ -94,11 +95,11 @@ def execute(self, **options):
entry = ldap.make_entry(dn)
entry['objectclass'] = ['nsContainer', 'pkiCA']
entry.single_value['cn'] = 'CAcert'
- entry.single_value['cACertificate;binary'] = ca_cert
+ entry.single_value['cACertificate;binary'] = dercert
ldap.add_entry(entry)
else:
if b'' in entry['cACertificate;binary']:
- entry.single_value['cACertificate;binary'] = ca_cert
+ entry.single_value['cACertificate;binary'] = dercert
ldap.update_entry(entry)
return False, []
diff --git a/ipaserver/install/server/upgrade.py b/ipaserver/install/server/upgrade.py
index 3e2abefc21..7f45a44a38 100644
--- a/ipaserver/install/server/upgrade.py
+++ b/ipaserver/install/server/upgrade.py
@@ -1583,7 +1583,7 @@ def disable_httpd_system_trust(http):
db = certs.CertDB(api.env.realm, nssdir=paths.HTTPD_ALIAS_DIR)
for nickname, trust_flags in db.list_certs():
if not trust_flags.has_key:
- cert = db.get_cert_from_db(nickname, pem=False)
+ cert = db.get_cert_from_db(nickname)
if cert:
ca_certs.append((cert, nickname, trust_flags))
diff --git a/ipaserver/install/service.py b/ipaserver/install/service.py
index 0523e914aa..785417f64f 100644
--- a/ipaserver/install/service.py
+++ b/ipaserver/install/service.py
@@ -32,7 +32,7 @@
from ipapython.dn import DN
from ipapython.ipa_log_manager import root_logger
from ipapython import kerberos
-from ipalib import api, errors
+from ipalib import api, errors, x509
from ipaplatform import services
from ipaplatform.paths import paths
@@ -244,7 +244,7 @@ def __init__(self, service_name, service_desc=None, sstore=None,
self.suffix = DN()
self.service_prefix = service_prefix
self.keytab = keytab
- self.dercert = None
+ self.cert = None
self.api = api
self.service_user = service_user
self.keytab_user = service_user
@@ -369,7 +369,8 @@ def add_cert_to_service(self):
dn = DN(('krbprincipalname', self.principal), ('cn', 'services'),
('cn', 'accounts'), self.suffix)
entry = api.Backend.ldap2.get_entry(dn)
- entry.setdefault('userCertificate', []).append(self.dercert)
+ entry.setdefault('userCertificate', []).append(
+ self.cert.public_bytes(x509.Encoding.DER))
try:
api.Backend.ldap2.update_entry(entry)
except Exception as e:
_______________________________________________
FreeIPA-devel mailing list -- freeipa-devel@lists.fedorahosted.org
To unsubscribe send an email to freeipa-devel-le...@lists.fedorahosted.org
