"""Registry v1 federation conformance harness.

Exercises every endpoint in ``docs/spec/registry-v1.md`` against a

running registry. Two modes:

- **In-process.** With no ``OVERSIGHT_REGISTRY_URL`` environment

  variable, the harness stands the reference Python registry up inside

  a FastAPI ``TestClient`` against a fresh SQLite database in a temp

  directory and runs every check there. This is the CI path.

- **Live operator URL.** When ``OVERSIGHT_REGISTRY_URL`` is set, the

  harness points an ``httpx.Client`` at that URL and runs the same

  checks. This is the acceptance gate an independent operator uses to

  claim v1 conformance.

The script fails loudly on any divergence from the spec. Each check

has a short name so a run log is a compact conformance report.

"""

from __future__ import annotations

import base64

import json

import os

import shutil

import sys

import tempfile

import time

import uuid

from dataclasses import asdict

from pathlib import Path

from typing import Any, Optional

ROOT = Path(__file__).resolve().parent.parent

sys.path.insert(0, str(ROOT))

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey

from cryptography.hazmat.primitives import serialization

from oversight_core.manifest import Manifest, Recipient, WatermarkRef

PASS = "[PASS]"

FAIL = "[FAIL]"

PASSED: list[str] = []

FAILED: list[tuple[str, str]] = []

def check(name: str, condition: bool, detail: str = "") -> None:

    if condition:

        PASSED.append(name)

        print(f"  {PASS} {name}")

    else:

        FAILED.append((name, detail))

        print(f"  {FAIL} {name}  ({detail})")

def check_error_envelope(name: str, response, expected_status: int, expected_code: str) -> None:

    try:

        body = response.json()

    except Exception:

        body = {}

    error = body.get("error") if isinstance(body, dict) else None

    ok = (

        response.status_code == expected_status

        and isinstance(error, dict)

        and error.get("code") == expected_code

        and isinstance(error.get("message"), str)

        and bool(error.get("message"))

    )

    check(name, ok, f"status={response.status_code} body={body}")

class Client:

    """Thin wrapper that presents the same get/post surface over a

    FastAPI TestClient or a live httpx.Client."""

    def __init__(self, impl, base_url: str = "", default_headers: Optional[dict[str, str]] = None):

        self._impl = impl

        self._base = base_url.rstrip("/")

        self._headers = default_headers or {}

    def _merge_headers(self, kwargs: dict[str, Any]) -> dict[str, Any]:

        if not self._headers:

            return kwargs

        merged = dict(self._headers)

        merged.update(kwargs.pop("headers", {}) or {})

        return {**kwargs, "headers": merged}

    def get(self, path: str, **kwargs):

        kwargs = self._merge_headers(kwargs)

        return self._impl.get(self._base + path, **kwargs) if self._base else self._impl.get(path, **kwargs)

    def post(self, path: str, **kwargs):

        kwargs = self._merge_headers(kwargs)

        return self._impl.post(self._base + path, **kwargs) if self._base else self._impl.post(path, **kwargs)

def operator_headers() -> dict[str, str]:

    token = os.environ.get("OVERSIGHT_OPERATOR_TOKEN", "").strip()

    return {"Authorization": f"Bearer {token}"} if token else {}

def build_in_process_client():

    """Spin up the reference registry in a fresh temp data dir."""

    from fastapi.testclient import TestClient

    tmp = tempfile.mkdtemp(prefix="oversight-conformance-")

    os.environ["OVERSIGHT_DATA_DIR"] = tmp

    os.environ.setdefault("OVERSIGHT_REKOR_ENABLED", "0")

    os.environ.setdefault("OVERSIGHT_AUTH_DISABLED", "1")

    os.environ["OVERSIGHT_DNS_EVENT_SECRET"] = "test-dns-secret-123"

    for mod in [m for m in list(sys.modules) if m.startswith("registry.")]:

        del sys.modules[mod]

    import registry.server as server

    server.DATA_DIR = Path(tmp)

    server.DB_PATH = Path(tmp) / "registry.sqlite"

    server.TLOG_DIR = Path(tmp) / "tlog"

    server.IDENTITY_PATH = Path(tmp) / "identity.json"

    server.DNS_EVENT_SECRET = "test-dns-secret-123"

    server.IDENTITY = server.load_or_create_identity()

    server.init_db()

    from oversight_core.tlog import TransparencyLog

    server.TLOG = TransparencyLog(server.TLOG_DIR, signing_key_hex=server.IDENTITY["ed25519_priv"])

    tc = TestClient(server.app)

    return Client(tc, default_headers=operator_headers()), tmp, server.IDENTITY["ed25519_pub"]

