openstack/octavia
Code Issues Proposed changes

Switch to using PKCS12 for TLS Term certs

Browse Source 
*NOT* deprecating the old way of storing these, as I believe that would
create a huge mess for anyone already using it.

Change-Id: I1fee174d8b8956f3d2053781a7f18c2940b21765
This commit is contained in:
Adam Harwell 2017-09-14 13:17:37 -06:00 committed by Michael Johnson
parent d7cc05be39
commit 8934a629df
22 changed files with 801 additions and 389 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
api-ref/source/parameters.yaml
View File

@ -245,17 +245,21 @@ default_pool_id-optional:
type: string
default_tls_container_ref:
description: |
The URI to the `key manager service
<https://docs.openstack.org/barbican/latest/>`__ secrets container
containing the certificate and key for ``TERMINATED_TLS`` listeners.
The URI of the `key manager service
<https://docs.openstack.org/castellan/latest/>`__ secret containing a
PKCS12 format certificate/key bundle for ``TERMINATED_TLS`` listeners.
DEPRECATED: A secret container of type "certificate" containing the
certificate and key for ``TERMINATED_TLS`` listeners.
in: body
required: true
type: string
default_tls_container_ref-optional:
description: |
The URI to the `key manager service
<https://docs.openstack.org/barbican/latest/>`__ secrets container
containing the certificate and key for ``TERMINATED_TLS`` listeners.
The URI of the `key manager service
<https://docs.openstack.org/castellan/latest/>`__ secret containing a
PKCS12 format certificate/key bundle for ``TERMINATED_TLS`` listeners.
DEPRECATED: A secret container of type "certificate" containing the
certificate and key for ``TERMINATED_TLS`` listeners.
in: body
required: false
type: string
@ -963,18 +967,20 @@ session_persistence_type:
sni_container_refs:
description: |
A list of URIs to the `key manager service
<https://docs.openstack.org/barbican/latest/>`__ secrets containers
containing the certificates and keys for ``TERMINATED_TLS`` the listener
using Server Name Indication.
<https://docs.openstack.org/barbican/latest/>`__ secrets containing
PKCS12 format certificate/key bundles for ``TERMINATED_TLS`` listeners.
(DEPRECATED) Secret containers of type "certificate" containing the
certificates and keys for ``TERMINATED_TLS`` listeners.
in: body
required: true
type: array
sni_container_refs-optional:
description: |
A list of URIs to the `key manager service
<https://docs.openstack.org/barbican/latest/>`__ secrets containers
containing the certificates and keys for ``TERMINATED_TLS`` the listener
using Server Name Indication.
<https://docs.openstack.org/barbican/latest/>`__ secrets containing
PKCS12 format certificate/key bundles for ``TERMINATED_TLS`` listeners.
(DEPRECATED) Secret containers of type "certificate" containing the
certificates and keys for ``TERMINATED_TLS`` listeners.
in: body
required: false
type: array

4
api-ref/source/v1/octaviaapi.rst
View File

@ -1,5 +1,5 @@
Octavia API v1 (SUPORTED)
=========================
Octavia API v1 (SUPPORTED)
==========================
Authentication
--------------

189
doc/source/user/guides/basic-cookbook.rst
View File

