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")