#!/usr/bin/env python3
# /// script
# dependencies = ["jsonschema>=4.10,<5"]
# ///
"""Validate platform-root custody records (NK-ADR-0017, NK-WP-0045-T03).

Checks the JSON Schema (`canon/schemas/custody-record_v0.1.schema.json`) and
the semantics the schema cannot express: unique entries, at most one active
grant per exact (issuer, subject), revision ordering, supersession links,
append-only successors, and secret-looking values. It also provides the
fail-closed read rule a consumer applies to an observation (15-minute bound).

Offline and read-only: it never contacts an issuer, OpenBao or a directory.
"""

from __future__ import annotations

import argparse
import datetime as dt
import json
import pathlib
import re
import sys
from typing import Any

import jsonschema

ROOT = pathlib.Path(__file__).resolve().parents[2]
SCHEMA = ROOT / "canon/schemas/custody-record_v0.1.schema.json"
MAX_CACHE_SECONDS = 900

# Value shapes that look like secret material (OpenBao/Vault tokens, PEM keys,
# JWTs). Field names are rejected by the schema; this catches smuggled values.
SECRET_VALUE = re.compile(
    r"(\bhv[sbr]\.[A-Za-z0-9_-]{8,}|\bs\.[A-Za-z0-9]{20,}"
    r"|-----BEGIN [A-Z ]*PRIVATE KEY-----"
    r"|\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.)"
)


def load_schema() -> dict[str, Any]:
    return json.loads(SCHEMA.read_text(encoding="utf-8"))


def parse_time(value: str) -> dt.datetime:
    return dt.datetime.fromisoformat(value.replace("Z", "+00:00"))


def key(entry: dict[str, Any]) -> tuple[str, str]:
    return (entry["subject"]["issuer"], entry["subject"]["subject"])


def is_active(entry: dict[str, Any]) -> bool:
    return "revoked" not in entry and "superseded_by" not in entry


def schema_errors(document: Any, schema: dict[str, Any]) -> list[str]:
    validator = jsonschema.Draft202012Validator(
        schema, format_checker=jsonschema.FormatChecker()
    )
    errors = []
    for error in sorted(validator.iter_errors(document), key=lambda e: list(e.path)):
        where = "/".join(str(part) for part in error.path) or "<root>"
        errors.append(f"schema: {where}: {error.message}")
    return errors


def _walk_strings(node: Any, path: str = ""):
    if isinstance(node, dict):
        for name, value in node.items():
            yield from _walk_strings(value, f"{path}/{name}")
    elif isinstance(node, list):
        for index, value in enumerate(node):
            yield from _walk_strings(value, f"{path}/{index}")
    elif isinstance(node, str):
        yield path or "<root>", node


def semantic_errors(record: dict[str, Any]) -> list[str]:
    errors: list[str] = []
    for where, value in _walk_strings(record):
        if SECRET_VALUE.search(value):
            errors.append(f"{where}: value looks like secret material")

    revision = record["revision"]
    entries = record["entries"]
    by_id: dict[str, dict[str, Any]] = {}
    for entry in entries:
        if entry["entry_id"] in by_id:
            errors.append(f"{entry['entry_id']}: duplicate entry_id")
        by_id[entry["entry_id"]] = entry

    active: dict[tuple[str, str], str] = {}
    for entry in entries:
        eid = entry["entry_id"]
        granted = entry["granted"]
        if granted["revision"] > revision:
            errors.append(f"{eid}: granted.revision is after record revision")
        if is_active(entry):
            if key(entry) in active:
                errors.append(
                    f"{eid}: second active grant for {key(entry)} "
                    f"(already {active[key(entry)]})"
                )
            active[key(entry)] = eid
        revoked = entry.get("revoked")
        if revoked:
            if not granted["revision"] < revoked["revision"] <= revision:
                errors.append(f"{eid}: revoked.revision must follow the grant and not exceed the record revision")
            if parse_time(revoked["at"]) < parse_time(granted["at"]):
                errors.append(f"{eid}: revoked before granted")
        link = entry.get("superseded_by")
        if link:
            successor = by_id.get(link["entry_id"])
            if successor is None or link["entry_id"] == eid:
                errors.append(f"{eid}: superseded_by names no other entry")
            else:
                if successor["granted"]["revision"] != link["revision"]:
                    errors.append(f"{eid}: superseded_by.revision must equal the successor's granted.revision")
                if parse_time(successor["granted"]["at"]) < parse_time(granted["at"]):
                    errors.append(f"{eid}: successor granted before the superseded entry")
            if not granted["revision"] < link["revision"] <= revision:
                errors.append(f"{eid}: superseded_by.revision must follow the grant and not exceed the record revision")
    return errors