@ -365,34 +365,30 @@ balancer features, like Layer 7 features and header manipulation.
* Back-end servers 192.0.2.10 and 192.0.2.11 on subnet *private-subnet* have
been configured with regular HTTP application on TCP port 80.
* These back-end servers have been configured with a health check at the URL
path "/healthcheck". See :ref:`http-heath-monitors` below.
* Subnet *public-subnet* is a shared external subnet created by the cloud
operator which is reachable from the internet.
* A TLS certificate, key, and intermediate certificate chain for
www.example.com have been obtained from an external certificate authority.
These now exist in the files server.crt, server.key, and ca-chain.p7b in the
These now exist in the files server.crt, server.key, and ca-chain.crt in the
current directory. The key and certificate are PEM-encoded, and the
intermediate certificate chain is PKCS7 PEM encoded. The key is not encrypted
with a passphrase.
intermediate certificate chain is multiple PEM-encoded certs concatenated
together. The key is not encrypted with a passphrase.
* The *admin* user on this cloud installation has keystone ID *admin_id*
* We want to configure a TLS-terminated HTTPS load balancer that is accessible
from the internet using the key and certificate mentioned above, which
distributes requests to the back-end servers over the non-encrypted HTTP
protocol.
* Octavia is configured to use barbican for key management.
**Solution**:
1. Create barbican *secret* resources for the certificate, key, and
intermediate certificate chain. We will call these *cert1*, *key1*, and
*intermediates1* respectively.
2. Create a *secret container* resource combining all of the above. We will
call this *tls_container1*.
3. Grant the *admin* user access to all the *secret* and *secret container*
barbican resources above.
1. Combine the individual cert/key/intermediates to a single PKCS12 file.
2. Create a barbican *secret* resource for the PKCS12 file. We will call
this *tls_secret1*.
3. Grant the *admin* user access to the *tls_secret1* barbican resource.
4. Create load balancer *lb1* on subnet *public-subnet*.
5. Create listener *listener1* as a TERMINATED_HTTPS listener referencing
*tls_container1* as its default TLS container.
*tls_secret1* as its default TLS container.
6. Create pool *pool1* as *listener1*'s default pool.
7. Add members 192.0.2.10 and 192.0.2.11 on *private-subnet* to *pool1*.
@ -400,21 +396,16 @@ balancer features, like Layer 7 features and header manipulation.
::
openstack secret store --name='cert1' --payload-content-type='text/plain' --payload="$(cat server.crt)"
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
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:
neutron lbaas-loadbalancer-show lb1
neutron lbaas-listener-create --loadbalancer lb1 --protocol-port 443 --protocol TERMINATED_HTTPS --name listener1 --default-tls-container=$(openstack secret container list | awk '/ tls_container1 / {print $2}')
neutron lbaas-pool-create --name pool1 --lb-algorithm ROUND_ROBIN --listener listener1 --protocol HTTP
neutron lbaas-member-create --subnet private-subnet --address 192.0.2.10 --protocol-port 80 pool1
neutron lbaas-member-create --subnet private-subnet --address 192.0.2.11 --protocol-port 80 pool1
openstack loadbalancer show lb1
openstack loadbalancer listener create --protocol-port 443 --protocol TERMINATED_HTTPS --name listener1 --default-tls-container=$(openstack secret list | awk '/ tls_secret1 / {print $2}' lb1
openstack loadbalancer pool create --name pool1 --lb-algorithm ROUND_ROBIN --listener listener1 --protocol HTTP
openstack loadbalancer member create --subnet-id private-subnet --address 192.0.2.10 --protocol-port 80 pool1
openstack loadbalancer member create --subnet-id private-subnet --address 192.0.2.11 --protocol-port 80 pool1
Deploy a TLS-terminated HTTPS load balancer with SNI
@ -427,18 +418,15 @@ listener using Server Name Indication (SNI) technology.
* Back-end servers 192.0.2.10 and 192.0.2.11 on subnet *private-subnet* have
been configured with regular HTTP application on TCP port 80.
* These back-end servers have been configured with a health check at the URL
path "/healthcheck". See :ref:`http-heath-monitors` below.
* Subnet *public-subnet* is a shared external subnet created by the cloud
operator which is reachable from the internet.
* TLS certificates, keys, and intermediate certificate chains for
www.example.com and www2.example.com have been obtained from an external
certificate authority. These now exist in the files server.crt, server.key,
ca-chain.p7b, server2.crt, server2-encrypted.key, and ca-chain2.p7b in the
ca-chain.crt, server2.crt, server2.key, and ca-chain2.crt in the
current directory. The keys and certificates are PEM-encoded, and the
intermediate certificate chains are PKCS7 PEM encoded.
* The key for www.example.com is not encrypted with a passphrase.
* The key for www2.example.com is encrypted with the passphrase "abc123".
intermediate certificate chains are multiple certs PEM-encoded and
concatenated together. Neither key is encrypted with a passphrase.
* The *admin* user on this cloud installation has keystone ID *admin_id*
* We want to configure a TLS-terminated HTTPS load balancer that is accessible
from the internet using the keys and certificates mentioned above, which
@ -449,50 +437,34 @@ listener using Server Name Indication (SNI) technology.
**Solution**:
1. Create barbican *secret* resources for the certificates, keys, and
intermediate certificate chains. We will call these *cert1*, *key1*,
*intermediates1*, *cert2*, *key2* and *intermediates2* respectively.
2. Create a barbican *secret* resource *passphrase2* for the passphrase for
*key2*
3. Create *secret container* resources combining the above appropriately. We
will call these *tls_container1* and *tls_container2*.
4. Grant the *admin* user access to all the *secret* and *secret container*
barbican resources above.
5. Create load balancer *lb1* on subnet *public-subnet*.
6. Create listener *listener1* as a TERMINATED_HTTPS listener referencing
*tls_container1* as its default TLS container, and referencing both
*tls_container1* and *tls_container2* using SNI.
7. Create pool *pool1* as *listener1*'s default pool.
8. Add members 192.0.2.10 and 192.0.2.11 on *private-subnet* to *pool1*.
1. Combine the individual cert/key/intermediates to single PKCS12 files.
2. Create barbican *secret* resources for the PKCS12 files. We will call them
*tls_secret1* and *tls_secret2*.
3. Grant the *admin* user access to both *tls_secret* barbican resources.
4. Create load balancer *lb1* on subnet *public-subnet*.
5. Create listener *listener1* as a TERMINATED_HTTPS listener referencing
*tls_secret1* as its default TLS container, and referencing both
*tls_secret1* and *tls_secret2* using SNI.
6. Create pool *pool1* as *listener1*'s default pool.
7. Add members 192.0.2.10 and 192.0.2.11 on *private-subnet* to *pool1*.
**CLI commands**:
::
openstack secret store --name='cert1' --payload-content-type='text/plain' --payload="$(cat server.crt)"
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 secret store --name='cert2' --payload-content-type='text/plain' --payload="$(cat server2.crt)"
openstack secret store --name='key2' --payload-content-type='text/plain' --payload="$(cat server2-encrypted.key)"
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
openssl pkcs12 -export -inkey server.key -in server.crt -certfile ca-chain.crt -passout pass: -out server.p12
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:
neutron lbaas-loadbalancer-show lb1
neutron lbaas-listener-create --loadbalancer lb1 --protocol-port 443 --protocol TERMINATED_HTTPS --name listener1 --default-tls-container=$(openstack secret container list | awk '/ tls_container1 / {print $2}') --sni-container_refs $(openstack secret container list | awk '/ tls_container1 / {print $2}') $(openstack secret container list | awk '/ tls_container2 / {print $2}')
neutron lbaas-pool-create --name pool1 --lb-algorithm ROUND_ROBIN --listener listener1 --protocol HTTP
neutron lbaas-member-create --subnet private-subnet --address 192.0.2.10 --protocol-port 80 pool1
neutron lbaas-member-create --subnet private-subnet --address 192.0.2.11 --protocol-port 80 pool1
openstack loadbalancer show lb1
openstack loadbalancer listener create --protocol-port 443 --protocol TERMINATED_HTTPS --name listener1 --default-tls-container=$(openstack secret list | awk '/ tls_secret1 / {print $2}' --sni-container_refs $(openstack secret list | awk '/ tls_secret1 / {print $2}') $(openstack secret list | awk '/ tls_secret2 / {print $2}') lb1
openstack loadbalancer pool create --name pool1 --lb-algorithm ROUND_ROBIN --listener listener1 --protocol HTTP
openstack loadbalancer member create --subnet-id private-subnet --address 192.0.2.10 --protocol-port 80 pool1
openstack loadbalancer member create --subnet-id private-subnet --address 192.0.2.11 --protocol-port 80 pool1
Deploy HTTP and TLS-terminated HTTPS load balancing on the same IP and backend
@ -512,16 +484,14 @@ HTTP just get redirected to the HTTPS listener), then please see `the example
* Back-end servers 192.0.2.10 and 192.0.2.11 on subnet *private-subnet* have
been configured with regular HTTP application on TCP port 80.
* These back-end servers have been configured with a health check at the URL
path "/healthcheck". See :ref:`http-heath-monitors` below.
* Subnet *public-subnet* is a shared external subnet created by the cloud
operator which is reachable from the internet.
* A TLS certificate, key, and intermediate certificate chain for
www.example.com have been obtained from an external certificate authority.
These now exist in the files server.crt, server.key, and ca-chain.p7b in the
These now exist in the files server.crt, server.key, and ca-chain.crt in the
current directory. The key and certificate are PEM-encoded, and the
intermediate certificate chain is PKCS7 PEM encoded. The key is not encrypted
with a passphrase.
intermediate certificate chain is multiple PEM-encoded certs concatenated
together. The key is not encrypted with a passphrase.
* The *admin* user on this cloud installation has keystone ID *admin_id*
* We want to configure a TLS-terminated HTTPS load balancer that is accessible
from the internet using the key and certificate mentioned above, which
@ -533,16 +503,13 @@ HTTP just get redirected to the HTTPS listener), then please see `the example
**Solution**:
1. Create barbican *secret* resources for the certificate, key, and
intermediate certificate chain. We will call these *cert1*, *key1*, and
*intermediates1* respectively.
2. Create a *secret container* resource combining all of the above. We will
call this *tls_container1*.
3. Grant the *admin* user access to all the *secret* and *secret container*
barbican resources above.
1. Combine the individual cert/key/intermediates to a single PKCS12 file.
2. Create a barbican *secret* resource for the PKCS12 file. We will call
this *tls_secret1*.
3. Grant the *admin* user access to the *tls_secret1* barbican resource.
4. Create load balancer *lb1* on subnet *public-subnet*.
5. Create listener *listener1* as a TERMINATED_HTTPS listener referencing
*tls_container1* as its default TLS container.
*tls_secret1* as its default TLS container.
6. Create pool *pool1* as *listener1*'s default pool.
7. Add members 192.0.2.10 and 192.0.2.11 on *private-subnet* to *pool1*.
8. Create listener *listener2* as an HTTP listener with *pool1* as its
@ -552,22 +519,18 @@ HTTP just get redirected to the HTTPS listener), then please see `the example
::
openstack secret store --name='cert1' --payload-content-type='text/plain' --payload="$(cat server.crt)"
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
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:
neutron lbaas-loadbalancer-show lb1
neutron lbaas-listener-create --loadbalancer lb1 --protocol-port 443 --protocol TERMINATED_HTTPS --name listener1 --default-tls-container=$(openstack secret container list | awk '/ tls_container1 / {print $2}')
neutron lbaas-pool-create --name pool1 --lb-algorithm ROUND_ROBIN --listener listener1 --protocol HTTP
neutron lbaas-member-create --subnet private-subnet --address 192.0.2.10 --protocol-port 80 pool1
neutron lbaas-member-create --subnet private-subnet --address 192.0.2.11 --protocol-port 80 pool1
neutron lbaas-listener-create --name listener2 --loadbalancer lb1 --protocol HTTP --protocol-port 80 --default-pool pool1
openstack loadbalancer show lb1
openstack loadbalancer listener create --protocol-port 443 --protocol TERMINATED_HTTPS --name listener1 --default-tls-container=$(openstack secret list | awk '/ tls_secret1 / {print $2}' lb1
openstack loadbalancer pool create --name pool1 --lb-algorithm ROUND_ROBIN --listener listener1 --protocol HTTP
openstack loadbalancer member create --subnet-id private-subnet --address 192.0.2.10 --protocol-port 80 pool1
openstack loadbalancer member create --subnet-id private-subnet --address 192.0.2.11 --protocol-port 80 pool1
openstack secret store --name='tls_secret1' --payload-content-type='text/plain' --payload="$(cat server.crt)"
openstack loadbalancer listener create --protocol-port 80 --protocol HTTP --name listener2 --default-pool pool1 lb1
.. _heath-monitor-best-practices:
@ -691,39 +654,33 @@ blocks of 64-character lines of ASCII text (that will look like gobbedlygook to
a human). These files are also typically named with a ``.crt`` or ``.pem``
extension.
To upload this type of intermediates chain to barbican, run a command similar
to the following (assuming "intermediates-chain.pem" is the name of the file):
::
openstack secret store --name='intermediates1' --payload-content-type='text/plain' --payload="$(cat intermediates-chain.pem)"
DER-encoded chains
------------------
If the intermediates chain provided to you is a file that contains what appears
to be random binary data, it is likely that it is a PKCS7 chain in DER format.
These files also may be named with a ``.p7b`` extension. In order to use this
intermediates chain, you can either convert it to a series of PEM-encoded
certificates with the following command:
These files also may be named with a ``.p7b`` extension.
You may use the binary DER file as-is when building your PKCS12 bundle:
::
openssl pkcs7 -in intermediates-chain.p7b -inform DER -print_certs -out intermediates-chain.pem
openssl pkcs12 -export -inkey server.key -in server.crt -certfile ca-chain.p7b -passout pass: -out server.p12
...or convert it into a PEM-encoded PKCS7 bundle with the following command:
... or you can convert it to a series of PEM-encoded certificates:
::
openssl pkcs7 -in intermediates-chain.p7b -inform DER -outform PEM -out intermediates-chain.pem
openssl pkcs7 -in intermediates-chain.p7b -inform DER -print_certs -out intermediates-chain.crt
...or simply upload the binary DER file to barbican without conversion:
... or you can convert it to a PEM-encoded PKCS7 bundle:
::
openstack secret store --name='intermediates1' --payload-content-type='application/octet-stream' --payload-content-encoding='base64' --payload="$(cat intermediates-chain.p7b | base64)"
openssl pkcs7 -in intermediates-chain.p7b -inform DER -outform PEM -out intermediates-chain.crt
In any case, if the file is not a PKCS7 DER bundle, then either of the above
two openssl commands will fail.
If the file is not a PKCS7 DER bundle, either of the two ``openssl pkcs7``
commands will fail.
Further reading
===============

1
etc/octavia.conf
View File

@ -108,6 +108,7 @@
# For the TLS management
# Certificate Manager options are local_cert_manager
# barbican_cert_manager
# castellan_cert_manager
# cert_manager = barbican_cert_manager
# For Barbican authentication (if using any Barbican based cert class)
# barbican_auth = barbican_acl_auth

5
octavia/certificates/common/barbican.py
View File

@ -19,7 +19,7 @@ Common classes for Barbican certificate handling
import abc
from barbicanclient import client as barbican_client
from barbicanclient.v1 import containers
from oslo_utils import encodeutils
import six
@ -31,8 +31,7 @@ from octavia.i18n import _
class BarbicanCert(cert.Cert):
"""Representation of a Cert based on the Barbican CertificateContainer."""
def __init__(self, cert_container):
if not isinstance(cert_container,
barbican_client.containers.CertificateContainer):
if not isinstance(cert_container, containers.CertificateContainer):
raise TypeError(_("Retrieved Barbican Container is not of the "
"correct type (certificate)."))
self._cert_container = cert_container

54
octavia/certificates/common/pkcs12.py Normal file
View File

@ -0,0 +1,54 @@
# 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.
"""
Common classes for pkcs12 based certificate handling
"""
from cryptography.hazmat.primitives import serialization
from OpenSSL import crypto
from octavia.certificates.common import cert
class PKCS12Cert(cert.Cert):
"""Representation of a Cert for local storage."""
def __init__(self, certbag):
p12 = crypto.load_pkcs12(certbag)
self.certificate = p12.get_certificate()
self.intermediates = p12.get_ca_certificates()
self.private_key = p12.get_privatekey()
def get_certificate(self):
return self.certificate.to_cryptography().public_bytes(
encoding=serialization.Encoding.PEM).strip()
def get_intermediates(self):
if self.intermediates:
int_data = [
ic.to_cryptography().public_bytes(
encoding=serialization.Encoding.PEM).strip()
for ic in self.intermediates
]
return int_data
def get_private_key(self):
return self.private_key.to_cryptography_key().private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption()).strip()
def get_private_key_passphrase(self):
return None

147
octavia/certificates/manager/barbican.py
View File

@ -1,4 +1,5 @@
# Copyright (c) 2014 Rackspace US, Inc
# Copyright (c) 2017 GoDaddy
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -14,15 +15,21 @@
# under the License.
"""
Cert manager implementation for Barbican
Cert manager implementation for Barbican using a single PKCS12 secret
"""
from OpenSSL import crypto
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import encodeutils
from oslo_utils import excutils
from stevedore import driver as stevedore_driver
from octavia.certificates.common import barbican as barbican_common
from octavia.certificates.common import pkcs12
from octavia.certificates.manager import barbican_legacy
from octavia.certificates.manager import cert_mgr
from octavia.common import exceptions
from octavia.common.tls_utils import cert_parser
LOG = logging.getLogger(__name__)
@ -38,11 +45,12 @@ class BarbicanCertManager(cert_mgr.CertManager):
invoke_on_load=True,
).driver
def store_cert(self, project_id, certificate, private_key,
intermediates=None, private_key_passphrase=None,
expiration=None, name='Octavia TLS Cert'):
def store_cert(self, context, certificate, private_key, intermediates=None,
private_key_passphrase=None, expiration=None,
name="PKCS12 Certificate Bundle"):
"""Stores a certificate in the certificate manager.
:param context: Oslo context of the request
:param certificate: PEM encoded TLS certificate
:param private_key: private key for the supplied certificate
:param intermediates: ordered and concatenated intermediate certs
@ -53,68 +61,42 @@ class BarbicanCertManager(cert_mgr.CertManager):
:returns: the container_ref of the stored cert
:raises Exception: if certificate storage fails
"""
connection = self.auth.get_barbican_client(project_id)
connection = self.auth.get_barbican_client(context.project_id)
LOG.info("Storing certificate container '%s' in Barbican.", name)
certificate_secret = None
private_key_secret = None
intermediates_secret = None
pkp_secret = None
LOG.info("Storing certificate secret '%s' in Barbican.", name)
p12 = crypto.PKCS12()
p12.set_friendlyname(encodeutils.to_utf8(name))
x509_cert = crypto.load_certificate(crypto.FILETYPE_PEM, certificate)
p12.set_certificate(x509_cert)
x509_pk = crypto.load_privatekey(crypto.FILETYPE_PEM, private_key)
p12.set_privatekey(x509_pk)
if intermediates:
cert_ints = list(cert_parser.get_intermediates_pems(intermediates))
x509_ints = [
crypto.load_certificate(crypto.FILETYPE_PEM, ci)
for ci in cert_ints]
p12.set_ca_certificates(x509_ints)
if private_key_passphrase:
raise exceptions.CertificateStorageException(
"Passphrase protected PKCS12 certificates are not supported.")
try:
certificate_secret = connection.secrets.create(
payload=certificate,
payload=p12.export(),
expiration=expiration,
name="Certificate"
name=name
)
private_key_secret = connection.secrets.create(
payload=private_key,
expiration=expiration,
name="Private Key"
)
certificate_container = connection.containers.create_certificate(
name=name,
certificate=certificate_secret,
private_key=private_key_secret
)
if intermediates:
intermediates_secret = connection.secrets.create(
payload=intermediates,
expiration=expiration,
name="Intermediates"
)
certificate_container.intermediates = intermediates_secret
if private_key_passphrase:
pkp_secret = connection.secrets.create(
payload=private_key_passphrase,
expiration=expiration,
name="Private Key Passphrase"
)
certificate_container.private_key_passphrase = pkp_secret
certificate_container.store()
return certificate_container.container_ref
certificate_secret.store()
return certificate_secret.secret_ref
except Exception as e:
for i in [certificate_secret, private_key_secret,
intermediates_secret, pkp_secret]:
if i and i.secret_ref:
old_ref = i.secret_ref
try:
i.delete()
LOG.info('Deleted secret %s (%s) during rollback.',
i.name, old_ref)
except Exception:
LOG.warning('Failed to delete %s (%s) during '
'rollback. This might not be a problem.',
i.name, old_ref)
with excutils.save_and_reraise_exception():
LOG.error('Error storing certificate data: %s', e)
def get_cert(self, project_id, cert_ref, resource_ref=None,
check_only=False, service_name='Octavia'):
def get_cert(self, context, cert_ref, resource_ref=None, check_only=False,
service_name=None):
"""Retrieves the specified cert and registers as a consumer.
:param context: Oslo context of the request
:param cert_ref: the UUID of the cert to retrieve
:param resource_ref: Full HATEOAS reference to the consuming resource
:param check_only: Read Certificate data without registering
@ -124,45 +106,40 @@ class BarbicanCertManager(cert_mgr.CertManager):
certificate data
:raises Exception: if certificate retrieval fails
"""
connection = self.auth.get_barbican_client(project_id)
connection = self.auth.get_barbican_client(context.project_id)
LOG.info('Loading certificate container %s from Barbican.', cert_ref)
LOG.info('Loading certificate secret %s from Barbican.', cert_ref)
try:
if check_only:
cert_container = connection.containers.get(
container_ref=cert_ref
)
else:
cert_container = connection.containers.register_consumer(
container_ref=cert_ref,
name=service_name,
url=resource_ref
)
return barbican_common.BarbicanCert(cert_container)
except Exception as e:
with excutils.save_and_reraise_exception():
LOG.error('Error getting %s: %s', cert_ref, e)
cert_secret = connection.secrets.get(secret_ref=cert_ref)
return pkcs12.PKCS12Cert(cert_secret.payload)
except Exception:
# If our get fails, try with the legacy driver.
# TODO(rm_work): Remove this code when the deprecation cycle for
# the legacy driver is complete.
legacy_mgr = barbican_legacy.BarbicanCertManager()
legacy_cert = legacy_mgr.get_cert(
context, cert_ref, resource_ref=resource_ref,
check_only=check_only, service_name=service_name
)
return legacy_cert
def delete_cert(self, project_id, cert_ref, resource_ref=None,
service_name='Octavia'):
def delete_cert(self, context, cert_ref, resource_ref, service_name=None):
"""Deregister as a consumer for the specified cert.
:param context: Oslo context of the request
:param cert_ref: the UUID of the cert to retrieve
:param resource_ref: Full HATEOAS reference to the consuming resource
:param service_name: Friendly name for the consuming service
:raises Exception: if deregistration fails
"""
connection = self.auth.get_barbican_client(project_id)
LOG.info('Deregistering as a consumer of %s in Barbican.', cert_ref)
# TODO(rm_work): We won't take any action on a delete in this driver,
# but for now try the legacy driver's delete and ignore failure.
try:
connection.containers.remove_consumer(
container_ref=cert_ref,
name=service_name,
url=resource_ref
)
except Exception as e:
with excutils.save_and_reraise_exception():
LOG.error('Error deregistering as a consumer of %s: %s',
cert_ref, e)
legacy_mgr = barbican_legacy.BarbicanCertManager(auth=self.auth)
legacy_mgr.delete_cert(
context, cert_ref, resource_ref, service_name=service_name)
except Exception:
# If the delete failed, it was probably because it isn't legacy
# (this will be fixed once Secrets have Consumer registration).
pass

172
octavia/certificates/manager/barbican_legacy.py Normal file
View File

@ -0,0 +1,172 @@
# Copyright (c) 2014 Rackspace US, Inc
# 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.
"""
Legacy cert manager implementation for Barbican (container+secrets)
"""
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import excutils
from stevedore import driver as stevedore_driver
from octavia.certificates.common import barbican as barbican_common
from octavia.certificates.manager import cert_mgr
LOG = logging.getLogger(__name__)
class BarbicanCertManager(cert_mgr.CertManager):
"""Certificate Manager that wraps the Barbican client API."""
def __init__(self, auth=None):
super(BarbicanCertManager, self).__init__()
if auth:
self.auth = auth
else:
self.auth = stevedore_driver.DriverManager(
namespace='octavia.barbican_auth',
name=cfg.CONF.certificates.barbican_auth,
invoke_on_load=True,
).driver
def store_cert(self, context, certificate, private_key, intermediates=None,
private_key_passphrase=None, expiration=None, name=None):
"""Stores a certificate in the certificate manager.
:param context: Oslo context of the request
:param certificate: PEM encoded TLS certificate
:param private_key: private key for the supplied certificate
:param intermediates: ordered and concatenated intermediate certs
:param private_key_passphrase: optional passphrase for the supplied key
:param expiration: the expiration time of the cert in ISO 8601 format
:param name: a friendly name for the cert
:returns: the container_ref of the stored cert
:raises Exception: if certificate storage fails
"""
connection = self.auth.get_barbican_client(context.project_id)
LOG.info("Storing certificate container '%s' in Barbican.", name)
certificate_secret = None
private_key_secret = None
intermediates_secret = None
pkp_secret = None
try:
certificate_secret = connection.secrets.create(
payload=certificate,
expiration=expiration,
name="Certificate"
)
private_key_secret = connection.secrets.create(
payload=private_key,
expiration=expiration,
name="Private Key"
)
certificate_container = connection.containers.create_certificate(
name=name,
certificate=certificate_secret,
private_key=private_key_secret
)
if intermediates:
intermediates_secret = connection.secrets.create(
payload=intermediates,
expiration=expiration,
name="Intermediates"
)
certificate_container.intermediates = intermediates_secret
if private_key_passphrase:
pkp_secret = connection.secrets.create(
payload=private_key_passphrase,
expiration=expiration,
name="Private Key Passphrase"
)
certificate_container.private_key_passphrase = pkp_secret
certificate_container.store()
return certificate_container.container_ref
except Exception as e:
for i in [certificate_secret, private_key_secret,
intermediates_secret, pkp_secret]:
if i and i.secret_ref:
old_ref = i.secret_ref
try:
i.delete()
LOG.info('Deleted secret %s (%s) during rollback.',
i.name, old_ref)
except Exception:
LOG.warning('Failed to delete %s (%s) during '
'rollback. This might not be a problem.',
i.name, old_ref)
with excutils.save_and_reraise_exception():
LOG.error('Error storing certificate data: %s', e)
def get_cert(self, context, cert_ref, resource_ref=None, check_only=False,
service_name=None):
"""Retrieves the specified cert and registers as a consumer.
:param context: Oslo context of the request
:param cert_ref: the UUID of the cert to retrieve
:param resource_ref: Full HATEOAS reference to the consuming resource
:param check_only: Read Certificate data without registering
:param service_name: Friendly name for the consuming service
:return: octavia.certificates.common.Cert representation of the
certificate data
:raises Exception: if certificate retrieval fails
"""
connection = self.auth.get_barbican_client(context.project_id)
LOG.info('Loading certificate container %s from Barbican.', cert_ref)
try:
if check_only:
cert_container = connection.containers.get(
container_ref=cert_ref
)
else:
cert_container = connection.containers.register_consumer(
container_ref=cert_ref,
name=service_name,
url=resource_ref
)
return barbican_common.BarbicanCert(cert_container)
except Exception as e:
with excutils.save_and_reraise_exception():
LOG.error('Error getting %s: %s', cert_ref, e)
def delete_cert(self, context, cert_ref, resource_ref, service_name=None):
"""Deregister as a consumer for the specified cert.
:param context: Oslo context of the request
:param cert_ref: the UUID of the cert to retrieve
:param resource_ref: Full HATEOAS reference to the consuming resource
:param service_name: Friendly name for the consuming service
:raises Exception: if deregistration fails
"""
connection = self.auth.get_barbican_client(context.project_id)
LOG.info('Deregistering as a consumer of %s in Barbican.', cert_ref)
try:
connection.containers.remove_consumer(
container_ref=cert_ref,
name=service_name,
url=resource_ref
)
except Exception as e:
with excutils.save_and_reraise_exception():
LOG.error('Error deregistering as a consumer of %s: %s',
cert_ref, e)

63
octavia/certificates/manager/castellan_mgr.py Normal file
View File

@ -0,0 +1,63 @@
# 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 OpenSSL import crypto
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
LOG = logging.getLogger(__name__)
class CastellanCertManager(cert_mgr.CertManager):
"""Certificate Manager for the Castellan library."""
def __init__(self):
super(CastellanCertManager, self).__init__()
self.manager = key_manager.API()
def store_cert(self, context, certificate, private_key, intermediates=None,
private_key_passphrase=None, expiration=None,
name="PKCS12 Certificate Bundle"):
p12 = crypto.PKCS12()
p12.set_certificate(certificate)
p12.set_privatekey(private_key)
if intermediates:
p12.set_ca_certificates(intermediates)
if private_key_passphrase:
raise exceptions.CertificateStorageException(
"Passphrases protected PKCS12 certificates are not supported.")
p12_data = opaque_data.OpaqueData(p12.export(), 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

12
octavia/certificates/manager/cert_mgr.py
View File

@ -29,9 +29,8 @@ class CertManager(object):
"""
@abc.abstractmethod
def store_cert(self, project_id, certificate, private_key,
intermediates=None, private_key_passphrase=None,
expiration=None, name=None):
def store_cert(self, context, certificate, private_key, intermediates=None,
private_key_passphrase=None, expiration=None, name=None):
"""Stores (i.e., registers) a cert with the cert manager.
This method stores the specified cert and returns its UUID that
@ -42,8 +41,8 @@ class CertManager(object):
pass
@abc.abstractmethod
def get_cert(self, project_id, cert_ref, resource_ref=None,
check_only=False, service_name=None):
def get_cert(self, context, cert_ref, resource_ref=None, check_only=False,
service_name=None):
"""Retrieves the specified cert.
If check_only is True, don't perform any sort of registration.
@ -53,8 +52,7 @@ class CertManager(object):
pass
@abc.abstractmethod
def delete_cert(self, project_id, cert_ref, resource_ref,
service_name=None):
def delete_cert(self, context, cert_ref, resource_ref, service_name=None):
"""Deletes the specified cert.
If the specified cert does not exist, a CertificateStorageException

12
octavia/certificates/manager/local.py
View File

@ -31,14 +31,14 @@ class LocalCertManager(cert_mgr.CertManager):
"""Cert Manager Interface that stores data locally."""
@staticmethod
def store_cert(project_id, certificate, private_key, intermediates=None,
def store_cert(context, certificate, private_key, intermediates=None,
private_key_passphrase=None, **kwargs):
"""Stores (i.e., registers) a cert with the cert manager.
This method stores the specified cert to the filesystem and returns
a UUID that can be used to retrieve it.
:param project_id: Ignored in this implementation
:param context: Ignored in this implementation
:param certificate: PEM encoded TLS certificate
:param private_key: private key for the supplied certificate
:param intermediates: ordered and concatenated intermediate certs
@ -82,10 +82,10 @@ class LocalCertManager(cert_mgr.CertManager):
return cert_ref
@staticmethod
def get_cert(project_id, cert_ref, **kwargs):
def get_cert(context, cert_ref, **kwargs):
"""Retrieves the specified cert.
:param project_id: Ignored in this implementation
:param context: Ignored in this implementation
:param cert_ref: the UUID of the cert to retrieve
:return: octavia.certificates.common.Cert representation of the
@ -134,10 +134,10 @@ class LocalCertManager(cert_mgr.CertManager):
return local_common.LocalCert(**cert_data)
@staticmethod
def delete_cert(project_id, cert_ref, **kwargs):
def delete_cert(context, cert_ref, **kwargs):
"""Deletes the specified cert.
:param project_id: Ignored in this implementation
:param context: Ignored in this implementation
:param cert_ref: the UUID of the cert to delete
:raises CertificateStorageException: if certificate deletion fails

9
octavia/common/tls_utils/cert_parser.py
View File

@ -18,6 +18,7 @@ import base64
from cryptography.hazmat import backends
from cryptography.hazmat.primitives import serialization
from cryptography import x509
from oslo_context import context as oslo_context
from oslo_log import log as logging
from pyasn1.codec.der import decoder as der_decoder
from pyasn1.codec.der import encoder as der_encoder
@ -332,23 +333,25 @@ def build_pem(tls_container):
return b'\n'.join(pem) + b'\n'
def load_certificates_data(cert_mngr, listener):
def load_certificates_data(cert_mngr, listener, context=None):
"""Load TLS certificate data from the listener.
return TLS_CERT and SNI_CERTS
"""
tls_cert = None
sni_certs = []
if not context:
context = oslo_context.RequestContext(project_id=listener.project_id)
if listener.tls_certificate_id:
tls_cert = _map_cert_tls_container(
cert_mngr.get_cert(listener.project_id,
cert_mngr.get_cert(context,
listener.tls_certificate_id,
check_only=True))
if listener.sni_containers:
for sni_cont in listener.sni_containers:
cert_container = _map_cert_tls_container(
cert_mngr.get_cert(listener.project_id,
cert_mngr.get_cert(context,
sni_cont.tls_container_id,
check_only=True))
sni_certs.append(cert_container)

15
octavia/tests/unit/certificates/common/test_barbican.py
View File

@ -12,7 +12,8 @@
# License for the specific language governing permissions and limitations
# under the License.
from barbicanclient import client as barbican_client
from barbicanclient.v1 import containers
from barbicanclient.v1 import secrets
import mock
import six
@ -24,19 +25,19 @@ import octavia.tests.unit.common.sample_configs.sample_certs as sample
class TestBarbicanCert(base.TestCase):
def _prepare(self):
self.certificate_secret = barbican_client.secrets.Secret(
self.certificate_secret = secrets.Secret(
api=mock.MagicMock(),
payload=self.certificate
)
self.intermediates_secret = barbican_client.secrets.Secret(
self.intermediates_secret = secrets.Secret(
api=mock.MagicMock(),
payload=sample.X509_IMDS
)
self.private_key_secret = barbican_client.secrets.Secret(
self.private_key_secret = secrets.Secret(
api=mock.MagicMock(),
payload=self.private_key
)
self.private_key_passphrase_secret = barbican_client.secrets.Secret(
self.private_key_passphrase_secret = secrets.Secret(
api=mock.MagicMock(),
payload=self.private_key_passphrase
)
@ -49,7 +50,7 @@ class TestBarbicanCert(base.TestCase):
self.private_key_passphrase = sample.X509_CERT_KEY_PASSPHRASE
self._prepare()
container = barbican_client.containers.CertificateContainer(
container = containers.CertificateContainer(
api=mock.MagicMock(),
certificate=self.certificate_secret,
intermediates=self.intermediates_secret,
@ -78,7 +79,7 @@ class TestBarbicanCert(base.TestCase):
sample.X509_CERT_KEY_PASSPHRASE)
self._prepare()
container = barbican_client.containers.CertificateContainer(
container = containers.CertificateContainer(
api=mock.MagicMock(),
certificate=self.certificate_secret,
intermediates=self.intermediates_secret,

196
octavia/tests/unit/certificates/manager/test_barbican.py
View File

@ -14,10 +14,8 @@
import uuid
from barbicanclient import containers
from barbicanclient import secrets
from barbicanclient.v1 import secrets
import mock
import six
import octavia.certificates.common.barbican as barbican_common
import octavia.certificates.common.cert as cert
@ -32,47 +30,21 @@ PROJECT_ID = "12345"
class TestBarbicanManager(base.TestCase):
def setUp(self):
# Make a fake Container and contents
# Make a fake Secret and contents
self.barbican_endpoint = 'http://localhost:9311/v1'
self.container_uuid = uuid.uuid4()
self.secret_uuid = uuid.uuid4()
self.container_ref = '{0}/containers/{1}'.format(
self.barbican_endpoint, self.container_uuid
self.secret_ref = '{0}/secrets/{1}'.format(
self.barbican_endpoint, self.secret_uuid
)
self.name = 'My Fancy Cert'
self.certificate = secrets.Secret(
self.secret = secrets.Secret(
api=mock.MagicMock(),
payload=sample.X509_CERT
)
self.intermediates = secrets.Secret(
api=mock.MagicMock(),
payload=sample.X509_IMDS
)
self.private_key = secrets.Secret(
api=mock.MagicMock(),
payload=sample.X509_CERT_KEY_ENCRYPTED
)
self.private_key_passphrase = secrets.Secret(
api=mock.MagicMock(),
payload=sample.X509_CERT_KEY_PASSPHRASE
payload=sample.PKCS12_BUNDLE
)
container = mock.Mock(spec=containers.CertificateContainer)
container.container_ref = self.container_ref
container.name = self.name
container.private_key = self.private_key
container.certificate = self.certificate
container.intermediates = self.intermediates
container.private_key_passphrase = self.private_key_passphrase
self.container = container
self.empty_container = mock.Mock(spec=containers.CertificateContainer)
self.secret1 = mock.Mock(spec=secrets.Secret)
self.secret2 = mock.Mock(spec=secrets.Secret)
self.secret3 = mock.Mock(spec=secrets.Secret)
self.secret4 = mock.Mock(spec=secrets.Secret)
self.empty_secret = mock.Mock(spec=secrets.Secret)
# Mock out the client
self.bc = mock.Mock()
@ -82,159 +54,97 @@ class TestBarbicanManager(base.TestCase):
self.cert_manager = barbican_cert_mgr.BarbicanCertManager()
self.cert_manager.auth = barbican_auth
self.context = mock.Mock()
self.context.project_id = PROJECT_ID
super(TestBarbicanManager, self).setUp()
def test_store_cert(self):
# Mock out the client
self.bc.containers.create_certificate.return_value = (
self.empty_container)
self.bc.secrets.create.return_value = (
self.empty_secret)
# Attempt to store a cert
container_ref = self.cert_manager.store_cert(
project_id=PROJECT_ID,
certificate=self.certificate,
private_key=self.private_key,
intermediates=self.intermediates,
private_key_passphrase=self.private_key_passphrase,
secret_ref = self.cert_manager.store_cert(
context=self.context,
certificate=sample.X509_CERT,
private_key=sample.X509_CERT_KEY,
intermediates=sample.X509_IMDS,
name=self.name
)
self.assertEqual(self.empty_container.container_ref, container_ref)
self.assertEqual(secret_ref, self.empty_secret.secret_ref)
# create_secret should be called four times with our data
# create_secret should be called once with our data
calls = [
mock.call(payload=self.certificate, expiration=None,
name=mock.ANY),
mock.call(payload=self.private_key, expiration=None,
name=mock.ANY),
mock.call(payload=self.intermediates, expiration=None,
name=mock.ANY),
mock.call(payload=self.private_key_passphrase, expiration=None,
name=mock.ANY)
mock.call(payload=mock.ANY, expiration=None,
name=self.name)
]
self.bc.secrets.create.assert_has_calls(calls, any_order=True)
# create_certificate should be called once
self.assertEqual(1, self.bc.containers.create_certificate.call_count)
self.bc.secrets.create.assert_has_calls(calls)
# Container should be stored once
self.empty_container.store.assert_called_once_with()
self.empty_secret.store.assert_called_once_with()
def test_store_cert_failure(self):
# Mock out the client
self.bc.containers.create_certificate.return_value = (
self.empty_container)
test_secrets = [
self.secret1,
self.secret2,
self.secret3,
self.secret4
]
self.bc.secrets.create.side_effect = test_secrets
self.empty_container.store.side_effect = ValueError()
self.bc.secrets.create.return_value = (
self.empty_secret)
self.empty_secret.store.side_effect = ValueError()
# Attempt to store a cert
self.assertRaises(
ValueError,
self.cert_manager.store_cert,
project_id=PROJECT_ID,
certificate=self.certificate,
private_key=self.private_key,
intermediates=self.intermediates,
private_key_passphrase=self.private_key_passphrase,
context=self.context,
certificate=sample.X509_CERT,
private_key=sample.X509_CERT_KEY,
intermediates=sample.X509_IMDS,
name=self.name
)
# create_secret should be called four times with our data
calls = [
mock.call(payload=self.certificate, expiration=None,
name=mock.ANY),
mock.call(payload=self.private_key, expiration=None,
name=mock.ANY),
mock.call(payload=self.intermediates, expiration=None,
name=mock.ANY),
mock.call(payload=self.private_key_passphrase, expiration=None,
name=mock.ANY)
]
self.bc.secrets.create.assert_has_calls(calls, any_order=True)
# create_certificate should be called once
self.assertEqual(1, self.bc.containers.create_certificate.call_count)
self.assertEqual(1, self.bc.secrets.create.call_count)
# Container should be stored once
self.empty_container.store.assert_called_once_with()
# All secrets should be deleted (or at least an attempt made)
for s in test_secrets:
s.delete.assert_called_once_with()
self.empty_secret.store.assert_called_once_with()
def test_get_cert(self):
# Mock out the client
self.bc.containers.register_consumer.return_value = self.container
self.bc.secrets.get.return_value = self.secret
# Get the container data
# Get the secret data
data = self.cert_manager.get_cert(
project_id=PROJECT_ID,
cert_ref=self.container_ref,
resource_ref=self.container_ref,
context=self.context,
cert_ref=self.secret_ref,
resource_ref=self.secret_ref,
service_name='Octavia'
)
# 'register_consumer' should be called once with the container_ref
self.bc.containers.register_consumer.assert_called_once_with(
container_ref=self.container_ref,
url=self.container_ref,
name='Octavia'
# 'get_secret' should be called once with the secret_ref
self.bc.secrets.get.assert_called_once_with(
secret_ref=self.secret_ref
)
# The returned data should be a Cert object with the correct values
self.assertIsInstance(data, cert.Cert)
self.assertEqual(data.get_private_key(),
self.private_key.payload)
self.assertEqual(data.get_certificate(),
self.certificate.payload)
self.assertEqual(data.get_intermediates(),
sample.X509_IMDS_LIST)
self.assertEqual(data.get_private_key_passphrase(),
six.b(self.private_key_passphrase.payload))
self.assertEqual(sample.X509_CERT_KEY, data.get_private_key())
self.assertEqual(sample.X509_CERT, data.get_certificate())
self.assertItemsEqual(sample.X509_IMDS_LIST, data.get_intermediates())
self.assertIsNone(data.get_private_key_passphrase())
def test_get_cert_no_registration(self):
self.bc.containers.get.return_value = self.container
# Get the container data
data = self.cert_manager.get_cert(
project_id=PROJECT_ID,
cert_ref=self.container_ref, check_only=True
)
# 'get' should be called once with the container_ref
self.bc.containers.get.assert_called_once_with(
container_ref=self.container_ref
)
# The returned data should be a Cert object with the correct values
self.assertIsInstance(data, cert.Cert)
self.assertEqual(data.get_private_key(),
self.private_key.payload)
self.assertEqual(data.get_certificate(),
self.certificate.payload)
self.assertEqual(data.get_intermediates(),
sample.X509_IMDS_LIST)
self.assertEqual(data.get_private_key_passphrase(),
six.b(self.private_key_passphrase.payload))
def test_delete_cert(self):
def test_delete_cert_legacy(self):
# Attempt to deregister as a consumer
self.cert_manager.delete_cert(
project_id=PROJECT_ID,
cert_ref=self.container_ref,
resource_ref=self.container_ref,
context=self.context,
cert_ref=self.secret_ref,
resource_ref=self.secret_ref,
service_name='Octavia'
)
# remove_consumer should be called once with the container_ref
# remove_consumer should be called once with the container_ref (legacy)
self.bc.containers.remove_consumer.assert_called_once_with(
container_ref=self.container_ref,
url=self.container_ref,
container_ref=self.secret_ref,
url=self.secret_ref,
name='Octavia'
)

242
octavia/tests/unit/certificates/manager/test_barbican_legacy.py Normal file
View File

@ -0,0 +1,242 @@
# Copyright 2014 Rackspace
#
# 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.
import uuid
from barbicanclient.v1 import containers
from barbicanclient.v1 import secrets
import mock
import six
import octavia.certificates.common.barbican as barbican_common
import octavia.certificates.common.cert as cert
import octavia.certificates.manager.barbican_legacy as barbican_cert_mgr
import octavia.tests.unit.base as base
import octavia.tests.unit.common.sample_configs.sample_certs as sample
PROJECT_ID = "12345"
class TestBarbicanManager(base.TestCase):
def setUp(self):
# Make a fake Container and contents
self.barbican_endpoint = 'http://localhost:9311/v1'
self.container_uuid = uuid.uuid4()
self.container_ref = '{0}/containers/{1}'.format(
self.barbican_endpoint, self.container_uuid
)
self.name = 'My Fancy Cert'
self.certificate = secrets.Secret(
api=mock.MagicMock(),
payload=sample.X509_CERT
)
self.intermediates = secrets.Secret(
api=mock.MagicMock(),
payload=sample.X509_IMDS
)
self.private_key = secrets.Secret(
api=mock.MagicMock(),
payload=sample.X509_CERT_KEY_ENCRYPTED
)
self.private_key_passphrase = secrets.Secret(
api=mock.MagicMock(),
payload=sample.X509_CERT_KEY_PASSPHRASE
)
container = mock.Mock(spec=containers.CertificateContainer)
container.container_ref = self.container_ref
container.name = self.name
container.private_key = self.private_key
container.certificate = self.certificate
container.intermediates = self.intermediates
container.private_key_passphrase = self.private_key_passphrase
self.container = container
self.empty_container = mock.Mock(spec=containers.CertificateContainer)
self.secret1 = mock.Mock(spec=secrets.Secret)
self.secret2 = mock.Mock(spec=secrets.Secret)
self.secret3 = mock.Mock(spec=secrets.Secret)
self.secret4 = mock.Mock(spec=secrets.Secret)
# Mock out the client
self.bc = mock.Mock()
barbican_auth = mock.Mock(spec=barbican_common.BarbicanAuth)
barbican_auth.get_barbican_client.return_value = self.bc
self.cert_manager = barbican_cert_mgr.BarbicanCertManager()
self.cert_manager.auth = barbican_auth
self.context = mock.Mock()
self.context.project_id = PROJECT_ID
super(TestBarbicanManager, self).setUp()
def test_store_cert(self):
# Mock out the client
self.bc.containers.create_certificate.return_value = (
self.empty_container)
# Attempt to store a cert
container_ref = self.cert_manager.store_cert(
context=self.context,
certificate=self.certificate,
private_key=self.private_key,
intermediates=self.intermediates,
private_key_passphrase=self.private_key_passphrase,
name=self.name
)
self.assertEqual(self.empty_container.container_ref, container_ref)
# create_secret should be called four times with our data
calls = [
mock.call(payload=self.certificate, expiration=None,
name=mock.ANY),
mock.call(payload=self.private_key, expiration=None,
name=mock.ANY),
mock.call(payload=self.intermediates, expiration=None,
name=mock.ANY),
mock.call(payload=self.private_key_passphrase, expiration=None,
name=mock.ANY)
]
self.bc.secrets.create.assert_has_calls(calls, any_order=True)
# create_certificate should be called once
self.assertEqual(1, self.bc.containers.create_certificate.call_count)
# Container should be stored once
self.empty_container.store.assert_called_once_with()
def test_store_cert_failure(self):
# Mock out the client
self.bc.containers.create_certificate.return_value = (
self.empty_container)
test_secrets = [
self.secret1,
self.secret2,
self.secret3,
self.secret4
]
self.bc.secrets.create.side_effect = test_secrets
self.empty_container.store.side_effect = ValueError()
# Attempt to store a cert
self.assertRaises(
ValueError,
self.cert_manager.store_cert,
context=self.context,
certificate=self.certificate,
private_key=self.private_key,
intermediates=self.intermediates,
private_key_passphrase=self.private_key_passphrase,
name=self.name
)
# create_secret should be called four times with our data
calls = [
mock.call(payload=self.certificate, expiration=None,
name=mock.ANY),
mock.call(payload=self.private_key, expiration=None,
name=mock.ANY),
mock.call(payload=self.intermediates, expiration=None,
name=mock.ANY),
mock.call(payload=self.private_key_passphrase, expiration=None,
name=mock.ANY)
]
self.bc.secrets.create.assert_has_calls(calls, any_order=True)
# create_certificate should be called once
self.assertEqual(1, self.bc.containers.create_certificate.call_count)
# Container should be stored once
self.empty_container.store.assert_called_once_with()
# All secrets should be deleted (or at least an attempt made)
for s in test_secrets:
s.delete.assert_called_once_with()
def test_get_cert(self):
# Mock out the client
self.bc.containers.register_consumer.return_value = self.container
# Get the container data
data = self.cert_manager.get_cert(
context=self.context,
cert_ref=self.container_ref,
resource_ref=self.container_ref,
service_name='Octavia'
)
# 'register_consumer' should be called once with the container_ref
self.bc.containers.register_consumer.assert_called_once_with(
container_ref=self.container_ref,
url=self.container_ref,
name='Octavia'
)
# The returned data should be a Cert object with the correct values
self.assertIsInstance(data, cert.Cert)
self.assertEqual(data.get_private_key(),
self.private_key.payload)
self.assertEqual(data.get_certificate(),
self.certificate.payload)
self.assertEqual(data.get_intermediates(),
sample.X509_IMDS_LIST)
self.assertEqual(data.get_private_key_passphrase(),
six.b(self.private_key_passphrase.payload))
def test_get_cert_no_registration(self):
self.bc.containers.get.return_value = self.container
# Get the container data
data = self.cert_manager.get_cert(
context=self.context,
cert_ref=self.container_ref, check_only=True
)
# 'get' should be called once with the container_ref
self.bc.containers.get.assert_called_once_with(
container_ref=self.container_ref
)
# The returned data should be a Cert object with the correct values
self.assertIsInstance(data, cert.Cert)
self.assertEqual(data.get_private_key(),
self.private_key.payload)
self.assertEqual(data.get_certificate(),
self.certificate.payload)
self.assertEqual(data.get_intermediates(),
sample.X509_IMDS_LIST)
self.assertEqual(data.get_private_key_passphrase(),
six.b(self.private_key_passphrase.payload))
def test_delete_cert(self):
# Attempt to deregister as a consumer
self.cert_manager.delete_cert(
context=self.context,
cert_ref=self.container_ref,
resource_ref=self.container_ref,
service_name='Octavia'
)
# remove_consumer should be called once with the container_ref
self.bc.containers.remove_consumer.assert_called_once_with(
container_ref=self.container_ref,
url=self.container_ref,
name='Octavia'
)

2
octavia/tests/unit/certificates/manager/test_local.py
View File

@ -44,7 +44,7 @@ class TestLocalManager(base.TestCase):
with mock.patch('os.open', open_mock), mock.patch.object(
os, 'fdopen', fd_mock):
cert_id = local_cert_mgr.LocalCertManager.store_cert(
None,
context=None,
certificate=self.certificate,
intermediates=self.intermediates,
private_key=self.private_key,

4
octavia/tests/unit/common/sample_configs/sample_certs.py
View File

@ -15,6 +15,7 @@
import base64
import pkg_resources
import six
@ -813,3 +814,6 @@ zfJ3Bo+P7In9fsHbyDAqIhMwDQYJKoZIhvcNAQELBQADQQBenkZ2k7RgZqgj+dxA
D7BF8MN1oUAOpyYqAjkGddSEuMyNmwtHKZI1dyQ0gBIQdiU9yAG2oTbUIK4msbBV
uJIQ
-----END CERTIFICATE-----"""
PKCS12_BUNDLE = pkg_resources.resource_string(
'octavia.tests.unit.common.sample_configs', 'sample_pkcs12.p12')

BIN
octavia/tests/unit/common/sample_configs/sample_pkcs12.p12 Normal file
View File

Binary file not shown.

10
octavia/tests/unit/common/tls_utils/test_cert_parser.py
View File

@ -134,18 +134,20 @@ class TestTLSParseUtils(base.TestCase):
def test_load_certificates(self):
listener = sample_configs.sample_listener_tuple(tls=True, sni=True)
client = mock.MagicMock()
context = mock.Mock()
context.project_id = '12345'
with mock.patch.object(cert_parser,
'get_host_names') as cp:
with mock.patch.object(cert_parser,
'_map_cert_tls_container'):
cp.return_value = {'cn': 'fakeCN'}
cert_parser.load_certificates_data(client, listener)
cert_parser.load_certificates_data(client, listener, context)
# Ensure upload_cert is called three times
calls_cert_mngr = [
mock.call.get_cert('12345', 'cont_id_1', check_only=True),
mock.call.get_cert('12345', 'cont_id_2', check_only=True),
mock.call.get_cert('12345', 'cont_id_3', check_only=True)
mock.call.get_cert(context, 'cont_id_1', check_only=True),
mock.call.get_cert(context, 'cont_id_2', check_only=True),
mock.call.get_cert(context, 'cont_id_3', check_only=True)
]
client.assert_has_calls(calls_cert_mngr)

21
releasenotes/notes/Support-PKCS12-certificate-objects-1c6e896be9d35977.yaml Normal file
View File

@ -0,0 +1,21 @@
---
features:
- |
Users can now use a reference to a single PKCS12 bundle as their
`default_tls_container_ref` instead of a Barbican container with
individual secret objects. PKCS12 supports bundling a private key,
certificate, and intermediates. Private keys can no longer be passphrase
protected when using PKCS12 bundles.
No configuration change is necessary to enable this feature. Users may
simply begin using this. Any use of the old style containers will be
detected and automatically fall back to using the old Barbican driver.
- |
Certificate bundles can now be stored in any backend Castellan supports,
and can be retrieved via a Castellan driver, even if Barbican is not
deployed.
security:
- |
Private keys can no longer be password protected, as PKCS12 does not
support storing a passphrase in an explicitly defined way. Note that this
is not noticeably less secure than storing a passphrase protected private
key in the same place as the passphrase, as was the case with Barbican.

1
requirements.txt
View File

@ -38,6 +38,7 @@ Jinja2!=2.9.0,!=2.9.1,!=2.9.2,!=2.9.3,!=2.9.4,>=2.8 # BSD License (3 clause)
taskflow>=2.7.0 # Apache-2.0
diskimage-builder!=1.6.0,!=1.7.0,!=1.7.1,>=1.1.2 # Apache-2.0
futures>=3.0.0;python_version=='2.7' or python_version=='2.6' # BSD
castellan>=0.16.0 # Apache-2.0
#for the amphora api
Flask!=0.11,<1.0,>=0.10 # BSD

1
setup.cfg
View File

@ -94,6 +94,7 @@ octavia.cert_generator =
octavia.cert_manager =
local_cert_manager = octavia.certificates.manager.local:LocalCertManager
barbican_cert_manager = octavia.certificates.manager.barbican:BarbicanCertManager
castellan_cert_manager = octavia.certificates.manager.castellan_mgr:CastellanCertManager
octavia.barbican_auth =
barbican_acl_auth = octavia.certificates.common.auth.barbican_acl:BarbicanACLAuth
octavia.plugins =