#!/usr/bin/env python3
"""Fetch every route this app declares with no cookie, and say what the gate answered.

Sources of truth: `app/services/access.py` (the policy the expectations are derived from),
`app/services/route_walk.py` (the walk), `tests/test_route_sweep.py`, `docs/RUNBOOK.md`
step 18, `docs/ACCOUNTS.md`.

The inventory test proves every route is in a class. It cannot prove the deployed server
behaves like the class, because a reverse proxy, a stale image or a mount added at startup
sits between the two. This walks the same list over HTTP, signed out, and prints one line
per route: what class it is in, what `access.decide()` says an anonymous caller must get,
and what the server actually gave. An internal route answering 200 to nobody is the finding
this exists to catch, and it exits non-zero so a launch step can depend on it.

    ../.venv-dev/bin/python scripts/route-sweep.py https://host          # members mode
    ../.venv-dev/bin/python scripts/route-sweep.py https://host --open   # go-live day
    python3 scripts/route-sweep.py --dry-run                             # the table, no network

`--open` states the expectations for `SITE_ACCESS=public`: the storefront answers 200 and
only the member and permission routes still refuse. The environment (`SITE_ROLE`) is read
from `/api/health` and can be forced with `--live` or `--staging`; on live a development
surface must answer 404, and a 302 to a sign-in page there is itself a finding.

Reads only. Every request is a GET with no cookie and no credential; a route whose method
is not GET is listed and skipped, because a sweep that posts is not a sweep. The one write
it could ever make is none: `agents.md`, "A GET must never write".

Exit: 0 all as expected, 1 some answer differed without disclosing anything, 2 an internal
route disclosed itself (200, or a redirect to somewhere that is not the sign-in page).
"""

from __future__ import annotations

import argparse
import json
import os
import pathlib
import sys
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass, field

MAIN = pathlib.Path(__file__).resolve().parents[1]
ROOT = MAIN.parent
sys.path.insert(0, str(MAIN))

try:
    import fastapi  # noqa: F401
except ImportError:  # re-exec under the venv check.sh uses
    venv_py = ROOT / ".venv-dev" / "bin" / "python"
    if venv_py.exists() and not os.environ.get("ROUTE_SWEEP_REEXEC"):
        os.environ["ROUTE_SWEEP_REEXEC"] = "1"
        os.execv(str(venv_py), [str(venv_py), *sys.argv])
    raise

from app.services import access  # noqa: E402
from app.services.route_walk import every_key, sample_path, spa_paths  # noqa: E402

USER_AGENT = "dfp-route-sweep (an unauthenticated check of our own site)"

OK = "ok"
LEAK = "LEAK"
MISMATCH = "MISMATCH"
GATE = "gate"
ERROR = "error"
SKIPPED = "skipped"
PENDING = "-"

#: A refusal the site gate (id-auth) gave instead of the app: the app's own answer is
#: unknown, so it is neither a pass nor a leak, and the report has to say which host spoke.
GATE_HOSTS = ("auth.bowden.works",)


@dataclass
class Row:
    key: str
    page: str | None
    path: str
    cls: str | None
    permission: str | None
    expected: int | None          # None: the gate lets it through, the handler answers
    status: int | None = None
    location: str | None = None
    headers: dict[str, str] = field(default_factory=dict)
    verdict: str = PENDING
    note: str = ""

    @property
    def expected_text(self) -> str:
        if self.verdict == SKIPPED:
            return "-"
        if self.expected is None:
            return "handler"
        if self.expected == 302:
            return "302 /login"
        return str(self.expected)


def build_app():
    """The real app with the built shell mounted if there is one, so the mounts exist."""
    from app import main as app_main

    static = MAIN / "web" / "dist"
    already = {getattr(r, "path", None) for r in app_main.app.routes}
    if "/assets" not in already and static.is_dir() and (static / "index.html").exists():
        try:
            app_main.mount_site(app_main.app, static)
        except Exception:  # a shell that will not mount is not this script's problem
            pass
    return app_main.app