def build_live_client(url: str):

    import httpx

    return Client(httpx.Client(timeout=15.0), base_url=url, default_headers=operator_headers()), None, None

def build_signed_manifest() -> tuple[dict, list[dict], list[dict], bytes]:

    """Return (manifest_dict, beacons, watermarks, issuer_priv_raw)."""

    issuer_sk = Ed25519PrivateKey.generate()

    issuer_pub_hex = (

        issuer_sk.public_key()

        .public_bytes(

            encoding=serialization.Encoding.Raw,

            format=serialization.PublicFormat.Raw,

        )

        .hex()

    )

    issuer_priv_raw = issuer_sk.private_bytes(

        encoding=serialization.Encoding.Raw,

        format=serialization.PrivateFormat.Raw,

        encryption_algorithm=serialization.NoEncryption(),

    )

    recipient_x25519 = X25519PrivateKey.generate().public_key().public_bytes(

        encoding=serialization.Encoding.Raw,

        format=serialization.PublicFormat.Raw,

    ).hex()

    recipient = Recipient(

        recipient_id="conformance-recipient",

        x25519_pub=recipient_x25519,

    )

    beacons = [

        {"token_id": uuid.uuid4().hex, "kind": "dns"},

        {"token_id": uuid.uuid4().hex, "kind": "http"},

    ]

    watermarks = [

        WatermarkRef(layer="L1_zero_width", mark_id="10" * 16),

        WatermarkRef(layer="L2_whitespace", mark_id="20" * 16),

    ]

    m = Manifest.new(

        original_filename="conformance.txt",

        content_hash="ab" * 32,

        size_bytes=4096,

        issuer_id="conformance-issuer",

        issuer_ed25519_pub_hex=issuer_pub_hex,

        recipient=recipient,

        registry_url="https://registry.example.org",

    )

    m.beacons = list(beacons)

    m.watermarks = list(watermarks)

    m.sign(issuer_priv_raw)

    manifest_dict = json.loads(m.to_json().decode("utf-8"))

    sidecar_beacons = list(beacons)

    sidecar_watermarks = [asdict(w) for w in watermarks]

    return manifest_dict, sidecar_beacons, sidecar_watermarks, issuer_priv_raw

def check_health(cli: Client) -> None:

    r = cli.get("/health")

    check("health-200", r.status_code == 200, f"status={r.status_code}")

    body = r.json() if r.status_code == 200 else {}

    check("health-has-status", body.get("status") in {"ok", "degraded"},

          f"status={body.get('status')!r}")

    check("health-service-prefix",

          str(body.get("service", "")).startswith("oversight-registry"),

          f"service={body.get('service')!r}")

    check("health-tlog-size-int", isinstance(body.get("tlog_size"), int))

def check_well_known(cli: Client) -> None:

    r = cli.get("/.well-known/oversight-registry")

    check("well-known-200", r.status_code == 200, f"status={r.status_code}")

    body = r.json() if r.status_code == 200 else {}

    pub = body.get("ed25519_pub")

    check("well-known-ed25519-hex",

          isinstance(pub, str) and len(pub) == 64 and all(c in "0123456789abcdef" for c in pub.lower()),

          f"ed25519_pub={pub!r}")

    check("well-known-has-version", isinstance(body.get("version"), str))

def check_register_roundtrip(cli: Client, manifest: dict, beacons: list, watermarks: list) -> Optional[str]:

    body = {"manifest": manifest, "beacons": beacons, "watermarks": watermarks}

    r = cli.post("/register", json=body)

    check("register-200", r.status_code == 200, f"status={r.status_code} body={r.text[:200]}")

    if r.status_code != 200:

        return None

    out = r.json()

    check("register-ok-true", out.get("ok") is True)

    check("register-file-id-echo", out.get("file_id") == manifest["file_id"])

    check("register-count", out.get("registered_beacons") == len(beacons))

    check("register-tlog-index-int", isinstance(out.get("tlog_index"), int))

    return out.get("file_id")

