#!/usr/bin/env python3
"""Verify the per-tenant event hash chain (migration 040) from a JSON export.

Published at https://intercis.io/evidence/verify_event_chain.py so a reviewer
can read and run it without asking us for anything. The code is the code in
scripts/verify_event_chain.py on the Intercis main branch; this paragraph and
the two usage lines below are the only edits, because the path changed.
Standard library only, no network calls. Questions: security@intercis.io.

Pure Python — no Postgres connection. Recomputes every event_hash from the
exported row fields and walks each tenant's chain in (created_at, id) order,
reporting the FIRST broken row id per tenant.

TWO FORMATS. A row's format is its `chain_version` field: missing, NULL or 1
is v1; 2 is v2; ANY other value fails that row — an unknown format is never
guessed at. Within one tenant's chain, a v1 row that FOLLOWS a v2 row is a
break (a downgrade is how an editor would escape the wider v2 hash). The prev
link works across the v1->v2 boundary exactly as it does inside a version, and
genesis is unchanged.

FORMAT v1 — ten fields
(must stay in lockstep with infra/supabase/migrations/040_event_hash_chain.sql):

    Every field is INJECTIVELY encoded before concatenation (security review
    2026-08-21: naive concatenation let cross-field character shifts and
    NULL<->'' flips produce identical hashes):

        enc(None) = 'N'
        enc(s)    = f"{len(s)}:{s}"        # length in characters

    'N' is distinct from enc('') == '0:', so NULL and empty string hash
    differently, and the length prefix makes the field split unambiguous.

    event_hash = hex(sha256(utf8(
        enc(prev_event_hash)
        || enc(id)                         # lowercase uuid text
        || enc(tenant_id)                  # lowercase uuid text, 'N' for NULL tenant
        || enc(agent_id)
        || enc(action)
        || enc(target)
        || enc(verdict)
        || enc(policy)
        || enc(injected: True->'t', False->'f')
        || enc(source)
        || enc(created_at as 'YYYY-MM-DDTHH:MM:SS.ffffffZ' in UTC (6-digit micros))
    )))
    genesis prev for a tenant's first row = hex(sha256(utf8(tenant_id or '')))
    (genesis keeps the coalesce: tenant_id is a uuid, so '' cannot collide
    with a real tenant's text form)

FORMAT v2 — the version marker, the same ten fields, and fourteen more
(ledger row 194aca0c; a later migration implements these exact bytes in SQL,
so treat the order and the text forms below as the specification):

    event_hash = hex(sha256(utf8(
        enc(prev_event_hash)
        || enc('2')                        # the version, as literal text
        || the TEN v1 fields, in v1 order, in v1 text forms:
             enc(id) || enc(tenant_id) || enc(agent_id) || enc(action)
          || enc(target) || enc(verdict) || enc(policy)
          || enc(injected: True->'t', False->'f') || enc(source)
          || enc(created_at as 'YYYY-MM-DDTHH:MM:SS.ffffffZ' in UTC)
        || then, in this order:
             enc(cloud_provider)            # cloud_provider_enum label as-is
          || enc(region)
          || enc(workload_type)             # workload_type_enum label as-is
          || enc(cloud_account_id)
          || enc(cluster_name)
          || enc(command_fingerprint)
          || enc(cc_session_id)
          || enc(cc_agent_id)
          || enc(cc_parent_agent_id)
          || enc(classifier_status)
          || enc(classifier_reason)
          || enc(usage_input_tokens)        # integer -> base-10 text, no
          || enc(usage_output_tokens)       #   padding, no '+' sign
          || enc(model)
    )))

    Same encoder as v1, so NULL ('N') stays distinguishable from the empty
    string ('0:') in every one of the fourteen. raw_request is NOT hashed in
    EITHER version. Fixed vectors for this formula, with the literal hash
    input for each, live in apps/proxy/tests/fixtures/event_chain_v2_vectors.json.

WHAT AN "ok" RESULT CERTIFIES — AND WHAT IT DOES NOT

It is now a per-version statement, and the script prints the count of each.

  * For a v1 row: the TEN chained fields — id, tenant_id, agent_id, action,
    target, verdict, policy, injected, source, created_at (plus the prev
    link) — are CONSISTENT WITH THE HASHES THE EXPORT ITSELF CARRIES. That
    is weaker than "unaltered", and the difference matters: the hashes are
    the export's own, not a head held anywhere this script can trust, so a
    row rewritten together with every hash after it verifies (see THE
    CEILING below). It says NOTHING about any other column of that
    row. That gap is permanent for rows already written under v1: re-hashing
    them was option B, rejected, because mass-updating the audit table is the
    act the chain exists to detect.
  * For a v2 row: EVERY column of public.events except raw_request is
    consistent with those same hashes, the format marker included.

raw_request is the one exclusion made deliberately at chain-design time, in
both formats: migration 035 NULLs it at 90 days (nightly pg_cron scrub), and
hashing a column the database itself rewrites on a schedule would break each
affected tenant's chain from its first scrubbed row onward. 035 is bounded —
rows older than 90 days, at most 10,000 per night by an id-subquery LIMIT
(infra/supabase/migrations/035_events_raw_request_retention.sql:16 and :52-55)
— so the breakage would arrive tenant by tenant over successive nights, not
as "every chain the night the first row aged out", which is how the 040
header and the records built on it put it. Recorded in DECISIONS.

The fourteen columns v2 adds are the ones v1 left outside — three migrations'
worth (013, 029, 032) that predate the chain and were dropped from 040's field
list with no recorded reason, and two (042, 046) added after it:

  * classifier_status, classifier_reason (042). Named first because these two
    exist precisely to prove a security control ran: they distinguish "the
    classifier ran and cleared this" from "the classifier could not answer".
    On a v1 row, a rewrite from classifier_status='unavailable' to 'safe'
    (service_role holds UPDATE; until migration 052 the only trigger on
    public.events was the BEFORE INSERT chain trigger) leaves event_hash
    valid and this verifier prints ok. On a v2 row it breaks the chain at
    that row. Since 052 — applied to prod 2026-09-19 — a BEFORE UPDATE guard
    refuses that edit in the database, but that is prevention, not evidence:
    a writer who can drop the guard is not stopped by it, and a v1 row gains
    no chain evidence from it.
  * cloud_provider, region, workload_type, cloud_account_id, cluster_name
    (013)
  * command_fingerprint (029)
  * cc_session_id, cc_agent_id, cc_parent_agent_id (032)
  * usage_input_tokens, usage_output_tokens, model (046)

(prev_event_hash, event_hash and chain_version are the chain itself rather
than payload; chain_version is hashed as the v2 version marker, so editing it
breaks the row.)

THE CEILING, UNCHANGED BY v2. There is no anchor outside the database. This
script holds no trusted head and no expected row count, so a writer who can
rewrite a row AND every hash after it — or drop the trigger, or delete the
newest rows — is not detected, in either format. v2 closes the quiet
single-column edit on new rows; it does not close a full re-chain, and it
gives the values already on v1 rows no evidence they did not have before.

Export formats accepted (auto-detected):
  * a JSON array of row objects  — e.g.
      psql -t -A -c "SELECT json_agg(to_jsonb(e) - 'raw_request'
        ORDER BY e.created_at, e.id) FROM public.events e" > export.json
    to_jsonb names no column, so this ONE query works both against today's
    prod, where chain_version does not exist yet (every row then arrives with
    no chain_version key and is read as v1 — see verify_chain), and against a
    database where the v2 migration has been applied. Naming the columns
    explicitly instead would make the query fail on prod the moment
    chain_version were added to the list, and a list that omitted the fourteen
    v2 columns would make a v2 row unverifiable. raw_request is subtracted
    because no format hashes it and it is the bulky, PII-bearing column.
  * an object with an "events" key holding that array
    (the docs/observe-events-rows-*.json convention)

Usage:
    python3 verify_event_chain.py export.json
    python3 verify_event_chain.py export.json --since 2026-08-01T00:00:00Z

--since only restricts which tenants/rows are REPORTED after full-chain
verification when the export itself is already windowed; the chain math always
starts from each tenant's first exported row, so a windowed export must start
at a chain boundary (export from the tenant's first row, or accept that the
first exported row's prev link cannot be checked against genesis).

Exit code 0 = every tenant chain verifies; 1 = at least one break; 2 = bad input.
Bad input includes an export with no rows (nothing was verified, so it is never
a 0) and a row that lacks the KEY for any hashed column: all ten v1 columns
(id, tenant_id, agent_id, action, target, verdict, policy, injected, source,
created_at) on every row, plus the fourteen v2 columns on a chain_version 2 row.
A present key with a null value is a stored NULL and hashes as 'N'; an absent
key is a lossy export (a wrong SELECT, a trimmed CSV) and is refused rather than
read as NULL, because a NULL-valued row would otherwise verify without the
column ever being checked.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import re
import sys
import uuid
from datetime import datetime, timezone

_TS_RE = re.compile(
    r"^(\d{4})-(\d{2})-(\d{2})[T ](\d{2}):(\d{2}):(\d{2})(?:\.(\d{1,6}))?"
    r"(Z|[+-]\d{2}(?::?\d{2})?)?$"
)


def parse_timestamp(value: str) -> datetime:
    """Parse the timestamptz string shapes Postgres exports produce.

    Handles json_agg ('2026-08-18T02:15:19.668209+00:00'), psql/COPY text
    ('2026-08-18 02:15:19.668209+00'), trimmed fractional digits, and 'Z'.
    A value with no offset is treated as UTC (COPY under TimeZone=UTC).
    """
    m = _TS_RE.match(value.strip())
    if not m:
        raise ValueError(f"unparseable timestamp: {value!r}")
    y, mo, d, h, mi, s = (int(x) for x in m.groups()[:6])
    frac = m.group(7) or ""
    micros = int(frac.ljust(6, "0")) if frac else 0
    tz = m.group(8)
    if tz is None or tz == "Z" or re.fullmatch(r"[+-]00:?0?0?|[+-]00", tz):
        tzinfo = timezone.utc
    else:
        sign = 1 if tz[0] == "+" else -1
        parts = tz[1:].replace(":", "")
        oh = int(parts[:2])
        om = int(parts[2:4]) if len(parts) >= 4 else 0
        from datetime import timedelta

        tzinfo = timezone(sign * timedelta(hours=oh, minutes=om))
    return datetime(y, mo, d, h, mi, s, micros, tzinfo=tzinfo)


def canonical_timestamp(value: str) -> str:
    """The exact string the DB trigger hashes: UTC, 6-digit micros, 'Z'."""
    dt = parse_timestamp(value).astimezone(timezone.utc)
    return dt.strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z"


def _sha256_hex(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def encode_field(value: str | None) -> str:
    """Injective field encoding: 'N' for NULL, length-prefixed otherwise.

    'N' is distinct from encode_field('') == '0:' so NULL and empty string
    hash differently; the length prefix keeps field boundaries unambiguous.
    Must stay in lockstep with public.events_chain_field() in migration 040.
    """
    if value is None:
        return "N"
    return f"{len(value)}:{value}"


def _bool_field(value) -> str | None:
    """Normalize an exported boolean to the 't'/'f' text the SQL hashes.

    json_agg exports true/false; psql -t -A text exports 't'/'f'.
    """
    if value is None:
        return None
    if isinstance(value, bool):
        return "t" if value else "f"
    if isinstance(value, str) and value in ("t", "f"):
        return value
    if isinstance(value, str) and value.lower() in ("true", "false"):
        return "t" if value.lower() == "true" else "f"
    raise ValueError(f"unparseable boolean: {value!r}")


# ── chain format v2 (ledger row 194aca0c) ───────────────────────────────────
#
# The fourteen columns v1 left outside the hash, in the order v2 hashes them.
# This tuple IS the format: a later SQL migration reproduces it byte for byte,
# so nothing here may be reordered or "tidied".
CHAIN_V2_COLUMNS = (
    "cloud_provider", "region", "workload_type", "cloud_account_id",
    "cluster_name", "command_fingerprint", "cc_session_id", "cc_agent_id",
    "cc_parent_agent_id", "classifier_status", "classifier_reason",
    "usage_input_tokens", "usage_output_tokens", "model",
)
# integer columns: base-10 text, no padding, no sign for non-negatives.
CHAIN_V2_INT_COLUMNS = frozenset({"usage_input_tokens", "usage_output_tokens"})

_INT_TEXT_RE = re.compile(r"^-?(?:0|[1-9][0-9]*)$")


def chain_version(row: dict) -> int | None:
    """The format of one row: 1, 2, or None for a value this script cannot read.

    Missing key, NULL and 1 are all v1 — that is today's prod, where the
    column does not exist. 2 is v2. Anything else returns None and the caller
    FAILS that row: an unknown format is never guessed at, and a v2 row must
    never be quietly re-read as v1.
    """
    value = row.get("chain_version")
    if value is None:
        return 1
    if isinstance(value, bool):
        return None
    if isinstance(value, int):
        return value if value in (1, 2) else None
    if isinstance(value, str) and value.strip() in ("1", "2"):
        return int(value.strip())
    return None


def _text_column(row: dict, column: str) -> str | None:
    """Text form of a v2 text/enum column. Enum labels are hashed as-is."""
    value = row[column]
    if value is None or isinstance(value, str):
        return value
    raise ValueError(f"{column} must be text or NULL, got {value!r}")


def _int_column(row: dict, column: str) -> str | None:
    """Text form of a v2 integer column: base-10, no padding, no '+'.

    An already-textual export ('1234', from psql -t -A) is accepted verbatim
    only when it is ALREADY canonical; '007', '+7' or ' 7' would hash to a
    different digest than the database's own int-to-text cast produces, so
    they are input errors rather than something to normalise silently.
    """
    value = row[column]
    if value is None:
        return None
    if isinstance(value, bool):
        raise ValueError(f"{column} must be an integer or NULL, got {value!r}")
    if isinstance(value, int):
        return str(value)
    if isinstance(value, str) and _INT_TEXT_RE.match(value):
        return value
    raise ValueError(f"{column} must be canonical base-10 integer text or NULL, "
                     f"got {value!r}")


def genesis_hash(tenant_id: str | None) -> str:
    return _sha256_hex(tenant_id or "")


# The ten v1 fields, in hash order. Every key must be PRESENT in an exported
# row; its value may be null (a stored NULL, hashed as 'N'). tenant_id, target,
# policy and source are the nullable four: reading them with .get() made an
# ABSENT key hash as 'N' too, so an export that dropped one of those columns
# still verified every row whose stored value was NULL (row 689eeaa1).
CHAIN_V1_COLUMNS = (
    "id", "tenant_id", "agent_id", "action", "target", "verdict", "policy",
    "injected", "source", "created_at",
)


def _missing_v1_columns(row: dict) -> list[str]:
    return [c for c in CHAIN_V1_COLUMNS if c not in row]


def _v1_field_encodings(row: dict) -> str:
    """The ten v1 fields, encoded and concatenated — identical in both formats."""
    missing = _missing_v1_columns(row)
    if missing:
        raise ValueError(
            f"row {row.get('id')!r} has no {missing[0]!r} key in the export; a "
            f"missing hashed column would hash as NULL (missing: {missing})"
        )
    return (
        encode_field(str(row["id"]))
        + encode_field(row["tenant_id"])
        + encode_field(row["agent_id"])
        + encode_field(row["action"])
        + encode_field(row["target"])
        + encode_field(row["verdict"])
        + encode_field(row["policy"])
        + encode_field(_bool_field(row["injected"]))
        + encode_field(row["source"])
        + encode_field(canonical_timestamp(row["created_at"]))
    )


def hash_input_v1(prev_hash: str, row: dict) -> str:
    return encode_field(prev_hash) + _v1_field_encodings(row)


def hash_input_v2(prev_hash: str, row: dict) -> str:
    """The exact string v2 digests — returned whole so vectors can record it."""
    parts = [encode_field(prev_hash), encode_field("2"), _v1_field_encodings(row)]
    for column in CHAIN_V2_COLUMNS:
        if column not in row:
            raise ValueError(
                f"row {row.get('id')!r} is chain_version 2 but the export has no "
                f"{column!r} key; a missing hashed column would hash as NULL"
            )
        if column in CHAIN_V2_INT_COLUMNS:
            parts.append(encode_field(_int_column(row, column)))
        else:
            parts.append(encode_field(_text_column(row, column)))
    return "".join(parts)


def compute_event_hash_v1(prev_hash: str, row: dict) -> str:
    return _sha256_hex(hash_input_v1(prev_hash, row))


def compute_event_hash_v2(prev_hash: str, row: dict) -> str:
    return _sha256_hex(hash_input_v2(prev_hash, row))


def compute_event_hash(prev_hash: str, row: dict) -> str:
    """Recompute one row's hash under the format its chain_version names."""
    version = chain_version(row)
    if version == 1:
        return compute_event_hash_v1(prev_hash, row)
    if version == 2:
        return compute_event_hash_v2(prev_hash, row)
    raise ValueError(
        f"row {row.get('id')!r} has unknown chain_version "
        f"{row.get('chain_version')!r}"
    )


