secrets-engine/src/secrets_engine/verify.py

"""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 _session_evidence(session: object | None) -> dict[str, Any]:
    """Extract only the session's explicit non-secret lifecycle record."""
    if session is None:
        return {"established": False}
    evidence = getattr(session, "evidence", None)
    if callable(evidence):
        return evidence()
    return {"established": True}


def verify_positive(client: OpenBaoClient, entry: CatalogEntry, field: str) -> VerifyResult:
    """Approved consumer token must be able to read the field."""
    if entry.delivery_auth_method != "approle" or not entry.has_delivery_auth:
        return VerifyResult(
            "positive",
            False,
            {"reason": "lane has no AppRole delivery auth", "path": entry.path},
        )
    session = None
    try:
        with client.approle_session(entry.role_name) as session:
            present = session.client.kv_field_present(entry.mount, entry.path, field)
    except Exception as e:  # backend errors -> failed verification, not a value leak
        return VerifyResult(
            "positive",
            False,
            {
                "reason": f"scoped verification session failed: {e}",
                "path": entry.path,
                "session": _session_evidence(session),
            },
        )
    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,
            "session": _session_evidence(session),
        },
    )


def verify_negative(
    client: OpenBaoClient,
    entry: CatalogEntry,
    *,
    unrelated_token: str | None,
) -> VerifyResult:
    """A real unrelated token must be denied the cataloged path."""
    if not unrelated_token:
        return VerifyResult(
            "negative",
            False,
            {
                "reason": "no real unrelated token supplied; denial not proven",
                "path": entry.path,
            },
        )
    denied = not client.kv_can_read(
        entry.mount, entry.path, token=unrelated_token
    )
    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."""
    missing: list[str] = []
    session = None
    try:
        with client.approle_session(entry.role_name) as session:
            for path in entry.auth_allowed_paths:
                caps = session.client.token_capabilities(
                    path, token=session.client.token
                )
                if "update" not in caps:
                    missing.append(path)
    except Exception as e:
        return VerifyResult(
            "positive",
            False,
            {
                "reason": f"scoped verification session failed: {e}",
                "role": entry.role_name,
                "session": _session_evidence(session),
            },
        )
    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,
            "session": _session_evidence(session),
        },
    )


def verify_auth_capability_negative(client: OpenBaoClient, entry: CatalogEntry) -> VerifyResult:
    """Approved AppRole token must not gain update outside denial probes."""
    leaks: list[str] = []
    session = None
    try:
        with client.approle_session(entry.role_name) as session:
            for path in entry.auth_denied_probe_paths:
                caps = session.client.token_capabilities(
                    path, token=session.client.token
                )
                if "update" in caps or "sudo" in caps or "root" in caps:
                    leaks.append(path)
    except Exception as e:
        return VerifyResult(
            "negative",
            False,
            {
                "reason": f"scoped verification session failed: {e}",
                "role": entry.role_name,
                "session": _session_evidence(session),
            },
        )
    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,
            "session": _session_evidence(session),
        },
    )


def run_verification(
    client: OpenBaoClient,
    entry: CatalogEntry,
    field: str,
    *,
    positive: bool,
    negative: bool,
    unrelated_token: str | None = None,
) -> list[VerifyResult]:
    if entry.kind == "kv" and field not in entry.fields:
        raise VerificationError(
            f"field '{field}' not declared in lane '{entry.id}' fields {entry.fields}"
        )
    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, unrelated_token=unrelated_token)
            )
    if not results:
        raise VerificationError("no verification check selected (use --positive/--negative)")
    return results