secrets-engine/tools/verify_service_jwt_boundary.py

"""Prepare or attend CCR-2026-0038 self-policy/capability acceptance.

Default is backend-free. --execute invokes existing attended in-memory custody;
never print backend response bodies, secret values, tokens or full accessors.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import re
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener

import yaml

from secrets_engine.config import Config
from secrets_engine.engine_auth import keycape_config, load_jwt_login_contract, login_service_jwt
from secrets_engine.errors import BackendError

ADMIN_DENIES = ("sys/policies/acl/*", "auth/*/role/*", "auth/token/create")
EXPECTED_GUARDS = (
    "platform/data/workloads/keycape/secrets-engine-openbao",
    "platform/metadata/workloads/keycape/secrets-engine-openbao",
)


def prepare(cfg, ccr_path):
    raw = ccr_path.read_bytes()
    ccr = yaml.safe_load(raw)
    if not isinstance(ccr, dict) or ccr.get("id") != "CCR-2026-0038" or ccr.get("status") != "applied":
        raise BackendError("expected applied CCR-2026-0038")
    bao = ccr.get("openbao", {})
    auth = bao.get("auth", {})
    contract = load_jwt_login_contract(cfg)
    kcfg = keycape_config(cfg)
    if (contract.mount, contract.role, contract.bound_issuer) != (
        auth.get("mount"), auth.get("role"), kcfg.issuer
    ):
        raise BackendError("login contract and issuer must match applied CCR")
    if bao.get("guarded_paths") != list(EXPECTED_GUARDS) or bao.get("policy_name") != "secrets-engine-login-self":
        raise BackendError("unexpected CCR guarded paths or self policy")
    return {
        "ccr_sha256": hashlib.sha256(raw).hexdigest(),
        "contract_sha256": hashlib.sha256(contract.path.read_bytes()).hexdigest(),
        "denied_paths": list(EXPECTED_GUARDS + ADMIN_DENIES),
        "expected_policy": "secrets-engine-login-self",
    }


MAX_RESPONSE_BYTES = 65536
SAFE_EVIDENCE = {
    "secret_source": str,
    "operator_session_handle": str,
    "session_handle": str,
    "established": bool,
    "operator_revocation_attempted": bool,
    "operator_revocation_succeeded": bool,
    "revocation_attempted": bool,
    "revocation_succeeded": bool,
}


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise BackendError("redirect refused")


def origin(url):
    parsed = urlsplit(url)
    if parsed.scheme not in ("http", "https") or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in ("", "/"):
        raise BackendError("invalid configured OpenBao origin")
    return (parsed.scheme, parsed.hostname, parsed.port or (443 if parsed.scheme == "https" else 80))


def request_json(client, path, payload=None, *, expected_addr=None):
    # expected_addr is captured from Config.load(), never inferred from a response.
    try:
        if expected_addr is None or origin(client.addr) != origin(expected_addr):
            raise BackendError("unexpected OpenBao origin")
        if path not in ("auth/token/lookup-self", "sys/capabilities-self"):
            raise BackendError("unexpected metadata endpoint")
        data = None if payload is None else json.dumps(payload).encode()
        req = Request(expected_addr.rstrip("/") + "/v1/" + path, data=data,
                      headers={"X-Vault-Token": client.token, "Content-Type": "application/json"})
        # Disable environment proxies and all redirect following before supplying token.
        opener = build_opener(ProxyHandler({}), NoRedirect())
        with opener.open(req, timeout=30) as response:
            if response.status != 200 or response.headers.get_content_type() != "application/json":
                raise BackendError("unexpected metadata response")
            length = response.headers.get("Content-Length")
            if length is not None and (not length.isdecimal() or int(length) > MAX_RESPONSE_BYTES):
                raise BackendError("oversize metadata response")
            raw = response.read(MAX_RESPONSE_BYTES + 1)
            if len(raw) > MAX_RESPONSE_BYTES:
                raise BackendError("oversize metadata response")
            result = json.loads(raw)
        if not isinstance(result, dict) or not isinstance(result.get("data"), dict) or "errors" in result:
            raise BackendError("malformed metadata response")
        return result["data"]
    except Exception:
        raise BackendError("boundary metadata request failed; response suppressed") from None


def safe_evidence(session):
    result = {}
    try:
        evidence = session.evidence()
        if not isinstance(evidence, dict):
            return result
        for key, kind in SAFE_EVIDENCE.items():
            value = evidence.get(key)
            if type(value) is not kind:
                continue
            if kind is str:
                if key == "secret_source" and value != "openbao-oidc-in-memory":
                    continue
                if key.endswith("handle") and not re.fullmatch(r"[0-9a-f]{12}", value):
                    continue
            result[key] = value
    except Exception:
        pass
    return result


def verify_session(session, plan, request=request_json):
    """Two metadata requests, then unconditional cleanup, within 8-use budget."""
    result = {**plan, "auth_provider": "service-jwt", "native_verified": False}
    cleanup_returned = False
    try:
        lookup = request(session.client, "auth/token/lookup-self")
        # Effective identity policies matter even when default policy is absent.
        policies = lookup.get("policies")
        identities = lookup.get("identity_policies", [])
        result["exact_self_policy"] = policies == [plan["expected_policy"]] and identities == []
        caps = request(session.client, "sys/capabilities-self", {"paths": plan["denied_paths"]})
        result["capability_denials"] = {path: caps.get(path) == ["deny"] for path in plan["denied_paths"]}
    except Exception:
        result["failure"] = "metadata verification failed; response suppressed"
    finally:
        try:
            session.close()
            cleanup_returned = True
        except Exception:
            result["failure"] = "service revocation failed"
        result.update(safe_evidence(session))
    evidence_complete = (
        result.get("secret_source") == "openbao-oidc-in-memory"
        and result.get("established") is True
        and result.get("operator_revocation_attempted") is True
        and result.get("operator_revocation_succeeded") is True
        and result.get("revocation_attempted") is True
        and result.get("revocation_succeeded") is True
        and all(isinstance(result.get(key), str) and re.fullmatch(r"[0-9a-f]{12}", result[key])
                for key in ("operator_session_handle", "session_handle"))
    )
    result["native_verified"] = bool(
        cleanup_returned and "failure" not in result and evidence_complete
        and result.get("exact_self_policy") and all(result.get("capability_denials", {}).values())
        and len(result.get("capability_denials", {})) == len(plan["denied_paths"])
        and result.get("operator_revocation_succeeded") and result.get("revocation_succeeded")
    )
    # These are deliberately separate from boundary acceptance.
    result["claim_negatives_proved"] = False
    result["protected_operation_adoption_proved"] = False
    return result


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--ccr", type=Path, required=True)
    parser.add_argument("--execute", action="store_true", help="attended live self metadata check; requires scoped authorization")
    args = parser.parse_args()
    try:
        cfg = Config.load()
        plan = prepare(cfg, args.ccr)
        result = verify_session(
            login_service_jwt(cfg), plan,
            lambda client, path, payload=None: request_json(client, path, payload, expected_addr=cfg.bao_addr),
        ) if args.execute else {
            **plan, "mode": "backend-free-preparation", "native_verified": False,
            "claim_negatives_proved": False, "protected_operation_adoption_proved": False,
        }
    except Exception:
        print(json.dumps({"native_verified": False, "failure": "preflight/login failed; details suppressed"}))
        return 1
    print(json.dumps(result, sort_keys=True))
    return 0 if not args.execute or result["native_verified"] else 1


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