def check_register_rejects_unsigned(cli: Client, manifest: dict, beacons: list, watermarks: list) -> None:

    tampered = dict(manifest)

    tampered["signature_ed25519"] = "00" * 64

    tampered["file_id"] = str(uuid.uuid4())

    r = cli.post("/register", json={"manifest": tampered, "beacons": beacons, "watermarks": watermarks})

    check("register-rejects-bad-sig", r.status_code == 400, f"status={r.status_code}")

    check_error_envelope("register-bad-sig-error-envelope", r, 400, "signature_invalid")

def check_register_rejects_sidecar_mismatch(cli: Client, manifest: dict, beacons: list, watermarks: list) -> None:

    bad = list(beacons) + [{"token_id": "sneaky", "kind": "dns"}]

    r = cli.post("/register", json={"manifest": manifest, "beacons": bad, "watermarks": watermarks})

    check("register-rejects-sidecar-mismatch", r.status_code == 400, f"status={r.status_code}")

    check_error_envelope("register-sidecar-error-envelope", r, 400, "sidecar_mismatch")

def check_attribute_by_token(cli: Client, beacons: list) -> None:

    r = cli.post("/attribute", json={"token_id": beacons[0]["token_id"]})

    check("attribute-200", r.status_code == 200, f"status={r.status_code}")

    body = r.json() if r.status_code == 200 else {}

    check("attribute-found", body.get("found") is True)

def check_attribute_miss(cli: Client) -> None:

    r = cli.post("/attribute", json={"token_id": "nonexistent-token-id"})

    check("attribute-miss-200", r.status_code == 200)

    check("attribute-miss-found-false", r.json().get("found") is False)

def check_attribute_missing_field_error(cli: Client) -> None:

    r = cli.post("/attribute", json={})

    check_error_envelope("attribute-missing-field-error-envelope", r, 400, "missing_field")

def check_evidence(cli: Client, file_id: str) -> None:

    r = cli.get(f"/evidence/{file_id}")

    check("evidence-200", r.status_code == 200, f"status={r.status_code}")

    body = r.json() if r.status_code == 200 else {}

    check("evidence-has-manifest", isinstance(body.get("manifest"), dict))

    check("evidence-has-events", isinstance(body.get("events"), list))

    check("evidence-has-beacons", isinstance(body.get("beacons"), list))

    check("evidence-has-watermarks", isinstance(body.get("watermarks"), list))

    check("evidence-has-registry-pub", isinstance(body.get("registry_pub"), str))

    check("evidence-has-tlog-head",

          "tlog_head" in body,

          f"keys={list(body)[:10]}")

    check("evidence-has-tlog-proofs",

          isinstance(body.get("tlog_proofs"), list))

    check("evidence-has-bundle-signature",

          isinstance(body.get("bundle_signature_ed25519"), str))

def check_evidence_missing_error(cli: Client) -> None:

    r = cli.get("/evidence/missing-file-id")

    check_error_envelope("evidence-missing-error-envelope", r, 404, "not_found")

def check_tlog_head(cli: Client) -> None:

    r = cli.get("/tlog/head")

    check("tlog-head-200", r.status_code == 200, f"status={r.status_code}")

def check_tlog_range(cli: Client) -> None:

    r = cli.get("/tlog/range?start=0&limit=10")

    body = r.json() if r.status_code == 200 else {}

    entries = body.get("entries")

    range_ok = (

        r.status_code == 200

        and isinstance(entries, list)

        and body.get("count") == len(entries)

        and all(

            isinstance(entry, dict)

            and isinstance(entry.get("index"), int)

            and isinstance(entry.get("leaf_hash"), str)

            and isinstance(entry.get("leaf_data"), str)

            for entry in entries

        )

    )

    check("tlog-range-200-shape", range_ok, f"status={r.status_code} body={body}")

def check_dns_event_requires_secret(cli: Client) -> None:

    token = "t-" + uuid.uuid4().hex

    r = cli.post(

        "/dns_event",

        json={"token_id": token, "client_ip": "198.51.100.8", "qtype": "A", "qname": "x.example"},

        headers={"X-Oversight-DNS-Secret": "wrong-secret"},

    )

    check(

        "dns-event-auth-enforced",

        r.status_code in (200, 401, 503),

        f"status={r.status_code}",

    )

