Automatically set Barbican ACLs

Story: 2002973
Task: 22981

Co-Authored-By: Carlos Goncalves <cgoncalves@redhat.com>

Change-Id: I51121c599f19a91a6755571abf1c6bd854e7d50f

doc/source/user/guides/basic-cookbook-neutron.rst

@@ -360,10 +360,6 @@ balancer features, like Layer 7 features and header manipulation.
     openstack secret store --name='key1' --payload-content-type='text/plain' --payload="$(cat server.key)"
     openstack secret store --name='intermediates1' --payload-content-type='text/plain' --payload="$(cat ca-chain.p7b)"
     openstack secret container create --name='tls_container1' --type='certificate' --secret="certificate=$(openstack secret list | awk '/ cert1 / {print $2}')" --secret="private_key=$(openstack secret list | awk '/ key1 / {print $2}')" --secret="intermediates=$(openstack secret list | awk '/ intermediates1 / {print $2}')"
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ cert1 / {print $2}')
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ key1 / {print $2}')
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ intermediates1 / {print $2}')
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ tls_container1 / {print $2}')
     neutron lbaas-loadbalancer-create --name lb1 public-subnet
     # Re-run the following until lb1 shows ACTIVE and ONLINE statuses:
     neutron lbaas-loadbalancer-show lb1
@@ -434,14 +430,6 @@ the same listener using Server Name Indication (SNI) technology.
     openstack secret store --name='intermediates2' --payload-content-type='text/plain' --payload="$(cat ca-chain2.p7b)"
     openstack secret store --name='passphrase2' --payload-content-type='text/plain' --payload="abc123"
     openstack secret container create --name='tls_container2' --type='certificate' --secret="certificate=$(openstack secret list | awk '/ cert2 / {print $2}')" --secret="private_key=$(openstack secret list | awk '/ key2 / {print $2}')" --secret="intermediates=$(openstack secret list | awk '/ intermediates2 / {print $2}')" --secret="private_key_passphrase=$(openstack secret list | awk '/ passphrase2 / {print $2}')"
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ cert1 / {print $2}')
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ key1 / {print $2}')
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ intermediates1 / {print $2}')
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ tls_container1 / {print $2}')
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ cert2 / {print $2}')
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ key2 / {print $2}')
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ intermediates2 / {print $2}')
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ tls_container2 / {print $2}')
     neutron lbaas-loadbalancer-create --name lb1 public-subnet
     # Re-run the following until lb1 shows ACTIVE and ONLINE statuses:
     neutron lbaas-loadbalancer-show lb1
@@ -513,10 +501,6 @@ HTTP just get redirected to the HTTPS listener), then please see `the example
     openstack secret store --name='key1' --payload-content-type='text/plain' --payload="$(cat server.key)"
     openstack secret store --name='intermediates1' --payload-content-type='text/plain' --payload="$(cat ca-chain.p7b)"
     openstack secret container create --name='tls_container1' --type='certificate' --secret="certificate=$(openstack secret list | awk '/ cert1 / {print $2}')" --secret="private_key=$(openstack secret list | awk '/ key1 / {print $2}')" --secret="intermediates=$(openstack secret list | awk '/ intermediates1 / {print $2}')"
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ cert1 / {print $2}')
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ key1 / {print $2}')
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ intermediates1 / {print $2}')
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ tls_container1 / {print $2}')
     neutron lbaas-loadbalancer-create --name lb1 public-subnet
     # Re-run the following until lb1 shows ACTIVE and ONLINE statuses:
     neutron lbaas-loadbalancer-show lb1

doc/source/user/guides/basic-cookbook.rst

@@ -398,7 +398,6 @@ balancer features, like Layer 7 features and header manipulation.
 
     openssl pkcs12 -export -inkey server.key -in server.crt -certfile ca-chain.crt -passout pass: -out server.p12
     openstack secret store --name='tls_secret1' -t 'application/octet-stream' -e 'base64' --payload="$(base64 < server.p12)"
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ tls_secret1 / {print $2}')
     openstack loadbalancer create --name lb1 --vip-subnet-id public-subnet
     # Re-run the following until lb1 shows ACTIVE and ONLINE statuses:
     openstack loadbalancer show lb1
@@ -456,8 +455,6 @@ listener using Server Name Indication (SNI) technology.
     openssl pkcs12 -export -inkey server2.key -in server2.crt -certfile ca-chain2.crt -passout pass: -out server2.p12
     openstack secret store --name='tls_secret1' -t 'application/octet-stream' -e 'base64' --payload="$(base64 < server.p12)"
     openstack secret store --name='tls_secret2' -t 'application/octet-stream' -e 'base64' --payload="$(base64 < server2.p12)"
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ tls_secret1 / {print $2}')
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ tls_secret2 / {print $2}')
     openstack loadbalancer create --name lb1 --vip-subnet-id public-subnet
     # Re-run the following until lb1 shows ACTIVE and ONLINE statuses:
     openstack loadbalancer show lb1
@@ -521,7 +518,6 @@ HTTP just get redirected to the HTTPS listener), then please see `the example
 
     openssl pkcs12 -export -inkey server.key -in server.crt -certfile ca-chain.crt -passout pass: -out server.p12
     openstack secret store --name='tls_secret1' -t 'application/octet-stream' -e 'base64' --payload="$(base64 < server.p12)"
