Compare commits
9 Commits
Author | SHA1 | Date | |
---|---|---|---|
bb5d88229a | |||
7975776398 | |||
9c8f1388ce | |||
4da8562871 | |||
009d9b15e0 | |||
56f5acfff8 | |||
a32c2743de | |||
c17775b715 | |||
4ba36b3a87 |
27
.gitea/workflows/ci.yaml
Normal file
27
.gitea/workflows/ci.yaml
Normal file
|
@ -0,0 +1,27 @@
|
|||
name: "CI"
|
||||
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- "release"
|
||||
|
||||
jobs:
|
||||
run-checks:
|
||||
name: "Checks"
|
||||
runs-on: "ubuntu-latest"
|
||||
steps:
|
||||
- uses: "actions/checkout@v4"
|
||||
- uses: "actions/setup-python@v5"
|
||||
with:
|
||||
python-version: "3.10"
|
||||
- uses: "Gr1N/setup-poetry@v8"
|
||||
with:
|
||||
poetry-version: "1.7.1"
|
||||
- name: "Install deps"
|
||||
run: |
|
||||
set -x
|
||||
poetry install
|
||||
- name: "Run CI task"
|
||||
run: |
|
||||
set -x
|
||||
poetry run inv ci
|
13
CHANGELOG.md
13
CHANGELOG.md
|
@ -1,5 +1,18 @@
|
|||
## Keep It Secret Changelog
|
||||
|
||||
#### v1.2.2 (2024-06-05)
|
||||
|
||||
* TeamCity integration for private builds.
|
||||
|
||||
#### v1.2.1 (2024-05-29)
|
||||
|
||||
* Gitea Actions integration.
|
||||
* `README.md` language fixes.
|
||||
|
||||
#### v1.2.0 (2024-02-08)
|
||||
|
||||
* Hashicorp Vault integration.
|
||||
|
||||
#### v1.1.0 (2024-01-18)
|
||||
|
||||
* `Secrets.resolve_dependency()` API for resolving field dependencies.
|
||||
|
|
88
Dockerfile
Normal file
88
Dockerfile
Normal file
|
@ -0,0 +1,88 @@
|
|||
ARG APP_USER_UID=10001
|
||||
ARG APP_USER_GID=10001
|
||||
ARG IMAGE_TAG=development.000000
|
||||
|
||||
FROM python:3.10.14-slim-bookworm AS base
|
||||
|
||||
ARG APP_USER_UID
|
||||
ARG APP_USER_GID
|
||||
ARG IMAGE_TAG
|
||||
|
||||
ENV PYTHONUNBUFFERED=1 \
|
||||
PYTHONDONTWRITEBYTECODE=1 \
|
||||
PIP_NO_CACHE_DIR=off \
|
||||
PIP_DISABLE_PIP_VERSION_CHECK=on \
|
||||
PIP_DEFAULT_TIMEOUT=100 \
|
||||
PIP_INDEX_URL="https://nexus.bthlabs.pl/repository/pypi/simple/" \
|
||||
POETRY_VERSION=1.7.1 \
|
||||
POETRY_HOME="/srv/poetry" \
|
||||
POETRY_NO_INTERACTION=1 \
|
||||
VIRTUAL_ENV="/srv/venv" \
|
||||
KEEP_IT_SECRET_IMAGE_TAG=${IMAGE_TAG}
|
||||
|
||||
RUN if [ ! $(getent group ${APP_USER_GID}) ];then groupadd -g ${APP_USER_GID} app; fi && \
|
||||
useradd -m -d /home/app -u ${APP_USER_UID} -g ${APP_USER_GID} app && \
|
||||
apt-get update && \
|
||||
apt-get install -y --no-install-recommends wait-for-it dumb-init curl && \
|
||||
(curl -sSL https://install.python-poetry.org | python -) && \
|
||||
python3.10 -m venv ${VIRTUAL_ENV} && \
|
||||
mkdir /srv/app /srv/bin /srv/lib /srv/log /srv/run && \
|
||||
chown -R ${APP_USER_UID}:${APP_USER_GID} /srv
|
||||
|
||||
ENV PATH="${VIRTUAL_ENV}/bin:/srv/bin:/srv/poetry/bin:${PATH}"
|
||||
|
||||
USER app
|
||||
WORKDIR /srv/app
|
||||
|
||||
FROM base AS development
|
||||
|
||||
ARG APP_USER_UID
|
||||
ARG APP_USER_GID
|
||||
ARG IMAGE_TAG
|
||||
|
||||
USER app
|
||||
WORKDIR /srv/app
|
||||
|
||||
FROM development AS deployment-build
|
||||
|
||||
ARG APP_USER_UID
|
||||
ARG APP_USER_GID
|
||||
ARG IMAGE_TAG
|
||||
|
||||
ADD --chown=$APP_USER_UID:$APP_USER_GID . /srv/app
|
||||
RUN poetry install --no-dev
|
||||
|
||||
FROM deployment-build AS ci
|
||||
|
||||
ARG APP_USER_UID
|
||||
ARG APP_USER_GID
|
||||
ARG IMAGE_TAG
|
||||
|
||||
RUN poetry install
|
||||
|
||||
FROM base AS deployment
|
||||
|
||||
ARG APP_USER_UID
|
||||
ARG APP_USER_GID
|
||||
ARG IMAGE_TAG
|
||||
|
||||
COPY --from=deployment-build /srv/app /srv/app
|
||||
COPY --from=deployment-build /srv/venv /srv/venv
|
||||
RUN chown -R $APP_USER_UID:$APP_USER_GID /srv
|
||||
|
||||
USER root
|
||||
|
||||
RUN apt-get clean autoclean && \
|
||||
apt-get autoremove --yes && \
|
||||
rm -rf /var/lib/apt /var/lib/dpkg && \
|
||||
rm -rf /home/app/.cache
|
||||
|
||||
USER app
|
||||
|
||||
ENV PYTHONPATH="/srv/app"
|
||||
ENV DJANGO_SETTINGS_MODULE="settings"
|
||||
|
||||
EXPOSE 8000
|
||||
|
||||
ENTRYPOINT ["/usr/bin/dumb-init"]
|
||||
CMD ["echo NOOP"]
|
|
@ -65,7 +65,7 @@ to provide secrets suitable for the development environment:
|
|||
```
|
||||
|
||||
The `ProductionSecrets` class uses environment variables and AWS Secrets
|
||||
Manager to provide secrets suitable for the development environment:
|
||||
Manager to provide secrets suitable for the production environment:
|
||||
|
||||
```
|
||||
>>> production_secrets = ProductionSecrets()
|
||||
|
|
|
@ -11,10 +11,10 @@
|
|||
project = 'Keep It Secret'
|
||||
copyright = '2023-present Tomek Wójcik'
|
||||
author = 'Tomek Wójcik'
|
||||
version = '1.1.0'
|
||||
version = '1.2.2'
|
||||
|
||||
# The full version, including alpha/beta/rc tags
|
||||
release = '1.1.0'
|
||||
release = '1.2.2'
|
||||
|
||||
# -- General configuration ---------------------------------------------------
|
||||
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration
|
||||
|
|
|
@ -25,6 +25,26 @@ to be installed:
|
|||
.. autoclass:: keep_it_secret.ext.aws.AWSSecretsManagerField
|
||||
:members:
|
||||
|
||||
Hashicorp Vault Wrapper
|
||||
-----------------------
|
||||
|
||||
**Installation**
|
||||
|
||||
Since Vault extension has external dependencies it needs to be explicitly named
|
||||
to be installed:
|
||||
|
||||
.. code-block:: shell
|
||||
|
||||
$ pip install keep_it_secret[vault]
|
||||
|
||||
**API**
|
||||
|
||||
.. autoclass:: keep_it_secret.ext.vault.VaultSecrets
|
||||
:members:
|
||||
|
||||
.. autoclass:: keep_it_secret.ext.vault.VaultKV2Field
|
||||
:members:
|
||||
|
||||
Basic secrets loader
|
||||
--------------------
|
||||
|
||||
|
|
|
@ -6,7 +6,7 @@ from .fields import ( # noqa: F401
|
|||
)
|
||||
from .secrets import Secrets # noqa: F401
|
||||
|
||||
__version__ = '1.1.0'
|
||||
__version__ = '1.2.2'
|
||||
|
||||
__all__ = [
|
||||
'AbstractField',
|
||||
|
|
177
keep_it_secret/ext/vault.py
Normal file
177
keep_it_secret/ext/vault.py
Normal file
|
@ -0,0 +1,177 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
from __future__ import annotations
|
||||
|
||||
import typing
|
||||
|
||||
import hvac
|
||||
|
||||
from keep_it_secret.fields import EnvField, Field
|
||||
from keep_it_secret.secrets import Secrets
|
||||
|
||||
|
||||
class VaultSecrets(Secrets):
|
||||
"""
|
||||
Concrete :py:class:`keep_it_secret.Secrets` subclass that maps environment
|
||||
variables to Vault credentials.
|
||||
"""
|
||||
|
||||
url: str = EnvField.new('VAULT_URL', required=True)
|
||||
"""
|
||||
Maps ``VAULT_URL`` environment variable.
|
||||
|
||||
:type: ``str``
|
||||
"""
|
||||
|
||||
token: str = EnvField.new('VAULT_TOKEN', required=True)
|
||||
"""
|
||||
Maps ``VAULT_TOKEN`` environment variable.
|
||||
|
||||
:type: ``str``
|
||||
"""
|
||||
|
||||
client_cert_path: str | None = EnvField.new('VAULT_CLIENT_CERT_PATH', required=False)
|
||||
"""
|
||||
Maps ``VAULT_CLIENT_CERT_PATH`` environment variable.
|
||||
|
||||
:type: ``str | None``
|
||||
"""
|
||||
|
||||
client_key_path: str | None = EnvField.new('VAULT_CLIENT_KEY_PATH', required=False)
|
||||
"""
|
||||
Maps ``VAULT_CLIENT_KEY_PATH`` environment variable.
|
||||
|
||||
:type: ``str | None``
|
||||
"""
|
||||
|
||||
server_cert_path: str | None = EnvField.new('VAULT_SERVER_CERT_PATH', required=False)
|
||||
"""
|
||||
Maps ``VAULT_SERVER_CERT_PATH`` environment variable.
|
||||
|
||||
:type: ``str | None``
|
||||
"""
|
||||
|
||||
def __init__(self, parent: Secrets | None = None):
|
||||
super().__init__(parent)
|
||||
|
||||
self.client: hvac.Client | None = None
|
||||
|
||||
def as_hvac_client_kwargs(self) -> dict[str, typing.Any]:
|
||||
"""
|
||||
Return representation of the mapped variables for use in
|
||||
``hvac.Client`` constructor.
|
||||
"""
|
||||
result: dict[str, typing.Any] = {
|
||||
'url': self.url,
|
||||
'token': self.token,
|
||||
}
|
||||
|
||||
if self.client_cert_path is not None and self.client_key_path is not None:
|
||||
result['cert'] = (self.client_cert_path, self.client_key_path)
|
||||
|
||||
if self.server_cert_path is not None:
|
||||
result['verify'] = self.server_cert_path
|
||||
|
||||
return result
|
||||
|
||||
def get_client(self) -> hvac.Client:
|
||||
"""
|
||||
Return the ``hvac.Client`` instance configured using the credentials.
|
||||
"""
|
||||
if self.client is None:
|
||||
self.client = hvac.Client(
|
||||
**self.as_hvac_client_kwargs(),
|
||||
)
|
||||
|
||||
return self.client
|
||||
|
||||
|
||||
class VaultKV2Field(Field):
|
||||
"""
|
||||
Concrete :py:class:`keep_it_secret.Field` subclass that uses Hashicorp
|
||||
Vault KV V2 secrets engine to resolve the value.
|
||||
|
||||
If ``as_type`` isn't provided, the fetched value will be returned as-is (
|
||||
i.e. as a dict). Otherwise, ``as_type`` should be a type which accepts
|
||||
kwargs representing the value's keys in its constructor.
|
||||
|
||||
:param mount_point: Mount path for the secret engine.
|
||||
:param path: Path to the secret to fetch.
|
||||
:param version: Version identifier. Defaults to ``None`` (aka the newest
|
||||
version).
|
||||
:param default: Default value. Defaults to ``None``.
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
mount_point: str,
|
||||
path: str,
|
||||
version: str | None = None,
|
||||
default: typing.Any = None,
|
||||
**field_options):
|
||||
super().__init__(**field_options)
|
||||
|
||||
as_type: type | None = field_options.pop('as_type', None)
|
||||
self.as_type = self.cast if as_type is not None else None # type: ignore[assignment]
|
||||
|
||||
self.mount_point = mount_point
|
||||
self.path = path
|
||||
self.version = version
|
||||
self.default = default
|
||||
self.cast_to = as_type
|
||||
|
||||
@classmethod
|
||||
def new(cls: type[VaultKV2Field], # type: ignore[override]
|
||||
mount_point: str,
|
||||
path: str,
|
||||
version: str | None = None,
|
||||
default: typing.Any = None,
|
||||
**field_options):
|
||||
return cls(mount_point, path, version=version, default=default, **field_options)
|
||||
|
||||
def cast(self, data: typing.Any) -> typing.Any:
|
||||
"""
|
||||
Cast ``data`` to the type specified in ``as_type`` argument of the
|
||||
constructor.
|
||||
"""
|
||||
if isinstance(data, dict) is False:
|
||||
return data
|
||||
|
||||
if self.cast_to is None:
|
||||
return data
|
||||
|
||||
return self.cast_to(**data)
|
||||
|
||||
def get_value(self, secrets: Secrets) -> typing.Any:
|
||||
"""
|
||||
Retrieve, decode and return the secret stored in a KV V2 secrets engine
|
||||
mounted at *mount_path* under the path *path*.
|
||||
|
||||
Depends on :py:class:`VaultSecrets` to be declared in ``vault`` field
|
||||
on ``secrets`` or one of its parents.
|
||||
|
||||
:raises DependencyMissing: Signal that ``secrets.aws`` field is
|
||||
missing.
|
||||
:raises RequiredValueMissing: Signal the field's value is required but
|
||||
*secret_id* is not present in the secrets engine.
|
||||
"""
|
||||
vault_secrets: VaultSecrets = secrets.resolve_dependency(
|
||||
'vault', include_parents=True,
|
||||
)
|
||||
if vault_secrets is secrets.UNRESOLVED_DEPENDENCY:
|
||||
raise self.DependencyMissing('vault')
|
||||
|
||||
client = vault_secrets.get_client()
|
||||
|
||||
try:
|
||||
secret_response = client.secrets.kv.v2.read_secret_version(
|
||||
self.path,
|
||||
version=self.version,
|
||||
mount_point=self.mount_point,
|
||||
raise_on_deleted_version=True,
|
||||
)
|
||||
|
||||
return secret_response['data']['data']
|
||||
except hvac.exceptions.InvalidPath as exception:
|
||||
if self.required is True:
|
||||
raise self.RequiredValueMissing(f'{self.mount_point}{self.path}') from exception
|
||||
else:
|
||||
return self.default
|
22
poetry.lock
generated
22
poetry.lock
generated
|
@ -1,4 +1,4 @@
|
|||
# This file is automatically @generated by Poetry 1.6.1 and should not be changed by hand.
|
||||
# This file is automatically @generated by Poetry 1.7.1 and should not be changed by hand.
|
||||
|
||||
[[package]]
|
||||
name = "alabaster"
|
||||
|
@ -429,6 +429,23 @@ pygments = ">=2.7"
|
|||
sphinx = ">=6.0,<8.0"
|
||||
sphinx-basic-ng = "*"
|
||||
|
||||
[[package]]
|
||||
name = "hvac"
|
||||
version = "2.1.0"
|
||||
description = "HashiCorp Vault API client"
|
||||
optional = false
|
||||
python-versions = ">=3.8,<4.0"
|
||||
files = [
|
||||
{file = "hvac-2.1.0-py3-none-any.whl", hash = "sha256:73bc91e58c3fc7c6b8107cdaca9cb71fa0a893dfd80ffbc1c14e20f24c0c29d7"},
|
||||
{file = "hvac-2.1.0.tar.gz", hash = "sha256:b48bcda11a4ab0a7b6c47232c7ba7c87fda318ae2d4a7662800c465a78742894"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
requests = ">=2.27.1,<3.0.0"
|
||||
|
||||
[package.extras]
|
||||
parser = ["pyhcl (>=0.4.4,<0.5.0)"]
|
||||
|
||||
[[package]]
|
||||
name = "idna"
|
||||
version = "3.6"
|
||||
|
@ -1692,8 +1709,9 @@ testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "p
|
|||
|
||||
[extras]
|
||||
aws = ["boto3"]
|
||||
vault = ["hvac"]
|
||||
|
||||
[metadata]
|
||||
lock-version = "2.0"
|
||||
python-versions = "^3.10"
|
||||
content-hash = "e0cabda70067d490af2f9c9e9251384fb43454c545da6a071e25b1b74cc9f951"
|
||||
content-hash = "e4959d95b206c17013e548f015fbf288d9bc754584bfc128458cadead137a863"
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
[tool.poetry]
|
||||
name = "keep-it-secret"
|
||||
version = "1.1.0"
|
||||
version = "1.2.2"
|
||||
description = "Keep It Secret by BTHLabs"
|
||||
authors = ["Tomek Wójcik <contact@bthlabs.pl>"]
|
||||
maintainers = ["BTHLabs <contact@bthlabs.pl>"]
|
||||
|
@ -13,12 +13,14 @@ documentation = "https://projects.bthlabs.pl/keep-it-secret/"
|
|||
[tool.poetry.dependencies]
|
||||
python = "^3.10"
|
||||
boto3 = {version = ">=1.34.0", optional = true}
|
||||
hvac = {version = ">=2.1.0", optional = true}
|
||||
|
||||
[tool.poetry.group.dev.dependencies]
|
||||
boto3 = "1.34.8"
|
||||
flake8 = "6.1.0"
|
||||
flake8-commas = "2.1.0"
|
||||
furo = "2023.9.10"
|
||||
hvac = "2.1.0"
|
||||
invoke = "1.7.3"
|
||||
ipdb = "0.13.13"
|
||||
ipython = "8.19.0"
|
||||
|
@ -32,6 +34,7 @@ twine = "4.0.2"
|
|||
|
||||
[tool.poetry.extras]
|
||||
aws = ["boto3"]
|
||||
vault = ["hvac"]
|
||||
|
||||
[build-system]
|
||||
requires = ["poetry-core"]
|
||||
|
|
|
@ -6,12 +6,15 @@ hang-closing = False
|
|||
|
||||
[tool:pytest]
|
||||
addopts = --disable-warnings
|
||||
junit_suite_name = keep_it_secret
|
||||
env =
|
||||
KEEP_IT_SECRET_TESTS_SPAM=spam
|
||||
AWS_ACCESS_KEY_ID=thisisntright
|
||||
AWS_SECRET_ACCESS_KEY=thisisntright
|
||||
AWS_SESSION_TOKEN=thisisntright
|
||||
AWS_DEFAULT_REGION=eu-central-1
|
||||
VAULT_URL=http://thisisntright:8200/
|
||||
VAULT_TOKEN=thisisntright
|
||||
|
||||
[mypy]
|
||||
|
||||
|
@ -21,5 +24,8 @@ ignore_missing_imports = True
|
|||
[mypy-boto3.*]
|
||||
ignore_missing_imports = True
|
||||
|
||||
[mypy-hvac.*]
|
||||
ignore_missing_imports = True
|
||||
|
||||
[mypy-moto.*]
|
||||
ignore_missing_imports = True
|
||||
|
|
16
tasks.py
16
tasks.py
|
@ -1,5 +1,7 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# type: ignore
|
||||
import os
|
||||
|
||||
from invoke import task
|
||||
|
||||
try:
|
||||
|
@ -20,13 +22,25 @@ def mypy(ctx, warn=False):
|
|||
|
||||
@task
|
||||
def tests(ctx, warn=False):
|
||||
return ctx.run('pytest -v', warn=warn)
|
||||
pytest_command_line = [
|
||||
'pytest',
|
||||
'-v',
|
||||
]
|
||||
|
||||
if 'KEEP_IT_SECRET_JUNIT_XML_PATH' in os.environ:
|
||||
pytest_command_line.append(
|
||||
f"--junit-xml={os.environ['KEEP_IT_SECRET_JUNIT_XML_PATH']}",
|
||||
)
|
||||
|
||||
return ctx.run(' '.join(pytest_command_line), warn=warn)
|
||||
|
||||
|
||||
@task
|
||||
def ci(ctx):
|
||||
result = True
|
||||
|
||||
ctx.run('mkdir -p build')
|
||||
|
||||
if flake8(ctx, warn=True).exited != 0:
|
||||
result = False
|
||||
|
||||
|
|
0
tests/ext/vault/__init__.py
Normal file
0
tests/ext/vault/__init__.py
Normal file
12
tests/ext/vault/conftest.py
Normal file
12
tests/ext/vault/conftest.py
Normal file
|
@ -0,0 +1,12 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# type: ignore
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from .fixtures import TestingVaultSecrets
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def testing_vault_secrets() -> TestingVaultSecrets:
|
||||
return TestingVaultSecrets()
|
13
tests/ext/vault/fixtures.py
Normal file
13
tests/ext/vault/fixtures.py
Normal file
|
@ -0,0 +1,13 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
from __future__ import annotations
|
||||
|
||||
# -*- coding: utf-8 -*-
|
||||
from __future__ import annotations
|
||||
|
||||
from keep_it_secret.ext.vault import VaultSecrets
|
||||
from keep_it_secret.fields import SecretsField
|
||||
from keep_it_secret.secrets import Secrets
|
||||
|
||||
|
||||
class TestingVaultSecrets(Secrets):
|
||||
vault = SecretsField.new(VaultSecrets)
|
212
tests/ext/vault/test_VaultKV2Field.py
Normal file
212
tests/ext/vault/test_VaultKV2Field.py
Normal file
|
@ -0,0 +1,212 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# type: ignore
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest import mock
|
||||
|
||||
import hvac
|
||||
import pytest
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from keep_it_secret import Secrets
|
||||
from keep_it_secret.ext import vault
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def field() -> vault.VaultKV2Field:
|
||||
return vault.VaultKV2Field('keep_it_secret/', 'tests/spam')
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def as_type() -> mock.Mock:
|
||||
return mock.Mock()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def field_with_as_type(as_type: mock.Mock) -> vault.VaultKV2Field:
|
||||
return vault.VaultKV2Field('keep_it_secret/', 'tests/spam', as_type=as_type)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def hvac_kv_v2_client(mocker: MockerFixture) -> mock.Mock:
|
||||
result = mock.Mock()
|
||||
result.secrets.kv.v2 = mock.Mock(spec=['read_secret_version'])
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def test_init():
|
||||
# When
|
||||
result = vault.VaultKV2Field('keep_it_secret/', 'tests/spam')
|
||||
|
||||
# Then
|
||||
assert result.mount_point == 'keep_it_secret/'
|
||||
assert result.path == 'tests/spam'
|
||||
assert result.version is None
|
||||
assert result.default is None
|
||||
assert result.cast_to is None
|
||||
assert result.as_type is None
|
||||
|
||||
|
||||
def test_init_with_version():
|
||||
# When
|
||||
result = vault.VaultKV2Field(
|
||||
'keep_it_secret/', 'tests/spam', version='1',
|
||||
)
|
||||
|
||||
# Then
|
||||
assert result.version == '1'
|
||||
|
||||
|
||||
def test_init_with_default():
|
||||
# When
|
||||
result = vault.VaultKV2Field(
|
||||
'keep_it_secret/', 'tests/spam', default='eggs',
|
||||
)
|
||||
|
||||
# Then
|
||||
assert result.default == 'eggs'
|
||||
|
||||
|
||||
def test_init_with_as_type():
|
||||
# When
|
||||
result = vault.VaultKV2Field(
|
||||
'keep_it_secret/', 'tests/spam', as_type=dict,
|
||||
)
|
||||
|
||||
# Then
|
||||
assert result.as_type == result.cast
|
||||
assert result.cast_to == dict
|
||||
|
||||
|
||||
def test_init_with_field_options():
|
||||
# When
|
||||
result = vault.VaultKV2Field(
|
||||
'keep_it_secret/', 'tests/spam', required=False, description='eggs',
|
||||
)
|
||||
|
||||
# Then
|
||||
assert result.required is False
|
||||
assert result.description == 'eggs'
|
||||
|
||||
|
||||
def test_new(mocker: MockerFixture):
|
||||
# Given
|
||||
mock_init = mocker.patch.object(
|
||||
vault.VaultKV2Field, '__init__', return_value=None,
|
||||
)
|
||||
|
||||
# When
|
||||
_ = vault.VaultKV2Field.new(
|
||||
'keep_it_secret/',
|
||||
'tests/spam',
|
||||
version='1',
|
||||
default=dict,
|
||||
as_type=dict,
|
||||
required=False,
|
||||
description='eggs',
|
||||
)
|
||||
|
||||
# Then
|
||||
mock_init.assert_called_once_with(
|
||||
'keep_it_secret/',
|
||||
'tests/spam',
|
||||
version='1',
|
||||
default=dict,
|
||||
as_type=dict,
|
||||
required=False,
|
||||
description='eggs',
|
||||
)
|
||||
|
||||
|
||||
def test_cast_not_dict(field: vault.VaultKV2Field):
|
||||
# When
|
||||
result = field.cast('spam')
|
||||
|
||||
# Then
|
||||
assert result == 'spam'
|
||||
|
||||
|
||||
def test_cast(field_with_as_type: vault.VaultKV2Field, as_type: mock.Mock):
|
||||
# Given
|
||||
as_type.return_value = 'spam'
|
||||
|
||||
# When
|
||||
result = field_with_as_type.cast({
|
||||
'spam': True,
|
||||
'eggs': False,
|
||||
})
|
||||
|
||||
# Then
|
||||
assert result == 'spam'
|
||||
|
||||
as_type.assert_called_once_with(
|
||||
spam=True,
|
||||
eggs=False,
|
||||
)
|
||||
|
||||
|
||||
def test_get_value_vault_dependency_missing(field: vault.VaultKV2Field,
|
||||
testing_secrets: Secrets):
|
||||
# Given
|
||||
with pytest.raises(field.DependencyMissing) as exception_info:
|
||||
# When
|
||||
_ = field(testing_secrets)
|
||||
|
||||
# Then
|
||||
assert exception_info.value.args[0] == 'vault'
|
||||
|
||||
|
||||
def test_get_value_required_value_not_found(testing_vault_secrets: Secrets,
|
||||
hvac_kv_v2_client: mock.Mock,
|
||||
field: vault.VaultKV2Field):
|
||||
# Given
|
||||
testing_vault_secrets.vault.client = hvac_kv_v2_client
|
||||
|
||||
hvac_kv_v2_client.secrets.kv.v2.read_secret_version.side_effect = hvac.exceptions.InvalidPath()
|
||||
|
||||
with pytest.raises(field.RequiredValueMissing) as exception_info:
|
||||
# When
|
||||
_ = field(testing_vault_secrets)
|
||||
|
||||
# Then
|
||||
assert exception_info.value.args[0] == 'keep_it_secret/tests/spam'
|
||||
|
||||
|
||||
def test_get_value_not_found_not_required(testing_vault_secrets: Secrets,
|
||||
hvac_kv_v2_client: mock.Mock,
|
||||
field: vault.VaultKV2Field):
|
||||
# Given
|
||||
testing_vault_secrets.vault.client = hvac_kv_v2_client
|
||||
|
||||
hvac_kv_v2_client.secrets.kv.v2.read_secret_version.side_effect = hvac.exceptions.InvalidPath()
|
||||
|
||||
field = vault.VaultKV2Field(
|
||||
'keep_it_secret/', 'tests/spam', required=False,
|
||||
)
|
||||
|
||||
result = field(testing_vault_secrets)
|
||||
|
||||
# Then
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_get_value(testing_vault_secrets: Secrets,
|
||||
hvac_kv_v2_client: mock.Mock,
|
||||
field: vault.VaultKV2Field):
|
||||
# Given
|
||||
testing_vault_secrets.vault.client = hvac_kv_v2_client
|
||||
|
||||
hvac_kv_v2_client.secrets.kv.v2.read_secret_version.return_value = {
|
||||
'data': {
|
||||
'data': {
|
||||
'spam': True,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
# When
|
||||
result = field(testing_vault_secrets)
|
||||
|
||||
# Then
|
||||
assert result == {'spam': True}
|
109
tests/ext/vault/test_VaultSecrets.py
Normal file
109
tests/ext/vault/test_VaultSecrets.py
Normal file
|
@ -0,0 +1,109 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# type: ignore
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from keep_it_secret.ext import vault
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_hvac_client(mocker: MockerFixture) -> mock.Mock:
|
||||
return mocker.patch.object(vault.hvac, 'Client')
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def hvac_client() -> mock.Mock:
|
||||
return mock.Mock()
|
||||
|
||||
|
||||
def test_init():
|
||||
# When
|
||||
result = vault.VaultSecrets()
|
||||
|
||||
# Then
|
||||
assert result.client is None
|
||||
|
||||
|
||||
@mock.patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
'VAULT_URL': 'https://vault.work/',
|
||||
'VAULT_TOKEN': 'test_vault_token',
|
||||
'VAULT_CLIENT_CERT_PATH': '/tmp/vault_client_cert.pem',
|
||||
'VAULT_CLIENT_KEY_PATH': '/tmp/vault_client_key.pem',
|
||||
'VAULT_SERVER_CERT_PATH': '/tmp/vault_server_cert.pem',
|
||||
},
|
||||
)
|
||||
def test_as_hvac_client_kwargs():
|
||||
# Given
|
||||
secrets = vault.VaultSecrets()
|
||||
|
||||
# When
|
||||
result = secrets.as_hvac_client_kwargs()
|
||||
|
||||
# Then
|
||||
assert result == {
|
||||
'url': 'https://vault.work/',
|
||||
'token': 'test_vault_token',
|
||||
'cert': ('/tmp/vault_client_cert.pem', '/tmp/vault_client_key.pem'),
|
||||
'verify': '/tmp/vault_server_cert.pem',
|
||||
}
|
||||
|
||||
|
||||
@mock.patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
'VAULT_URL': 'https://vault.work/',
|
||||
'VAULT_TOKEN': 'test_vault_token',
|
||||
},
|
||||
)
|
||||
def test_as_hvac_client_kwargs_without_optional_fields():
|
||||
# Given
|
||||
secrets = vault.VaultSecrets()
|
||||
|
||||
# When
|
||||
result = secrets.as_hvac_client_kwargs()
|
||||
|
||||
# Then
|
||||
assert result == {
|
||||
'url': 'https://vault.work/',
|
||||
'token': 'test_vault_token',
|
||||
}
|
||||
|
||||
|
||||
def test_get_client_cache_miss(mock_hvac_client: mock.Mock,
|
||||
hvac_client: mock.Mock):
|
||||
# Given
|
||||
mock_hvac_client.return_value = hvac_client
|
||||
|
||||
secrets = vault.VaultSecrets()
|
||||
|
||||
# When
|
||||
result = secrets.get_client()
|
||||
|
||||
# Then
|
||||
assert result == hvac_client
|
||||
|
||||
assert secrets.client == hvac_client
|
||||
|
||||
mock_hvac_client.assert_called_once_with(**secrets.as_hvac_client_kwargs())
|
||||
|
||||
|
||||
def test_get_client_cache_hit(mock_hvac_client: mock.Mock,
|
||||
hvac_client: mock.Mock):
|
||||
# Given
|
||||
secrets = vault.VaultSecrets()
|
||||
secrets.client = hvac_client
|
||||
|
||||
# When
|
||||
result = secrets.get_client()
|
||||
|
||||
# Then
|
||||
assert result == hvac_client
|
||||
|
||||
mock_hvac_client.assert_not_called()
|
Loading…
Reference in New Issue
Block a user