From 9a5fe2de8feb8c650f652c286bf1cf842ce7bcbc Mon Sep 17 00:00:00 2001 From: Esta Nagy Date: Wed, 7 Jan 2026 00:03:37 +0100 Subject: [PATCH] feat: New Container: Lowkey Vault - Add Lowkey Vault container implementation - Cover the new container with tests - Add basic example for secret use Resolves #948 --- modules/lowkey-vault/README.rst | 2 + modules/lowkey-vault/example_basic.py | 38 +++ .../testcontainers/lowkeyvault/__init__.py | 116 ++++++++ .../samples/network_container/Dockerfile | 20 ++ .../network_container/netowrk_container.py | 206 +++++++++++++++ .../lowkey-vault/tests/test_lowkey_vault.py | 249 ++++++++++++++++++ poetry.lock | 163 +++++++++++- pyproject.toml | 14 + 8 files changed, 801 insertions(+), 7 deletions(-) create mode 100644 modules/lowkey-vault/README.rst create mode 100644 modules/lowkey-vault/example_basic.py create mode 100644 modules/lowkey-vault/testcontainers/lowkeyvault/__init__.py create mode 100644 modules/lowkey-vault/tests/samples/network_container/Dockerfile create mode 100644 modules/lowkey-vault/tests/samples/network_container/netowrk_container.py create mode 100644 modules/lowkey-vault/tests/test_lowkey_vault.py diff --git a/modules/lowkey-vault/README.rst b/modules/lowkey-vault/README.rst new file mode 100644 index 00000000..6f986627 --- /dev/null +++ b/modules/lowkey-vault/README.rst @@ -0,0 +1,2 @@ +.. autoclass:: testcontainers.lowkeyvault.LowkeyVaultContainer +.. title:: testcontainers.lowkeyvault.LowkeyVaultContainer diff --git a/modules/lowkey-vault/example_basic.py b/modules/lowkey-vault/example_basic.py new file mode 100644 index 00000000..2459e1c9 --- /dev/null +++ b/modules/lowkey-vault/example_basic.py @@ -0,0 +1,38 @@ +import urllib3 +from azure.core.pipeline.transport._requests_basic import RequestsTransport +from azure.keyvault.secrets import SecretClient + +from testcontainers.lowkeyvault import LowkeyVaultContainer + + +def basic_example(): + with LowkeyVaultContainer() as lowkey_vault_container: + # get connection details + connection_url = lowkey_vault_container.get_connection_url() + print(f"Lowkey Vault is running: {connection_url}") + token = lowkey_vault_container.get_token() + print("Obtained token") + # prepare a transport ignoring self-signed certificate issues + transport = RequestsTransport(connection_verify=False) + # make sure to turn off challenge resource verification + secret_client: SecretClient = SecretClient( + vault_url=connection_url, credential=token, verify_challenge_resource=False, transport=transport + ) + + # set a secret + secret_client.set_secret(name="test-secret", value="a secret message") + print("The secret has been set.") + + # get the value of the secret + actual: str = secret_client.get_secret(name="test-secret").value + print(f"The secret has been retrieved with value: '{actual}'") + + # close the secret client + secret_client.close() + + +if __name__ == "__main__": + # ignore cert errors + urllib3.disable_warnings() + # run the code + basic_example() diff --git a/modules/lowkey-vault/testcontainers/lowkeyvault/__init__.py b/modules/lowkey-vault/testcontainers/lowkeyvault/__init__.py new file mode 100644 index 00000000..79ee6f4b --- /dev/null +++ b/modules/lowkey-vault/testcontainers/lowkeyvault/__init__.py @@ -0,0 +1,116 @@ +import os +from enum import Enum +from typing import Any, NamedTuple, Optional + +import requests + +from testcontainers.core.container import DockerContainer +from testcontainers.core.wait_strategies import LogMessageWaitStrategy + +# This comment can be removed (Used for testing) + + +class StaticToken(NamedTuple): + """Represents an OAuth access token.""" + + token: str + """The token string.""" + expires_on: int + """The token's expiration time in Unix time.""" + + +class StaticTokenCredential: + def __init__(self, json_token: str): + self.access_token = StaticToken(token=json_token.get("access_token"), expires_on=json_token.get("expires_on")) + + def get_token( + self, + *scopes: str, + claims: Optional[str] = None, + tenant_id: Optional[str] = None, + enable_cae: bool = False, + **kwargs: Any, + ) -> StaticToken: + return self.access_token + + +class NetworkType(Enum): + NETWORK = "network" + LOCAL = "local" + + +class LowkeyVaultContainer(DockerContainer): + """ + Container for a Lowkey Vault instance for emulating Azure Key Vault. + Supports Key, Secret and Certificate APIs. + + Example: + + .. doctest:: + + >>> from azure.core.pipeline.transport._requests_basic import RequestsTransport + >>> from azure.keyvault.secrets import SecretClient + >>> from testcontainers.lowkeyvault import LowkeyVaultContainer + + >>> with LowkeyVaultContainer() as lowkey_vault: + ... connection_url = lowkey_vault.get_connection_url() + ... token = lowkey_vault.get_token() + ... # don't fail due to the self-signed certificate + ... transport = RequestsTransport(connection_verify=False) + ... # make sure to turn off challenge resource verification + ... secret_client = SecretClient( + ... vault_url=connection_url, + ... credential=token, + ... verify_challenge_resource=False, + ... transport=transport + ... ) + """ + + def __init__( + self, image: str = "nagyesta/lowkey-vault:7.0.9-ubi10-minimal", container_alias: Optional[str] = None, **kwargs + ) -> None: + super().__init__(image, **kwargs) + self.api_port = 8443 + self.metadata_port = 8080 + self.with_exposed_ports(self.api_port, self.metadata_port) + self.with_env("LOWKEY_VAULT_RELAXED_PORTS", "true") + container_host_ip: str = self.get_container_host_ip() + if container_alias is not None: + self.with_network_aliases(container_alias) + self.container_alias = container_alias + self.with_env("LOWKEY_VAULT_ALIASES", f"localhost={container_alias}:") + elif container_host_ip != "localhost": + self.with_env("LOWKEY_VAULT_ALIASES", f"localhost={container_host_ip}:") + self.waiting_for(LogMessageWaitStrategy("Started LowkeyVaultApp.")) + + def _configure(self) -> None: + return + + def get_connection_url(self, network_type: NetworkType = NetworkType.LOCAL) -> str: + if network_type == NetworkType.LOCAL: + return f"https://{self.get_container_host_ip()}:{self.get_exposed_port(self.api_port)}" + else: + return f"https://{self.container_alias}:{self.api_port}" + + def get_imds_endpoint(self, network_type: NetworkType = NetworkType.LOCAL) -> str: + if network_type == NetworkType.LOCAL: + return f"http://{self.get_container_host_ip()}:{self.get_exposed_port(self.metadata_port)}" + else: + return f"http://{self.container_alias}:{self.metadata_port}" + + def auto_set_local_managed_identity_env_variables(self): + imds_endpoint: str = self.get_imds_endpoint(network_type=NetworkType.LOCAL) + token_url: str = self.get_token_url(network_type=NetworkType.LOCAL) + os.environ["AZURE_POD_IDENTITY_AUTHORITY_HOST"] = imds_endpoint + os.environ["IMDS_ENDPOINT"] = imds_endpoint + os.environ["IDENTITY_ENDPOINT"] = token_url + + def get_token_url(self, network_type: NetworkType = NetworkType.LOCAL) -> str: + base_url = self.get_imds_endpoint(network_type=network_type) + return f"{base_url}/metadata/identity/oauth2/token" + + def get_token(self, network_type: NetworkType = NetworkType.LOCAL) -> StaticTokenCredential: + resource = self.get_connection_url(network_type=network_type) + token_url = self.get_token_url(network_type=network_type) + json_response = requests.get(f"{token_url}?resource={resource}").json() + return StaticTokenCredential(json_token=json_response) diff --git a/modules/lowkey-vault/tests/samples/network_container/Dockerfile b/modules/lowkey-vault/tests/samples/network_container/Dockerfile new file mode 100644 index 00000000..2c4be65b --- /dev/null +++ b/modules/lowkey-vault/tests/samples/network_container/Dockerfile @@ -0,0 +1,20 @@ +# Use an official Python runtime as a parent image +FROM python:3.10-slim + +# Set the working directory in the container +WORKDIR /app + +# Install dependencies and create a dummy key file for the authentication to work +RUN \ + pip install azure-keyvault-certificates==4.10.0 && \ + pip install azure-keyvault-secrets==4.10.0 && \ + pip install azure-keyvault-keys==4.11.0 && \ + pip install cryptography==46.0.3 && \ + pip install azure-identity==1.25.1 && \ + mkdir -p /var/opt/azcmagent/tokens/ && \ + touch /var/opt/azcmagent/tokens/assumed-identity.key + +COPY ./netowrk_container.py netowrk_container.py +EXPOSE 80 +# Define the command to run the application +CMD ["python", "netowrk_container.py"] diff --git a/modules/lowkey-vault/tests/samples/network_container/netowrk_container.py b/modules/lowkey-vault/tests/samples/network_container/netowrk_container.py new file mode 100644 index 00000000..7e279995 --- /dev/null +++ b/modules/lowkey-vault/tests/samples/network_container/netowrk_container.py @@ -0,0 +1,206 @@ +import base64 +import os + +import urllib3 +from azure.core.pipeline.transport._requests_basic import RequestsTransport +from azure.identity import DefaultAzureCredential +from azure.keyvault.certificates import CertificateClient, CertificatePolicy +from azure.keyvault.keys import KeyClient, KeyOperation +from azure.keyvault.keys.crypto import CryptographyClient, EncryptionAlgorithm, EncryptResult, DecryptResult +from azure.keyvault.secrets import SecretClient +from cryptography import x509 +from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePrivateKey +from cryptography.hazmat.primitives.serialization.pkcs12 import load_key_and_certificates + +API_VERSION = "7.6" + + +def hello_secrets_from_an_external_container(): + """ + Entry point function for a custom Docker container to test connectivity + and "secrets" functionality with Lowkey Vault. + + This function is designed to run inside a separate container within the + same Docker network as a Lowkey Vault instance. + """ + vault_url = os.environ["CONNECTION_URL"] + secret_message = os.environ["SECRET_VALUE"] + secret_name = os.environ["SECRET_NAME"] + # test secrets API to see that the container works + secret_client = None + try: + # ignore SSL errors because we are using a self-signed certificate + transport_secrets = RequestsTransport(connection_verify=False) + # create the client we need + secret_client = SecretClient( + vault_url=vault_url, + credential=DefaultAzureCredential(), + verify_challenge_resource=False, + transport=transport_secrets, + api_version=API_VERSION, + ) + # set the result as a secret + secret_client.set_secret(name=secret_name, value=secret_message) + # get back the value + actual = secret_client.get_secret(name=secret_name).value + + # verify the result + assert actual == secret_message + print("Lowkey Vault Container created.") + except Exception as e: + print(f"Something went wrong : {e}") + finally: + # close client + if secret_client is not None: + secret_client.close() + + +def hello_keys_from_an_external_container(): + """ + Entry point function for a custom Docker container to test connectivity + and "keys" functionality with Lowkey Vault. + + This function is designed to run inside a separate container within the + same Docker network as a Lowkey Vault instance. + """ + vault_url = os.environ["CONNECTION_URL"] + secret_message = os.environ["SECRET_VALUE"] + key_name = os.environ["KEY_NAME"] + # test key API to see that the container works + key_client = None + crypto_client = None + try: + # ignore SSL errors because we are using a self-signed certificate + transport_keys = RequestsTransport(connection_verify=False) + transport_crypto = RequestsTransport(connection_verify=False) + # create the clients we need + key_client = KeyClient( + vault_url=vault_url, + credential=DefaultAzureCredential(), + verify_challenge_resource=False, + transport=transport_keys, + api_version=API_VERSION, + ) + # create a new key + key_client.create_rsa_key( + name=key_name, + size=2048, + key_operations=[KeyOperation.encrypt, KeyOperation.decrypt, KeyOperation.wrap_key, KeyOperation.unwrap_key], + ) + + crypto_client = CryptographyClient( + key=key_client.get_key(name=key_name).id, + credential=DefaultAzureCredential(), + verify_challenge_resource=False, + transport=transport_crypto, + api_version=API_VERSION, + ) + + # encode the text + text_as_bytes: bytes = bytes(secret_message.encode("utf-8")) + encrypted: EncryptResult = crypto_client.encrypt( + algorithm=EncryptionAlgorithm.rsa_oaep_256, plaintext=text_as_bytes + ) + cipher_text: bytes = encrypted.ciphertext + + # decode the cipher text + decrypted: DecryptResult = crypto_client.decrypt( + algorithm=EncryptionAlgorithm.rsa_oaep_256, ciphertext=cipher_text + ) + decrypted_text: str = decrypted.plaintext.decode("utf-8") + + # verify the result + assert decrypted_text == secret_message + print("Lowkey Vault Container created.") + except Exception as e: + print(f"Something went wrong : {e}") + finally: + # close clients + if key_client is not None: + key_client.close() + if crypto_client is not None: + crypto_client.close() + + +def hello_certificates_from_an_external_container(): + """ + Entry point function for a custom Docker container to test connectivity + and "certificates" functionality with Lowkey Vault. + + This function is designed to run inside a separate container within the + same Docker network as a Lowkey Vault instance. + """ + vault_url = os.environ["CONNECTION_URL"] + cert_name = os.environ["CERT_NAME"] + # test certificates API to see that the container works + certificate_client = None + secret_client = None + try: + # ignore SSL errors because we are using a self-signed certificate + transport_certs = RequestsTransport(connection_verify=False) + transport_secrets = RequestsTransport(connection_verify=False) + # create the clients we need + certificate_client = CertificateClient( + vault_url=vault_url, + credential=DefaultAzureCredential(), + verify_challenge_resource=False, + transport=transport_certs, + api_version=API_VERSION, + ) + secret_client = SecretClient( + vault_url=vault_url, + credential=DefaultAzureCredential(), + verify_challenge_resource=False, + transport=transport_secrets, + api_version=API_VERSION, + ) + + subject_name: str = "CN=example.com" + policy: CertificatePolicy = CertificatePolicy( + issuer_name="Self", + subject=subject_name, + key_curve_name="P-256", + key_type="EC", + validity_in_months=12, + content_type="application/x-pkcs12", + ) + certificate_client.begin_create_certificate(certificate_name=cert_name, policy=policy).wait() + + cert_value = secret_client.get_secret(name=cert_name).value + + # decode base64 secret + decoded = base64.b64decode(cert_value) + # open decoded secret as PKCS12 file + pkcs12 = load_key_and_certificates(decoded, b"") + + # get the components + ec_key: EllipticCurvePrivateKey = pkcs12[0] + x509_cert: x509.Certificate = pkcs12[1] + + # verify the result + assert subject_name == x509_cert.subject.rdns[0].rfc4514_string() + assert "secp256r1" == ec_key.curve.name + + print("Lowkey Vault Container created.") + except Exception as e: + print(f"Something went wrong : {e}") + finally: + # close clients + if certificate_client is not None: + certificate_client.close() + if secret_client is not None: + secret_client.close() + + +if __name__ == "__main__": + mode = os.getenv("TEST") + # ignore cert errors + urllib3.disable_warnings() + if mode == "secrets": + hello_secrets_from_an_external_container() + elif mode == "keys": + hello_keys_from_an_external_container() + elif mode == "certificates": + hello_certificates_from_an_external_container() + else: + print("The TEST env variable must be 'secrets', 'keys', or 'certificates'.") diff --git a/modules/lowkey-vault/tests/test_lowkey_vault.py b/modules/lowkey-vault/tests/test_lowkey_vault.py new file mode 100644 index 00000000..03fa9dce --- /dev/null +++ b/modules/lowkey-vault/tests/test_lowkey_vault.py @@ -0,0 +1,249 @@ +import base64 +import logging +from pathlib import Path + +from azure.core.pipeline.transport._requests_basic import RequestsTransport +from azure.keyvault.certificates import CertificateClient, CertificatePolicy +from azure.keyvault.keys import KeyClient, KeyOperation +from azure.keyvault.keys.crypto import CryptographyClient, EncryptResult, DecryptResult, EncryptionAlgorithm +from azure.keyvault.secrets import SecretClient +from cryptography import x509 +from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePrivateKey +from cryptography.hazmat.primitives.serialization.pkcs12 import load_key_and_certificates + +from testcontainers.core.image import DockerImage +from testcontainers.core.container import DockerContainer +from testcontainers.core.network import Network +from testcontainers.core.waiting_utils import wait_for_logs +from testcontainers.lowkeyvault import LowkeyVaultContainer, NetworkType + +logger = logging.getLogger(__name__) + +DOCKER_FILE_PATH = ".modules/lowkeyvault/tests/external_container_sample" +IMAGE_TAG = "external_container:test" + +TEST_DIR = Path(__file__).parent + +API_VERSION = "7.6" + + +def test_docker_run_lowkey_vault_with_secrets(): + with LowkeyVaultContainer() as lowkey_vault_container: + local_connection_url = lowkey_vault_container.get_connection_url() + token = lowkey_vault_container.get_token() + transport = RequestsTransport(connection_verify=False) + secret_client: SecretClient = SecretClient( + vault_url=local_connection_url, + credential=token, + verify_challenge_resource=False, + transport=transport, + api_version=API_VERSION, + ) + message: str = "a secret message" + name: str = "test-secret" + + secret_client.set_secret(name=name, value=message) + actual: str = secret_client.get_secret(name=name).value + + # close the client + secret_client.close() + # verify the result + assert actual == message + + +def test_docker_run_lowkey_vault_with_keys(): + with LowkeyVaultContainer() as lowkey_vault_container: + local_connection_url = lowkey_vault_container.get_connection_url() + token = lowkey_vault_container.get_token() + secret_message: str = "a secret message" + key_name: str = "rsa-key-name" + # test key API to see that the container works + # ignore SSL errors because we are using a self-signed certificate + transport_keys = RequestsTransport(connection_verify=False) + transport_crypto = RequestsTransport(connection_verify=False) + # create the clients we need + key_client = KeyClient( + vault_url=local_connection_url, + credential=token, + verify_challenge_resource=False, + transport=transport_keys, + api_version=API_VERSION, + ) + # create a new key + key_client.create_rsa_key( + name=key_name, + size=2048, + key_operations=[KeyOperation.encrypt, KeyOperation.decrypt, KeyOperation.wrap_key, KeyOperation.unwrap_key], + ) + + crypto_client = CryptographyClient( + key=key_client.get_key(name=key_name).id, + credential=token, + verify_challenge_resource=False, + transport=transport_crypto, + api_version=API_VERSION, + ) + + # encode the text + text_as_bytes: bytes = bytes(secret_message.encode("utf-8")) + encrypted: EncryptResult = crypto_client.encrypt( + algorithm=EncryptionAlgorithm.rsa_oaep_256, plaintext=text_as_bytes + ) + cipher_text: bytes = encrypted.ciphertext + + # decode the cipher text + decrypted: DecryptResult = crypto_client.decrypt( + algorithm=EncryptionAlgorithm.rsa_oaep_256, ciphertext=cipher_text + ) + decrypted_text: str = decrypted.plaintext.decode("utf-8") + + # close the clients + key_client.close() + crypto_client.close() + # verify the result + assert decrypted_text == secret_message + + +def test_docker_run_lowkey_vault_with_certificates(): + with LowkeyVaultContainer() as lowkey_vault_container: + local_connection_url = lowkey_vault_container.get_connection_url() + token = lowkey_vault_container.get_token() + # test certificates API to see that the container works + cert_name: str = "ec-cert-name" + # ignore SSL errors because we are using a self-signed certificate + transport_certs = RequestsTransport(connection_verify=False) + transport_secrets = RequestsTransport(connection_verify=False) + # create the clients we need + certificate_client = CertificateClient( + vault_url=local_connection_url, + credential=token, + verify_challenge_resource=False, + transport=transport_certs, + api_version=API_VERSION, + ) + secret_client = SecretClient( + vault_url=local_connection_url, + credential=token, + verify_challenge_resource=False, + transport=transport_secrets, + api_version=API_VERSION, + ) + + subject_name: str = "CN=example.com" + policy: CertificatePolicy = CertificatePolicy( + issuer_name="Self", + subject=subject_name, + key_curve_name="P-256", + key_type="EC", + validity_in_months=12, + content_type="application/x-pkcs12", + ) + certificate_client.begin_create_certificate(certificate_name=cert_name, policy=policy).wait() + + cert_value = secret_client.get_secret(name=cert_name).value + + # decode base64 secret + decoded = base64.b64decode(cert_value) + # open decoded secret as PKCS12 file + pkcs12 = load_key_and_certificates(decoded, b"") + + # get the components + ec_key: EllipticCurvePrivateKey = pkcs12[0] + x509_cert: x509.Certificate = pkcs12[1] + + # close the clients + secret_client.close() + certificate_client.close() + + # verify the result + assert subject_name == x509_cert.subject.rdns[0].rfc4514_string() + assert "secp256r1" == ec_key.curve.name + + +def test_docker_run_lowkey_vault_inter_container_communication_with_secrets(): + """ + Tests inter-container communication between a Lowkey Vault container and + a custom application container within the same Docker network using the + secrets API. + """ + with Network() as network: + with LowkeyVaultContainer(container_alias="lowkey-vault").with_network(network) as lowkey_vault_container: + network_connection_url = lowkey_vault_container.get_connection_url(network_type=NetworkType.NETWORK) + imds_endpoint = lowkey_vault_container.get_imds_endpoint(network_type=NetworkType.NETWORK) + token_url = lowkey_vault_container.get_token_url(network_type=NetworkType.NETWORK) + with DockerImage(path=TEST_DIR / "samples/network_container", tag=IMAGE_TAG) as image: + with ( + DockerContainer(image=str(image)) + .with_env("CONNECTION_URL", network_connection_url) + .with_env("AZURE_POD_IDENTITY_AUTHORITY_HOST", imds_endpoint) + .with_env("IMDS_ENDPOINT", imds_endpoint) + .with_env("IDENTITY_ENDPOINT", token_url) + .with_env("TEST", "secrets") + .with_env("SECRET_NAME", "secret-name") + .with_env("SECRET_VALUE", "secret-value") + .with_network(network) + .with_network_aliases("network_container") + .with_exposed_ports(80, 80) as container + ): + wait_for_logs(container=container, predicate="Lowkey Vault Container created.") + # make sure the container was actually created + assert lowkey_vault_container is not None + + +def test_docker_run_lowkey_vault_inter_container_communication_with_keys(): + """ + Tests inter-container communication between a Lowkey Vault container and + a custom application container within the same Docker network using the + keys API. + """ + with Network() as network: + with LowkeyVaultContainer(container_alias="lowkey-vault").with_network(network) as lowkey_vault_container: + network_connection_url = lowkey_vault_container.get_connection_url(network_type=NetworkType.NETWORK) + imds_endpoint = lowkey_vault_container.get_imds_endpoint(network_type=NetworkType.NETWORK) + token_url = lowkey_vault_container.get_token_url(network_type=NetworkType.NETWORK) + with DockerImage(path=TEST_DIR / "samples/network_container", tag=IMAGE_TAG) as image: + with ( + DockerContainer(image=str(image)) + .with_env("CONNECTION_URL", network_connection_url) + .with_env("AZURE_POD_IDENTITY_AUTHORITY_HOST", imds_endpoint) + .with_env("IMDS_ENDPOINT", imds_endpoint) + .with_env("IDENTITY_ENDPOINT", token_url) + .with_env("TEST", "keys") + .with_env("KEY_NAME", "rsa-key") + .with_env("SECRET_VALUE", "secret-value") + .with_network(network) + .with_network_aliases("network_container") + .with_exposed_ports(80, 80) as container + ): + wait_for_logs(container=container, predicate="Lowkey Vault Container created.") + # make sure the container was actually created + assert lowkey_vault_container is not None + + +def test_docker_run_lowkey_vault_inter_container_communication_with_certs(): + """ + Tests inter-container communication between a Lowkey Vault container and + a custom application container within the same Docker network using the + certificate API. + """ + with Network() as network: + with LowkeyVaultContainer(container_alias="lowkey-vault").with_network(network) as lowkey_vault_container: + network_connection_url = lowkey_vault_container.get_connection_url(network_type=NetworkType.NETWORK) + imds_endpoint = lowkey_vault_container.get_imds_endpoint(network_type=NetworkType.NETWORK) + token_url = lowkey_vault_container.get_token_url(network_type=NetworkType.NETWORK) + with DockerImage(path=TEST_DIR / "samples/network_container", tag=IMAGE_TAG) as image: + with ( + DockerContainer(image=str(image)) + .with_env("CONNECTION_URL", network_connection_url) + .with_env("AZURE_POD_IDENTITY_AUTHORITY_HOST", imds_endpoint) + .with_env("IMDS_ENDPOINT", imds_endpoint) + .with_env("IDENTITY_ENDPOINT", token_url) + .with_env("TEST", "certificates") + .with_env("CERT_NAME", "ec-cert-example-com") + .with_network(network) + .with_network_aliases("network_container") + .with_exposed_ports(80, 80) as container + ): + wait_for_logs(container=container, predicate="Lowkey Vault Container created.") + # make sure the container was actually created + assert lowkey_vault_container is not None diff --git a/poetry.lock b/poetry.lock index b25e8143..8ea0a2b7 100644 --- a/poetry.lock +++ b/poetry.lock @@ -430,7 +430,7 @@ description = "Microsoft Azure Core Library for Python" optional = true python-versions = ">=3.9" groups = ["main"] -markers = "extra == \"azurite\" or extra == \"cosmosdb\"" +markers = "extra == \"lowkey-vault\" or extra == \"azurite\" or extra == \"cosmosdb\"" files = [ {file = "azure_core-1.37.0-py3-none-any.whl", hash = "sha256:b3abe2c59e7d6bb18b38c275a5029ff80f98990e7c90a5e646249a56630fcc19"}, {file = "azure_core-1.37.0.tar.gz", hash = "sha256:7064f2c11e4b97f340e8e8c6d923b822978be3016e46b7bc4aa4b337cfb48aee"}, @@ -471,6 +471,101 @@ type = "legacy" url = "https://pypi.org/simple" reference = "PyPI-public" +[[package]] +name = "azure-identity" +version = "1.25.1" +description = "Microsoft Azure Identity Library for Python" +optional = true +python-versions = ">=3.9" +groups = ["main"] +markers = "extra == \"lowkey-vault\"" +files = [ + {file = "azure_identity-1.25.1-py3-none-any.whl", hash = "sha256:e9edd720af03dff020223cd269fa3a61e8f345ea75443858273bcb44844ab651"}, + {file = "azure_identity-1.25.1.tar.gz", hash = "sha256:87ca8328883de6036443e1c37b40e8dc8fb74898240f61071e09d2e369361456"}, +] + +[package.dependencies] +azure-core = ">=1.31.0" +cryptography = ">=2.5" +msal = ">=1.30.0" +msal-extensions = ">=1.2.0" +typing-extensions = ">=4.0.0" + +[package.source] +type = "legacy" +url = "https://pypi.org/simple" +reference = "PyPI-public" + +[[package]] +name = "azure-keyvault-certificates" +version = "4.10.0" +description = "Microsoft Corporation Key Vault Certificates Client Library for Python" +optional = true +python-versions = ">=3.9" +groups = ["main"] +markers = "extra == \"lowkey-vault\"" +files = [ + {file = "azure_keyvault_certificates-4.10.0-py3-none-any.whl", hash = "sha256:fa76cbc329274cb5f4ab61b0ed7d209d44377df4b4d6be2fd01e741c2fbb83a9"}, + {file = "azure_keyvault_certificates-4.10.0.tar.gz", hash = "sha256:004ff47a73152f9f40f678e5a07719b753a3ca86f0460bfeaaf6a23304872e05"}, +] + +[package.dependencies] +azure-core = ">=1.31.0" +isodate = ">=0.6.1" +typing-extensions = ">=4.6.0" + +[package.source] +type = "legacy" +url = "https://pypi.org/simple" +reference = "PyPI-public" + +[[package]] +name = "azure-keyvault-keys" +version = "4.11.0" +description = "Microsoft Corporation Key Vault Keys Client Library for Python" +optional = true +python-versions = ">=3.9" +groups = ["main"] +markers = "extra == \"lowkey-vault\"" +files = [ + {file = "azure_keyvault_keys-4.11.0-py3-none-any.whl", hash = "sha256:fa5febd5805f0fed4c0a1d13c9096081c72a6fa36ccae1299a137f34280eda53"}, + {file = "azure_keyvault_keys-4.11.0.tar.gz", hash = "sha256:f257b1917a2c3a88983e3f5675a6419449eb262318888d5b51e1cb3bed79779a"}, +] + +[package.dependencies] +azure-core = ">=1.31.0" +cryptography = ">=2.1.4" +isodate = ">=0.6.1" +typing-extensions = ">=4.6.0" + +[package.source] +type = "legacy" +url = "https://pypi.org/simple" +reference = "PyPI-public" + +[[package]] +name = "azure-keyvault-secrets" +version = "4.10.0" +description = "Microsoft Corporation Key Vault Secrets Client Library for Python" +optional = true +python-versions = ">=3.9" +groups = ["main"] +markers = "extra == \"lowkey-vault\"" +files = [ + {file = "azure_keyvault_secrets-4.10.0-py3-none-any.whl", hash = "sha256:9dbde256077a4ee1a847646671580692e3f9bea36bcfc189c3cf2b9a94eb38b9"}, + {file = "azure_keyvault_secrets-4.10.0.tar.gz", hash = "sha256:666fa42892f9cee749563e551a90f060435ab878977c95265173a8246d546a36"}, +] + +[package.dependencies] +azure-core = ">=1.31.0" +isodate = ">=0.6.1" +typing-extensions = ">=4.6.0" + +[package.source] +type = "legacy" +url = "https://pypi.org/simple" +reference = "PyPI-public" + [[package]] name = "azure-storage-blob" version = "12.27.1" @@ -910,7 +1005,7 @@ files = [ {file = "cffi-2.0.0-cp39-cp39-win_amd64.whl", hash = "sha256:b882b3df248017dba09d6b16defe9b5c407fe32fc7c65a9c69798e6175601be9"}, {file = "cffi-2.0.0.tar.gz", hash = "sha256:44d1b5909021139fe36001ae048dbdde8214afa20200eda0f64c068cac5d5529"}, ] -markers = {main = "python_version >= \"3.10\" and python_version <= \"3.13\" and platform_python_implementation != \"PyPy\" and (extra == \"keycloak\" or extra == \"azurite\" or extra == \"mysql\" or extra == \"oracle\" or extra == \"oracle-free\" or extra == \"weaviate\" or extra == \"mailpit\" or extra == \"sftp\") or extra == \"minio\" and python_version >= \"3.10\" or os_name == \"nt\" and implementation_name != \"pypy\" and extra == \"selenium\" and python_version >= \"3.10\" or python_version < \"4.0\" and extra == \"keycloak\" and python_version >= \"3.14\" and platform_python_implementation != \"PyPy\" or (extra == \"azurite\" or extra == \"mysql\" or extra == \"oracle\" or extra == \"oracle-free\" or extra == \"weaviate\" or extra == \"mailpit\" or extra == \"sftp\") and python_version >= \"3.14\" and platform_python_implementation != \"PyPy\"", dev = "platform_python_implementation != \"PyPy\""} +markers = {main = "python_version >= \"3.10\" and python_version <= \"3.13\" and platform_python_implementation != \"PyPy\" and (extra == \"keycloak\" or extra == \"lowkey-vault\" or extra == \"azurite\" or extra == \"mysql\" or extra == \"oracle\" or extra == \"oracle-free\" or extra == \"weaviate\" or extra == \"mailpit\" or extra == \"sftp\") or extra == \"minio\" and python_version >= \"3.10\" or os_name == \"nt\" and implementation_name != \"pypy\" and extra == \"selenium\" and python_version >= \"3.10\" or python_version < \"4.0\" and extra == \"keycloak\" and python_version >= \"3.14\" and platform_python_implementation != \"PyPy\" or (extra == \"lowkey-vault\" or extra == \"azurite\" or extra == \"mysql\" or extra == \"oracle\" or extra == \"oracle-free\" or extra == \"weaviate\" or extra == \"mailpit\" or extra == \"sftp\") and python_version >= \"3.14\" and platform_python_implementation != \"PyPy\"", dev = "platform_python_implementation != \"PyPy\""} [package.dependencies] pycparser = {version = "*", markers = "implementation_name != \"PyPy\""} @@ -1435,7 +1530,7 @@ files = [ {file = "cryptography-46.0.3-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:6b5063083824e5509fdba180721d55909ffacccc8adbec85268b48439423d78c"}, {file = "cryptography-46.0.3.tar.gz", hash = "sha256:a8b17438104fed022ce745b362294d9ce35b4c2e45c1d958ad4a4b019285f4a1"}, ] -markers = {main = "(python_version < \"4.0\" or extra == \"azurite\" or extra == \"mysql\" or extra == \"oracle\" or extra == \"oracle-free\" or extra == \"weaviate\" or extra == \"mailpit\" or extra == \"sftp\") and (extra == \"keycloak\" or extra == \"azurite\" or extra == \"mysql\" or extra == \"oracle\" or extra == \"oracle-free\" or extra == \"weaviate\" or extra == \"mailpit\" or extra == \"sftp\")"} +markers = {main = "(python_version < \"4.0\" or extra == \"lowkey-vault\" or extra == \"azurite\" or extra == \"mysql\" or extra == \"oracle\" or extra == \"oracle-free\" or extra == \"weaviate\" or extra == \"mailpit\" or extra == \"sftp\") and (extra == \"keycloak\" or extra == \"lowkey-vault\" or extra == \"azurite\" or extra == \"mysql\" or extra == \"oracle\" or extra == \"oracle-free\" or extra == \"weaviate\" or extra == \"mailpit\" or extra == \"sftp\")"} [package.dependencies] cffi = {version = ">=2.0.0", markers = "python_full_version >= \"3.9.0\" and platform_python_implementation != \"PyPy\""} @@ -2809,7 +2904,7 @@ description = "An ISO 8601 date/time/duration parser and formatter" optional = true python-versions = ">=3.7" groups = ["main"] -markers = "extra == \"azurite\"" +markers = "extra == \"lowkey-vault\" or extra == \"azurite\"" files = [ {file = "isodate-0.7.2-py3-none-any.whl", hash = "sha256:28009937d8031054830160fce6d409ed342816b543597cece116d966c6d99e15"}, {file = "isodate-0.7.2.tar.gz", hash = "sha256:4cd1aa0f43ca76f4a6c6c0292a85f40b35ec2e43e315b59f06e6d32171a953e6"}, @@ -3494,6 +3589,56 @@ type = "legacy" url = "https://pypi.org/simple" reference = "PyPI-public" +[[package]] +name = "msal" +version = "1.34.0" +description = "The Microsoft Authentication Library (MSAL) for Python library enables your app to access the Microsoft Cloud by supporting authentication of users with Microsoft Azure Active Directory accounts (AAD) and Microsoft Accounts (MSA) using industry standard OAuth2 and OpenID Connect." +optional = true +python-versions = ">=3.8" +groups = ["main"] +markers = "extra == \"lowkey-vault\"" +files = [ + {file = "msal-1.34.0-py3-none-any.whl", hash = "sha256:f669b1644e4950115da7a176441b0e13ec2975c29528d8b9e81316023676d6e1"}, + {file = "msal-1.34.0.tar.gz", hash = "sha256:76ba83b716ea5a6d75b0279c0ac353a0e05b820ca1f6682c0eb7f45190c43c2f"}, +] + +[package.dependencies] +cryptography = ">=2.5,<49" +PyJWT = {version = ">=1.0.0,<3", extras = ["crypto"]} +requests = ">=2.0.0,<3" + +[package.extras] +broker = ["pymsalruntime (>=0.14,<0.19) ; python_version >= \"3.6\" and platform_system == \"Windows\"", "pymsalruntime (>=0.17,<0.19) ; python_version >= \"3.8\" and platform_system == \"Darwin\"", "pymsalruntime (>=0.18,<0.19) ; python_version >= \"3.8\" and platform_system == \"Linux\""] + +[package.source] +type = "legacy" +url = "https://pypi.org/simple" +reference = "PyPI-public" + +[[package]] +name = "msal-extensions" +version = "1.3.1" +description = "Microsoft Authentication Library extensions (MSAL EX) provides a persistence API that can save your data on disk, encrypted on Windows, macOS and Linux. Concurrent data access will be coordinated by a file lock mechanism." +optional = true +python-versions = ">=3.9" +groups = ["main"] +markers = "extra == \"lowkey-vault\"" +files = [ + {file = "msal_extensions-1.3.1-py3-none-any.whl", hash = "sha256:96d3de4d034504e969ac5e85bae8106c8373b5c6568e4c8fa7af2eca9dbe6bca"}, + {file = "msal_extensions-1.3.1.tar.gz", hash = "sha256:c5b0fd10f65ef62b5f1d62f4251d51cbcaf003fcedae8c91b040a488614be1a4"}, +] + +[package.dependencies] +msal = ">=1.29,<2" + +[package.extras] +portalocker = ["portalocker (>=1.4,<4)"] + +[package.source] +type = "legacy" +url = "https://pypi.org/simple" +reference = "PyPI-public" + [[package]] name = "msgpack" version = "1.1.2" @@ -5457,7 +5602,7 @@ files = [ {file = "pycparser-2.23-py3-none-any.whl", hash = "sha256:e5c6e8d3fbad53479cab09ac03729e0a9faf2bee3db8208a550daf5af81a5934"}, {file = "pycparser-2.23.tar.gz", hash = "sha256:78816d4f24add8f10a06d6f05b4d424ad9e96cfebf68a4ddc99c65c0720d00c2"}, ] -markers = {main = "((python_version < \"4.0\" and extra == \"keycloak\" or extra == \"azurite\" or extra == \"mysql\" or extra == \"oracle\" or extra == \"oracle-free\" or extra == \"weaviate\" or extra == \"mailpit\" or extra == \"sftp\") and platform_python_implementation != \"PyPy\" or extra == \"minio\" and python_version < \"3.14\" or os_name == \"nt\" and implementation_name != \"pypy\" and extra == \"selenium\") and implementation_name != \"PyPy\" and python_version == \"3.10\" or python_version < \"4.0\" and extra == \"keycloak\" and python_version >= \"3.14\" and platform_python_implementation != \"PyPy\" and implementation_name != \"PyPy\" or (extra == \"azurite\" or extra == \"mysql\" or extra == \"oracle\" or extra == \"oracle-free\" or extra == \"weaviate\" or extra == \"mailpit\" or extra == \"sftp\") and python_version >= \"3.14\" and platform_python_implementation != \"PyPy\" and implementation_name != \"PyPy\" or extra == \"minio\" and implementation_name != \"PyPy\" and python_version >= \"3.11\" or os_name == \"nt\" and implementation_name != \"pypy\" and implementation_name != \"PyPy\" and extra == \"selenium\" and python_version >= \"3.11\" or python_version >= \"3.11\" and python_version <= \"3.13\" and platform_python_implementation != \"PyPy\" and implementation_name != \"PyPy\" and (extra == \"keycloak\" or extra == \"azurite\" or extra == \"mysql\" or extra == \"oracle\" or extra == \"oracle-free\" or extra == \"weaviate\" or extra == \"mailpit\" or extra == \"sftp\")", dev = "platform_python_implementation != \"PyPy\" and implementation_name != \"PyPy\""} +markers = {main = "python_version >= \"3.10\" and python_version <= \"3.13\" and platform_python_implementation != \"PyPy\" and implementation_name != \"PyPy\" and (extra == \"keycloak\" or extra == \"lowkey-vault\" or extra == \"azurite\" or extra == \"mysql\" or extra == \"oracle\" or extra == \"oracle-free\" or extra == \"weaviate\" or extra == \"mailpit\" or extra == \"sftp\") or extra == \"minio\" and implementation_name != \"PyPy\" and python_version >= \"3.10\" or os_name == \"nt\" and implementation_name != \"pypy\" and implementation_name != \"PyPy\" and extra == \"selenium\" and python_version >= \"3.10\" or python_version < \"4.0\" and extra == \"keycloak\" and python_version >= \"3.14\" and platform_python_implementation != \"PyPy\" and implementation_name != \"PyPy\" or (extra == \"lowkey-vault\" or extra == \"azurite\" or extra == \"mysql\" or extra == \"oracle\" or extra == \"oracle-free\" or extra == \"weaviate\" or extra == \"mailpit\" or extra == \"sftp\") and python_version >= \"3.14\" and platform_python_implementation != \"PyPy\" and implementation_name != \"PyPy\"", dev = "platform_python_implementation != \"PyPy\" and implementation_name != \"PyPy\""} [package.source] type = "legacy" @@ -5714,12 +5859,15 @@ description = "JSON Web Token implementation in Python" optional = true python-versions = ">=3.9" groups = ["main"] -markers = "extra == \"arangodb\"" +markers = "extra == \"arangodb\" or extra == \"lowkey-vault\"" files = [ {file = "PyJWT-2.10.1-py3-none-any.whl", hash = "sha256:dcdd193e30abefd5debf142f9adfcdd2b58004e644f25406ffaebd50bd98dacb"}, {file = "pyjwt-2.10.1.tar.gz", hash = "sha256:3cc5772eb20009233caf06e9d8a0577824723b44e6648ee0a2aedb6cf9381953"}, ] +[package.dependencies] +cryptography = {version = ">=3.4.0", optional = true, markers = "extra == \"crypto\""} + [package.extras] crypto = ["cryptography (>=3.4.0)"] dev = ["coverage[toml] (==5.0.4)", "cryptography (>=3.4.0)", "pre-commit", "pytest (>=6.0.0,<7.0.0)", "sphinx", "sphinx-rtd-theme", "zope.interface"] @@ -8400,6 +8548,7 @@ k3s = ["kubernetes", "pyyaml"] kafka = [] keycloak = ["python-keycloak"] localstack = ["boto3"] +lowkey-vault = ["azure-identity", "azure-keyvault-certificates", "azure-keyvault-keys", "azure-keyvault-secrets", "cryptography"] mailpit = ["cryptography"] memcached = [] milvus = [] @@ -8432,4 +8581,4 @@ weaviate = ["weaviate-client"] [metadata] lock-version = "2.1" python-versions = ">=3.10" -content-hash = "b52b8f3b08e45d7d0e41006651ccdf1f88d04db0ec1cc2a25b56438408195fce" +content-hash = "5eeb6bb97768d6fe6af352bed5f4c772bed5ca2b86d2eaf4b95c4c26b86d7688" diff --git a/pyproject.toml b/pyproject.toml index 7d936084..61c0d3bd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,6 +49,7 @@ packages = [ { include = "testcontainers", from = "modules/kafka" }, { include = "testcontainers", from = "modules/keycloak" }, { include = "testcontainers", from = "modules/localstack" }, + { include = "testcontainers", from = "modules/lowkey-vault" }, { include = "testcontainers", from = "modules/mailpit" }, { include = "testcontainers", from = "modules/memcached" }, { include = "testcontainers", from = "modules/minio" }, @@ -91,6 +92,10 @@ python-dotenv = "*" # community modules python-arango = { version = "^8", optional = true } +azure-identity = { version = ">=1.25.1", optional = true } +azure-keyvault-certificates = { version = ">=4.10.0", optional = true } +azure-keyvault-keys = { version = ">=4.11.0", optional = true } +azure-keyvault-secrets = { version = ">=4.10.0", optional = true } azure-storage-blob = { version = "^12", optional = true } cassandra-driver = { version = "^3", optional = true } #clickhouse-driver = { version = "*", optional = true } @@ -166,6 +171,14 @@ k3s = ["kubernetes", "pyyaml"] kafka = [] keycloak = ["python-keycloak"] localstack = ["boto3"] +lowkey-vault = [ + "azure-identity", + "azure-keyvault-certificates", + "azure-keyvault-keys", + "azure-keyvault-secrets", + "cryptography", + +] mailpit = ["cryptography"] memcached = [] minio = ["minio"] @@ -344,6 +357,7 @@ mypy_path = [ # "modules/kafka", # "modules/keycloak", # "modules/localstack", + # "modules/lowkeyvault", "modules/mailpit", # "modules/minio", # "modules/mongodb",