barbican/barbican/plugin/castellan_secret_store.py
Takashi Kajinami 6d34865522 Remove six
Python 2 is no longer supported, thus usage of six can be removed.

Also, This removes B314 test from documentation because its actual
implementation was already removed[1].

[1] 9dbeefb55eb55a9baa481ec6d6293527d46e04a5

Change-Id: Ib01714e6462470dd5c3f6f06b52a3afeff573696
2022-10-28 14:02:00 +02:00

217 lines
7.8 KiB
Python

# Copyright (c) 2018 Red Hat Inc.
#
# 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 abc
import base64
from castellan.common.objects import key
from castellan.common.objects import opaque_data
from castellan import key_manager
from oslo_context import context
from oslo_log import log
from barbican.plugin.interface import secret_store as ss
LOG = log.getLogger(__name__)
class CastellanSecretStore(ss.SecretStoreBase, metaclass=abc.ABCMeta):
KEY_ID = "key_id"
ALG = "alg"
BIT_LENGTH = "bit_length"
METADATA_VERSION = "version"
CURRENT_VERSION = 1
def _set_params(self, conf):
self.key_manager = key_manager.API(conf)
self.context = context.get_current()
def _meta_dict(self, key_id, bit_length=None, algorithm=None):
"""Return the current version of the metadata dict
Builds the metadata dict to be stored in the database.
"""
meta = {
self.KEY_ID: key_id,
self.METADATA_VERSION: self.CURRENT_VERSION,
}
if bit_length is not None:
meta[self.BIT_LENGTH] = bit_length
if algorithm is not None:
meta[self.ALG] = algorithm
return meta
def _ensure_legacy_base64(self, secret):
"""Ensure secret data is base64 encoded
This method ensures that secrets that were stored prior to the fix
for Story 2008335 are base64 encoded.
"""
payload = secret.get_encoded()
if isinstance(secret, key.Key):
# Keys generated by Castellan are not base64-encoded.
# Both symmetric and asymmetric keys returned by Castellan
# are subclasses of key.Key
LOG.debug("Encoding legacy Castellan-generated key")
return base64.b64encode(payload)
else:
# Objects stored by Barbican are stored as opaque_data.OpaqueData
# in Castellan. They should already be base64-encoded so we
# check here to make sure.
LOG.debug("Validating base64 encoding")
try:
_ = base64.b64decode(payload)
return payload
except UnicodeDecodeError:
# Data can't be decoded. Not sure how we ended up here,
# but we can encode now to prevent issues when we attempt
# to base64-decode the DTO later.
LOG.warning("Legacy secret data assumed to be plaintext")
return base64.b64encode(payload)
@abc.abstractmethod
def get_conf(self, conf):
"""Get plugin configuration
This method is supposed to be implemented by the relevant
subclass. This method reads in the config for the plugin
in barbican.conf -- which should look like the way other
barbican plugins are configured, and convert them to the
proper oslo.config object to be passed to the keymanager
API. (keymanager.API(conf)
@returns oslo.config object
"""
raise NotImplementedError # pragma: no cover
@abc.abstractmethod
def get_plugin_name(self):
"""Get plugin name
This method is implemented by the subclass.
Note that this name must be unique across the deployment.
"""
raise NotImplementedError # pragma: no cover
def get_secret(self, secret_type, secret_metadata):
secret_ref = secret_metadata[CastellanSecretStore.KEY_ID]
try:
secret = self.key_manager.get(
self.context,
secret_ref)
meta_version = secret_metadata.get(self.METADATA_VERSION)
if meta_version is None:
# Secrets without a metadata version were stored prior to fix
# for Story 2008335. They may or may not be base64-encoded.
LOG.debug("Retrieving legacy secret")
data = self._ensure_legacy_base64(secret)
else:
# Version 1 - secret payload data is stored in plaintext in
# the backend. We need to base64 encode them for the DTO.
data = base64.b64encode(secret.get_encoded())
return ss.SecretDTO(secret_type, data, ss.KeySpec(), None)
except Exception as e:
LOG.exception("Error retrieving secret {}: {}".format(
secret_ref, str(e)))
raise ss.SecretGeneralException(e)
def store_secret(self, secret_dto):
if not self.store_secret_supports(secret_dto.key_spec):
raise ss.SecretAlgorithmNotSupportedException(
secret_dto.key_spec.alg)
plaintext = base64.b64decode(secret_dto.secret)
try:
secret_id = self.key_manager.store(
self.context,
opaque_data.OpaqueData(plaintext)
)
return self._meta_dict(secret_id)
except Exception as e:
LOG.exception("Error storing secret: {}".format(
str(e)))
raise ss.SecretGeneralException(e)
def delete_secret(self, secret_metadata):
secret_ref = secret_metadata[CastellanSecretStore.KEY_ID]
try:
self.key_manager.delete(
self.context,
secret_ref)
except KeyError:
LOG.warning("Attempting to delete a non-existent secret {}".format(
secret_ref))
except Exception as e:
LOG.exception("Error deleting secret: {}".format(
str(e)))
raise ss.SecretGeneralException(e)
def generate_symmetric_key(self, key_spec):
if not self.generate_supports(key_spec):
raise ss.SecretAlgorithmNotSupportedException(
key_spec.alg)
try:
secret_id = self.key_manager.create_key(
self.context,
key_spec.alg,
key_spec.bit_length
)
return self._meta_dict(secret_id)
except Exception as e:
LOG.exception("Error generating symmetric key: {}".format(
str(e)))
raise ss.SecretGeneralException(e)
def generate_asymmetric_key(self, key_spec):
if not self.generate_supports(key_spec):
raise ss.SecretAlgorithmNotSupportedException(
key_spec.alg)
if key_spec.passphrase:
raise ss.GeneratePassphraseNotSupportedException()
try:
private_id, public_id = self.key_manager.create_key_pair(
self.context,
key_spec.alg,
key_spec.bit_length
)
private_key_metadata = self._meta_dict(
private_id, key_spec.bit_length, key_spec.alg
)
public_key_metadata = self._meta_dict(
public_id, key_spec.bit_length, key_spec.alg
)
return ss.AsymmetricKeyMetadataDTO(
private_key_metadata,
public_key_metadata,
None
)
except Exception as e:
LOG.exception("Error generating asymmetric key: {}".format(
str(e)))
raise ss.SecretGeneralException(e)
@abc.abstractmethod
def store_secret_supports(self, key_spec):
raise NotImplementedError # pragma: no cover
@abc.abstractmethod
def generate_supports(self, key_spec):
raise NotImplementedError # pragma: no cover