#!/usr/bin/env python3
# Copyright 2026 SZL Holdings - SPDX-License-Identifier: Apache-2.0
"""Verify installed A11oy signing authority against the pinned runtime key.

Founder decision 2026-09-30: this module is the supported consumer of
installed credential authority for the hf-sync manual-prerequisite gate.

What it measures (all public, no secret value is read, printed or stored):

* ``{origin}/cosign.pub`` is fetched without redirects, bounded to 4 KiB and
  parsed as an ECDSA P-256 SubjectPublicKeyInfo. The runtime serves the public
  half derived from the private key its signer loaded (``a11oy_signing_key``:
  ``SZL_COSIGN_PRIVATE_PEM`` first, then configured persistent sources).
* The served key's DER SHA-256 must equal BOTH independent pins: the committed
  key ``ayllu/keys/council-runtime-2026-07-21.pub`` and the hardcoded
  fingerprint below. If the two pins disagree the result fails closed.
* The served key must NOT equal the static embedded fallback
  ``szl_dsse.COSIGN_PUBLIC_PEM`` (read from source, never imported); serving
  it means no private key is loaded (``SIGNING_KEY_NOT_INSTALLED``).
* ``GDW_CREDENTIALS_JSON`` must be present by name. Its live authority remains
  proven downstream by ``scripts/prove_hf_gdw_runtime.py``.
* The GitHub public reader is OPTIONAL: absent is honestly labelled
  ``PUBLIC_ANONYMOUS``; present by name is ``INSTALLED_NAME_ONLY``.

What it does NOT prove: no fresh signature is requested (the signing route is
operator-gated), so a hostile runtime that copied the public key could pass.
That residual is bounded by the exact-source, single-writer and live-proof
guards of hf-sync, not by this module.

Network access goes through an injectable ``get(url) -> (status, headers,
body)`` so tests are offline. ``huggingface_hub`` is never imported here.
"""

from __future__ import annotations

import ast
import hashlib
import json
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Iterable, Mapping

REPO_ROOT = Path(__file__).resolve().parent.parent
SCHEMA = "szl.hf-installed-authority/v1"
CANONICAL_ORIGIN = "https://szlholdings-a11oy.hf.space"
PINNED_SIGNING_PUBLIC_KEY_PATH = "ayllu/keys/council-runtime-2026-07-21.pub"
# Second, independent pin: DER SHA-256 of the pinned SubjectPublicKeyInfo.
PINNED_SIGNING_KEY_DER_SHA256 = "8e2d106c6995e11dbf7cbedfa9e5800bb50c82a635756e40dcd330364f6ea8ba"
STATIC_FALLBACK_SOURCE = "szl_dsse.py"
STATIC_FALLBACK_NAME = "COSIGN_PUBLIC_PEM"
PUBLIC_KEY_ROUTE = "/cosign.pub"
HONEST_ROUTE = "/api/a11oy/v1/honest"
MAX_PUBLIC_KEY_BYTES = 4 * 1024
MAX_HONEST_BYTES = 64 * 1024
MAX_PIN_FILE_BYTES = 4 * 1024
REQUEST_TIMEOUT_SECONDS = 15

SIGNING_SECRET = "SZL_COSIGN_PRIVATE_PEM"
GDW_SECRET = "GDW_CREDENTIALS_JSON"
GITHUB_PUBLIC_READ_SECRET = "A11OY_GITHUB_PUBLIC_READ_TOKEN"
REQUIRED_SECRET_NAMES = (SIGNING_SECRET, GDW_SECRET)
OPTIONAL_SECRET_NAMES = (GITHUB_PUBLIC_READ_SECRET,)

VERIFIED_PINNED_RUNTIME_KEY = "VERIFIED_PINNED_RUNTIME_KEY"
SIGNING_KEY_NOT_INSTALLED = "SIGNING_KEY_NOT_INSTALLED"
SIGNING_KEY_MISMATCH = "SIGNING_KEY_MISMATCH"
ORIGIN_UNAVAILABLE = "ORIGIN_UNAVAILABLE"
SIGNING_STATES = (VERIFIED_PINNED_RUNTIME_KEY, SIGNING_KEY_NOT_INSTALLED,
                  SIGNING_KEY_MISMATCH, ORIGIN_UNAVAILABLE)
