#!/usr/bin/env python3
"""Executor audit sender against audit-core's real ingestion (NK-WP-0056-T15).

audit-core's source (IngestionApplication, SQLite backend) is served on loopback with its sender registry built the
way production builds it: a Secret-shaped AUDIT_CORE_SENDERS entry plus the Git-declared scope overlay
deploy/senders-scope.json. The executor side is real: executor.Run.execute() on in-memory operations, emitting
through clients.AuditCore over HTTP with a token file.

Checks:
  - a completed run and a reverted run: started and receipt accepted (202), and both pairs pass the receipt validator
    (tools/execution-attribution-receipt, schema v0.1)
  - a retry of the same event is a duplicate (200); the same id with another body is a conflict (409), which the
    client reports as a run-id-reuse defect
  - wrong source, wrong tenant (`tenant:platform`), no token and a wrong token are refused; nothing is stored for them
  - a started event whose admission window has passed, with no receipt, becomes an opening_without_closing finding
    held to that window; the closed runs do not
Synthetic tokens only. Writes a value-free receipt. Exit 0 when every check passes.

  python3 bounded-executor/exercise_audit_sender.py --receipt FILE [--audit-core-source ~/audit-core]
"""
from __future__ import annotations

import argparse
import copy
import datetime as dt
import hashlib
import importlib.util
import io
import json
import os
import pathlib
import secrets
import subprocess
import sys
import tempfile
import threading
import urllib.error
import urllib.request
import uuid
from wsgiref.simple_server import WSGIRequestHandler, make_server

HERE = pathlib.Path(__file__).resolve().parent
REPO = HERE.parent
TOOL = REPO / "tools/execution-attribution-receipt"


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


G = _load("ear_generate", TOOL / "generate_fixtures.py")  # loads executor.py as `executor`
V = _load("ear_validate", TOOL / "validate.py")
X = G.X
C = _load("clients", HERE / "clients.py")


def commit(path: pathlib.Path) -> str:
    return subprocess.run(["git", "-C", str(path), "rev-parse", "HEAD"], capture_output=True, text=True,
                          check=True).stdout.strip()


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


def run_executor(name: str, audit, fail_checks=()) -> list[dict]:
    """One real executor run on the example admission with a window around now; returns the events it emitted."""
    doc = G.X.bea.load(REPO / "examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml")
    now = dt.datetime.now(dt.UTC).replace(microsecond=0)
    doc["spec"]["id"] = name
    doc["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")}
    doc["spec"]["budget"]["attempts"] = 1  # a failed check then ends the run as reverted, not retried
    doc["spec"]["plan_digest"] = X.bea.plan_digest(doc)
    fake = G.Fake(doc, fail_checks)
    seen: list[dict] = []

    class Tee:
        def emit(self, event):
            audit.emit(event)  # raises unless audit-core stored it
            seen.append(event)

    plan = doc["spec"]["plan"]
    deps = X.Deps(carrier=fake, operations={s["operation"]: fake.op for s in plan["steps"] + plan["revert_steps"]},
                  checks={c["id"]: fake.check for c in doc["spec"]["checks"]}, trust=lambda _s: True, audit=Tee(),
                  alert=lambda *_a: None, sleep=lambda _s: None)
    cfg = X.Config(executor_identity=doc["spec"]["executor"]["identity"],
                   runtime_principal="system:serviceaccount:sso:nk-bind-executor", tenant="platform",
                   environment="TEST", approver_subjects={"admin": "kc-sub-0f3a"},
                   approver_issuer="https://kc.coulomb.social")
    run = X.Run(doc, G.APPROVAL, cfg, deps)
    run.preflight = lambda: run._check_approval(fake.get(G.APPROVAL), consumed_ok=False)  # pins are the example's
    result = run.execute()
    assert result["outcome"] in ("completed", "reverted") and "reason" not in result, result.get("reason")
    if G.MARKER in json.dumps(seen):
        raise RuntimeError("an operation value reached an event")
    return seen