def load_rows(path: str) -> list[dict]:
    with open(path, "r", encoding="utf-8") as fh:
        data = json.load(fh)
    if isinstance(data, dict) and "events" in data:
        data = data["events"]
    if data is None:
        return []
    if not isinstance(data, list):
        raise ValueError("export must be a JSON array of rows or {'events': [...]}")
    return data


REQUIRED_FIELDS = ("id", "agent_id", "action", "verdict", "injected",
                   "created_at", "prev_event_hash", "event_hash")


def verify_chain(rows: list[dict]) -> dict:
    """Verify every tenant chain in `rows`.

    Returns {tenant_key: {"ok": bool, "count": int,
                          "first_break": row_id | None,
                          "reason": str | None,
                          "v1_verified": int, "v2_verified": int,
                          "boundary": row_id | None}}
    tenant_key is the tenant uuid string, or "" for NULL-tenant rows.
    "boundary" is the id of the tenant's first v2 row when the tenant also has
    verified v1 rows before it — the one place the format changes.

    Raises ValueError (bad input, not tampering) when a row is missing a field
    the format needs: the v1 REQUIRED_FIELDS (present and non-null), the KEY
    of every one of the ten v1 hashed columns (CHAIN_V1_COLUMNS; the value may
    be null), or, for a chain_version 2 row, any of the fourteen v2 columns. A
    hashed column absent from the export would otherwise hash as NULL and read
    as a clean row.
    """
    for row in rows:
        missing = [f for f in REQUIRED_FIELDS if row.get(f) is None]
        if missing:
            raise ValueError(f"row {row.get('id')!r} missing fields: {missing}")
        # Checked for EVERY row before any chain is walked, so a missing
        # nullable column is reported even on a row that sits after a break.
        missing = _missing_v1_columns(row)
        if missing:
            raise ValueError(
                f"row {row.get('id')!r} has no {missing[0]!r} key in the export; "
                f"a missing hashed column would hash as NULL (missing: {missing})"
            )
        if chain_version(row) == 2:
            missing = [c for c in CHAIN_V2_COLUMNS if c not in row]
            if missing:
                raise ValueError(
                    f"row {row.get('id')!r} is chain_version 2 but the export has "
                    f"no key for: {missing}"
                )

    tenants: dict[str, list[dict]] = {}
    for row in rows:
        tenants.setdefault(row.get("tenant_id") or "", []).append(row)

    results: dict[str, dict] = {}
    for tenant_key, tenant_rows in sorted(tenants.items()):
        ordered = sorted(
            tenant_rows,
            key=lambda r: (parse_timestamp(r["created_at"]), uuid.UUID(str(r["id"]))),
        )
        prev = genesis_hash(tenant_key or None)
        result = {"ok": True, "count": len(ordered), "first_break": None,
                  "reason": None, "v1_verified": 0, "v2_verified": 0,
                  "boundary": None}
        seen_v2 = False
        for row in ordered:
            version = chain_version(row)
            if version is None:
                result.update(
                    ok=False, first_break=str(row["id"]),
                    reason=(f"unknown chain_version {row.get('chain_version')!r} "
                            f"— this script verifies formats 1 and 2 only and "
                            f"does not guess at a row's format"),
                )
                break
            if version == 1 and seen_v2:
                result.update(
                    ok=False, first_break=str(row["id"]),
                    reason=("chain version downgrade: a v1 row follows a v2 row in "
                            "this tenant's chain, which would drop the fourteen "
                            "columns v2 hashes"),
                )
                break
            if row["prev_event_hash"] != prev:
                result.update(
                    ok=False, first_break=str(row["id"]),
                    reason=(f"prev_event_hash mismatch: stored "
                            f"{row['prev_event_hash'][:16]}…, chain expects {prev[:16]}…"),
                )
                break
            recomputed = compute_event_hash(prev, row)
            if row["event_hash"] != recomputed:
                result.update(
                    ok=False, first_break=str(row["id"]),
                    reason=(f"event_hash mismatch under format v{version}: stored "
                            f"{row['event_hash'][:16]}…, recomputed "
                            f"{recomputed[:16]}… (row fields altered)"),
                )
                break
            if version == 2:
                if not seen_v2 and result["v1_verified"]:
                    result["boundary"] = str(row["id"])
                seen_v2 = True
                result["v2_verified"] += 1
            else:
                result["v1_verified"] += 1
            prev = row["event_hash"]
        results[tenant_key] = result
    return results


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("export", help="JSON export of events rows")
    parser.add_argument("--since", default=None,
                        help="only report on rows at/after this timestamp "
                             "(chain math still starts at each tenant's first exported row)")
    args = parser.parse_args(argv)

    try:
        rows = load_rows(args.export)
    except (OSError, ValueError, json.JSONDecodeError) as exc:
        print(f"error: cannot load export: {exc}", file=sys.stderr)
        return 2

    if not rows:
        # An empty export certifies nothing, so it is never a success: a wrong
        # SELECT or an emptied file must not read as a clean chain (row 689eeaa1).
        print("error: the export has no rows — nothing was verified",
              file=sys.stderr)
        return 2

    try:
        results = verify_chain(rows)
    except ValueError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    since_dt = parse_timestamp(args.since) if args.since else None
    exit_code = 0
    v1_total = v2_total = 0
    for tenant_key, res in results.items():
        label = tenant_key or "<no tenant>"
        if since_dt is not None:
            tenant_rows = [r for r in rows if (r.get("tenant_id") or "") == tenant_key
                           and parse_timestamp(r["created_at"]) >= since_dt]
            if not tenant_rows and res["ok"]:
                continue
        v1_total += res["v1_verified"]
        v2_total += res["v2_verified"]
        versions = f"{res['v1_verified']} v1, {res['v2_verified']} v2"
        boundary = (f"; v2 begins at event id {res['boundary']}"
                    if res["boundary"] else "")
        if res["ok"]:
            print(f"tenant {label}: ok ({res['count']} events, chain verified "
                  f"from genesis; {versions}{boundary})")
        else:
            exit_code = 1
            print(f"tenant {label}: FIRST BREAK at event id {res['first_break']} "
                  f"— {res['reason']} ({res['count']} events checked; "
                  f"{versions} verified before the break{boundary})")

    print(f"verified {v1_total} row{'' if v1_total == 1 else 's'} under format "
          f"v1 and {v2_total} under v2.")
    if not any("chain_version" in row for row in rows):
        print("This export carries no chain_version column, so every row was "
              "read as format v1.")
    print("An ok verifies, for a v1 row, the ten chained fields (id, tenant_id, "
          "agent_id, action, target, verdict, policy, injected, source, "
          "created_at) and nothing else about that row; for a v2 row, every "
          "events column except raw_request.")
    print("Ceiling, unchanged: there is no external anchor. A writer who can "
          "rewrite a row and every hash after it — or drop the trigger — is "
          "not detected here, in either format.")
    return exit_code


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