def check_cors_headers(cli: Client) -> None:

    """A browser inspector hosted at an Oversight-approved origin must be able

    to read /health and /.well-known; confirm the CORS middleware is present."""

    origin = "https://oversight-protocol.github.io"

    try:

        r = cli.get("/health", headers={"Origin": origin})

    except TypeError:

        r = cli.get("/health")

    acao = r.headers.get("access-control-allow-origin") if hasattr(r, "headers") else None

    check(

        "cors-allows-github-pages-origin",

        acao in (origin, "*"),

        f"Access-Control-Allow-Origin={acao!r}",

    )

def check_beacon_endpoints(cli: Client, beacons: list) -> None:

    token = beacons[0]["token_id"]

    r = cli.get(f"/p/{token}.png")

    check("beacon-http-img-200", r.status_code == 200, f"status={r.status_code}")

    r = cli.get(f"/r/{token}")

    check("beacon-ocsp-200", r.status_code == 200, f"status={r.status_code}")

    r = cli.get(f"/v/{token}")

    check("beacon-license-200", r.status_code == 200, f"status={r.status_code}")

def run(cli: Client) -> None:

    print("[*] Oversight registry v1 conformance harness")

    print("\n[*] Identity and liveness")

    check_health(cli)

    check_well_known(cli)

    print("\n[*] Registration")

    manifest, beacons, watermarks, _ = build_signed_manifest()

    file_id = check_register_roundtrip(cli, manifest, beacons, watermarks)

    check_register_rejects_unsigned(cli, manifest, beacons, watermarks)

    check_register_rejects_sidecar_mismatch(cli, manifest, beacons, watermarks)

    if file_id:

        print("\n[*] Attribution and evidence")

        check_attribute_by_token(cli, beacons)

        check_attribute_miss(cli)

        check_attribute_missing_field_error(cli)

        check_evidence(cli, file_id)

        check_evidence_missing_error(cli)

        print("\n[*] Transparency log")

        check_tlog_head(cli)

        check_tlog_range(cli)

        print("\n[*] CORS")

        check_cors_headers(cli)

        print("\n[*] Beacons and DNS event")

        check_beacon_endpoints(cli, beacons)

        check_dns_event_requires_secret(cli)

    print()

    print(f"[summary] passed={len(PASSED)} failed={len(FAILED)}")

    if FAILED:

        for name, detail in FAILED:

            print(f"  -> {name}: {detail}")

        raise SystemExit(1)

    print("[ok] conformance harness green")

def main() -> None:

    url = os.environ.get("OVERSIGHT_REGISTRY_URL", "").strip()

    tmp = None

    try:

        if url:

            print(f"[*] target: live registry at {url}")

            cli, tmp, _ = build_live_client(url)

        else:

            print("[*] target: in-process reference registry")

            cli, tmp, _ = build_in_process_client()

        run(cli)

    finally:

        if tmp and os.path.isdir(tmp):

            shutil.rmtree(tmp, ignore_errors=True)

def test_registry_v1_conformance_harness() -> None:

    """Pytest entry point for the registry v1 conformance harness.

    The harness is intentionally a single pytest case: the checks share state

    (the registered file_id drives the subsequent attribution, evidence, tlog,

    and beacon checks) and the question under test is one yes/no question,

    "does this registry meet v1 conformance?" Per-check pass/fail is still

    printed to stdout so a CI log is a compact conformance report.

    """

    PASSED.clear()

    FAILED.clear()

    url = os.environ.get("OVERSIGHT_REGISTRY_URL", "").strip()

    tmp = None

    try:

        if url:

            cli, tmp, _ = build_live_client(url)

        else:

            cli, tmp, _ = build_in_process_client()

        run(cli)

    finally:

        if tmp and os.path.isdir(tmp):

            shutil.rmtree(tmp, ignore_errors=True)

    assert not FAILED, f"{len(FAILED)} conformance check(s) failed: " + ", ".join(

        name for name, _ in FAILED

    )

if __name__ == "__main__":

    main()