From 3390fe551b927149186feee68aace034f087d9de Mon Sep 17 00:00:00 2001 From: Joseph Magly <1159087+jmagly@users.noreply.github.com> Date: Sat, 22 Aug 2026 10:02:49 -0400 Subject: [PATCH] feat(secrets): add provider-neutral credential hooks --- README.md | 47 ++++++ app.py | 22 ++- ci/test-risk-map.json | 25 ++++ obliteratus/abliterate.py | 3 +- obliteratus/auto_obliterate.py | 4 +- obliteratus/credential_sources.py | 180 +++++++++++++++++++++++ obliteratus/local_ui.py | 5 +- obliteratus/models/loader.py | 5 +- obliteratus/telemetry.py | 10 +- obliteratus/tourney.py | 3 +- tests/test_auto_obliterate.py | 21 ++- tests/test_secrets.py | 229 ++++++++++++++++++++++++++++++ 12 files changed, 530 insertions(+), 24 deletions(-) create mode 100644 obliteratus/credential_sources.py create mode 100644 tests/test_secrets.py diff --git a/README.md b/README.md index 84d20c5..f2e1f27 100644 --- a/README.md +++ b/README.md @@ -289,6 +289,53 @@ output_dir: results/my_run obliteratus run my_study.yaml ``` +## Credentials and optional vault integration + +Environment variables remain the default for quick local development and tests: + +```bash +export HF_TOKEN=hf_example +export HF_PUSH_TOKEN=hf_example +export OPENROUTER_API_KEY=sk-or-v1-example +``` + +OBLITERATUS also provides a provider-neutral secret hook for deployments backed by +OpenBao, HashiCorp Vault, Kubernetes, Docker, systemd credentials, or another secret +manager. No vault SDK is required. Each existing credential name resolves in this +order: + +1. explicit UI or Python argument, where supported; +2. its existing environment variable; +3. `_FILE`, such as `OPENROUTER_API_KEY_FILE`; +4. a file in `OBLITERATUS_SECRET_DIR`; +5. a systemd credential in `CREDENTIALS_DIRECTORY`; +6. the trusted executable configured by `OBLITERATUS_SECRET_COMMAND`. + +Mounted directories use normalized lowercase filenames. For example, +`OPENROUTER_API_KEY` maps to `openrouter-api-key`, and `HF_TOKEN` maps to +`hf-token`. Existing uppercase filenames are also accepted. A Vault/OpenBao Agent +can render those files into a private tmpfs mount: + +```bash +export OBLITERATUS_SECRET_DIR=/run/secrets/obliteratus +obliteratus ui +``` + +For brokers that fetch values on demand, configure one absolute executable path: + +```bash +export OBLITERATUS_SECRET_COMMAND=/usr/local/libexec/obliteratus-secret +obliteratus ui +``` + +The executable receives exactly one argument—the normalized environment-variable +name—and writes only the secret value to stdout. Exit `0` returns a value, exit `2` +means the credential is unavailable, and other statuses fail closed. It runs without +a shell and with a bounded timeout. Treat the executable as privileged deployment +configuration; never point it at contributor-controlled code or emit diagnostics on +stdout. Secrets are resolved at use time so mounted-file and broker rotations do not +require restarting the application. + ## Two intervention paradigms OBLITERATUS supports both permanent and reversible liberation: diff --git a/app.py b/app.py index 722ae06..2253906 100644 --- a/app.py +++ b/app.py @@ -58,6 +58,7 @@ if "HF_HOME" not in os.environ: import gradio as gr import torch from obliteratus import device as dev +from obliteratus.credential_sources import resolve_first, resolve_secret, secret_available from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer # ── ZeroGPU support ───────────────────────────────────────────────── @@ -347,7 +348,6 @@ METHODS = { # Shared org + token so users can auto-push without their own HF_TOKEN. # Set OBLITERATUS_HUB_TOKEN as a Space secret with write access to the org. _HUB_COMMUNITY_ORG = os.environ.get("OBLITERATUS_HUB_ORG", "OBLITERATUS") -_HUB_COMMUNITY_TOKEN = os.environ.get("OBLITERATUS_HUB_TOKEN") # Import preset configs for Advanced Settings defaults from obliteratus.abliterate import METHODS as _PRESET_CONFIGS # noqa: E402 @@ -466,7 +466,6 @@ def _on_dataset_change(dataset_label: str): def _validate_hub_repo(hub_repo: str) -> str: """Validate Hub repo ID format and check HF_TOKEN. Returns warning HTML or empty string.""" - import os import re repo = hub_repo.strip() if hub_repo else "" if not repo: @@ -477,7 +476,7 @@ def _validate_hub_repo(hub_repo: str) -> str: "Invalid repo format — use `username/model-name` " "(letters, numbers, hyphens, dots only)" ) - if not os.environ.get("HF_TOKEN") and not os.environ.get("HF_PUSH_TOKEN") and not _HUB_COMMUNITY_TOKEN: + if not secret_available("HF_TOKEN", "HF_PUSH_TOKEN", "OBLITERATUS_HUB_TOKEN"): warnings.append( "No Hub token available — push will fail. " "Set HF_PUSH_TOKEN, HF_TOKEN, or OBLITERATUS_HUB_TOKEN." @@ -591,7 +590,6 @@ def push_session_to_hub( progress=gr.Progress(), ): """Push a session model to HuggingFace Hub, with optional refinement.""" - import os import re if not session_label or session_label.startswith("("): @@ -622,7 +620,7 @@ def push_session_to_hub( # Resolve token token = hub_token_input.strip() if hub_token_input else None if not token: - token = os.environ.get("HF_PUSH_TOKEN") or os.environ.get("HF_TOKEN") or _HUB_COMMUNITY_TOKEN + token = resolve_first("HF_PUSH_TOKEN", "HF_TOKEN", "OBLITERATUS_HUB_TOKEN") if not token: yield ( "**Error:** No Hub token available. Enter a token above, " @@ -720,7 +718,7 @@ def _should_quantize(model_id: str, is_preset: bool = False) -> str | None: try: from obliteratus.models.loader import _estimate_model_memory_gb, _available_gpu_memory_gb from transformers import AutoConfig - token = os.environ.get("HF_TOKEN") or None + token = resolve_secret("HF_TOKEN") config = AutoConfig.from_pretrained(model_id, trust_remote_code=is_preset, token=token) # Skip if model already ships with native quantization (e.g. Mxfp4Config) if getattr(config, "quantization_config", None) is not None: @@ -1835,8 +1833,6 @@ def obliterate(model_choice: str, method_choice: str, 5 minutes). The @spaces.GPU decorator allocates a GPU at call time and releases it when the function returns. """ - import os - model_id = MODELS.get(model_choice, model_choice) is_preset = model_choice in MODELS method = METHODS.get(method_choice, "advanced") @@ -1874,7 +1870,7 @@ def obliterate(model_choice: str, method_choice: str, # Early validation: gated model access from obliteratus.presets import is_gated - if is_gated(model_id) and not (os.environ.get("HF_TOKEN") or os.environ.get("HF_PUSH_TOKEN")): + if is_gated(model_id) and not secret_available("HF_TOKEN", "HF_PUSH_TOKEN"): yield ( f"**Error: Gated model requires authentication.**\n\n" f"`{model_id}` is a gated HuggingFace repo. To use it:\n\n" @@ -2969,7 +2965,7 @@ def ab_chat_respond(message: str, history_left: list[dict], history_right: list[ model_id, torch_dtype=torch.float16, trust_remote_code=is_preset, low_cpu_mem_usage=True, - token=os.environ.get("HF_TOKEN") or None, + token=resolve_secret("HF_TOKEN"), ) streamer_orig = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True, timeout=stream_timeout) @@ -4991,8 +4987,10 @@ To opt out, set the environment variable `OBLITERATUS_TELEMETRY=0` before launch diag.append(f"- Telemetry enabled: `{is_enabled()}`") diag.append(f"- On HF Spaces: `{_ON_HF_SPACES}`") diag.append(f"- Repo: `{_TELEMETRY_REPO or '(not set)'}`") - diag.append(f"- HF_TOKEN set: `{bool(os.environ.get('HF_TOKEN'))}`") - diag.append(f"- HF_PUSH_TOKEN set: `{bool(os.environ.get('HF_PUSH_TOKEN'))}`") + diag.append(f"- HF token available: `{secret_available('HF_TOKEN')}`") + diag.append( + f"- HF push token available: `{secret_available('HF_PUSH_TOKEN')}`" + ) diag.append(f"- Local file: `{TELEMETRY_FILE}`") diag.append(f"- Local file exists: `{TELEMETRY_FILE.exists()}`") n_records = len(read_telemetry()) if TELEMETRY_FILE.exists() else 0 diff --git a/ci/test-risk-map.json b/ci/test-risk-map.json index de745b7..f1be700 100644 --- a/ci/test-risk-map.json +++ b/ci/test-risk-map.json @@ -139,6 +139,22 @@ "tests/conditional/test_jetson_runtime.py" ] }, + { + "id": "credential-resolution", + "owner": "runtime security maintainers", + "description": "Provider-neutral environment, mounted-file, systemd credential, and executable-broker secret resolution", + "contract_types": [ + "configuration", + "external-service", + "public-interface" + ], + "paths": [ + "obliteratus/credential_sources.py" + ], + "required_tests": [ + "tests/test_secrets.py" + ] + }, { "id": "public-interface-and-services", "owner": "operator interface maintainers", @@ -711,6 +727,15 @@ "network-services" ] }, + { + "path": "obliteratus/credential_sources.py", + "risk_class": "cpu-contract", + "risk": "credential source precedence, mounted-secret validation, and broker execution boundaries", + "required_tests": [ + "tests/test_secrets.py" + ], + "conditional_gates": [] + }, { "path": "obliteratus/service_contracts.py", "risk_class": "cpu-contract", diff --git a/obliteratus/abliterate.py b/obliteratus/abliterate.py index d77921f..e5a07e4 100644 --- a/obliteratus/abliterate.py +++ b/obliteratus/abliterate.py @@ -94,6 +94,7 @@ from obliteratus.runtime_contracts import ( # noqa: E402 is_quantized_parameter, norm_restoration_ratio, ) +from obliteratus.credential_sources import resolve_first from obliteratus.strategies.utils import ( # noqa: E402 get_attention_module, get_ffn_module, @@ -7201,7 +7202,7 @@ class AbliterationPipeline: if self.push_to_hub: from huggingface_hub import HfApi - _fallback_token = os.environ.get("HF_PUSH_TOKEN") or os.environ.get("HF_TOKEN") or None + _fallback_token = resolve_first("HF_PUSH_TOKEN", "HF_TOKEN") api = HfApi(token=self.hub_token) if self.hub_token else (HfApi(token=_fallback_token) if _fallback_token else HfApi()) # Resolve "auto" → {namespace}/{short_model}-OBLITERATED diff --git a/obliteratus/auto_obliterate.py b/obliteratus/auto_obliterate.py index 9522ecc..a105ec1 100644 --- a/obliteratus/auto_obliterate.py +++ b/obliteratus/auto_obliterate.py @@ -27,6 +27,8 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any, Generator +from obliteratus.credential_sources import resolve_secret + logger = logging.getLogger(__name__) # ── State persistence ───────────────────────────────────────────────── @@ -223,7 +225,7 @@ class AutoObliterator: Claude to judge whether each response is a refusal. Returns metrics dict. """ - api_key = os.environ.get("OPENROUTER_API_KEY") + api_key = resolve_secret("OPENROUTER_API_KEY") if not api_key: return {"method": "skipped", "reason": "no OPENROUTER_API_KEY"} diff --git a/obliteratus/credential_sources.py b/obliteratus/credential_sources.py new file mode 100644 index 0000000..b9c1f3a --- /dev/null +++ b/obliteratus/credential_sources.py @@ -0,0 +1,180 @@ +"""Provider-neutral runtime credential resolution. + +OBLITERATUS does not embed a client for any particular vault. Instead, each +credential keeps its existing environment-variable name and can be supplied by +an environment value, a mounted file, or a trusted executable broker. This +works with Vault/OpenBao Agent templates, Kubernetes and Docker secret mounts, +systemd credentials, and other secret managers without adding provider SDKs to +the application. +""" + +from __future__ import annotations + +import os +import re +import stat +import subprocess +from pathlib import Path + + +_SECRET_NAME_RE = re.compile(r"^[A-Z][A-Z0-9_]*$") +_MAX_SECRET_BYTES = 64 * 1024 +_DEFAULT_BROKER_TIMEOUT_SECONDS = 5.0 + + +class SecretResolutionError(RuntimeError): + """A configured secret source could not be resolved safely.""" + + +def _validate_name(name: str) -> str: + if not isinstance(name, str) or not _SECRET_NAME_RE.fullmatch(name): + raise ValueError("secret names must use uppercase environment-variable syntax") + return name + + +def _normalize_value(raw: str, *, source: str) -> str: + value = raw.rstrip("\r\n") + if not value: + raise SecretResolutionError(f"configured secret source is empty: {source}") + if "\x00" in value: + raise SecretResolutionError(f"configured secret source contains a NUL byte: {source}") + return value + + +def _read_secret_file(path: Path, *, source: str) -> str: + try: + resolved = path.expanduser().resolve(strict=True) + metadata = resolved.stat() + if not stat.S_ISREG(metadata.st_mode): + raise SecretResolutionError(f"configured secret source is not a file: {source}") + if metadata.st_size > _MAX_SECRET_BYTES: + raise SecretResolutionError(f"configured secret source exceeds 64 KiB: {source}") + raw = resolved.read_bytes() + if len(raw) > _MAX_SECRET_BYTES: + raise SecretResolutionError(f"configured secret source exceeds 64 KiB: {source}") + return _normalize_value(raw.decode("utf-8"), source=source) + except SecretResolutionError: + raise + except (OSError, UnicodeError) as exc: + raise SecretResolutionError(f"configured secret source is unreadable: {source}") from exc + + +def _directory_candidates(directory: str, name: str) -> tuple[Path, ...]: + root = Path(directory).expanduser() + normalized = name.lower().replace("_", "-") + return root / normalized, root / name + + +def _broker_timeout() -> float: + raw = os.environ.get("OBLITERATUS_SECRET_COMMAND_TIMEOUT", "").strip() + if not raw: + return _DEFAULT_BROKER_TIMEOUT_SECONDS + try: + timeout = float(raw) + except ValueError as exc: + raise SecretResolutionError( + "OBLITERATUS_SECRET_COMMAND_TIMEOUT must be a number", + ) from exc + if not 0.1 <= timeout <= 30.0: + raise SecretResolutionError( + "OBLITERATUS_SECRET_COMMAND_TIMEOUT must be between 0.1 and 30 seconds", + ) + return timeout + + +def _resolve_from_broker(name: str, command: str) -> str | None: + executable = Path(command).expanduser() + if not executable.is_absolute(): + raise SecretResolutionError("OBLITERATUS_SECRET_COMMAND must be an absolute path") + try: + metadata = executable.stat() + except OSError as exc: + raise SecretResolutionError("OBLITERATUS_SECRET_COMMAND is unavailable") from exc + if not stat.S_ISREG(metadata.st_mode) or not os.access(executable, os.X_OK): + raise SecretResolutionError( + "OBLITERATUS_SECRET_COMMAND must be an executable regular file", + ) + if metadata.st_mode & (stat.S_IWGRP | stat.S_IWOTH): + raise SecretResolutionError("OBLITERATUS_SECRET_COMMAND must not be group/world writable") + + try: + completed = subprocess.run( + [str(executable), name], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + check=False, + close_fds=True, + timeout=_broker_timeout(), + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise SecretResolutionError("secret broker execution failed") from exc + + # Exit 2 is the normalized "secret not found" result. Broker stderr is + # intentionally never included because provider errors may contain secrets. + if completed.returncode == 2: + return None + if completed.returncode != 0: + raise SecretResolutionError( + f"secret broker failed for {name} with exit status {completed.returncode}", + ) + if len(completed.stdout) > _MAX_SECRET_BYTES: + raise SecretResolutionError("secret broker output exceeds 64 KiB") + try: + raw = completed.stdout.decode("utf-8") + except UnicodeError as exc: + raise SecretResolutionError("secret broker output is not UTF-8") from exc + return _normalize_value(raw, source=f"broker:{name}") + + +def resolve_secret(name: str, *, explicit: str | None = None) -> str | None: + """Resolve one credential without coupling the app to a vault provider. + + Resolution order is explicit value, existing environment value, ``NAME_FILE``, + ``OBLITERATUS_SECRET_DIR``, systemd ``CREDENTIALS_DIRECTORY``, then + ``OBLITERATUS_SECRET_COMMAND``. A configured but broken source fails closed. + The broker is executed directly (never through a shell) and receives only the + normalized credential name as its sole argument. + """ + + name = _validate_name(name) + if explicit is not None and explicit.rstrip("\r\n"): + return _normalize_value(explicit, source="explicit value") + + environment_value = os.environ.get(name) + if environment_value: + return _normalize_value(environment_value, source=name) + + file_variable = f"{name}_FILE" + configured_file = os.environ.get(file_variable, "").strip() + if configured_file: + return _read_secret_file(Path(configured_file), source=file_variable) + + for directory_variable in ("OBLITERATUS_SECRET_DIR", "CREDENTIALS_DIRECTORY"): + directory = os.environ.get(directory_variable, "").strip() + if not directory: + continue + for candidate in _directory_candidates(directory, name): + if candidate.exists(): + return _read_secret_file(candidate, source=directory_variable) + + command = os.environ.get("OBLITERATUS_SECRET_COMMAND", "").strip() + if command: + return _resolve_from_broker(name, command) + return None + + +def resolve_first(*names: str) -> str | None: + """Return the first available credential in caller-defined priority order.""" + + for name in names: + value = resolve_secret(name) + if value is not None: + return value + return None + + +def secret_available(*names: str) -> bool: + """Return whether any requested credential resolves successfully.""" + + return resolve_first(*names) is not None diff --git a/obliteratus/local_ui.py b/obliteratus/local_ui.py index b7fec89..5f8e7e9 100644 --- a/obliteratus/local_ui.py +++ b/obliteratus/local_ui.py @@ -21,6 +21,8 @@ from rich.console import Console from rich.panel import Panel from rich.table import Table +from obliteratus.credential_sources import secret_available + console = Console() # ── ASCII banner ──────────────────────────────────────────────────────────── @@ -178,8 +180,7 @@ def _print_system_info(gpus: list[dict]) -> None: table.add_row("GPU", "[yellow]None detected — CPU mode[/yellow]") # HF Token - hf_token = os.environ.get("HF_TOKEN", "") - if hf_token: + if secret_available("HF_TOKEN"): table.add_row("HF Token", "[green]set[/green]") else: table.add_row("HF Token", "[dim]not set (gated models won't work)[/dim]") diff --git a/obliteratus/models/loader.py b/obliteratus/models/loader.py index c30c2c6..35afc9f 100644 --- a/obliteratus/models/loader.py +++ b/obliteratus/models/loader.py @@ -13,6 +13,7 @@ import sys as _sys import torch from obliteratus import device as dev from obliteratus.models import quant_dequant as qd +from obliteratus.credential_sources import resolve_secret from obliteratus.runtime_contracts import ( effective_model_memory_gb, quantized_model_fits_gpu, @@ -618,8 +619,8 @@ def _available_gpu_memory_gb() -> float: def _hf_token() -> str | None: - """Return the HF_TOKEN from environment, or None.""" - return os.environ.get("HF_TOKEN") or None + """Resolve the optional Hugging Face read token.""" + return resolve_secret("HF_TOKEN") def load_model( diff --git a/obliteratus/telemetry.py b/obliteratus/telemetry.py index 98d5a28..b4ecc50 100644 --- a/obliteratus/telemetry.py +++ b/obliteratus/telemetry.py @@ -43,6 +43,8 @@ from datetime import datetime, timezone from pathlib import Path, PureWindowsPath from typing import Any +from obliteratus.credential_sources import resolve_first, resolve_secret + logger = logging.getLogger(__name__) BENCHMARK_SCHEMA_VERSION = 1 @@ -410,7 +412,7 @@ def _ensure_hub_repo(repo_id: str) -> bool: return True try: from huggingface_hub import HfApi - api = HfApi(token=os.environ.get("HF_PUSH_TOKEN") or os.environ.get("HF_TOKEN")) + api = HfApi(token=resolve_first("HF_PUSH_TOKEN", "HF_TOKEN")) # First try create_repo (works if we own the namespace) try: api.create_repo( @@ -465,7 +467,7 @@ def _sync_to_hub_bg() -> None: from huggingface_hub import HfApi if not _ensure_hub_repo(repo): return - api = HfApi(token=os.environ.get("HF_PUSH_TOKEN") or os.environ.get("HF_TOKEN")) + api = HfApi(token=resolve_first("HF_PUSH_TOKEN", "HF_TOKEN")) slug = _instance_slug() api.upload_file( path_or_fileobj=str(TELEMETRY_FILE), @@ -548,7 +550,7 @@ def _fetch_via_hf_api(repo: str, max_records: int) -> list[dict[str, Any]]: """Fetch telemetry via huggingface_hub API.""" from huggingface_hub import HfApi, hf_hub_download - api = HfApi(token=os.environ.get("HF_TOKEN")) + api = HfApi(token=resolve_secret("HF_TOKEN")) try: all_files = api.list_repo_files(repo, repo_type="dataset") except Exception: @@ -1003,7 +1005,7 @@ def push_to_hub(repo_id: str | None = None) -> bool: if not _ensure_hub_repo(repo): return False - api = HfApi(token=os.environ.get("HF_PUSH_TOKEN") or os.environ.get("HF_TOKEN")) + api = HfApi(token=resolve_first("HF_PUSH_TOKEN", "HF_TOKEN")) slug = _instance_slug() api.upload_file( path_or_fileobj=str(TELEMETRY_FILE), diff --git a/obliteratus/tourney.py b/obliteratus/tourney.py index 1d8913e..65fc1b9 100644 --- a/obliteratus/tourney.py +++ b/obliteratus/tourney.py @@ -26,6 +26,7 @@ from datetime import datetime from pathlib import Path from typing import Any, Callable +from obliteratus.credential_sources import resolve_first from obliteratus.tourney_contracts import parse_checkpoint_document # --------------------------------------------------------------------------- @@ -1460,7 +1461,7 @@ class TourneyRunner: self.log(f"\nPushing winner to Hub: {repo_id}") - _token = os.environ.get("HF_PUSH_TOKEN") or os.environ.get("HF_TOKEN") or None + _token = resolve_first("HF_PUSH_TOKEN", "HF_TOKEN") api = HfApi(token=_token) if _token else HfApi() api.create_repo(repo_id, exist_ok=True) diff --git a/tests/test_auto_obliterate.py b/tests/test_auto_obliterate.py index d82b658..43fe0e9 100644 --- a/tests/test_auto_obliterate.py +++ b/tests/test_auto_obliterate.py @@ -234,7 +234,14 @@ def test_auto_loop_records_failures_and_completes_without_success( def test_prompt_expansion_and_benchmark_fallbacks(monkeypatch): - monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + for name in ( + "OPENROUTER_API_KEY", + "OPENROUTER_API_KEY_FILE", + "OBLITERATUS_SECRET_DIR", + "CREDENTIALS_DIRECTORY", + "OBLITERATUS_SECRET_COMMAND", + ): + monkeypatch.delenv(name, raising=False) assert AutoObliterator._quick_benchmark_claude("missing", "model") == { "method": "skipped", "reason": "no OPENROUTER_API_KEY", @@ -253,6 +260,18 @@ def test_prompt_expansion_and_benchmark_fallbacks(monkeypatch): } +def test_openrouter_benchmark_accepts_mounted_secret_file(monkeypatch, tmp_path): + token_file = tmp_path / "openrouter-api-key" + token_file.write_text("mounted-openrouter-key\n", encoding="utf-8") + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + monkeypatch.setenv("OPENROUTER_API_KEY_FILE", str(token_file)) + + assert AutoObliterator._quick_benchmark_claude("missing", "model") == { + "method": "skipped", + "reason": "output_dir not found", + } + + def test_reset_clears_persisted_state(tmp_path): auto = AutoObliterator("local/model", output_base=str(tmp_path / "run")) auto._result.iterations.append(IterationResult(1, "aggressive", 1)) diff --git a/tests/test_secrets.py b/tests/test_secrets.py new file mode 100644 index 0000000..642d7d0 --- /dev/null +++ b/tests/test_secrets.py @@ -0,0 +1,229 @@ +"""Provider-neutral runtime secret resolution contracts.""" + +from __future__ import annotations + +import os +import subprocess + +import pytest + +from obliteratus.credential_sources import ( + SecretResolutionError, + resolve_first, + resolve_secret, + secret_available, +) + + +pytestmark = pytest.mark.cpu + +_TEST_ENVIRONMENT = { + "HF_TOKEN", + "HF_TOKEN_FILE", + "HF_PUSH_TOKEN", + "HF_PUSH_TOKEN_FILE", + "OPENROUTER_API_KEY", + "OPENROUTER_API_KEY_FILE", + "OBLITERATUS_HUB_TOKEN", + "OBLITERATUS_HUB_TOKEN_FILE", + "OBLITERATUS_SECRET_DIR", + "OBLITERATUS_SECRET_COMMAND", + "OBLITERATUS_SECRET_COMMAND_TIMEOUT", + "CREDENTIALS_DIRECTORY", +} + + +@pytest.fixture(autouse=True) +def _clean_secret_environment(monkeypatch): + for name in _TEST_ENVIRONMENT: + monkeypatch.delenv(name, raising=False) + + +def test_environment_is_the_default_and_precedes_advanced_sources(monkeypatch, tmp_path): + missing_file = tmp_path / "not-used" + monkeypatch.setenv("HF_TOKEN", "environment-value") + monkeypatch.setenv("HF_TOKEN_FILE", str(missing_file)) + monkeypatch.setenv("OBLITERATUS_SECRET_COMMAND", "relative-command") + + assert resolve_secret("HF_TOKEN") == "environment-value" + + +def test_explicit_value_precedes_environment(monkeypatch): + monkeypatch.setenv("HF_TOKEN", "environment-value") + + assert resolve_secret("HF_TOKEN", explicit="explicit-value\n") == "explicit-value" + + +def test_per_secret_file_supports_vault_agent_and_docker_mounts(monkeypatch, tmp_path): + secret_file = tmp_path / "openrouter" + secret_file.write_text("mounted-value\n", encoding="utf-8") + monkeypatch.setenv("OPENROUTER_API_KEY_FILE", str(secret_file)) + + assert resolve_secret("OPENROUTER_API_KEY") == "mounted-value" + + +def test_configured_file_fails_closed_instead_of_falling_through(monkeypatch, tmp_path): + monkeypatch.setenv("HF_TOKEN_FILE", str(tmp_path / "missing")) + monkeypatch.setenv("OBLITERATUS_SECRET_COMMAND", "relative-command") + + with pytest.raises(SecretResolutionError, match="unreadable"): + resolve_secret("HF_TOKEN") + + +def test_configured_file_must_be_regular(monkeypatch, tmp_path): + monkeypatch.setenv("HF_TOKEN_FILE", str(tmp_path)) + + with pytest.raises(SecretResolutionError, match="not a file"): + resolve_secret("HF_TOKEN") + + +@pytest.mark.parametrize("directory_variable", ["OBLITERATUS_SECRET_DIR", "CREDENTIALS_DIRECTORY"]) +def test_normalized_mounted_secret_directories(monkeypatch, tmp_path, directory_variable): + secret_file = tmp_path / "openrouter-api-key" + secret_file.write_text("directory-value", encoding="utf-8") + monkeypatch.setenv(directory_variable, str(tmp_path)) + + assert resolve_secret("OPENROUTER_API_KEY") == "directory-value" + + +def test_uppercase_filename_is_supported_for_existing_secret_mounts(monkeypatch, tmp_path): + secret_file = tmp_path / "HF_TOKEN" + secret_file.write_text("uppercase-file", encoding="utf-8") + monkeypatch.setenv("OBLITERATUS_SECRET_DIR", str(tmp_path)) + + assert resolve_secret("HF_TOKEN") == "uppercase-file" + + +@pytest.mark.skipif(os.name == "nt", reason="executable broker fixture is POSIX-specific") +def test_executable_broker_receives_only_normalized_name(monkeypatch, tmp_path): + broker = tmp_path / "secret-broker" + broker.write_text( + "#!/bin/sh\n" + "[ \"$#\" -eq 1 ] || exit 9\n" + "[ \"$1\" = OPENROUTER_API_KEY ] || exit 2\n" + "printf 'broker-value\\n'\n", + encoding="utf-8", + ) + broker.chmod(0o700) + monkeypatch.setenv("OBLITERATUS_SECRET_COMMAND", str(broker)) + + assert resolve_secret("OPENROUTER_API_KEY") == "broker-value" + assert resolve_secret("HF_TOKEN") is None + + +def test_broker_requires_absolute_executable(monkeypatch): + monkeypatch.setenv("OBLITERATUS_SECRET_COMMAND", "vault read secret") + + with pytest.raises(SecretResolutionError, match="absolute path"): + resolve_secret("HF_TOKEN") + + +def test_broker_timeout_is_bounded(monkeypatch, tmp_path): + broker = tmp_path / "broker" + broker.write_text("placeholder", encoding="utf-8") + broker.chmod(0o700) + monkeypatch.setenv("OBLITERATUS_SECRET_COMMAND", str(broker)) + monkeypatch.setenv("OBLITERATUS_SECRET_COMMAND_TIMEOUT", "31") + + with pytest.raises(SecretResolutionError, match="between 0.1 and 30"): + resolve_secret("HF_TOKEN") + + +def test_broker_timeout_must_be_numeric(monkeypatch, tmp_path): + broker = tmp_path / "broker" + broker.write_text("placeholder", encoding="utf-8") + broker.chmod(0o700) + monkeypatch.setenv("OBLITERATUS_SECRET_COMMAND", str(broker)) + monkeypatch.setenv("OBLITERATUS_SECRET_COMMAND_TIMEOUT", "fast") + + with pytest.raises(SecretResolutionError, match="must be a number"): + resolve_secret("HF_TOKEN") + + +def test_missing_and_non_executable_brokers_fail_closed(monkeypatch, tmp_path): + missing = tmp_path / "missing" + monkeypatch.setenv("OBLITERATUS_SECRET_COMMAND", str(missing)) + with pytest.raises(SecretResolutionError, match="unavailable"): + resolve_secret("HF_TOKEN") + + broker = tmp_path / "broker" + broker.write_text("placeholder", encoding="utf-8") + broker.chmod(0o600) + monkeypatch.setenv("OBLITERATUS_SECRET_COMMAND", str(broker)) + with pytest.raises(SecretResolutionError, match="executable regular file"): + resolve_secret("HF_TOKEN") + + broker.chmod(0o722) + with pytest.raises(SecretResolutionError, match="group/world writable"): + resolve_secret("HF_TOKEN") + + +@pytest.mark.parametrize( + ("completed", "message"), + [ + (subprocess.CompletedProcess([], 9, stdout=b"", stderr=b"private"), "exit status 9"), + (subprocess.CompletedProcess([], 0, stdout=b"x" * (64 * 1024 + 1)), "64 KiB"), + (subprocess.CompletedProcess([], 0, stdout=b"\xff"), "not UTF-8"), + ], +) +def test_broker_errors_never_surface_stderr(monkeypatch, tmp_path, completed, message): + broker = tmp_path / "broker" + broker.write_text("placeholder", encoding="utf-8") + broker.chmod(0o700) + monkeypatch.setenv("OBLITERATUS_SECRET_COMMAND", str(broker)) + monkeypatch.setattr( + "obliteratus.credential_sources.subprocess.run", + lambda *_a, **_k: completed, + ) + + with pytest.raises(SecretResolutionError, match=message) as failure: + resolve_secret("HF_TOKEN") + assert "private" not in str(failure.value) + + +def test_broker_execution_failure_is_normalized(monkeypatch, tmp_path): + broker = tmp_path / "broker" + broker.write_text("placeholder", encoding="utf-8") + broker.chmod(0o700) + monkeypatch.setenv("OBLITERATUS_SECRET_COMMAND", str(broker)) + monkeypatch.setattr( + "obliteratus.credential_sources.subprocess.run", + lambda *_a, **_k: (_ for _ in ()).throw(subprocess.TimeoutExpired("broker", 1)), + ) + + with pytest.raises(SecretResolutionError, match="execution failed"): + resolve_secret("HF_TOKEN") + + +@pytest.mark.parametrize("payload", ["", "value\x00suffix"]) +def test_invalid_file_payloads_are_rejected(monkeypatch, tmp_path, payload): + secret_file = tmp_path / "invalid" + secret_file.write_bytes(payload.encode("utf-8")) + monkeypatch.setenv("HF_TOKEN_FILE", str(secret_file)) + + with pytest.raises(SecretResolutionError, match="empty|NUL"): + resolve_secret("HF_TOKEN") + + +def test_oversized_secret_file_is_rejected(monkeypatch, tmp_path): + secret_file = tmp_path / "oversized" + secret_file.write_bytes(b"x" * (64 * 1024 + 1)) + monkeypatch.setenv("HF_TOKEN_FILE", str(secret_file)) + + with pytest.raises(SecretResolutionError, match="64 KiB"): + resolve_secret("HF_TOKEN") + + +def test_resolve_first_preserves_caller_priority_and_availability(monkeypatch): + monkeypatch.setenv("HF_TOKEN", "read-token") + monkeypatch.setenv("HF_PUSH_TOKEN", "push-token") + + assert resolve_first("HF_PUSH_TOKEN", "HF_TOKEN") == "push-token" + assert secret_available("OPENROUTER_API_KEY", "HF_TOKEN") is True + assert secret_available("OPENROUTER_API_KEY") is False + + +@pytest.mark.parametrize("name", ["hf_token", "HF-TOKEN", "", "1TOKEN"]) +def test_secret_names_are_normalized_and_not_shell_fragments(name): + with pytest.raises(ValueError, match="uppercase environment-variable"): + resolve_secret(name)