ops-warden/scripts/check_policy_caller_identity.py

#!/usr/bin/env python3
"""Readiness gate for the flex-auth ``policy.enabled`` flip (WARDEN-WP-0031).

flex-auth deployed ``flex-auth-ops-warden`` (FLEX-WP-0016) in ``callerAuth.mode:
warn``: it authenticates the caller with a Kubernetes TokenReview and binds
``resource.system: ops-warden`` to ``system:serviceaccount:ops-warden:ops-warden``,
but a caller that sends no ``Authorization`` header only produces a
``caller authentication warning`` and is still served. That pin cannot move to
``enforce`` — and therefore ``policy.enabled: true`` cannot be set — until
ops-warden's calling side actually presents a token.

This script asserts the calling side *without* flipping anything:

  * warden.yaml loads and ``policy.caller_auth.mode`` is not ``none``,
  * a caller token can actually be obtained (file / env / command),
  * (optional, ``--url``) a live ``/v1/check`` against the warn pin returns a
    decision **and** the response is reached with the header attached.

Exit 0 = ready to ask flex-auth to enforce, 1 = not ready, 2 = bad input.
The token is never printed, logged, or written anywhere — only its length and a
truncated SHA-256 fingerprint, which are safe to paste into a handoff message.

Usage:
  python scripts/check_policy_caller_identity.py [--config ~/.config/warden/warden.yaml]
  python scripts/check_policy_caller_identity.py --url http://127.0.0.1:19090
"""
from __future__ import annotations

import argparse
import hashlib
import sys
from pathlib import Path
from typing import List, Optional, Tuple

_SRC = Path(__file__).resolve().parent.parent / "src"
if _SRC.is_dir() and str(_SRC) not in sys.path:
    sys.path.insert(0, str(_SRC))

from warden.caller_identity import (  # noqa: E402
    CallerIdentityError,
    resolve_caller_token,
)
from warden.config import ConfigError, load_config  # noqa: E402

Check = Tuple[str, str, str]


def _fingerprint(token: str) -> str:
    return "sha256:" + hashlib.sha256(token.encode()).hexdigest()[:12]


def run_checks(config_path: Optional[Path], url: Optional[str]) -> List[Check]:
    checks: List[Check] = []
    try:
        cfg = load_config(config_path)
    except ConfigError as e:
        return [("fail", "warden.yaml", str(e))]

    policy = cfg.policy
    checks.append(
        ("ok", "warden.yaml", f"loaded; policy.enabled={str(policy.enabled).lower()}")
    )

    mode = policy.caller_auth.mode
    if mode == "none":
        checks.append(
            (
                "fail",
                "caller_auth.mode",
                "none — no Authorization header is sent; the flex-auth pin stays in warn",
            )
        )
        return checks
    checks.append(("ok", "caller_auth.mode", mode))

    try:
        token = resolve_caller_token(policy.caller_auth)
    except CallerIdentityError as e:
        checks.append(("fail", "caller token", str(e)))
        return checks
    assert token is not None
    checks.append(
        ("ok", "caller token", f"obtained, {len(token)} chars, {_fingerprint(token)}")
    )

    target = url or policy.flex_auth_url
    if url is None and not policy.enabled:
        checks.append(
            (
                "skip",
                "live /v1/check",
                f"policy.enabled=false; pass --url to smoke {target} anyway",
            )
        )
        return checks

    import httpx  # local import: the offline checks above must not need it

    probe = {
        "subject": {
            "id": "agt-state-hub-bridge",
            "type": "agt",
            "tenant": policy.tenant,
        },
        "action": "sign",
        "resource": {
            "id": "ssh-cert:actor/agt-state-hub-bridge",
            "type": "ssh-certificate",
            "system": policy.system,
            "tenant": policy.tenant,
        },
        "context": {
            # A structurally complete context, so a deny means the policy said
            # no — not that the probe was malformed. What is under test here is
            # the caller identity, and that is answered by the HTTP status.
            "actor_name": "agt-state-hub-bridge",
            "actor_type": "agt",
            "principals": ["agt-task-bridge"],
            "ttl_hours": 24,
            "pubkey_fingerprint": "sha256:" + "0" * 64,
            "readiness_probe": True,
        },
    }
    try:
        response = httpx.post(
            target.rstrip("/") + "/v1/check",
            json=probe,
            headers={"Authorization": f"Bearer {token}"},
            timeout=10.0,
        )
    except httpx.RequestError as e:
        checks.append(("fail", "live /v1/check", f"unreachable at {target}: {e}"))
        return checks

    if response.status_code == 401:
        checks.append(
            (
                "fail",
                "live /v1/check",
                "401 — the token was sent but flex-auth did not accept it "
                "(check the TokenReview audience and the ServiceAccount binding)",
            )
        )
    elif response.status_code == 403:
        checks.append(
            (
                "fail",
                "live /v1/check",
                f"403 — authenticated, but the principal may not represent "
                f"system {policy.system!r}",
            )
        )
    elif response.status_code >= 400:
        checks.append(
            ("fail", "live /v1/check", f"HTTP {response.status_code} from {target}")
        )
    else:
        try:
            decision = response.json()
        except ValueError:
            checks.append(("fail", "live /v1/check", "non-JSON decision"))
            return checks
        effect = str(decision.get("effect", "?"))
        decision_id = decision.get("id") or decision.get("request_id") or "?"
        checks.append(
            ("ok", "live /v1/check", f"HTTP 200, effect={effect}, decision={decision_id}")
        )
    return checks


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", type=Path, default=None, help="path to warden.yaml")
    parser.add_argument(
        "--url",
        default=None,
        help="flex-auth base URL to smoke (e.g. a port-forward of the warn pin)",
    )
    args = parser.parse_args()

    checks = run_checks(args.config, args.url)
    glyph = {"ok": "✓", "fail": "✗", "skip": "·"}
    print("flex-auth caller-identity readiness\n")
    for status, label, detail in checks:
        print(f"  {glyph[status]} {label}: {detail}")

    failed = [c for c in checks if c[0] == "fail"]
    if failed:
        print(
            f"\nNOT READY — {len(failed)} check(s) failed. "
            "Do not ask flex-auth to enforce, and do not set policy.enabled: true."
        )
        return 1
    print(
        "\nREADY — the calling side presents an identity. Next: tell flex-auth to set "
        "callerAuth.mode: enforce on flex-auth-ops-warden, re-run this check, then set "
        "policy.enabled: true with fail_closed: true."
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())