-    openstack acl user add -u admin_id $(openstack secret list | awk '/ tls_secret1 / {print $2}')
     openstack loadbalancer create --name lb1 --vip-subnet-id public-subnet
     # Re-run the following until lb1 shows ACTIVE and ONLINE statuses:
     openstack loadbalancer show lb1

octavia/amphorae/drivers/haproxy/rest_api_driver.py

@@ -190,7 +190,8 @@ class HaproxyAmphoraLoadBalancerDriver(
 
     def _upload_cert(self, amp, listener_id, pem, md5, name):
         try:
-            if self.client.get_cert_md5sum(amp, listener_id, name) == md5:
+            if self.client.get_cert_md5sum(
+                    amp, listener_id, name, ignore=(404,)) == md5:
                 return
         except exc.NotFound:
             pass
@@ -343,11 +344,11 @@ class AmphoraAPIClient(object):
         r = self.put(amp, 'certificate', data=pem_file)
         return exc.check_exception(r)
 
-    def get_cert_md5sum(self, amp, listener_id, pem_filename):
+    def get_cert_md5sum(self, amp, listener_id, pem_filename, ignore=tuple()):
         r = self.get(amp,
                      'listeners/{listener_id}/certificates/{filename}'.format(
                          listener_id=listener_id, filename=pem_filename))
-        if exc.check_exception(r):
+        if exc.check_exception(r, ignore):
             return r.json().get("md5sum")
         return None
 

octavia/api/v2/controllers/listener.py

@@ -132,6 +132,7 @@ class ListenersController(base.BaseController):
         bad_refs = []
         for ref in tls_refs:
             try:
+                self.cert_manager.set_acls(context, ref)
                 self.cert_manager.get_cert(context, ref, check_only=True)
             except Exception:
                 bad_refs.append(ref)
@@ -371,6 +372,45 @@ class ListenersController(base.BaseController):
                      driver.name)
             driver_utils.call_provider(driver.name, driver.listener_delete, id)
 
+        # Revoke access of octavia service user to certificates
+        tls_refs = []
+
+        for sni in db_listener.sni_containers:
+            filters = {'tls_container_id': sni.tls_container_id}
+            snis = self.repositories.sni.get_all(context.session, **filters)[0]
+
+            if len(snis) == 1:
+                # referred only once, enqueue for access revoking
+                tls_refs.append(sni.tls_container_id)
+            else:
+                blocking_listeners = [s.listener_id for s in snis if
+                                      s.listener_id != id]
+                LOG.debug("Listeners %s using TLS ref %s. Access to TLS ref "
+                          "will not be revoked.", blocking_listeners,
+                          sni.tls_container_id)
+
+        if db_listener.tls_certificate_id:
+            filters = {'tls_certificate_id': db_listener.tls_certificate_id}
+            # Note get_all returns the list and links. We only want the list.
+            listeners = self.repositories.listener.get_all(
+                context.session, show_deleted=False, **filters)[0]
+
+            if len(listeners) == 1:
+                # referred only once, enqueue for access revoking
+                tls_refs.append(db_listener.tls_certificate_id)
+            else:
+                blocking_listeners = [l.id for l in listeners if l.id != id]
+                LOG.debug("Listeners %s using TLS ref %s. Access to TLS ref "
+                          "will not be revoked.", blocking_listeners,
+                          db_listener.tls_certificate_id)
+
+        for ref in tls_refs:
+            try:
+                self.cert_manager.unset_acls(context, ref)
+            except Exception:
+                # certificate may have been removed already
+                pass
+
     @pecan.expose()
     def _lookup(self, id, *remainder):
         """Overridden pecan _lookup method for custom routing.

octavia/certificates/common/auth/barbican_acl.py

@@ -17,6 +17,9 @@
 Barbican ACL auth class for Barbican certificate handling
 """
 from barbicanclient import client as barbican_client
