"""Verification: prove access (or denial) without printing the value. Positive: the approved consumer (via its approle-scoped token) CAN read the lane. Negative: an unrelated/unscoped token CANNOT read the lane. Each check returns a boolean + a non-secret evidence dict. The secret value is never read into the result. """ from __future__ import annotations from dataclasses import dataclass from typing import Any from secrets_engine.catalog import CatalogEntry from secrets_engine.errors import VerificationError from secrets_engine.openbao import OpenBaoClient @dataclass class VerifyResult: check: str # "positive" | "negative" passed: bool detail: dict[str, Any] def render(self) -> str: status = "PASS" if self.passed else "FAIL" return f" {self.check} check: {status} ({self.detail.get('reason', '')})" def verify_positive(client: OpenBaoClient, entry: CatalogEntry, field: str) -> VerifyResult: """Approved consumer token must be able to read the field.""" try: token = client.approle_login_token(entry.role_name) except Exception as e: # backend errors -> failed verification, not a value leak return VerifyResult( "positive", False, {"reason": f"could not obtain approle token: {e}", "path": entry.path}, ) present = client.kv_field_present(entry.mount, entry.path, field, token=token) return VerifyResult( "positive", present, { "reason": "approved consumer can read lane field" if present else "approved consumer could NOT read field", "path": entry.path, "field": field, "role": entry.role_name, }, ) def verify_negative(client: OpenBaoClient, entry: CatalogEntry) -> VerifyResult: """An unrelated token must be denied. Uses an empty (invalid) token.""" # An empty/garbage token stands in for an unrelated consumer. denied = not client.kv_can_read(entry.mount, entry.path, token="se-unrelated-denied") return VerifyResult( "negative", denied, { "reason": "unrelated token denied read" if denied else "unrelated token was ABLE to read (LEAK RISK)", "path": entry.path, }, ) def verify_auth_capability_positive(client: OpenBaoClient, entry: CatalogEntry) -> VerifyResult: """Approved AppRole token must carry update on every exact allowed path.""" try: token = client.approle_login_token(entry.role_name) except Exception as e: return VerifyResult( "positive", False, {"reason": f"could not obtain approle token: {e}", "role": entry.role_name}, ) missing: list[str] = [] for path in entry.auth_allowed_paths: try: caps = client.token_capabilities(path, token=token) except Exception as e: return VerifyResult( "positive", False, {"reason": f"could not inspect capabilities: {e}", "path": path}, ) if "update" not in caps: missing.append(path) passed = not missing return VerifyResult( "positive", passed, { "reason": "approle token can update every allowlisted path" if passed else "approle token lacks update on allowlisted paths", "role": entry.role_name, "allowed_paths": sorted(entry.auth_allowed_paths), "missing_update": missing, }, ) def verify_auth_capability_negative(client: OpenBaoClient, entry: CatalogEntry) -> VerifyResult: """Approved AppRole token must not gain update outside denial probes.""" try: token = client.approle_login_token(entry.role_name) except Exception as e: return VerifyResult( "negative", False, {"reason": f"could not obtain approle token: {e}", "role": entry.role_name}, ) leaks: list[str] = [] for path in entry.auth_denied_probe_paths: try: caps = client.token_capabilities(path, token=token) except Exception as e: return VerifyResult( "negative", False, {"reason": f"could not inspect capabilities: {e}", "path": path}, ) if "update" in caps or "sudo" in caps or "root" in caps: leaks.append(path) passed = not leaks return VerifyResult( "negative", passed, { "reason": "denial probes lack update capability" if passed else "approle token can update outside the allowlist", "role": entry.role_name, "denied_probe_paths": entry.auth_denied_probe_paths, "leaks": leaks, }, ) def run_verification( client: OpenBaoClient, entry: CatalogEntry, field: str, *, positive: bool, negative: bool ) -> list[VerifyResult]: results: list[VerifyResult] = [] if entry.kind == "auth-capability": if positive: results.append(verify_auth_capability_positive(client, entry)) if negative: results.append(verify_auth_capability_negative(client, entry)) else: if positive: results.append(verify_positive(client, entry, field)) if negative: results.append(verify_negative(client, entry)) if not results: raise VerificationError("no verification check selected (use --positive/--negative)") return results