freeipa/ipaserver/plugins/ca.py

#
# Copyright (C) 2016  FreeIPA Contributors see COPYING for license
#

import base64

import six

from ipalib import api, errors, messages, output
from ipalib import Bytes, DNParam, Flag, Str, Int
from ipalib.constants import IPA_CA_CN
from ipalib.plugable import Registry
from ipapython.dn import ATTR_NAME_BY_OID
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, x509


__doc__ = _("""
Manage Certificate Authorities
""") + _("""
Subordinate Certificate Authorities (Sub-CAs) can be added for scoped issuance
of X.509 certificates.
""") + _("""
CAs are enabled on creation, but their use is subject to CA ACLs unless the
operator has permission to bypass CA ACLs.
""") + _("""
All CAs except the 'IPA' CA can be disabled or re-enabled.  Disabling a CA
prevents it from issuing certificates but does not affect the validity of its
certificate.
""") + _("""
CAs (all except the 'IPA' CA) can be deleted.  Deleting a CA causes its signing
certificate to be revoked and its private key deleted.
""") + _("""
EXAMPLES:
""") + _("""
  Create new CA, subordinate to the IPA CA (requires permission
  "System: Add CA"):

    ipa ca-add puppet --desc "Puppet" \\
        --subject "CN=Puppet CA,O=EXAMPLE.COM"
""") + _("""
  Disable a CA (requires permission "System: Modify CA"):

    ipa ca-disable puppet
""") + _("""
  Re-enable a CA (requires permission "System: Modify CA"):

    ipa ca-enable puppet
""") + _("""
  Delete a CA (requires permission "System: Delete CA"; also requires
  CA to be disabled first):

    ipa ca-del puppet
""")


register = Registry()


@register()
class ca(LDAPObject):
    """
    Lightweight CA Object
    """
    container_dn = api.env.container_ca
    object_name = _('Certificate Authority')
    object_name_plural = _('Certificate Authorities')
    object_class = ['ipaca']
    permission_filter_objectclasses = ['ipaca']
    default_attributes = [
        'cn', 'description', 'ipacaid', 'ipacaissuerdn', 'ipacasubjectdn',
        'ipacarandomserialnumberversion',
    ]
    rdn_attribute = 'cn'
    allow_rename = True
    label = _('Certificate Authorities')
    label_singular = _('Certificate Authority')

    takes_params = (
        Str('cn',
            primary_key=True,
            cli_name='name',
            label=_('Name'),
            doc=_('Name for referencing the CA'),
        ),
        Str('description?',
            cli_name='desc',
            label=_('Description'),
            doc=_('Description of the purpose of the CA'),
        ),
        Str('ipacaid',
            cli_name='id',
            label=_('Authority ID'),
            doc=_('Dogtag Authority ID'),
            flags=['no_create', 'no_update'],
        ),
        DNParam('ipacasubjectdn',
            cli_name='subject',
            label=_('Subject DN'),
            doc=_('Subject Distinguished Name'),
            flags=['no_update'],
        ),
        DNParam('ipacaissuerdn',
            cli_name='issuer',
            label=_('Issuer DN'),
            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'},
        ),
        Int(
            'ipacarandomserialnumberversion',
            cli_name='randomserialnumberversion',
            label=_('RSN Version'),
            doc=_('Random Serial Number Version'),
            flags={'no_create', 'no_update'},
        ),
    )

    permission_filter_objectclasses = ['ipaca']
    managed_permissions = {
        'System: Read CAs': {
            'replaces_global_anonymous_aci': True,
            'ipapermbindruletype': 'all',
            'ipapermright': {'read', 'search', 'compare'},
            'ipapermdefaultattr': {
                'cn',
                'description',
                'ipacaid',
                'ipacaissuerdn',
                'ipacasubjectdn',
                'ipacarandomserialnumberversion',
                'objectclass',
            },
        },
        'System: Add CA': {
            'ipapermright': {'add'},
            'replaces': [
                '(target = "ldap:///cn=*,cn=cas,cn=ca,$SUFFIX")(version 3.0;acl "permission:Add CA";allow (add) groupdn = "ldap:///cn=Add CA,cn=permissions,cn=pbac,$SUFFIX";)',
            ],
            'default_privileges': {'CA Administrator'},
        },
        'System: Delete CA': {
            'ipapermright': {'delete'},
            'replaces': [
                '(target = "ldap:///cn=*,cn=cas,cn=ca,$SUFFIX")(version 3.0;acl "permission:Delete CA";allow (delete) groupdn = "ldap:///cn=Delete CA,cn=permissions,cn=pbac,$SUFFIX";)',
            ],
            'default_privileges': {'CA Administrator'},
        },
        'System: Modify CA': {
            'ipapermright': {'write'},
            'ipapermdefaultattr': {
                'cn',
                'description',
            },
            'replaces': [
                '(targetattr = "cn || description")(target = "ldap:///cn=*,cn=cas,cn=ca,$SUFFIX")(version 3.0;acl "permission:Modify CA";allow (write) groupdn = "ldap:///cn=Modify CA,cn=permissions,cn=pbac,$SUFFIX";)',
            ],
            'default_privileges': {'CA Administrator'},
        },
    }