+from keystoneauth1.identity.generic import token
+from keystoneauth1 import session
+
 from oslo_config import cfg
 from oslo_log import log as logging
 from oslo_utils import excutils
@@ -35,9 +38,9 @@ class BarbicanACLAuth(barbican_common.BarbicanAuth):
     def get_barbican_client(cls, project_id=None):
         if not cls._barbican_client:
             try:
-                session = keystone.KeystoneSession().get_session()
+                ksession = keystone.KeystoneSession()
                 cls._barbican_client = barbican_client.Client(
-                    session=session,
+                    session=ksession.get_session(),
                     region_name=CONF.certificates.region_name,
                     interface=CONF.certificates.endpoint_type
                 )
@@ -45,3 +48,46 @@ class BarbicanACLAuth(barbican_common.BarbicanAuth):
                 with excutils.save_and_reraise_exception():
                     LOG.exception("Error creating Barbican client")
         return cls._barbican_client
+
+    @classmethod
+    def ensure_secret_access(cls, context, ref):
+        # get a normal session
+        ksession = keystone.KeystoneSession()
+        user_id = ksession.get_service_user_id()
+
+        # use barbican client to set the ACLs
+        bc = cls.get_barbican_client_user_auth(context)
+        acl = bc.acls.get(ref)
+        read_oper = acl.get('read')
+        if user_id not in read_oper.users:
+            read_oper.users.append(user_id)
+            acl.submit()
+
+    @classmethod
+    def revoke_secret_access(cls, context, ref):
+        # get a normal session
+        ksession = keystone.KeystoneSession()
+        user_id = ksession.get_service_user_id()
+
+        # use barbican client to set the ACLs
+        bc = cls.get_barbican_client_user_auth(context)
+        acl = bc.acls.get(ref)
+        read_oper = acl.get('read')
+        if user_id in read_oper.users:
+            read_oper.users.remove(user_id)
+            acl.submit()
+
+    @classmethod
+    def get_barbican_client_user_auth(cls, context):
+        # get a normal session
+        ksession = keystone.KeystoneSession()
+        service_auth = ksession.get_auth()
+
+        # make our own auth and swap it in
+        user_auth = token.Token(auth_url=service_auth.auth_url,
+                                token=context.auth_token,
+                                project_id=context.project_id)
+        user_session = session.Session(auth=user_auth)
+
+        # create a special barbican client with our user's session
+        return barbican_client.Client(session=user_session)

octavia/certificates/common/barbican.py

