Add options to write lightweight CA cert or chain to file

API.txt

@@ -445,10 +445,11 @@ option: Str('version?')
 output: Output('count', type=[<type 'int'>])
 output: Output('results', type=[<type 'list'>, <type 'tuple'>])
 command: ca_add/1
-args: 1,7,3
+args: 1,8,3
 arg: Str('cn', cli_name='name')
 option: Str('addattr*', cli_name='addattr')
 option: Flag('all', autofill=True, cli_name='all', default=False)
+option: Flag('chain', autofill=True, default=False)
 option: Str('description?', cli_name='desc')
 option: DNParam('ipacasubjectdn', cli_name='subject')
 option: Flag('raw', autofill=True, cli_name='raw', default=False)
@@ -519,9 +520,10 @@ output: Entry('result')
 output: Output('summary', type=[<type 'unicode'>, <type 'NoneType'>])
 output: PrimaryKey('value')
 command: ca_show/1
-args: 1,4,3
+args: 1,5,3
 arg: Str('cn', cli_name='name')
 option: Flag('all', autofill=True, cli_name='all', default=False)
+option: Flag('chain', autofill=True, default=False)
 option: Flag('raw', autofill=True, cli_name='raw', default=False)
 option: Flag('rights', autofill=True, default=False)
 option: Str('version?')

VERSION.m4

@@ -73,8 +73,8 @@ define(IPA_DATA_VERSION, 20100614120000)
 #                                                      #
 ########################################################
 define(IPA_API_VERSION_MAJOR, 2)
-define(IPA_API_VERSION_MINOR, 216)
-# Last change: DNS: Support URI resource record type
+define(IPA_API_VERSION_MINOR, 217)
+# Last change: Add options to write lightweight CA cert or chain to file
 
 
 ########################################################

ipaclient/plugins/ca.py

@@ -0,0 +1,53 @@
+#
+# Copyright (C) 2016  FreeIPA Contributors see COPYING for license
+#
+
+import base64
+from ipaclient.frontend import MethodOverride
+from ipalib import util, x509, Str
+from ipalib.plugable import Registry
+from ipalib.text import _
+
+register = Registry()
+
+
+class WithCertOutArgs(MethodOverride):
+
+    takes_options = (
+        Str(
+            'certificate_out?',
+            doc=_('Write certificate (chain if --chain used) to file'),
+            include='cli',
+            cli_metavar='FILE',
+        ),
+    )
+
+    def forward(self, *keys, **options):
+        filename = None
+        if 'certificate_out' in options:
+            filename = options.pop('certificate_out')
+            util.check_writable_file(filename)
+
+        result = super(WithCertOutArgs, self).forward(*keys, **options)
+        if filename:
+            def to_pem(x):
+                return x509.make_pem(x)
+            if options.get('chain', False):
+                ders = result['result']['certificate_chain']
+                data = '\n'.join(to_pem(base64.b64encode(der)) for der in ders)
+            else:
+                data = to_pem(result['result']['certificate'])
+            with open(filename, 'wb') as f:
+                f.write(data)
+
+        return result
+
+
+@register(override=True, no_fail=True)
+class ca_add(WithCertOutArgs):
+    pass
+
+
+@register(override=True, no_fail=True)
+class ca_show(WithCertOutArgs):
+    pass

ipalib/x509.py

@@ -49,14 +49,24 @@
 from ipalib import util
 from ipalib import errors
 from ipapython.dn import DN
+from ipapython import ipautil
+
+try:
+    from ipaplatform.paths import paths
+except ImportError:
+    OPENSSL = '/usr/bin/openssl'
+else:
+    OPENSSL = paths.OPENSSL
 
 if six.PY3:
     unicode = str
 
 PEM = 0
 DER = 1
 
-PEM_REGEX = re.compile(r'(?<=-----BEGIN CERTIFICATE-----).*?(?=-----END CERTIFICATE-----)', re.DOTALL)
+PEM_REGEX = re.compile(
+    r'-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----',
+    re.DOTALL)
 
 EKU_SERVER_AUTH = '1.3.6.1.5.5.7.3.1'
 EKU_CLIENT_AUTH = '1.3.6.1.5.5.7.3.2'
@@ -145,6 +155,23 @@ def load_certificate_list_from_file(filename):
         return load_certificate_list(f.read())
 
 