def expected_status(key: str, page: str | None, *, site_open: bool, live: bool) -> int | None:
    """What `access.decide()` gives an anonymous caller; None means the handler answers.

    Derived, never typed: the script cannot disagree with the policy it is checking.
    """
    verdict = access.decide(
        method="GET", key=key, page=page, site_open=site_open,
        signed_in=False, active=False, must_change=False,
        holds=lambda _permission: False, read_only=False, origin_ok=True,
        wants_html=access.wants_html(key), live=live,
    )
    return None if verdict is None else verdict.status


def policy_keys() -> set[str]:
    """Every key the policy names. Some routes exist only where their directory is mounted
    (the explainer pages live under a container path), so a walk of the app on a host is
    short of what the deployed app declares; the union of the two is the honest list, and
    the inventory test is what keeps the policy's own names from going stale."""
    return (set(access.PUBLIC_ALWAYS) | set(access.PUBLIC_WHEN_OPEN) | set(access.MEMBER)
            | set(access.PERMISSION))


def plan(app, *, site_open: bool, live: bool) -> list[Row]:
    """One row per thing to fetch: every route key, plus every SPA path the catch-all serves."""
    rows: list[Row] = []
    for key in sorted(every_key(app) | policy_keys()):
        method, _, template = key.partition(" ")
        if key == access.CATCH_ALL:
            continue  # covered path by path below
        cls, permission = access.classify(key, None)
        if method == "MOUNT":
            path = template.rstrip("/") + "/"
        elif method != "GET":
            rows.append(Row(key=key, page=None, path=sample_path(template), cls=cls,
                            permission=permission, expected=None, verdict=SKIPPED,
                            note="not a read; a sweep never posts"))
            continue
        else:
            path = sample_path(template)
        rows.append(Row(key=key, page=None, path=path, cls=cls, permission=permission,
                        expected=expected_status(key, None, site_open=site_open, live=live)))

    for template in sorted(spa_paths()):
        page = sample_path(template)
        cls, permission = access.classify(access.CATCH_ALL, page)
        rows.append(Row(key=f"{access.CATCH_ALL} -> {template}", page=page, path=page, cls=cls,
                        permission=permission,
                        expected=expected_status(access.CATCH_ALL, page,
                                                 site_open=site_open, live=live)))
    return rows


def fetch(base: str, path: str, timeout: float) -> tuple[int | None, dict[str, str], str]:
    """A GET with no cookie, redirects not followed. Returns status, headers, any error."""
    url = urllib.parse.urljoin(base + "/", path.lstrip("/"))
    request = urllib.request.Request(url, method="GET", headers={
        "User-Agent": USER_AGENT, "Accept": "*/*", "Cache-Control": "no-cache",
    })
    opener = urllib.request.build_opener(_NoRedirect)
    try:
        with opener.open(request, timeout=timeout) as answer:
            return answer.status, dict(answer.headers), ""
    except urllib.error.HTTPError as exc:
        headers = dict(exc.headers or {})
        exc.close()
        return exc.code, headers, ""
    except Exception as exc:  # a timeout or a DNS failure is a result, not a crash
        return None, {}, f"{type(exc).__name__}: {exc}"


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def is_login_location(location: str | None) -> bool:
    """A redirect that sends an anonymous caller to this site's own sign-in page."""
    if not location:
        return False
    parts = urllib.parse.urlsplit(location)
    if parts.netloc:
        return False
    return (parts.path or "/").rstrip("/") in ("", access.LOGIN_PATH, access.ACCOUNT_PATH)


def gate_host(location: str | None) -> str | None:
    """The site gate's own host, when the redirect left this site entirely."""
    if not location:
        return None
    host = urllib.parse.urlsplit(location).netloc.split(":")[0]
    return host if host in GATE_HOSTS else None


def judge(row: Row) -> None:
    """The verdict for one answered row: ok, MISMATCH, LEAK, gate or error."""
    if row.verdict == SKIPPED:
        return
    if row.status is None:
        row.verdict = ERROR
        return
    host = gate_host(row.location)
    if host:
        row.verdict = GATE
        row.note = f"the site gate answered ({host}), not the app"
        return
    internal = row.cls != "public_always"
    if row.expected is None:
        # public_always: the handler answers. Anything that refuses an anonymous caller here
        # is the gate speaking where it should not, which is a finding of its own.
        if row.status == 401 or (300 <= row.status < 400 and is_login_location(row.location)):
            row.verdict = MISMATCH
            row.note = "a route that is public in every mode asked for a sign-in"
        else:
            row.verdict = OK
        return
    if row.status == row.expected:
        if row.status == 302 and not is_login_location(row.location):
            row.verdict = LEAK if internal else MISMATCH
            row.note = f"302 to {row.location!r}, which is not the sign-in page"
        else:
            row.verdict = OK
        return
    if internal and (row.status == 200 or (300 <= row.status < 400
                                           and not is_login_location(row.location))):
        row.verdict = LEAK
        row.note = f"answered {row.status} to a caller with no session"
        return
    row.verdict = MISMATCH