@@ -73,3 +73,19 @@ class BarbicanAuth(object):
         :return: a Barbican Client object
         :raises Exception: if the client cannot be created
         """
+
+    @abc.abstractmethod
+    def ensure_secret_access(self, context, ref):
+        """Do whatever steps are necessary to ensure future access to a secret.
+
+        :param context: pecan context object
+        :param ref: Reference to a Barbican object
+        """
+
+    @abc.abstractmethod
+    def revoke_secret_access(self, context, ref):
+        """Revoke access of Octavia keystone user to a secret.
+
+        :param context: pecan context object
+        :param ref: Reference to a Barbican object
+        """

octavia/certificates/manager/barbican.py

@@ -143,3 +143,11 @@ class BarbicanCertManager(cert_mgr.CertManager):
             # If the delete failed, it was probably because it isn't legacy
             # (this will be fixed once Secrets have Consumer registration).
             pass
+
+    def set_acls(self, context, cert_ref):
+        LOG.debug('Setting project ACL for certificate secret...')
+        self.auth.ensure_secret_access(context, cert_ref)
+
+    def unset_acls(self, context, cert_ref):
+        LOG.debug('Unsetting project ACL for certificate secret...')
+        self.auth.revoke_secret_access(context, cert_ref)

octavia/certificates/manager/barbican_legacy.py

@@ -144,6 +144,7 @@ class BarbicanCertManager(cert_mgr.CertManager):
                     url=resource_ref
                 )
             barbican_cert = barbican_common.BarbicanCert(cert_container)
+
             LOG.debug('Validating certificate data for %s.', cert_ref)
             cert_parser.validate_cert(
                 barbican_cert.get_certificate(),
@@ -152,6 +153,7 @@ class BarbicanCertManager(cert_mgr.CertManager):
                     barbican_cert.get_private_key_passphrase()),
                 intermediates=barbican_cert.get_intermediates())
             LOG.debug('Certificate data validated for %s.', cert_ref)
+
             return barbican_cert
         except Exception as e:
             with excutils.save_and_reraise_exception():
@@ -180,3 +182,43 @@ class BarbicanCertManager(cert_mgr.CertManager):
             with excutils.save_and_reraise_exception():
                 LOG.error('Error deregistering as a consumer of %s: %s',
                           cert_ref, e)
+
+    def set_acls(self, context, cert_ref):
+        LOG.debug('Setting project ACLs for certificate secrets...')
+        self.auth.ensure_secret_access(context, cert_ref)
+
+        connection = self.auth.get_barbican_client(context.project_id)
+        cert_container = connection.containers.get(
+            container_ref=cert_ref
+        )
+        self.auth.ensure_secret_access(
+            context, cert_container.certificate.secret_ref)
+        self.auth.ensure_secret_access(
+            context, cert_container.private_key.secret_ref)
+        if cert_container.private_key_passphrase:
+            self.auth.ensure_secret_access(
+                context,
+                cert_container.private_key_passphrase.secret_ref)
+        if cert_container.intermediates:
+            self.auth.ensure_secret_access(
+                context, cert_container.intermediates.secret_ref)
+
+    def unset_acls(self, context, cert_ref):
+        LOG.debug('Unsetting project ACLs for certificate secrets...')
+        self.auth.revoke_secret_access(context, cert_ref)
+
+        connection = self.auth.get_barbican_client(context.project_id)
+        cert_container = connection.containers.get(
+            container_ref=cert_ref
+        )
+        self.auth.revoke_secret_access(
+            context, cert_container.certificate.secret_ref)
+        self.auth.revoke_secret_access(
+            context, cert_container.private_key.secret_ref)
+        if cert_container.private_key_passphrase:
+            self.auth.revoke_secret_access(
+                context,
+                cert_container.private_key_passphrase.secret_ref)
+        if cert_container.intermediates:
+            self.auth.revoke_secret_access(
+                context, cert_container.intermediates.secret_ref)

octavia/certificates/manager/castellan_mgr.py

@@ -61,3 +61,13 @@ class CastellanCertManager(cert_mgr.CertManager):
         # 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

octavia/certificates/manager/cert_mgr.py

@@ -38,7 +38,6 @@ class CertManager(object):
         If storage of the certificate data fails, a CertificateStorageException
         should be raised.
         """