+def pkcs7_to_pems(data, datatype=PEM):
+    """
+    Extract certificates from a PKCS #7 object.
+
+    Return a ``list`` of X.509 PEM strings.
+
+    May throw ``ipautil.CalledProcessError`` on invalid data.
+
+    """
+    cmd = [
+        OPENSSL, "pkcs7", "-print_certs",
+        "-inform", "PEM" if datatype == PEM else "DER",
+    ]
+    result = ipautil.run(cmd, stdin=data, capture_output=True)
+    return PEM_REGEX.findall(result.output)
+
+
 def is_self_signed(certificate, datatype=PEM):
     cert = load_certificate(certificate, datatype)
     return cert.issuer == cert.subject

ipapython/certdb.py

@@ -203,7 +203,7 @@ def import_files(self, files, db_password_filename, import_keys=False,
         """
         key_file = None
         extracted_key = None
-        extracted_certs = ''
+        extracted_certs = []
 
         for filename in files:
             try:
@@ -234,18 +234,13 @@ def import_files(self, files, db_password_filename, import_keys=False,
                                     filename, line, e)
                                 continue
                         else:
-                            extracted_certs += body + '\n'
+                            extracted_certs.append(body)
                             loaded = True
                             continue
 
                     if label in ('PKCS7', 'PKCS #7 SIGNED DATA', 'CERTIFICATE'):
-                        args = [
-                            OPENSSL, 'pkcs7',
-                            '-print_certs',
-                        ]
                         try:
-                            result = ipautil.run(
-                                args, stdin=body, capture_output=True)
+                            certs = x509.pkcs7_to_pems(body)
                         except ipautil.CalledProcessError as e:
                             if label == 'CERTIFICATE':
                                 root_logger.warning(
@@ -257,7 +252,7 @@ def import_files(self, files, db_password_filename, import_keys=False,
                                     filename, line, e)
                             continue
                         else:
-                            extracted_certs += result.output + '\n'
+                            extracted_certs.extend(certs)
                             loaded = True
                             continue
 
@@ -307,7 +302,7 @@ def import_files(self, files, db_password_filename, import_keys=False,
                 pass
             else:
                 data = x509.make_pem(base64.b64encode(data))
-                extracted_certs += data + '\n'
+                extracted_certs.append(data)
                 continue
 
             # Try to import the file as PKCS#12 file
@@ -348,14 +343,15 @@ def import_files(self, files, db_password_filename, import_keys=False,
             raise RuntimeError(
                 "No server certificates found in %s" % (', '.join(files)))
 
-        certs = x509.load_certificate_list(extracted_certs)
-        for cert in certs:
+        for cert_pem in extracted_certs:
+            cert = x509.load_certificate(cert_pem)
             nickname = str(DN(cert.subject))
             data = cert.public_bytes(serialization.Encoding.DER)
             self.add_cert(data, nickname, ',,')
 
         if extracted_key:
-            in_file = ipautil.write_tmp_file(extracted_certs + extracted_key)
+            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)

ipaserver/install/cainstance.py

@@ -749,44 +749,30 @@ def __import_ca_chain(self):
         # makes openssl throw up.
         data = base64.b64decode(chain)
 
-        result = ipautil.run(
-            [paths.OPENSSL,
-             "pkcs7",
-             "-inform",
-             "DER",
-             "-print_certs",
-             ], stdin=data, capture_output=True)
-        certlist = result.output
+        certlist = x509.pkcs7_to_pems(data, x509.DER)
 
         # Ok, now we have all the certificates in certs, walk through it
         # and pull out each certificate and add it to our database
 
-        st = 1
-        en = 0
-        subid = 0
         ca_dn = DN(('CN','Certificate Authority'), self.subject_base)
-        while st > 0:
-            st = certlist.find('-----BEGIN', en)
-            en = certlist.find('-----END', en+1)
-            if st > 0:
-                try:
-                    (chain_fd, chain_name) = tempfile.mkstemp()
-                    os.write(chain_fd, certlist[st:en+25])
-                    os.close(chain_fd)
-                    (_rdn, subject_dn) = certs.get_cert_nickname(certlist[st:en+25])
-                    if subject_dn == ca_dn:
-                        nick = get_ca_nickname(self.realm)
-                        trust_flags = 'CT,C,C'
-                    else:
-                        nick = str(subject_dn)
-                        trust_flags = ',,'
-                    self.__run_certutil(
-                        ['-A', '-t', trust_flags, '-n', nick, '-a',
-                         '-i', chain_name]
-                    )
-                finally:
-                    os.remove(chain_name)
-                    subid += 1
+        for cert in certlist:
+            try:
+                chain_fd, chain_name = tempfile.mkstemp()
+                os.write(chain_fd, cert)
+                os.close(chain_fd)
+                (_rdn, subject_dn) = certs.get_cert_nickname(cert)
+                if subject_dn == ca_dn:
+                    nick = get_ca_nickname(self.realm)
+                    trust_flags = 'CT,C,C'
+                else:
+                    nick = str(subject_dn)
+                    trust_flags = ',,'
+                self.__run_certutil(
+                    ['-A', '-t', trust_flags, '-n', nick, '-a',
+                     '-i', chain_name]
+                )
+            finally:
+                os.remove(chain_name)
 
         # Restore NSS trust flags of all previously existing certificates
         for nick, trust_flags in cert_backup_list:

ipaserver/plugins/ca.py

@@ -2,14 +2,18 @@
 # Copyright (C) 2016  FreeIPA Contributors see COPYING for license
 #
 
-from ipalib import api, errors, output, DNParam, Str
+import base64
+
+import six
+
+from ipalib import api, errors, output, Bytes, DNParam, Flag, Str
 from ipalib.constants import IPA_CA_CN
 from ipalib.plugable import Registry
 from ipaserver.plugins.baseldap import (
     LDAPObject, LDAPSearch, LDAPCreate, LDAPDelete,
     LDAPUpdate, LDAPRetrieve, LDAPQuery, pkey_to_value)
 from ipaserver.plugins.cert import ca_enabled_check
-from ipalib import _, ngettext
+from ipalib import _, ngettext, x509
 
 
 __doc__ = _("""
@@ -100,6 +104,18 @@ class ca(LDAPObject):
             doc=_('Issuer Distinguished Name'),
             flags=['no_create', 'no_update'],
         ),
