# octavia.certificates.manager.castellan_mgr

# Source code for octavia.certificates.manager.castellan_mgr

    # Copyright (c) 2017 GoDaddy
    # All Rights Reserved.
    #
    #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    #    not use this file except in compliance with the License. You may obtain
    #    a copy of the License at
    #
    #         http://www.apache.org/licenses/LICENSE-2.0
    #
    #    Unless required by applicable law or agreed to in writing, software
    #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
    #    License for the specific language governing permissions and limitations
    #    under the License.

    """
    Cert manager implementation for Castellan
    """
    from castellan.common.objects import opaque_data
    from castellan import key_manager
    from cryptography.hazmat.primitives import serialization
    from cryptography.hazmat.primitives.serialization import pkcs12 as c_pkcs12
    from oslo_config import cfg
    from oslo_log import log as logging

    from octavia.certificates.common import pkcs12
    from octavia.certificates.manager import cert_mgr
    from octavia.common import exceptions

    CONF = cfg.CONF

    LOG = logging.getLogger(__name__)

    class CastellanCertManager(cert_mgr.CertManager):
        """Certificate Manager for the Castellan library."""

        def __init__(self):
            super().__init__()
            self.manager = key_manager.API(CONF)

        def store_cert(self, context, certificate, private_key, intermediates=None,
                       private_key_passphrase=None, expiration=None,
                       name="PKCS12 Certificate Bundle"):
            if private_key_passphrase:
                raise exceptions.CertificateStorageException(
                    "Passphrases protected PKCS12 certificates are not supported.")

            p12_data = opaque_data.OpaqueData(
                c_pkcs12.serialize_key_and_certificates(
                    name=None,
                    key=private_key,
                    cert=certificate,
                    cas=intermediates,
                    encryption_algorithm=serialization.NoEncryption()
                ),
                name=name
            )
            self.manager.store(context, p12_data)

        def get_cert(self, context, cert_ref, resource_ref=None, check_only=False,
                     service_name=None):
            certbag = self.manager.get(context, cert_ref)
            certbag_data = certbag.get_encoded()
            cert = pkcs12.PKCS12Cert(certbag_data)
            return cert

        def delete_cert(self, context, cert_ref, resource_ref, service_name=None):
            # Delete is not a great name for this -- we don't delete anything
            # in reality, we just do cleanup here. For castellan, none is required
            pass

        def set_acls(self, context, cert_ref):
            # We don't manage ACL based access for things retrieved via Castellan
            # because we assume we have elevated access to the secret store.
            pass

        def unset_acls(self, context, cert_ref):
            # We don't manage ACL based access for things retrieved via Castellan
            # because we assume we have elevated access to the secret store.
            pass

        def get_secret(self, context, secret_ref):
            try:
                certbag = self.manager.get(context, secret_ref)
                certbag_data = certbag.get_encoded()
            except Exception as e:
                LOG.error("Failed to access secret for %s due to: %s.",
                          secret_ref, str(e))
                raise exceptions.CertificateRetrievalException(ref=secret_ref)
            return certbag_data

this page last updated: 2026-04-13 15:26:39

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.

found an error? report a bug