-        pass
 
     @abc.abstractmethod
     def get_cert(self, context, cert_ref, resource_ref=None, check_only=False,
@@ -49,7 +48,6 @@ class CertManager(object):
         If the specified cert does not exist, a CertificateStorageException
         should be raised.
         """
-        pass
 
     @abc.abstractmethod
     def delete_cert(self, context, cert_ref, resource_ref, service_name=None):
@@ -58,4 +56,19 @@ class CertManager(object):
         If the specified cert does not exist, a CertificateStorageException
         should be raised.
         """
-        pass
+
+    @abc.abstractmethod
+    def set_acls(self, context, cert_ref):
+        """Adds ACLs so Octavia can access the cert objects.
+
+        If the specified cert does not exist or the addition of ACLs fails for
+        any reason, a CertificateStorageException should be raised.
+        """
+
+    @abc.abstractmethod
+    def unset_acls(self, context, cert_ref):
+        """Remove ACLs so Octavia can access the cert objects.
+
+        If the specified cert does not exist or the removal of ACLs fails for
+        any reason, a CertificateStorageException should be raised.
+        """

octavia/certificates/manager/local.py

@@ -160,3 +160,11 @@ class LocalCertManager(cert_mgr.CertManager):
         except IOError as ioe:
             LOG.error("Failed to delete certificate %s", cert_ref)
             raise exceptions.CertificateStorageException(message=ioe.message)
+
+    def set_acls(self, context, cert_ref):
+        # There is no security on this store, because it's really dumb
+        pass
+
+    def unset_acls(self, context, cert_ref):
+        # There is no security on this store, because it's really dumb
+        pass

octavia/common/keystone.py

@@ -28,6 +28,7 @@ class KeystoneSession(object):
 
     def __init__(self, section=constants.SERVICE_AUTH):
         self._session = None
+        self._auth = None
 
         self.section = section
         ks_loading.register_auth_conf_options(cfg.CONF, self.section)
@@ -39,13 +40,20 @@ class KeystoneSession(object):
         :return: a Keystone Session object
         """
         if not self._session:
-            self._auth = ks_loading.load_auth_from_conf_options(
-                cfg.CONF, self.section)
             self._session = ks_loading.load_session_from_conf_options(
-                cfg.CONF, self.section, auth=self._auth)
+                cfg.CONF, self.section, auth=self.get_auth())
 
         return self._session
 
+    def get_auth(self):
+        if not self._auth:
+            self._auth = ks_loading.load_auth_from_conf_options(
+                cfg.CONF, self.section)
+        return self._auth
+
+    def get_service_user_id(self):
+        return self.get_auth().get_user_id(self.get_session())
+
 
 class SkippingAuthProtocol(auth_token.AuthProtocol):
     """SkippingAuthProtocol to reach special endpoints

octavia/tests/unit/amphorae/drivers/haproxy/test_rest_api_driver.py

@@ -108,11 +108,12 @@ class TestHaproxyAmphoraLoadBalancerDriverTest(base.TestCase):
         # this is called 3 times
         gcm_calls = [
             mock.call(self.amp, self.sl.id,
-                      self.sl.default_tls_container.id + '.pem'),
+                      self.sl.default_tls_container.id + '.pem',
+                      ignore=(404,)),
             mock.call(self.amp, self.sl.id,
-                      sconts[0].id + '.pem'),
+                      sconts[0].id + '.pem', ignore=(404,)),
             mock.call(self.amp, self.sl.id,
-                      sconts[1].id + '.pem')
+                      sconts[1].id + '.pem', ignore=(404,))
         ]
         self.driver.client.get_cert_md5sum.assert_has_calls(gcm_calls,
                                                             any_order=True)

octavia/tests/unit/certificates/common/auth/test_barbican_acl.py

@@ -12,10 +12,12 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from barbicanclient.v1 import acls
 import mock
 from oslo_config import cfg
 from oslo_config import fixture as oslo_fixture
 
+
 import octavia.certificates.common.auth.barbican_acl as barbican_acl
 import octavia.certificates.manager.barbican as barbican_cert_mgr
 from octavia.common import keystone
@@ -27,12 +29,12 @@ CONF = cfg.CONF
 class TestBarbicanACLAuth(base.TestCase):
 
     def setUp(self):
+        super(TestBarbicanACLAuth, self).setUp()
         # Reset the client
         keystone._SESSION = None
-        conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
-        conf.config(group="certificates", region_name=None)
-        conf.config(group="certificates", endpoint_type='publicURL')
-        super(TestBarbicanACLAuth, self).setUp()
+        self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
+        self.conf.config(group="certificates", region_name=None)
+        self.conf.config(group="certificates", endpoint_type='publicURL')
 
     @mock.patch('keystoneauth1.session.Session', mock.Mock())
     def test_get_barbican_client(self):
@@ -56,3 +58,36 @@ class TestBarbicanACLAuth(base.TestCase):
     def test_load_auth_driver(self):
         bcm = barbican_cert_mgr.BarbicanCertManager()
         self.assertIsInstance(bcm.auth, barbican_acl.BarbicanACLAuth)