+        Bytes(
+            'certificate',
+            label=_("Certificate"),
+            doc=_("Base-64 encoded certificate."),
+            flags={'no_create', 'no_update', 'no_search'},
+        ),
+        Bytes(
+            'certificate_chain*',
+            label=_("Certificate chain"),
+            doc=_("X.509 certificate chain"),
+            flags={'no_create', 'no_update', 'no_search'},
+        ),
     )
 
     permission_filter_objectclasses = ['ipaca']
@@ -145,6 +161,21 @@ class ca(LDAPObject):
     }
 
 
+def set_certificate_attrs(entry, options, always_include_cert=True):
+    ca_id = entry['ipacaid'][0]
+    full = options.get('all', False)
+    with api.Backend.ra_lightweight_ca as ca_api:
+        if always_include_cert or full:
+            der = ca_api.read_ca_cert(ca_id)
+            entry['certificate'] = six.text_type(base64.b64encode(der))
+
+        if options.get('chain', False) 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]
+            entry['certificate_chain'] = ders
+
+
 @register()
 class ca_find(LDAPSearch):
     __doc__ = _("Search for CAs.")
@@ -154,23 +185,43 @@ class ca_find(LDAPSearch):
 
     def execute(self, *keys, **options):
         ca_enabled_check()
-        return super(ca_find, self).execute(*keys, **options)
+        result = super(ca_find, self).execute(*keys, **options)
+        for entry in result['result']:
+            set_certificate_attrs(entry, options, always_include_cert=False)
+        return result
+
+
+_chain_flag = Flag(
+    'chain',
+    default=False,
+    doc=_('Include certificate chain in output'),
+)
 
 
 @register()
 class ca_show(LDAPRetrieve):
     __doc__ = _("Display the properties of a CA.")
 
-    def execute(self, *args, **kwargs):
+    takes_options = LDAPRetrieve.takes_options + (
+        _chain_flag,
+    )
+
+    def execute(self, *keys, **options):
         ca_enabled_check()
-        return super(ca_show, self).execute(*args, **kwargs)
+        result = super(ca_show, self).execute(*keys, **options)
+        set_certificate_attrs(result['result'], options)
+        return result
 
 
 @register()
 class ca_add(LDAPCreate):
     __doc__ = _("Create a CA.")
     msg_summary = _('Created CA "%(value)s"')
 
+    takes_options = LDAPCreate.takes_options + (
+        _chain_flag,
+    )
+
     def pre_callback(self, ldap, dn, entry, entry_attrs, *keys, **options):
         ca_enabled_check()
         if not ldap.can_add(dn[1:]):
@@ -203,6 +254,10 @@ def pre_callback(self, ldap, dn, entry, entry_attrs, *keys, **options):
         entry['ipacasubjectdn'] = [resp['dn']]
         return dn
 
+    def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
+        set_certificate_attrs(entry_attrs, options)
+        return dn
+
 
 @register()
 class ca_del(LDAPDelete):