def exercise(args) -> dict:
    sys.path.insert(0, str(args.audit_core_source))
    from audit_core.ingestion import IngestionApplication
    from audit_core.senders import SenderRegistry
    from audit_core.sqlite_backend import SQLiteAuditBackend

    scope = args.audit_core_source / "deploy/senders-scope.json"
    receipt = {
        "schema": "netkingdom.bounded-executor.audit-sender-exercise/v1",
        "task": "NK-WP-0056-T15",
        "target": "disposable local processes; synthetic tokens",
        "started_at": dt.datetime.now(dt.UTC).isoformat(timespec="seconds"),
        "audit_core_commit": commit(args.audit_core_source),
        "senders_scope_sha256": hashlib.sha256(scope.read_bytes()).hexdigest(),
        "net_kingdom_commit": commit(REPO),
        "executor_source_sha256": {n: hashlib.sha256((HERE / n).read_bytes()).hexdigest()
                                   for n in ("executor.py", "clients.py")},
        "receipt_schema_sha256": hashlib.sha256(V.SCHEMA.read_bytes()).hexdigest(),
        "limitations": [
            "audit-core runs from source in-process (SQLite backend), not the released receiver image",
            "the executor's operations are in-memory; the bind rotation itself is T12's synthetic pilot",
            "plain HTTP on loopback; the live path is the ESO-delivered bearer over the sso ingress (T16)",
        ],
        "checks": {},
    }
    checks = receipt["checks"]
    tokens = {"executor": secrets.token_urlsafe(32), "operator": secrets.token_urlsafe(32),
              "other": secrets.token_urlsafe(32)}
    with tempfile.TemporaryDirectory(prefix="nk56-t15-") as temporary:
        root = pathlib.Path(temporary)
        registry = SenderRegistry.from_env({
            # Minted with the defaults on purpose, as audit-core's own test does: the Git overlay must narrow it.
            "AUDIT_CORE_SENDERS": json.dumps([
                {"name": "nk-bounded-executor", "tokens": [tokens["executor"]], "sources": ["*"], "tenants": ["*"]},
                {"name": "user-engine", "tokens": [tokens["other"]], "sources": ["user-engine"]},
                {"name": "operator", "tokens": [tokens["operator"]], "sources": ["*"], "tenants": ["*"],
                 "may_write": False, "may_read": True}]),
            "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope)})
        backend = SQLiteAuditBackend(str(root / "audit.db"))
        app = IngestionApplication(backend, registry)

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

        server = make_server("127.0.0.1", 0, app, handler_class=Quiet)
        threading.Thread(target=server.serve_forever, daemon=True).start()
        base = f"http://127.0.0.1:{server.server_port}"
        try:
            for name, value in tokens.items():
                private_write(root / f"{name}.token", value + "\n")
            audit = C.AuditCore(base, str(root / "executor.token"))

            def raw(event, token: str | None = "executor", path="/v1/events", method="POST"):
                headers = {"Content-Type": "application/json"}
                if token:
                    headers["Authorization"] = "Bearer " + (root / f"{token}.token").read_text().strip() \
                        if token in tokens else "Bearer " + token
                if event is not None:
                    headers["Idempotency-Key"] = event["id"]
                req = urllib.request.Request(base + path, method=method, headers=headers,
                                             data=json.dumps(event).encode() if event is not None else None)
                try:
                    with urllib.request.urlopen(req, timeout=5) as response:
                        return response.status, json.loads(response.read() or b"{}")
                except urllib.error.HTTPError as error:
                    with error:
                        return error.code, json.loads(error.read() or b"{}")

            # ---- real runs through the real client
            first_status = []
            original = audit.http.call

            def recording(method, path, body=None, headers=None, scope=None):
                status, data = original(method, path, body, headers, scope)
                first_status.append(status)
                return status, data

            audit.http.call = recording
            completed = run_executor("BEA-2026-9201", audit)
            reverted = run_executor("BEA-2026-9202", audit, fail_checks=["synthetic-journey"])
            audit.http.call = original
            assert first_status == [202, 202, 202, 202], first_status
            assert completed[1]["data"]["outcome"] == "completed" and reverted[1]["data"]["outcome"] == "reverted"
            checks["completed_run_started_and_receipt_accepted_202"] = True
            checks["reverted_run_started_and_receipt_accepted_202"] = True
            for started, final in (completed, reverted):
                assert V.evaluate_pair(started, final) == {"valid": True, "findings": []}
            checks["emitted_pairs_pass_receipt_schema_v0_1"] = True

            assert raw(completed[0])[0] == 200 and raw(completed[1])[0] == 200
            audit.emit(completed[1])  # the client's own retry path counts 200 as stored
            checks["retry_is_duplicate_200"] = True

            changed = copy.deepcopy(completed[1])
            changed["data"]["outcome"] = "stopped"
            assert raw(changed)[0] == 409
            try:
                audit.emit(changed)
            except C.ClientError as error:
                assert "409" in str(error) and "run id reused" in str(error)
            else:
                raise AssertionError("conflict accepted")
            checks["same_id_other_body_conflict_409_reported_as_defect"] = True

            def fresh(event, **over):
                e = copy.deepcopy(event)
                rid = str(uuid.uuid4())
                e.update(id=f"run:{rid}:started", correlation_id=rid, **over)
                e["data"]["run_id"] = rid
                return e

            refused = {
                "wrong_source": (fresh(completed[0], source="user-engine"), "executor"),
                "wrong_tenant": (fresh(completed[0], tenant="tenant:platform"), "executor"),
                "other_sender_token": (fresh(completed[0]), "other"),
                "no_token": (fresh(completed[0]), None),
                "unknown_token": (fresh(completed[0]), "synthetic-unknown-token"),
            }
            for label, (event, token) in refused.items():
                status, _ = raw(event, token)
                assert status in (400, 401, 403), (label, status)
            for event, _ in refused.values():  # nothing stored under a refused event's run id
                status, body = raw(None, "operator", "/v1/events?correlation_id=" + event["correlation_id"], "GET")
                assert status == 200 and body["events"] == [], status
            status, body = raw(None, "operator", "/v1/events?correlation_id=" + completed[0]["correlation_id"], "GET")
            assert status == 200 and len(body["events"]) == 2, status  # while the accepted run's events are held
            checks["wrong_source_tenant_or_token_refused_and_not_stored"] = True

            # ---- a started event without a receipt, whose admission window closed two hours ago
            orphan = fresh(completed[0])
            orphan["data"]["window"]["not_after"] = (dt.datetime.now(dt.UTC) - dt.timedelta(hours=2)) \
                .isoformat(timespec="seconds").replace("+00:00", "Z")
            audit.emit(orphan)
            status, body = raw(None, "operator", "/v1/stream-findings", "GET")
            assert status == 200
            opened = [f for f in body.get("stream_findings", []) if f.get("kind") == "opening_without_closing"
                      and f.get("class") == "execution.run.started"]
            flagged = json.dumps(opened)
            mine = [f for f in opened if f.get("correlation_id") == orphan["correlation_id"]]
            assert len(mine) == 1 and mine[0]["bound"] == "window.not_after", "orphan not flagged by its window"
            assert completed[0]["correlation_id"] not in flagged and reverted[0]["correlation_id"] not in flagged
            checks["started_without_receipt_flagged_by_its_own_window"] = True
            checks["closed_runs_not_flagged"] = True
        finally:
            server.shutdown()
            server.server_close()

    receipt.update(status="passed", cleanup_complete=True,
                   finished_at=dt.datetime.now(dt.UTC).isoformat(timespec="seconds"))
    if any(v in json.dumps(receipt) for v in tokens.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])
    parser.add_argument("--audit-core-source", type=pathlib.Path, default=pathlib.Path.home() / "audit-core")
    parser.add_argument("--receipt", required=True, type=pathlib.Path)
    parser.add_argument("--trace", action="store_true", help="print the failing line (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:
        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
            detail = str(error) if isinstance(error, AssertionError) else ""  # own messages: statuses, labels
            where = f" at line {line} {detail}".rstrip()
        print(f"Audit sender exercise failed: {type(error).__name__}{where}", file=sys.stderr)
        return 1
    private_write(args.receipt, json.dumps(receipt, indent=2) + "\n")
    print(f"Audit sender exercise passed: {len(receipt['checks'])} checks; temporary resources removed.")
    return 0


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