def set_certificate_attrs(entry, options, want_cert=True):
    """
    Set certificate attributes into the entry.  Depending on
    options, this may contact Dogtag to retrieve certificate or
    chain.  If the retrieval fails with 404 (which can occur under
    normal operation due to lightweight CA key replication delay),
    return a message object that should be set in the response.

    """
    try:
        ca_id = entry['ipacaid'][0]
    except KeyError:
        return None
    full = options.get('all', False)
    want_chain = options.get('chain', False)

    want_data = want_cert or want_chain or full
    if not want_data:
        return None

    msg = None

    with api.Backend.ra_lightweight_ca as ca_api:
        if want_cert or full:
            try:
                der = ca_api.read_ca_cert(ca_id)
                entry['certificate'] = base64.b64encode(der).decode('ascii')
            except errors.HTTPRequestError as e:
                if e.status == 404:  # pylint: disable=no-member
                    msg = messages.LightweightCACertificateNotAvailable(
                        ca=entry['cn'][0])
                else:
                    raise e

        if want_chain or full:
            try:
                pkcs7_der = ca_api.read_ca_chain(ca_id)
                certs = x509.pkcs7_to_certs(pkcs7_der, x509.DER)
                ders = [cert.public_bytes(x509.Encoding.DER) for cert in certs]
                entry['certificate_chain'] = ders
            except errors.HTTPRequestError as e:
                if e.status == 404:  # pylint: disable=no-member
                    msg = messages.LightweightCACertificateNotAvailable(
                        ca=entry['cn'][0])
                else:
                    raise e

    return msg

@register()
class ca_find(LDAPSearch):
    __doc__ = _("Search for CAs.")
    msg_summary = ngettext(
        '%(count)d CA matched', '%(count)d CAs matched', 0
    )

    def execute(self, *keys, **options):
        ca_enabled_check(self.api)
        result = super(ca_find, self).execute(*keys, **options)
        if not options.get('pkey_only', False):
            for entry in result['result']:
                msg = set_certificate_attrs(entry, options, want_cert=False)
                if msg:
                    self.add_message(msg)
        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.")

    takes_options = LDAPRetrieve.takes_options + (
        _chain_flag,
    )

    def execute(self, *keys, **options):
        ca_enabled_check(self.api)
        result = super(ca_show, self).execute(*keys, **options)
        msg = set_certificate_attrs(result['result'], options)
        if msg:
            self.add_message(msg)
        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(self.api)
        if not ldap.can_add(dn[1:], 'ipaca'):
            raise errors.ACIError(
                info=_("Insufficient 'add' privilege for entry '%s'.") % dn)

        # check that DN only includes standard naming attributes
        dn_attrs = {
            ava.attr.lower()
            for rdn in options['ipacasubjectdn']
            for ava in rdn
        }
        x509_attrs = {
            attr.lower()
            for attr in six.viewvalues(ATTR_NAME_BY_OID)
        }
        unknown_attrs = dn_attrs - x509_attrs
        if len(unknown_attrs) > 0:
            raise errors.ValidationError(
                name=_("Subject DN"),
                error=_("Unrecognized attributes: %(attrs)s")
                    % dict(attrs=", ".join(unknown_attrs))
            )

        # check for name collision before creating CA in Dogtag
        try:
            api.Object.ca.get_dn_if_exists(keys[-1])
            self.obj.handle_duplicate_entry(*keys)
        except errors.NotFound:
            pass

        # check for subject collision before creating CA in Dogtag
        result = api.Command.ca_find(ipacasubjectdn=options['ipacasubjectdn'])
        if result['count'] > 0:
            raise errors.DuplicateEntry(message=_(
                "Subject DN is already used by CA '%s'"
                ) % result['result'][0]['cn'][0])

        # Create the CA in Dogtag.
        with self.api.Backend.ra_lightweight_ca as ca_api:
            resp = ca_api.create_ca(options['ipacasubjectdn'])
        entry['ipacaid'] = [resp['id']]
        entry['ipacaissuerdn'] = [resp['issuerDN']]

        # In the event that the issued certificate's subject DN
        # differs from what was requested, record the actual DN.
        #
        entry['ipacasubjectdn'] = [resp['dn']]
        return dn

    def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
        msg = set_certificate_attrs(entry_attrs, options)
        if msg:
            self.add_message(msg)
        return dn