def validate(record: Any, schema: dict[str, Any] | None = None) -> list[str]:
    errors = schema_errors(record, schema or load_schema())
    if errors:
        return errors
    return semantic_errors(record)


def successor_errors(previous: dict[str, Any], current: dict[str, Any]) -> list[str]:
    """A later read must be a strictly newer, append-only revision."""
    if validate(previous) or validate(current):
        return ["invalid record in revision comparison"]
    errors: list[str] = []
    if previous["record_id"] != current["record_id"]:
        return ["record_id changed between revisions"]
    if current["revision"] < previous["revision"]:
        return ["revision went backwards: stale read, keep the newer observation"]
    if current["revision"] == previous["revision"]:
        if current != previous:
            errors.append("same revision with different content")
        return errors
    now = {entry["entry_id"]: entry for entry in current["entries"]}
    before = {entry["entry_id"]: entry for entry in previous["entries"]}
    if parse_time(current["updated_at"]) < parse_time(previous["updated_at"]):
        errors.append("updated_at went backwards")
    for entry in current["entries"]:
        earlier = before.get(entry["entry_id"], {})
        for field in ("granted", "revoked", "superseded_by"):
            if field in entry and field not in earlier:
                if entry[field]["revision"] <= previous["revision"]:
                    errors.append(f"{entry['entry_id']}: new {field} must follow the previous record revision")
    for entry in previous["entries"]:
        later = now.get(entry["entry_id"])
        if later is None:
            errors.append(f"{entry['entry_id']}: entry removed; custody actions append, never delete")
            continue
        for field in ("subject", "grant", "evidence", "granted"):
            if later[field] != entry[field]:
                errors.append(f"{entry['entry_id']}: {field} rewritten")
        for field in ("revoked", "superseded_by"):
            if field in entry and later.get(field) != entry[field]:
                errors.append(f"{entry['entry_id']}: {field} rewritten")
    return errors


def observe_root_grant(
    record: dict[str, Any] | None,
    issuer: str,
    subject: str,
    tenant: str,
    read_at: dt.datetime,
    now: dt.datetime,
    previous_record: dict[str, Any] | None = None,
) -> dict[str, Any] | None:
    """Return the ADR-0017 section 3 observation, or None (fail closed).

    None when: the tenant is not `platform`; the record is missing, invalid or
    older than the cache bound; or no active entry matches the exact
    (issuer, subject).
    """
    if tenant != "platform" or record is None or validate(record):
        return None
    if previous_record is not None and successor_errors(previous_record, record):
        return None
    if read_at.utcoffset() is None or now.utcoffset() is None:
        return None
    if parse_time(record["updated_at"]) > read_at:
        return None
    bound = min(record["freshness"]["max_cache_seconds"], MAX_CACHE_SECONDS)
    if not dt.timedelta(0) <= now - read_at <= dt.timedelta(seconds=bound):
        return None
    for entry in record["entries"]:
        if (key(entry) == (issuer, subject) and is_active(entry)
                and parse_time(entry["granted"]["at"]) <= read_at):
            return {
                "subject": dict(entry["subject"]),
                "grant": dict(entry["grant"]),
                "source": {"record_id": record["record_id"], "revision": record["revision"]},
                "observed_at": read_at.isoformat().replace("+00:00", "Z"),
                "expires_at": (read_at + dt.timedelta(seconds=bound)).isoformat().replace("+00:00", "Z"),
            }
    return None


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("paths", nargs="+", type=pathlib.Path)
    args = parser.parse_args(argv)
    schema = load_schema()
    failed = False
    for path in args.paths:
        try:
            errors = validate(json.loads(path.read_text(encoding="utf-8")), schema)
        except (OSError, ValueError):
            errors = ["unreadable input"]
        # jsonschema messages may contain the rejected secret value. Never
        # expose those messages (or input paths) through the command line.
        if errors:
            print(f"custody record: rejected ({len(errors)} errors)", file=sys.stderr)
        failed = failed or bool(errors)
        if not errors:
            print("custody record: ok")
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())