GDW_NAME_PRESENT = "NAME_PRESENT_RUNTIME_PROVEN_DOWNSTREAM"
GDW_CREDENTIALS_MISSING = "GDW_CREDENTIALS_MISSING"
READER_PUBLIC_ANONYMOUS = "PUBLIC_ANONYMOUS"
READER_INSTALLED_NAME_ONLY = "INSTALLED_NAME_ONLY"

# Fixed public diagnostics. Nothing from a provider, response body or
# exception text is ever copied into a report.
DIAGNOSTIC_CODES = {
    "INSTALLED_AUTHORITY_VERIFIED", "INSTALLED_AUTHORITY_UNKNOWN",
    "AUTHORITY_ORIGIN_NOT_CANONICAL", "AUTHORITY_ORIGIN_UNAVAILABLE",
    "SIGNING_KEY_NOT_INSTALLED", "SIGNING_KEY_MISMATCH", "PINNED_KEY_INCONSISTENT",
    "SIGNING_SECRET_MISSING", "GDW_CREDENTIALS_MISSING", "PUBLIC_VARIABLE_COLLISION",
}

Getter = Callable[[str], tuple[int, Mapping[str, str], bytes]]


class _NoRedirect(Exception):
    pass


def default_get(url: str, *, max_bytes: int = MAX_PUBLIC_KEY_BYTES) -> tuple[int, Mapping[str, str], bytes]:
    """Unauthenticated GET, no redirects, bounded body. Never sends credentials."""
    from urllib import request
    from urllib.error import HTTPError

    class Handler(request.HTTPRedirectHandler):
        def redirect_request(self, *_args, **_kwargs):  # noqa: D401 - urllib hook
            return None  # a 3xx surfaces as HTTPError and is rejected by the caller

    opener = request.build_opener(Handler)
    req = request.Request(url, method="GET", headers={
        "Accept": "text/plain, application/json",
        "User-Agent": "SZL-Installed-Authority-Verifier/1.0",
        "Cache-Control": "no-cache",
    })
    try:
        with opener.open(req, timeout=REQUEST_TIMEOUT_SECONDS) as response:
            body = response.read(max_bytes + 1)
            return int(response.status), dict(response.headers.items()), body
    except HTTPError as error:
        return int(error.code), dict(error.headers.items()) if error.headers else {}, b""


def _utc_now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _der_sha256_of_pem(pem: bytes) -> str:
    """Parse a PUBLIC ECDSA P-256 SPKI PEM and return its DER SHA-256."""
    from cryptography.hazmat.primitives import serialization
    from cryptography.hazmat.primitives.asymmetric import ec

    key = serialization.load_pem_public_key(pem)
    if not isinstance(key, ec.EllipticCurvePublicKey) or getattr(key.curve, "name", "") != "secp256r1":
        raise ValueError("public key is not ECDSA P-256")
    der = key.public_bytes(serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo)
    return hashlib.sha256(der).hexdigest()


def pinned_fingerprint(pinned_pem_path: Path | str | None = None) -> str:
    """Return the pinned DER SHA-256 iff the pin file and hardcoded pin agree."""
    path = Path(pinned_pem_path) if pinned_pem_path is not None else REPO_ROOT / PINNED_SIGNING_PUBLIC_KEY_PATH
    with path.open("rb") as stream:
        raw = stream.read(MAX_PIN_FILE_BYTES + 1)
    if len(raw) > MAX_PIN_FILE_BYTES:
        raise ValueError("pinned key file is oversized")
    from_file = _der_sha256_of_pem(raw)
    if from_file != PINNED_SIGNING_KEY_DER_SHA256:
        raise ValueError("pinned key file disagrees with the hardcoded fingerprint")
    return from_file