@register()
class ca_del(LDAPDelete):
    __doc__ = _('Delete a CA (must be disabled first).')

    msg_summary = _('Deleted CA "%(value)s"')

    def pre_callback(self, ldap, dn, *keys, **options):
        ca_enabled_check(self.api)

        # ensure operator has permission to delete CA
        # before contacting Dogtag
        if not ldap.can_delete(dn):
            raise errors.ACIError(info=_(
                "Insufficient privilege to delete a CA."))

        if keys[0] == IPA_CA_CN:
            raise errors.ProtectedEntryError(
                label=_("CA"),
                key=keys[0],
                reason=_("IPA CA cannot be deleted"))

        ca_id = self.api.Command.ca_show(keys[0])['result']['ipacaid'][0]
        with self.api.Backend.ra_lightweight_ca as ca_api:
            data = ca_api.read_ca(ca_id)
            if data['enabled']:
                raise errors.ProtectedEntryError(
                    label=_("CA"),
                    key=keys[0],
                    reason=_("Must be disabled first"))
            ca_api.delete_ca(ca_id)

        return dn


@register()
class ca_mod(LDAPUpdate):
    __doc__ = _("Modify CA configuration.")
    msg_summary = _('Modified CA "%(value)s"')

    def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
        ca_enabled_check(self.api)

        if 'rename' in options or 'cn' in entry_attrs:
            if keys[0] == IPA_CA_CN:
                raise errors.ProtectedEntryError(
                    label=_("CA"),
                    key=keys[0],
                    reason=u'IPA CA cannot be renamed')

        return dn


class CAQuery(LDAPQuery):
    has_output = output.standard_value

    def execute(self, cn, **options):
        ca_enabled_check(self.api)

        ca_obj = self.api.Command.ca_show(cn)['result']

        # ensure operator has permission to modify CAs
        if not self.api.Backend.ldap2.can_write(ca_obj['dn'], 'description'):
            raise errors.ACIError(info=_(
                "Insufficient privilege to modify a CA."))

        with self.api.Backend.ra_lightweight_ca as ca_api:
            self.perform_action(ca_api, ca_obj['ipacaid'][0])

        return dict(
            result=True,
            value=pkey_to_value(cn, options),
        )

    def perform_action(self, ca_api, ca_id):
        raise NotImplementedError


@register()
class ca_disable(CAQuery):
    __doc__ = _('Disable a CA.')
    msg_summary = _('Disabled CA "%(value)s"')

    def execute(self, cn, **options):
        if cn == IPA_CA_CN:
            raise errors.ProtectedEntryError(
                label=_("CA"),
                key=cn,
                reason=_("IPA CA cannot be disabled"))

        return super(ca_disable, self).execute(cn, **options)

    def perform_action(self, ca_api, ca_id):
        ca_api.disable_ca(ca_id)


@register()
class ca_enable(CAQuery):
    __doc__ = _('Enable a CA.')
    msg_summary = _('Enabled CA "%(value)s"')

    def perform_action(self, ca_api, ca_id):
        ca_api.enable_ca(ca_id)