def read_role(base: str, timeout: float) -> str | None:
    """`/api/health` prints the environment since the environment line landed (AW6.2)."""
    url = urllib.parse.urljoin(base + "/", "api/health")
    try:
        with urllib.request.urlopen(
            urllib.request.Request(url, headers={"User-Agent": USER_AGENT}), timeout=timeout
        ) as answer:
            return json.loads(answer.read().decode()).get("role")
    except Exception:
        return None


def render(rows: list[Row], *, show_skipped: bool) -> str:
    width = max((len(r.key) for r in rows), default=20)
    lines = []
    for row in rows:
        if row.verdict == SKIPPED and not show_skipped:
            continue
        status = "-" if row.status is None else str(row.status)
        line = (f"{row.verdict:<8} {row.key:<{width}}  {row.cls or 'UNCLASSIFIED':<18} "
                f"got {status:<4} want {row.expected_text}")
        if row.note:
            line += f"   [{row.note}]"
        lines.append(line)
    return "\n".join(lines)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("base_url", nargs="?", help="https://host of the site to sweep")
    parser.add_argument("--open", action="store_true",
                        help="expect SITE_ACCESS=public (go-live day), not members mode")
    parser.add_argument("--live", dest="live", action="store_true", default=None,
                        help="expect SITE_ROLE=live (development surfaces answer 404)")
    parser.add_argument("--staging", dest="live", action="store_false",
                        help="expect SITE_ROLE=staging")
    parser.add_argument("--dry-run", action="store_true",
                        help="print the expectation table and make no request")
    parser.add_argument("--json", action="store_true", help="one JSON object per row")
    parser.add_argument("--show-skipped", action="store_true",
                        help="list the routes whose method is not GET")
    parser.add_argument("--timeout", type=float, default=15.0)
    args = parser.parse_args(argv)

    if not args.dry_run and not args.base_url:
        parser.error("a base URL is required unless --dry-run")

    live = args.live
    if live is None:
        role = read_role(args.base_url, args.timeout) if args.base_url else None
        live = role == "live"
        if args.base_url and role is None:
            print("note: /api/health named no role; assuming staging (--live to force)",
                  file=sys.stderr)

    rows = plan(build_app(), site_open=args.open, live=live)
    if not args.dry_run:
        for row in rows:
            if row.verdict == SKIPPED:
                continue
            row.status, row.headers, error = fetch(args.base_url, row.path, args.timeout)
            row.location = row.headers.get("Location") or row.headers.get("location")
            if error:
                row.note = error
            judge(row)

    if args.json:
        for row in rows:
            print(json.dumps({
                "key": row.key, "path": row.path, "class": row.cls,
                "permission": row.permission, "expected": row.expected_text,
                "status": row.status, "location": row.location, "verdict": row.verdict,
                "note": row.note,
                "via": row.headers.get("Via") or row.headers.get("via"),
                "cache_control": row.headers.get("Cache-Control") or row.headers.get("cache-control"),
            }))
    else:
        print(render(rows, show_skipped=args.dry_run or args.show_skipped))

    counts = {v: sum(1 for r in rows if r.verdict == v)
              for v in (OK, MISMATCH, LEAK, GATE, ERROR, SKIPPED)}
    mode = "public" if args.open else "members"
    where = "live" if live else "staging"
    print(f"route-sweep: {len(rows)} routes, {mode} mode, {where} role; "
          + ", ".join(f"{n} {v}" for v, n in counts.items() if n))
    if args.dry_run:
        return 0
    if counts[LEAK]:
        return 2
    return 1 if counts[MISMATCH] or counts[ERROR] else 0


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