def static_fallback_fingerprint(source_path: Path | str | None = None) -> str:
    """DER SHA-256 of ``szl_dsse.COSIGN_PUBLIC_PEM``, read statically (never imported)."""
    path = Path(source_path) if source_path is not None else REPO_ROOT / STATIC_FALLBACK_SOURCE
    tree = ast.parse(path.read_text(encoding="utf-8"))
    values = [node.value.value for node in tree.body
              if isinstance(node, ast.Assign) and len(node.targets) == 1
              and isinstance(node.targets[0], ast.Name) and node.targets[0].id == STATIC_FALLBACK_NAME
              and isinstance(node.value, ast.Constant) and type(node.value.value) is str]
    if len(values) != 1:
        raise ValueError("static fallback public key is not uniquely defined")
    return _der_sha256_of_pem(values[0].strip().encode("ascii"))


def _canonical_origin(origin: str) -> str | None:
    value = (origin or "").strip().rstrip("/")
    return value if value == CANONICAL_ORIGIN else None


def verify_signing_authority(
    origin: str,
    get: Getter | None = None,
    pinned_pem_path: Path | str | None = None,
    *,
    static_source_path: Path | str | None = None,
) -> dict[str, Any]:
    """Compare the live runtime public key with both pins; never reads a secret."""
    result: dict[str, Any] = {
        "state": ORIGIN_UNAVAILABLE,
        "diagnostic_code": "AUTHORITY_ORIGIN_UNAVAILABLE",
        "route": PUBLIC_KEY_ROUTE,
        "pinned_key_path": PINNED_SIGNING_PUBLIC_KEY_PATH,
        "pinned_fingerprint_sha256": PINNED_SIGNING_KEY_DER_SHA256,
        "served_fingerprint_sha256": None,
        "fresh_signature_verified": False,
    }
    try:
        pinned = pinned_fingerprint(pinned_pem_path)
        static = static_fallback_fingerprint(static_source_path)
    except Exception:
        result.update(state=SIGNING_KEY_MISMATCH, diagnostic_code="PINNED_KEY_INCONSISTENT")
        return result
    if pinned == static:
        # A pin equal to the static fallback could never distinguish an
        # installed key from an absent one.
        result.update(state=SIGNING_KEY_MISMATCH, diagnostic_code="PINNED_KEY_INCONSISTENT")
        return result
    base = _canonical_origin(origin)
    if base is None:
        result["diagnostic_code"] = "AUTHORITY_ORIGIN_NOT_CANONICAL"
        return result
    get = get or default_get
    try:
        status, _headers, body = get(base + PUBLIC_KEY_ROUTE)
    except Exception:
        return result
    if type(status) is not int or status != 200 or not isinstance(body, (bytes, bytearray)):
        return result
    if len(body) > MAX_PUBLIC_KEY_BYTES or not body:
        return result
    try:
        served = _der_sha256_of_pem(bytes(body))
    except Exception:
        result.update(state=SIGNING_KEY_MISMATCH, diagnostic_code="SIGNING_KEY_MISMATCH")
        return result
    result["served_fingerprint_sha256"] = served
    if served == static:
        result.update(state=SIGNING_KEY_NOT_INSTALLED, diagnostic_code="SIGNING_KEY_NOT_INSTALLED")
    elif served == pinned == PINNED_SIGNING_KEY_DER_SHA256:
        result.update(state=VERIFIED_PINNED_RUNTIME_KEY, diagnostic_code="INSTALLED_AUTHORITY_VERIFIED")
    else:
        result.update(state=SIGNING_KEY_MISMATCH, diagnostic_code="SIGNING_KEY_MISMATCH")
    return result


def verify_gdw_authority(secret_names: Iterable[str]) -> dict[str, Any]:
    """Pre-deploy admission needs only the name; live proof stays downstream."""
    present = GDW_SECRET in set(secret_names)
    return {
        "state": GDW_NAME_PRESENT if present else GDW_CREDENTIALS_MISSING,
        "secret_name": GDW_SECRET,
        "blocking": not present,
        "live_proof": "scripts/prove_hf_gdw_runtime.py (post-deploy)",
    }