+
+    @mock.patch('barbicanclient.v1.acls.ACLManager.get')
+    @mock.patch('octavia.common.keystone.KeystoneSession')
+    def test_ensure_secret_access(self, mock_ksession, mock_aclm):
+        acl = mock.MagicMock(spec=acls.SecretACL)
+        mock_aclm.return_value = acl
+
+        acl_auth_object = barbican_acl.BarbicanACLAuth()
+        acl_auth_object.ensure_secret_access(mock.Mock(), mock.Mock())
+        acl.submit.assert_called_once()
+
+    @mock.patch('barbicanclient.v1.acls.ACLManager.get')
+    @mock.patch('octavia.common.keystone.KeystoneSession')
+    def test_revoke_secret_access(self, mock_ksession, mock_aclm):
+        service_user_id = 'uuid1'
+
+        mock_ksession().get_service_user_id.return_value = service_user_id
+        acl = mock.MagicMock(spec=acls.SecretACL)
+        poacl = mock.MagicMock(spec=acls._PerOperationACL)
+        type(poacl).users = mock.PropertyMock(return_value=[service_user_id])
+        acl.get.return_value = poacl
+        mock_aclm.return_value = acl
+
+        acl_auth_object = barbican_acl.BarbicanACLAuth()
+        acl_auth_object.revoke_secret_access(mock.Mock(), mock.Mock())
+        acl.submit.assert_called_once()
+
+    @mock.patch('octavia.common.keystone.KeystoneSession')
+    def test_get_barbican_client_user_auth(self, mock_ksession):
+        acl_auth_object = barbican_acl.BarbicanACLAuth()
+        bc = acl_auth_object.get_barbican_client_user_auth(mock.Mock())
+        self.assertTrue(hasattr(bc, 'containers') and
+                        hasattr(bc.containers, 'register_consumer'))

octavia/tests/unit/certificates/manager/test_barbican.py

@@ -148,3 +148,14 @@ class TestBarbicanManager(base.TestCase):
             url=self.secret_ref,
             name='Octavia'
         )
+
+    def test_set_acls(self):
+        self.cert_manager.set_acls(
+            context=self.context,
+            cert_ref=self.secret_ref
+        )
+
+        # our mock_bc should have one call to ensure_secret_access
+        self.cert_manager.auth.ensure_secret_access.assert_called_once_with(
+            self.context, self.secret_ref
+        )

octavia/tests/unit/certificates/manager/test_barbican_legacy.py

@@ -85,6 +85,7 @@ class TestBarbicanManager(base.TestCase):
 
         # Mock out the client
         self.bc = mock.Mock()
+        self.bc.containers.get.return_value = self.container
         barbican_auth = mock.Mock(spec=barbican_common.BarbicanAuth)
         barbican_auth.get_barbican_client.return_value = self.bc
 
@@ -267,3 +268,19 @@ class TestBarbicanManager(base.TestCase):
             url=self.container_ref,
             name='Octavia'
         )
+
+    def test_set_acls(self):
+        self.cert_manager.set_acls(
+            context=self.context,
+            cert_ref=self.container_ref
+        )
+
+        # our mock_bc should have one call to ensure_secret_access for each
+        # of our secrets, and the container
+        self.cert_manager.auth.ensure_secret_access.assert_has_calls([
+            mock.call(self.context, self.container_ref),
+            mock.call(self.context, self.certificate_uuid),
+            mock.call(self.context, self.intermediates_uuid),
+            mock.call(self.context, self.private_key_uuid),
+            mock.call(self.context, self.private_key_passphrase_uuid)
+        ], any_order=True)

releasenotes/notes/add-ability-setting-barbican-acls-85f36747d4284035.yaml

@@ -0,0 +1,8 @@
+---
+features:
+  - |
+    Added ability for Octavia to automatically set Barbican ACLs on behalf of
+    the user. Such enables users to create TLS-terminated listeners without
+    having to add the Octavia keystone user id to the ACL list. Octavia will
+    also automatically revoke access to secrets whenever load balancing
+    resources no longer require access to them.