From ff96348e9243a525bf72f74b1c860c5c63705856 Mon Sep 17 00:00:00 2001 From: Daniil Anfimov Date: Tue, 25 Aug 2026 18:31:30 +0300 Subject: [PATCH] feat: fetch GPG passphrases from HashiCorp Vault Adds a HashiCorp Vault KV v2 provider as an alternative to Bitwarden for resolving GPG key passphrases at startup, so unattended restarts do not fall back to an interactive prompt. Passphrase resolution now goes through sign_node/utils/secrets.py, which picks the enabled provider and refuses to run with more than one enabled: signing keys should have a single unambiguous source of truth, so that is a configuration error rather than a fallback chain. Each keyid maps to a secret at // holding the passphrase in a configurable field ('passphrase' by default). Supports a static token (inline or from a file), AppRole, and the VAULT_ADDR/VAULT_TOKEN environment variables for hosts already running a Vault agent. hvac is imported lazily so it is only needed when the provider is enabled. Passphrases are read once at startup, so a short-lived token is enough and no Vault session is renewed while the node runs. Resolves: AlmaLinux/build-system#548 --- README.md | 58 +++++- almalinux_sign_node.py | 18 +- node-config/sign_node.yml | 26 +++ requirements.txt | 1 + sign_node/config.py | 34 ++++ sign_node/utils/secrets.py | 74 ++++++++ sign_node/utils/vault.py | 157 +++++++++++++++++ tests/sign_node/utils/test_secrets.py | 106 +++++++++++ tests/sign_node/utils/test_vault.py | 243 ++++++++++++++++++++++++++ 9 files changed, 697 insertions(+), 20 deletions(-) create mode 100644 sign_node/utils/secrets.py create mode 100644 sign_node/utils/vault.py create mode 100644 tests/sign_node/utils/test_secrets.py create mode 100644 tests/sign_node/utils/test_vault.py diff --git a/README.md b/README.md index 3790d70..bd9bbfd 100644 --- a/README.md +++ b/README.md @@ -28,12 +28,21 @@ Pre-requisites: To start the system, run the following command: `docker compose up -d`. To rebuild images after your local changes, just run `docker compose up -d --build`. -# Fetching GPG passphrases from Bitwarden +# Fetching GPG passphrases from a secret provider By default the sign node asks for each PGP key passphrase interactively at startup (or uses `dev_pgp_key_password` in development mode). Instead, it can -fetch passphrases from a Bitwarden vault using -[py-bitwarden-wrapper](https://github.com/AlmaLinux/py-bitwarden-wrapper). +fetch them from Bitwarden or from HashiCorp Vault. + +Only **one** provider may be enabled at a time — signing keys should have a +single unambiguous source of truth, so enabling both is a configuration error +rather than a fallback chain. Whichever provider is enabled takes precedence +over the development password and interactive prompts, and startup fails fast +if any keyid is missing from it or its passphrase does not unlock the GPG key. + +## Bitwarden + +Uses [py-bitwarden-wrapper](https://github.com/AlmaLinux/py-bitwarden-wrapper). Requirements: * The Bitwarden CLI (`bw`) must be installed and on `PATH`. @@ -54,13 +63,48 @@ bitwarden_password_file: /run/secrets/bw_master # bitwarden_collection_id: ``` -When `bitwarden_enabled` is true, fetched passphrases take precedence over the -development password and interactive prompts. Startup fails fast if any keyid -is missing from the vault or its passphrase does not unlock the GPG key. - `bitwarden-wrapper` is not published on PyPI — it is installed directly from GitHub via `requirements.txt`. +## HashiCorp Vault + +Reads passphrases from a KV v2 store using +[hvac](https://github.com/hvac/hvac). For each keyid listed in `pgp_keys`, +create a secret at `//` holding the +passphrase in the `passphrase` field: + +``` +vault kv put secret/albs/sign-keys/7C3955C2A345DA89 passphrase='...' +``` + +Enable it in the node config (`sign_node.yml`): + +```yaml +vault_enabled: yes +vault_addr: https://vault.example.com:8200 +vault_mount: secret # KV v2 mount point +vault_path_prefix: albs/sign-keys +# Authenticate with a static token from a file (preferred) ... +vault_token_file: /run/secrets/vault_token +# ... or inline (less safe): +# vault_token: "..." +# ... or via AppRole: +# vault_role_id: +# vault_secret_id_file: /run/secrets/vault_secret_id +# Optional: Vault Enterprise / HCP namespace and a custom CA bundle. +# vault_namespace: admin/albs +# vault_ca_cert: /etc/pki/vault-ca.pem +# Optional: read a different field, for an existing secret layout. +# vault_passphrase_field: passphrase +``` + +`VAULT_ADDR` and `VAULT_TOKEN` from the environment are used as a fallback when +the corresponding options are unset, so a host already running a Vault agent +needs no credentials in the config file. + +Passphrases are read once at startup, so a short-lived token is sufficient and +no Vault session is renewed while the node runs. + # Reporting issues All issues should be reported to the [Build System project](https://github.com/AlmaLinux/build-system). diff --git a/almalinux_sign_node.py b/almalinux_sign_node.py index 7501668..94cda62 100755 --- a/almalinux_sign_node.py +++ b/almalinux_sign_node.py @@ -16,10 +16,10 @@ from sign_node.config import SignNodeConfig from sign_node.errors import ConfigurationError from sign_node.signer import Signer -from sign_node.utils.bitwarden import fetch_passphrases from sign_node.utils.config import locate_config_file from sign_node.utils.file_utils import clean_dir, safe_mkdir from sign_node.utils.pgp_utils import PGPPasswordDB, init_gpg +from sign_node.utils.secrets import resolve_passphrases def init_arg_parser(): @@ -78,18 +78,10 @@ def main(): init_sentry(config) gpg = init_gpg() - preloaded_passwords = None - if config.bitwarden_enabled: - try: - preloaded_passwords = fetch_passphrases( - keyids=config.pgp_keys, - username=config.bitwarden_username, - password=config.bitwarden_password, - password_file=config.bitwarden_password_file, - collection_id=config.bitwarden_collection_id, - ) - except ConfigurationError as e: - args_parser.error(str(e)) + try: + preloaded_passwords = resolve_passphrases(config) + except ConfigurationError as e: + args_parser.error(str(e)) password_db = PGPPasswordDB( gpg, key_ids_from_config=config.pgp_keys.copy(), diff --git a/node-config/sign_node.yml b/node-config/sign_node.yml index 49eee95..b326e2a 100644 --- a/node-config/sign_node.yml +++ b/node-config/sign_node.yml @@ -19,3 +19,29 @@ is_community_sign_node: true # bitwarden_password: "..." # Optional: restrict the lookup to a single collection (real UUID only). # bitwarden_collection_id: + +# Alternatively, fetch the passphrases from a HashiCorp Vault KV v2 store. +# For each keyid in 'pgp_keys', create a secret at +# '//' holding the passphrase in the +# 'passphrase' field: +# vault kv put secret/albs/sign-keys/ passphrase='...' +# Only one secret provider may be enabled at a time: switching this on while +# 'bitwarden_enabled' is also set is a configuration error. +# vault_enabled: yes +# vault_addr: https://vault.example.com:8200 +# vault_mount: secret +# vault_path_prefix: albs/sign-keys +# Authenticate with a static token from a file (preferred) ... +# vault_token_file: /run/secrets/vault_token +# ... or inline (less safe): +# vault_token: "..." +# ... or via AppRole: +# vault_role_id: +# vault_secret_id_file: /run/secrets/vault_secret_id +# VAULT_ADDR and VAULT_TOKEN from the environment are used as a fallback, +# so a host running a Vault agent needs no credentials here. +# Optional: Vault Enterprise / HCP namespace, custom CA bundle, and a +# different field name for an existing secret layout. +# vault_namespace: admin/albs +# vault_ca_cert: /etc/pki/vault-ca.pem +# vault_passphrase_field: passphrase diff --git a/requirements.txt b/requirements.txt index 96958b3..aaee9bb 100644 --- a/requirements.txt +++ b/requirements.txt @@ -16,5 +16,6 @@ websocket-client==1.8.0 # https://jasonralph.org/?p=997 cryptography==43.0.3 pgpy==0.6.0 +hvac==2.4.0 git+https://github.com/AlmaLinux/immudb-wrapper.git@0.1.9#egg=immudb_wrapper git+https://github.com/AlmaLinux/py-bitwarden-wrapper.git@0.1.17#egg=bitwarden_wrapper diff --git a/sign_node/config.py b/sign_node/config.py index af850b4..b9ee5d2 100644 --- a/sign_node/config.py +++ b/sign_node/config.py @@ -8,6 +8,8 @@ from .utils.config import BaseConfig from .utils.file_utils import normalize_path +from .utils.vault import DEFAULT_FIELD as DEFAULT_VAULT_FIELD +from .utils.vault import DEFAULT_MOUNT as DEFAULT_VAULT_MOUNT __all__ = ["SignNodeConfig"] @@ -82,6 +84,18 @@ def __init__(self, config_file=None, **cmd_args): 'bitwarden_password': None, 'bitwarden_password_file': None, 'bitwarden_collection_id': None, + 'vault_enabled': False, + 'vault_addr': None, + 'vault_token': None, + 'vault_token_file': None, + 'vault_role_id': None, + 'vault_secret_id': None, + 'vault_secret_id_file': None, + 'vault_namespace': None, + 'vault_mount': DEFAULT_VAULT_MOUNT, + 'vault_path_prefix': '', + 'vault_passphrase_field': DEFAULT_VAULT_FIELD, + 'vault_ca_cert': None, } schema = { "development_mode": {"type": "boolean", "default": False}, @@ -118,6 +132,26 @@ def __init__(self, config_file=None, **cmd_args): 'bitwarden_password': {'type': 'string', 'nullable': True}, 'bitwarden_password_file': {'type': 'string', 'nullable': True}, 'bitwarden_collection_id': {'type': 'string', 'nullable': True}, + 'vault_enabled': {'type': 'boolean', 'default': False}, + 'vault_addr': {'type': 'string', 'nullable': True}, + 'vault_token': {'type': 'string', 'nullable': True}, + 'vault_token_file': {'type': 'string', 'nullable': True}, + 'vault_role_id': {'type': 'string', 'nullable': True}, + 'vault_secret_id': {'type': 'string', 'nullable': True}, + 'vault_secret_id_file': {'type': 'string', 'nullable': True}, + 'vault_namespace': {'type': 'string', 'nullable': True}, + 'vault_mount': { + 'type': 'string', + 'default': DEFAULT_VAULT_MOUNT, + 'empty': False, + }, + 'vault_path_prefix': {'type': 'string', 'nullable': True}, + 'vault_passphrase_field': { + 'type': 'string', + 'default': DEFAULT_VAULT_FIELD, + 'empty': False, + }, + 'vault_ca_cert': {'type': 'string', 'nullable': True}, } super(SignNodeConfig, self).__init__( default_config, config_file, schema, **cmd_args diff --git a/sign_node/utils/secrets.py b/sign_node/utils/secrets.py new file mode 100644 index 0000000..4052811 --- /dev/null +++ b/sign_node/utils/secrets.py @@ -0,0 +1,74 @@ +# -*- mode:python; coding:utf-8; -*- + +"""Resolve GPG key passphrases from an external secret provider. + +Exactly one provider may be enabled at a time: signing keys should have a +single unambiguous source of truth, so enabling several is a configuration +error rather than a merge of their results. +""" + +import logging +from typing import Dict, Optional + +from ..errors import ConfigurationError +from . import bitwarden, vault + +__all__ = ["resolve_passphrases", "enabled_providers"] + +logger = logging.getLogger(__name__) + + +def enabled_providers(config) -> list: + """Names of the secret providers switched on in the configuration.""" + flags = ( + ("bitwarden", config.bitwarden_enabled), + ("vault", config.vault_enabled), + ) + return [name for name, enabled in flags if enabled] + + +def _from_vault(config) -> Dict[str, str]: + return vault.fetch_passphrases( + keyids=config.pgp_keys, + addr=config.vault_addr, + token=config.vault_token, + token_file=config.vault_token_file, + role_id=config.vault_role_id, + secret_id=config.vault_secret_id, + secret_id_file=config.vault_secret_id_file, + namespace=config.vault_namespace, + mount=config.vault_mount, + path_prefix=config.vault_path_prefix, + field=config.vault_passphrase_field, + ca_cert=config.vault_ca_cert, + ) + + +def _from_bitwarden(config) -> Dict[str, str]: + return bitwarden.fetch_passphrases( + keyids=config.pgp_keys, + username=config.bitwarden_username, + password=config.bitwarden_password, + password_file=config.bitwarden_password_file, + collection_id=config.bitwarden_collection_id, + ) + + +def resolve_passphrases(config) -> Optional[Dict[str, str]]: + """Fetch passphrases from the configured provider. + + Returns ``None`` when no provider is enabled, leaving the caller to fall + back to development mode or interactive prompts. + """ + providers = enabled_providers(config) + if len(providers) > 1: + raise ConfigurationError( + "Only one secret provider may be enabled at a time, but these " + "are enabled: " + ", ".join(providers) + ) + if not providers: + return None + provider = providers[0] + logger.info("Using the %s secret provider for GPG passphrases", provider) + fetchers = {"bitwarden": _from_bitwarden, "vault": _from_vault} + return fetchers[provider](config) diff --git a/sign_node/utils/vault.py b/sign_node/utils/vault.py new file mode 100644 index 0000000..f0e33c7 --- /dev/null +++ b/sign_node/utils/vault.py @@ -0,0 +1,157 @@ +# -*- mode:python; coding:utf-8; -*- + +"""Fetch GPG key passphrases from a HashiCorp Vault KV v2 store. + +Each GPG keyid must correspond to a secret at +``//`` holding the passphrase in a single field +(``passphrase`` by default). +""" + +import logging +import os +from typing import Dict, List, Optional + +from ..errors import ConfigurationError + +__all__ = ["fetch_passphrases", "DEFAULT_FIELD", "DEFAULT_MOUNT"] + +logger = logging.getLogger(__name__) + +DEFAULT_MOUNT = "secret" +DEFAULT_FIELD = "passphrase" + + +def _read_secret_file(path: str, what: str) -> str: + try: + with open(path, "r", encoding="utf-8") as fd: + value = fd.read().strip() + except OSError as e: + raise ConfigurationError( + f"Cannot read Vault {what} from {path}: {e}" + ) from e + if not value: + raise ConfigurationError(f"Vault {what} file {path} is empty") + return value + + +def _authenticate( + client, + hvac_exceptions, + *, + token: Optional[str], + token_file: Optional[str], + role_id: Optional[str], + secret_id: Optional[str], + secret_id_file: Optional[str], +): + """Log the client in, preferring a static token over AppRole.""" + if token_file: + client.token = _read_secret_file(token_file, "token") + elif token: + client.token = token + elif role_id: + if secret_id_file: + secret_id = _read_secret_file(secret_id_file, "secret_id") + if not secret_id: + raise ConfigurationError( + "Vault AppRole authentication requires a secret id " + "(set vault_secret_id or vault_secret_id_file)" + ) + try: + client.auth.approle.login(role_id=role_id, secret_id=secret_id) + except hvac_exceptions.VaultError as e: + raise ConfigurationError( + f"Vault AppRole authentication failed: {e}" + ) from e + elif os.environ.get("VAULT_TOKEN"): + # Honour the ambient environment, e.g. a host running a Vault agent. + client.token = os.environ["VAULT_TOKEN"] + else: + raise ConfigurationError( + "No Vault credentials provided (set vault_token_file, " + "vault_token, an AppRole via vault_role_id, or VAULT_TOKEN " + "in the environment)" + ) + + +def fetch_passphrases( + keyids: List[str], + *, + addr: Optional[str] = None, + token: Optional[str] = None, + token_file: Optional[str] = None, + role_id: Optional[str] = None, + secret_id: Optional[str] = None, + secret_id_file: Optional[str] = None, + namespace: Optional[str] = None, + mount: str = DEFAULT_MOUNT, + path_prefix: str = "", + field: str = DEFAULT_FIELD, + ca_cert: Optional[str] = None, +) -> Dict[str, str]: + """Return a keyid -> passphrase mapping read from Vault.""" + try: + import hvac + from hvac import exceptions as hvac_exceptions + except ImportError as e: + raise ConfigurationError( + "hvac is not installed. Install it with: pip install hvac" + ) from e + + addr = addr or os.environ.get("VAULT_ADDR") + if not addr: + raise ConfigurationError( + "Vault address must be provided (set vault_addr or VAULT_ADDR)" + ) + + logger.info( + "Fetching GPG passphrases from Vault %s for %d keys", addr, len(keyids) + ) + client = hvac.Client( + url=addr, + namespace=namespace, + verify=ca_cert if ca_cert else True, + ) + _authenticate( + client, + hvac_exceptions, + token=token, + token_file=token_file, + role_id=role_id, + secret_id=secret_id, + secret_id_file=secret_id_file, + ) + + prefix = path_prefix.strip("/") + result: Dict[str, str] = {} + missing: List[str] = [] + for keyid in keyids: + path = f"{prefix}/{keyid}" if prefix else keyid + try: + response = client.secrets.kv.v2.read_secret_version( + path=path, + mount_point=mount, + raise_on_deleted_version=True, + ) + except hvac_exceptions.InvalidPath: + missing.append(keyid) + continue + except hvac_exceptions.VaultError as e: + # Sealed vault, bad token, denied policy: not a missing key, so + # fail loudly instead of degrading to an interactive prompt. + raise ConfigurationError( + f"Cannot read Vault secret {mount}/{path}: {e}" + ) from e + passphrase = (response.get("data", {}).get("data") or {}).get(field) + if not passphrase: + missing.append(keyid) + continue + result[keyid] = passphrase + + if missing: + raise ConfigurationError( + f"Vault is missing passphrases (field '{field}') for keys: " + + ", ".join(missing) + ) + + return result diff --git a/tests/sign_node/utils/test_secrets.py b/tests/sign_node/utils/test_secrets.py new file mode 100644 index 0000000..4a164c7 --- /dev/null +++ b/tests/sign_node/utils/test_secrets.py @@ -0,0 +1,106 @@ +from types import SimpleNamespace + +import pytest + +from sign_node.errors import ConfigurationError +from sign_node.utils import secrets + +KEY_A = "AAAA1111BBBB2222" + + +def make_config(**overrides): + config = SimpleNamespace( + pgp_keys=[KEY_A], + bitwarden_enabled=False, + bitwarden_username=None, + bitwarden_password=None, + bitwarden_password_file=None, + bitwarden_collection_id=None, + vault_enabled=False, + vault_addr=None, + vault_token=None, + vault_token_file=None, + vault_role_id=None, + vault_secret_id=None, + vault_secret_id_file=None, + vault_namespace=None, + vault_mount="secret", + vault_path_prefix="", + vault_passphrase_field="passphrase", + vault_ca_cert=None, + ) + for key, value in overrides.items(): + setattr(config, key, value) + return config + + +def test_no_provider_returns_none(): + assert secrets.resolve_passphrases(make_config()) is None + + +def test_both_providers_enabled_raises(): + config = make_config(bitwarden_enabled=True, vault_enabled=True) + with pytest.raises(ConfigurationError, match="Only one secret provider"): + secrets.resolve_passphrases(config) + + +def test_vault_provider_receives_config(monkeypatch): + captured = {} + + def fake_fetch(**kwargs): + captured.update(kwargs) + return {KEY_A: "from-vault"} + + monkeypatch.setattr(secrets.vault, "fetch_passphrases", fake_fetch) + config = make_config( + vault_enabled=True, + vault_addr="https://vault.example.com:8200", + vault_token_file="/run/secrets/vault_token", + vault_path_prefix="albs/sign-keys", + ) + + assert secrets.resolve_passphrases(config) == {KEY_A: "from-vault"} + assert captured["keyids"] == [KEY_A] + assert captured["addr"] == "https://vault.example.com:8200" + assert captured["token_file"] == "/run/secrets/vault_token" + assert captured["path_prefix"] == "albs/sign-keys" + assert captured["field"] == "passphrase" + + +def test_bitwarden_provider_receives_config(monkeypatch): + captured = {} + + def fake_fetch(**kwargs): + captured.update(kwargs) + return {KEY_A: "from-bw"} + + monkeypatch.setattr(secrets.bitwarden, "fetch_passphrases", fake_fetch) + config = make_config( + bitwarden_enabled=True, + bitwarden_username="signer@example.com", + bitwarden_password_file="/run/secrets/bw_master", + ) + + assert secrets.resolve_passphrases(config) == {KEY_A: "from-bw"} + assert captured["keyids"] == [KEY_A] + assert captured["username"] == "signer@example.com" + assert captured["password_file"] == "/run/secrets/bw_master" + + +def test_enabled_providers_lists_only_active(): + assert secrets.enabled_providers(make_config()) == [] + assert secrets.enabled_providers(make_config(vault_enabled=True)) == [ + "vault" + ] + assert secrets.enabled_providers( + make_config(bitwarden_enabled=True, vault_enabled=True) + ) == ["bitwarden", "vault"] + + +def test_provider_error_propagates(monkeypatch): + def boom(**kwargs): + raise ConfigurationError("vault is sealed") + + monkeypatch.setattr(secrets.vault, "fetch_passphrases", boom) + with pytest.raises(ConfigurationError, match="sealed"): + secrets.resolve_passphrases(make_config(vault_enabled=True)) diff --git a/tests/sign_node/utils/test_vault.py b/tests/sign_node/utils/test_vault.py new file mode 100644 index 0000000..5dfca89 --- /dev/null +++ b/tests/sign_node/utils/test_vault.py @@ -0,0 +1,243 @@ +import sys +import types +from unittest.mock import MagicMock + +import pytest + +from sign_node.errors import ConfigurationError + +KEY_A = "AAAA1111BBBB2222" +KEY_B = "CCCC3333DDDD4444" +ADDR = "https://vault.example.com:8200" + + +class InvalidPath(Exception): + pass + + +class VaultError(Exception): + pass + + +@pytest.fixture +def fake_hvac(monkeypatch): + """Install a stub ``hvac`` module with the bits the fetcher touches.""" + package = types.ModuleType("hvac") + exceptions = types.ModuleType("hvac.exceptions") + exceptions.InvalidPath = InvalidPath + exceptions.VaultError = VaultError + package.exceptions = exceptions + package.Client = MagicMock() + monkeypatch.setitem(sys.modules, "hvac", package) + monkeypatch.setitem(sys.modules, "hvac.exceptions", exceptions) + monkeypatch.delenv("VAULT_TOKEN", raising=False) + monkeypatch.delenv("VAULT_ADDR", raising=False) + yield package + + +def _import_fetcher(): + from sign_node.utils.vault import fetch_passphrases + + return fetch_passphrases + + +def _kv(fake_hvac): + return fake_hvac.Client.return_value.secrets.kv.v2 + + +def _secrets(mapping, field="passphrase"): + """Build a read_secret_version side effect from a path -> value map.""" + + def _read(path, mount_point, raise_on_deleted_version): + if path not in mapping: + raise InvalidPath(path) + return {"data": {"data": {field: mapping[path]}}} + + return _read + + +def test_fetch_passphrases_returns_keyid_map(fake_hvac): + _kv(fake_hvac).read_secret_version.side_effect = _secrets({ + f"albs/sign-keys/{KEY_A}": "secret-a", + f"albs/sign-keys/{KEY_B}": "secret-b", + }) + + fetch_passphrases = _import_fetcher() + result = fetch_passphrases( + keyids=[KEY_A, KEY_B], + addr=ADDR, + token="t", + path_prefix="albs/sign-keys", + ) + + assert result == {KEY_A: "secret-a", KEY_B: "secret-b"} + fake_hvac.Client.assert_called_once_with( + url=ADDR, namespace=None, verify=True + ) + + +def test_fetch_passphrases_without_path_prefix(fake_hvac): + _kv(fake_hvac).read_secret_version.side_effect = _secrets({KEY_A: "x"}) + + fetch_passphrases = _import_fetcher() + assert fetch_passphrases(keyids=[KEY_A], addr=ADDR, token="t") == { + KEY_A: "x" + } + + +def test_fetch_passphrases_custom_mount_and_field(fake_hvac): + _kv(fake_hvac).read_secret_version.side_effect = _secrets( + {KEY_A: "x"}, field="password" + ) + + fetch_passphrases = _import_fetcher() + result = fetch_passphrases( + keyids=[KEY_A], addr=ADDR, token="t", mount="kv", field="password" + ) + + assert result == {KEY_A: "x"} + _, kwargs = _kv(fake_hvac).read_secret_version.call_args + assert kwargs["mount_point"] == "kv" + + +def test_fetch_passphrases_missing_secret_raises(fake_hvac): + _kv(fake_hvac).read_secret_version.side_effect = _secrets({KEY_A: "a"}) + + fetch_passphrases = _import_fetcher() + with pytest.raises(ConfigurationError, match=KEY_B): + fetch_passphrases(keyids=[KEY_A, KEY_B], addr=ADDR, token="t") + + +def test_fetch_passphrases_wrong_field_treated_as_missing(fake_hvac): + _kv(fake_hvac).read_secret_version.side_effect = _secrets( + {KEY_A: "a"}, field="other" + ) + + fetch_passphrases = _import_fetcher() + with pytest.raises(ConfigurationError, match=KEY_A): + fetch_passphrases(keyids=[KEY_A], addr=ADDR, token="t") + + +def test_fetch_passphrases_sealed_vault_raises(fake_hvac): + _kv(fake_hvac).read_secret_version.side_effect = VaultError("sealed") + + fetch_passphrases = _import_fetcher() + with pytest.raises(ConfigurationError, match="Cannot read Vault secret"): + fetch_passphrases(keyids=[KEY_A], addr=ADDR, token="t") + + +def test_fetch_passphrases_reads_token_file(fake_hvac, tmp_path): + token_file = tmp_path / "token" + token_file.write_text("s.tokenvalue\n") + _kv(fake_hvac).read_secret_version.side_effect = _secrets({KEY_A: "x"}) + + fetch_passphrases = _import_fetcher() + fetch_passphrases(keyids=[KEY_A], addr=ADDR, token_file=str(token_file)) + + assert fake_hvac.Client.return_value.token == "s.tokenvalue" + + +def test_fetch_passphrases_empty_token_file_raises(fake_hvac, tmp_path): + token_file = tmp_path / "token" + token_file.write_text("\n") + + fetch_passphrases = _import_fetcher() + with pytest.raises(ConfigurationError, match="is empty"): + fetch_passphrases(keyids=[KEY_A], addr=ADDR, token_file=str(token_file)) + + +def test_fetch_passphrases_unreadable_token_file_raises(fake_hvac, tmp_path): + fetch_passphrases = _import_fetcher() + with pytest.raises(ConfigurationError, match="Cannot read Vault token"): + fetch_passphrases( + keyids=[KEY_A], addr=ADDR, token_file=str(tmp_path / "nope") + ) + + +def test_fetch_passphrases_approle_login(fake_hvac, tmp_path): + secret_id_file = tmp_path / "secret_id" + secret_id_file.write_text("sid-value\n") + _kv(fake_hvac).read_secret_version.side_effect = _secrets({KEY_A: "x"}) + + fetch_passphrases = _import_fetcher() + fetch_passphrases( + keyids=[KEY_A], + addr=ADDR, + role_id="rid", + secret_id_file=str(secret_id_file), + ) + + fake_hvac.Client.return_value.auth.approle.login.assert_called_once_with( + role_id="rid", secret_id="sid-value" + ) + + +def test_fetch_passphrases_approle_without_secret_id_raises(fake_hvac): + fetch_passphrases = _import_fetcher() + with pytest.raises(ConfigurationError, match="requires a secret id"): + fetch_passphrases(keyids=[KEY_A], addr=ADDR, role_id="rid") + + +def test_fetch_passphrases_approle_login_failure_raises(fake_hvac): + fake_hvac.Client.return_value.auth.approle.login.side_effect = VaultError( + "denied" + ) + + fetch_passphrases = _import_fetcher() + with pytest.raises(ConfigurationError, match="AppRole authentication"): + fetch_passphrases( + keyids=[KEY_A], addr=ADDR, role_id="rid", secret_id="sid" + ) + + +def test_fetch_passphrases_falls_back_to_env(fake_hvac, monkeypatch): + monkeypatch.setenv("VAULT_ADDR", ADDR) + monkeypatch.setenv("VAULT_TOKEN", "env-token") + _kv(fake_hvac).read_secret_version.side_effect = _secrets({KEY_A: "x"}) + + fetch_passphrases = _import_fetcher() + assert fetch_passphrases(keyids=[KEY_A]) == {KEY_A: "x"} + + assert fake_hvac.Client.return_value.token == "env-token" + fake_hvac.Client.assert_called_once_with( + url=ADDR, namespace=None, verify=True + ) + + +def test_fetch_passphrases_requires_addr(fake_hvac): + fetch_passphrases = _import_fetcher() + with pytest.raises(ConfigurationError, match="Vault address"): + fetch_passphrases(keyids=[KEY_A], token="t") + + +def test_fetch_passphrases_requires_credentials(fake_hvac): + fetch_passphrases = _import_fetcher() + with pytest.raises(ConfigurationError, match="No Vault credentials"): + fetch_passphrases(keyids=[KEY_A], addr=ADDR) + + +def test_fetch_passphrases_passes_namespace_and_ca_cert(fake_hvac): + _kv(fake_hvac).read_secret_version.side_effect = _secrets({KEY_A: "x"}) + + fetch_passphrases = _import_fetcher() + fetch_passphrases( + keyids=[KEY_A], + addr=ADDR, + token="t", + namespace="admin/albs", + ca_cert="/etc/pki/vault-ca.pem", + ) + + fake_hvac.Client.assert_called_once_with( + url=ADDR, namespace="admin/albs", verify="/etc/pki/vault-ca.pem" + ) + + +def test_fetch_passphrases_without_hvac_installed(monkeypatch): + monkeypatch.setitem(sys.modules, "hvac", None) + monkeypatch.setitem(sys.modules, "hvac.exceptions", None) + + from sign_node.utils.vault import fetch_passphrases + + with pytest.raises(ConfigurationError, match="hvac is not installed"): + fetch_passphrases(keyids=[KEY_A], addr=ADDR, token="t")