def github_public_reader_state(secret_names: Iterable[str]) -> dict[str, Any]:
    """Optional reader (founder decision): reported, never required."""
    present = GITHUB_PUBLIC_READ_SECRET in set(secret_names)
    return {
        "state": READER_INSTALLED_NAME_ONLY if present else READER_PUBLIC_ANONYMOUS,
        "secret_name": GITHUB_PUBLIC_READ_SECRET,
        "blocking": False,
        "value_verified": False,
    }


def read_live_git_sha(origin: str, get: Getter | None = None) -> str | None:
    """Informational: the runtime's self-reported source revision (public)."""
    base = _canonical_origin(origin)
    if base is None:
        return None
    try:
        status, _headers, body = (get or (lambda url: default_get(url, max_bytes=MAX_HONEST_BYTES)))(base + HONEST_ROUTE)
        if status != 200 or not isinstance(body, (bytes, bytearray)) or len(body) > MAX_HONEST_BYTES:
            return None
        value = json.loads(bytes(body).decode("utf-8")).get("git_sha")
    except Exception:
        return None
    return value if type(value) is str and re.fullmatch(r"[0-9a-f]{40}", value) else None


def verify_installed_authority(
    secret_names: Iterable[str],
    public_variable_names: Iterable[str] = (),
    *,
    origin: str = CANONICAL_ORIGIN,
    get: Getter | None = None,
    pinned_pem_path: Path | str | None = None,
    static_source_path: Path | str | None = None,
) -> dict[str, Any]:
    """Combine signing, GDW and reader evidence into one secret-free report."""
    names = {name for name in secret_names if type(name) is str}
    public_names = {name for name in public_variable_names if type(name) is str}
    started = _utc_now()
    signing = verify_signing_authority(origin, get, pinned_pem_path, static_source_path=static_source_path)
    gdw = verify_gdw_authority(names)
    reader = github_public_reader_state(names)
    collision = bool(set(REQUIRED_SECRET_NAMES + OPTIONAL_SECRET_NAMES) & public_names)
    signing_secret_present = SIGNING_SECRET in names

    if collision:
        state, diagnostic = "BLOCKED", "PUBLIC_VARIABLE_COLLISION"
    elif not signing_secret_present:
        state, diagnostic = "BLOCKED", "SIGNING_SECRET_MISSING"
    elif signing["state"] == ORIGIN_UNAVAILABLE:
        state, diagnostic = "UNKNOWN", signing["diagnostic_code"]
    elif signing["state"] != VERIFIED_PINNED_RUNTIME_KEY:
        state, diagnostic = "BLOCKED", signing["diagnostic_code"]
    elif gdw["blocking"]:
        state, diagnostic = "BLOCKED", "GDW_CREDENTIALS_MISSING"
    else:
        state, diagnostic = "VERIFIED", "INSTALLED_AUTHORITY_VERIFIED"
    if diagnostic not in DIAGNOSTIC_CODES:
        state, diagnostic = "UNKNOWN", "INSTALLED_AUTHORITY_UNKNOWN"

    return {
        "schema": SCHEMA,
        "origin": _canonical_origin(origin) or "NOT_CANONICAL",
        "observed_at_start": started,
        "observed_at_end": _utc_now(),
        "live_git_sha": read_live_git_sha(origin, get),
        "credential_authority_state": state,
        "diagnostic_code": diagnostic,
        "signing": signing,
        "signing_secret_name_present": signing_secret_present,
        "gdw": gdw,
        "github_public_reader": reader,
        "public_variable_collision": collision,
        "evidence_class": "measured-public-key-match",
        "proves": "the runtime's loaded signer derives the pinned public key",
        "does_not_prove": "fresh proof-of-possession (no signature requested)",
        "secret_values_read": False,
        "secret_values_written": False,
    }


def main(argv: list[str] | None = None) -> int:
    """Public-only CLI: checks the live key match (no provider metadata, no secrets)."""
    import argparse
    import os

    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--origin", default=os.environ.get("CANONICAL_ORIGIN", CANONICAL_ORIGIN))
    args = parser.parse_args(argv)
    report = verify_signing_authority(args.origin)
    print(json.dumps(report, indent=2, sort_keys=True))
    return 0 if report["state"] == VERIFIED_PINNED_RUNTIME_KEY else 1


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