diff --git a/CHANGELOG.md b/CHANGELOG.md index 9ac5c26..bd9b874 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,9 @@ Entries are newest-last within a release, matching the order they were written. - **the one edge-declaration path that still deferred its error.** `add_conditional_edge` passed the router and its mapping straight through to LangGraph, so a mapping pointing at a node nobody added was accepted, an empty mapping was accepted, and the first run to take that branch died on `self.ends[key]` — a bare `KeyError` raised from inside LangGraph's branch machinery, naming neither the graph, the source node, nor the router that produced the key. Everywhere else this kernel fails at declaration: an undeclared write raises at `add_node`, a write to a field the schema does not have raises at `add_node`, a cycle is refused at `compile()`. The mapping's targets were knowable all along. They are checked now, at `add_conditional_edge`, with an empty mapping refused and every unreachable target named alongside the key that leads to it; a router that annotates what it returns — a `Literal`, an `Enum` — has those members held against the mapping's keys, using the same hash lookup LangGraph will use, so the check predicts the failure rather than approximating it. A router that annotates nothing is still not second-guessed: predicting an arbitrary function's return value is not a check, and inventing a requirement would be worse than the gap. That last case is no longer a `KeyError`, though — the router is wrapped so an unmapped key raises `GraphRoutingError` naming the node, the key and the keys that were declared, which is what the rest of the kernel raises for a transition it cannot make. The wrapper keeps the router's name and annotations, because LangGraph names the branch after the one and infers the branch's input schema from the other. - a **reused `--run-id` silently welded two runs into one record.** Every executing command appends to its `--trace` file — by design, since `grapharc diff` reads two runs out of one file — and nothing checked whether the id the operator passed was already in there. Running the same `plan` twice with one `--trace`/`--run-id` pair produced a single "run" whose `metrics` summed both runs' tokens and node counts, whose `viz` drew the second path welded onto the end of the first, and whose `replay` reconstructed a chimera; the operator got no signal at any point, and the trace is documented as the record the metrics cannot disagree with. The file being appendable was never the defect — the id being reused was, so the guard sits at the start of the run rather than in the recorder: `plan`, `run` and `agent` (both executors) refuse an explicit `--run-id` that already has events in the target trace, with exit 2 naming the id, the count and the file, before a single event is written. Fail closed rather than auto-renaming, because a run id is the name an operator will look the run up under later and picking a different one silently is the same class of surprise. Generated ids are untouched — fresh by construction, so they pay for no scan — and different ids in one file stay exactly as they were. - the planner's system prompt **withheld the edge policy**, so a model had to learn it one refusal at a time. The prompt states the catalog, the START/END literals and the structural rules, and its own comments say why — "stating the rule up front is cheaper than three wasted rounds" — but the rule models actually trip over was the one it never stated. Observed with qwen3:8b against the incident registry: the goal said "find the cause and propose a fix", the policy denied `*->deploy`, and the planner proposed an edge into `deploy` in all three rounds (`edge_denied`; `edge_denied` + `cycle`; `edge_denied`) until the loop stopped `admission_refused` — about 3.5 minutes of local inference spent discovering one sentence, and a run that reads as a model failure when it is an information failure. The refusal came back every round and `edge_denied` names the check, not the rule, so "no edge may enter `deploy`, ever" was never on the page. `EdgePolicy.disclosure()` and `NodePolicy.disclosure()` now render a policy's deny rules as one line each (`edges into 'deploy' are denied by policy — do not propose them`), `PlannerNode(edge_policy=…, node_policy=…)` puts them directly under the catalog, and the shipped loop builders hand the planner the same policy *object* the checker holds, so the prompt cannot describe a policy the gate is not applying. Allow rules and the default are left out — they say what is permitted, which the catalog already covers — and so is `ask`, whose remedy is an approval rather than a different proposal. The refusal side is enriched to match: `EdgeRule` carries the `reason` `NodeRule` already had, `PolicyEngine.edge_policy()` compiles it out of the document instead of dropping it on the floor, and `policy/edge_denied` quotes it, so a planner reads why and not only what. **None of this is enforcement.** No check consults the disclosure, the admission gate is byte-identical, and a model that ignores what it was told is refused exactly as one that was never told — pinned by a test that compares the rejections of a disclosed and an undisclosed planner field by field, and by the shipped demo, whose scripted round 1 still proposes the denied deploy and is still refused. +- the `/live` **token was accepted in the query string on every route**, and a URL is the one place a secret cannot be taken back from: the uvicorn request line, the nginx access log, browser history, and the referrer of anything the page opens. The index made it worse by writing the token into every link it rendered, so clicking a trace filed the secret in history a second time. It is refused off `/live/api/stream` now — that route keeps it because a browser `EventSource` cannot set a header and has no other way in — with a 401 whose reason says *where* to put the token rather than that it is wrong. A browser gets a sign-in page instead of a bare 401 and trades the token for a cookie: a SHA-256 digest of it rather than the token itself, `HttpOnly`, `SameSite=Strict`, scoped to `/live`, and always ASCII, so a non-ASCII secret survives the latin-1 header encoding that a `Bearer` header cannot. Links carry no token at all. The residual exposure — the SSE request line — is now named in the cookbook next to `--live-token`, with what to scrub. Every confinement the reader already enforced is untouched: `../`, `%2e%2e%2f`, absolute paths, NUL bytes and symlinked traces are the same 404s, and a hostile token is still a 401 rather than a crash. (#41) +- the live page was **blind for the whole planning phase**, which is where a governed run spends its budget and does its refusing. `plan`, `admission` and `round` events were on disk — 2,081 tokens spent before any node ran, in the report — and the page rendered none of them, because it keys the graph off the `topology` event that only lands once a round is admitted and materialised. A run refused on every round produces no topology at all, so the most governance-relevant run there is showed nothing from start to "finished". The snapshot now carries a `planning` block folded from those same events (no new trace events): per round, the proposal size, the admission status, the checks that failed and the rejection codes, the planner tokens, and whether it executed; plus the loop's stop reason and detail when it stopped without a graph. The page renders it as a panel, and a round that has begun and not closed reads as *active* rather than idle — a planner mid-inference writes nothing for a minute at a time, which is exactly the "is it thinking or is it wedged?" the report describes. A run that never planned has no `planning` field and renders exactly as before. (#47) +- a finished trace **rendered as a done deal**: instantly all-green, with the amber `running` styling unreachable for every run that is already over — and for any live run whose nodes finish between two SSE polls. `?replay=1` on the stream walks the recorded events in timestamp order and emits the snapshots the run would have sent, so a node is amber for its recorded window and green after; `&speed=N` divides the wall clock and the whole replay is capped at 40 seconds, so a 40-minute incident trace is watchable. Frames are rebuilt by the same snapshot code a live stream uses, pointed at a prefix of the file, and depend on no clock: a trace replayed twice renders identically. Without the parameter nothing changed. (#48) ## 0.1.3 diff --git a/docs/cookbook/06-serving-and-ops.md b/docs/cookbook/06-serving-and-ops.md index cae9922..07ddaa3 100644 --- a/docs/cookbook/06-serving-and-ops.md +++ b/docs/cookbook/06-serving-and-ops.md @@ -1322,6 +1322,23 @@ status — each time the file grows. The server recomputes the snapshot; the page only renders it. Add `&run=ID` to pin one run in a file that holds several; without it the view follows the newest. +A planner run has no graph for as long as it takes the model to propose one, +so the snapshot also carries a `planning` block — round number, proposal size, +admitted or rejected with the checks that failed, planner tokens spent — and +the page renders it as a panel from the `plan`, `admission` and `round` events +already in the trace. A run refused on every round (`admission_refused`) never +produces a topology at all; it shows its rounds and its stop reason instead of +an empty page. A round that has begun and not closed also counts as activity, +because a planner mid-inference writes nothing for a minute at a time. + +`GET /live/view?trace=REL&replay=1` replays a finished trace instead of +rendering its final state: the recorded events are walked in timestamp order +and each one emits the snapshot a live run would have sent, so nodes go amber +then green in the order and at the pace they really ran. `&speed=N` divides the +wall clock, and a whole replay is capped at 40 seconds however slow the +recording was, so yesterday's 40-minute incident trace is watchable. Without +the parameter nothing changes: one snapshot per file change, as before. + This composes with the Slack bot, which gives every tracing command a trace path under its working directory: run `grapharc serve --live-root` over that same directory, set `GRAPHARC_SLACK_LIVE_URL`, and the bot posts a @@ -1338,7 +1355,26 @@ response; the exposure is what `viz` already prints. The bind stays `127.0.0.1` unless you say otherwise; binding wider prints a warning, because reachability is meant to come from a tunnel or tailnet in front, optionally with `--live-token TOKEN` (or `GRAPHARC_LIVE_TOKEN`) required on every -`/live` request. The diagram renders with mermaid.js from a pinned CDN; with +`/live` request. + +**Where that token is allowed to travel matters.** A URL is copied into places +with much weaker access control than the traces it protects: the uvicorn +request line, an nginx access log, browser history, and the referrer of +anything the page links out to. So the token goes in an +`Authorization: Bearer TOKEN` header, or in the cookie that `POST /live/auth` +sets when you paste it into the sign-in page a browser gets instead of a 401. +`?token=` is accepted on `/live/api/stream` and nowhere else — a browser +`EventSource` cannot set a header, so that one route has no alternative — and +any other `/live` route refuses a query-string token with a 401 that says so +rather than accepting the secret into your logs. The tradeoff that remains: +the SSE request line still carries the token, so if you terminate TLS at nginx +and log request URIs, scrub `token=` from that one path (or log +`$request_method $uri` rather than `$request`). The cookie is a digest of the +token, not the token, is `HttpOnly` and `SameSite=Strict`, and is scoped to +`/live`. Sign-in and the SSE exemption both apply only when a token is +configured at all; without one, nothing about `/live` is authenticated. + +The diagram renders with mermaid.js from a pinned CDN; with no CDN reachable the page falls back to the raw Mermaid source plus the same mermaid.live fragment link the Slack bot posts. diff --git a/docs/cookbook/07-slack.md b/docs/cookbook/07-slack.md index 1e11588..3929bec 100644 --- a/docs/cookbook/07-slack.md +++ b/docs/cookbook/07-slack.md @@ -182,7 +182,9 @@ contents — what the page shows is what `viz` and `metrics` already show. Reachability is deliberately your problem, not the bot's: the bot never opens a port (that is the whole point of Socket Mode), and `serve` still binds loopback by default. Put a tailnet or tunnel (Tailscale, cloudflared) in front -for the person on the phone, and add `--live-token` if the URL is guessable. +for the person on the phone, and add `--live-token` if the URL is guessable — +the person then signs in once on the page rather than carrying the token in +the link, which is what keeps it out of access logs and browser history. Details in [06-serving-and-ops.md](06-serving-and-ops.md). ## A `plan` that reads diff --git a/grapharc/server/live.py b/grapharc/server/live.py index b2d67a2..df79566 100644 --- a/grapharc/server/live.py +++ b/grapharc/server/live.py @@ -18,27 +18,35 @@ Reachability is the operator's problem by design: bind stays loopback unless they choose otherwise, and the recommended remote path is a tunnel or tailnet in front, optionally with the shared-secret `token` (a convenience lock, not a -perimeter — it rides in the query string because `EventSource` cannot set -headers). +perimeter). The token travels in a header, or in the cookie `POST /live/auth` +sets for a browser; `?token=` is accepted on `/live/api/stream` alone, because +`EventSource` cannot set headers and that route has no other way in. Everywhere +else a query-string token is refused outright, with its own reason: a URL is +copied into access logs, browser history and referrers, and a secret that +protects read access to every trace under the root should not accrue copies in +places with weaker access control than the traces. """ from __future__ import annotations import asyncio +import hashlib import html +import math import secrets +from datetime import datetime from pathlib import Path from time import monotonic, time from typing import Any -from urllib.parse import quote +from urllib.parse import quote, urlencode -from fastapi import APIRouter, HTTPException, Query, Request -from fastapi.responses import HTMLResponse, StreamingResponse +from fastapi import APIRouter, HTTPException, Query, Request, Response +from fastapi.responses import HTMLResponse, RedirectResponse, StreamingResponse from pydantic import BaseModel from grapharc.observe.metrics import RunMetrics, summarize, to_mermaid from grapharc.observe.replay import replay -from grapharc.observe.trace import TailRecorder +from grapharc.observe.trace import TailRecorder, TraceEvent from grapharc.slack.format import mermaid_live_url #: How often the stream re-stats the trace file. Coarser than the session @@ -55,6 +63,25 @@ #: "running". A delegated executor is silent mid-flight, so quiet ≠ idle — but #: past this, an open node is a run that died mid-node, not one still working. OPEN_NODE_GRACE_SECONDS = 900.0 +#: The cookie `POST /live/auth` sets, so a browser can authenticate every later +#: navigation without the token ever entering a URL. +LIVE_COOKIE = "grapharc_live_token" +#: Why a token in the query string is refused off the SSE route. Distinct from +#: "missing or wrong token" on purpose: the caller holds the right secret and +#: needs to be told *where* to put it, not that it is wrong. +QUERY_TOKEN_REFUSED = ( + "the token may not travel in the query string on this route — it would be " + "copied into access logs, browser history and referrers. Send it as an " + "`Authorization: Bearer` header, or sign in once to set a cookie." +) +#: A replay never runs longer than this however slow the recording was, so a +#: 40-minute run is watchable. A `speed=` multiplier faster than the cap wins. +REPLAY_MAX_SECONDS = 40.0 +#: An upper bound on frames per replay: each one re-renders the whole prefix, +#: so a 50k-event trace would otherwise cost quadratic work to watch. +REPLAY_MAX_FRAMES = 400 +#: Clamp on `speed=`; past this the replay is instant anyway. +REPLAY_MAX_SPEED = 10_000.0 class LivePathError(Exception): @@ -78,6 +105,39 @@ def resolve_trace(root: Path, raw: str) -> Path: return resolved +class PlanningRound(BaseModel): + """One governed-loop round, as far as the trace has got with it. + + Counts and labels only — the same exposure the rest of the live view has. + """ + + round: int + status: str = "" + nodes: int = 0 + proposals: int = 0 + tokens: int = 0 + failed_checks: list[str] = [] + rejections: list[str] = [] + executed: bool = False + #: The round has `plan`/`admission` events but no closing `round` event: + #: the planner is thinking right now, which is exactly the window the page + #: used to render as "waiting for the run to start…". + in_flight: bool = False + error: str | None = None + + +class PlanningView(BaseModel): + """What the planner did before (or instead of) producing a graph.""" + + rounds: list[PlanningRound] = [] + planner_tokens: int = 0 + stop: str | None = None + stop_detail: str | None = None + #: A graph was admitted and materialised, so the diagram is the main event + #: and this is history. Without it, this *is* the run so far. + has_topology: bool = False + + class LiveSnapshot(BaseModel): """Everything one page render needs; recomputed server-side per change.""" @@ -94,6 +154,94 @@ class LiveSnapshot(BaseModel): active: bool = False done: bool = False awaiting_approval: bool = False + #: None for a run that never planned — an ordinary graph invocation renders + #: exactly as it did before this field existed. + planning: PlanningView | None = None + + +def _as_int(value: Any, fallback: int = 0) -> int: + """A count out of a trace file written by someone else. Never raises.""" + try: + return int(value) + except (TypeError, ValueError): + return fallback + + +def summarize_planning(run_events: list[TraceEvent]) -> PlanningView | None: + """Fold `plan`/`admission`/`round` (and the loop's `stop`) into a panel. + + No new trace events: everything here is already on disk while the page is + showing nothing. A round is opened by the first `plan` or `admission` event + that follows the last closed one, so an in-flight round — the 30-45 seconds + of local inference this exists for — is a row too. + + Returns None when the run has no planning at all, which keeps the field + absent for every non-planner run. + """ + rounds: list[PlanningRound] = [] + view = PlanningView() + pending: PlanningRound | None = None + + def slot() -> PlanningRound: + nonlocal pending + if pending is None: + pending = PlanningRound(round=len(rounds) + 1, in_flight=True) + return pending + + for event in run_events: + delta = event.state_delta or {} + if event.phase == "topology": + view.has_topology = True + elif event.phase == "plan": + entry = slot() + entry.proposals += 1 + entry.nodes = _as_int(delta.get("nodes"), entry.nodes) + entry.tokens += event.tokens or 0 + entry.error = event.error or entry.error + elif event.phase == "admission": + entry = slot() + entry.status = str(delta.get("status") or entry.status) + entry.nodes = _as_int(delta.get("nodes"), entry.nodes) + entry.failed_checks = [str(c) for c in delta.get("failed_checks") or []] + elif event.phase == "round": + entry = slot() + entry.in_flight = False + entry.round = _as_int(delta.get("round"), entry.round) + entry.status = str(delta.get("status") or entry.status) + entry.nodes = _as_int(delta.get("nodes"), entry.nodes) + entry.rejections = [str(r) for r in delta.get("rejections") or []] + entry.executed = bool(delta.get("executed")) + entry.error = event.error or entry.error + rounds.append(entry) + pending = None + elif event.phase == "stop" and "stop" in delta: + # The governed loop's own terminal event — an agent's `stop` carries + # a `termination_reason` instead, and is not a planning outcome. + view.stop = str(delta.get("stop") or "") or None + view.stop_detail = str(delta.get("detail") or "") or None + if pending is not None: + rounds.append(pending) + if not rounds and view.stop is None: + return None + view.rounds = rounds + view.planner_tokens = sum(r.tokens for r in rounds) + return view + + +class _FrozenRecorder(TailRecorder): + """A reader over a fixed slice of events rather than the file's current one. + + `to_mermaid`, `summarize` and `replay` all take a recorder and call + `read_events`, so a replay frame is the ordinary snapshot computation + pointed at a prefix of the trace. + """ + + def __init__(self, path: str | Path, events: list[TraceEvent]) -> None: + super().__init__(path) + self._events = list(events) + + def read_events(self, run_id: str | None = None) -> list[TraceEvent]: + return [e for e in self._events if run_id is None or e.run_id == run_id] def build_snapshot(root: Path, rel: str, run_id: str | None) -> LiveSnapshot: @@ -107,16 +255,34 @@ def build_snapshot(root: Path, rel: str, run_id: str | None) -> LiveSnapshot: events = recorder.read_events() if not events: return LiveSnapshot(trace=rel) + try: + stat = path.stat() + size, quiet_for = stat.st_size, time() - stat.st_mtime + except OSError: + size, quiet_for = 0, float("inf") + return compose_snapshot(rel, recorder, events, run_id, size=size, quiet_for=quiet_for) + +def compose_snapshot( + rel: str, + recorder: TailRecorder, + events: list[TraceEvent], + run_id: str | None, + *, + size: int, + quiet_for: float, +) -> LiveSnapshot: + """The snapshot for one already-read set of events. Pure; run it in a thread. + + Split out of `build_snapshot` so a replay frame — a prefix of a finished + file, with no file mtime to consult — is computed by the same code that + computes a live one. + """ + if not events: + return LiveSnapshot(trace=rel) run_ids = list(dict.fromkeys(e.run_id for e in events)) chosen = run_id if run_id in run_ids else run_ids[-1] run_events = [e for e in events if e.run_id == chosen] - - try: - size = path.stat().st_size - quiet_for = time() - path.stat().st_mtime - except OSError: - size, quiet_for = 0, float("inf") active = quiet_for < ACTIVE_WINDOW_SECONDS mermaid = to_mermaid(recorder, chosen) @@ -138,6 +304,18 @@ def build_snapshot(root: Path, rel: str, run_id: str | None) -> LiveSnapshot: not e.completed for e in run.executions ): active = True + # A planning round that has begun and not closed is the same shape one step + # earlier: the planner writes nothing between the request and the model's + # reply, and 30-45 seconds of local inference is longer than the activity + # window. Bounded by the same grace, for the same reason. + planning = summarize_planning(run_events) + if ( + not done + and quiet_for < OPEN_NODE_GRACE_SECONDS + and planning is not None + and any(r.in_flight for r in planning.rounds) + ): + active = True # An approval request with no later response: the run is deliberately # parked, and the page should say "awaiting approval", not "idle". awaiting = False @@ -162,9 +340,64 @@ def build_snapshot(root: Path, rel: str, run_id: str | None) -> LiveSnapshot: active=active, done=done, awaiting_approval=awaiting, + planning=planning, ) +def _elapsed_seconds(events: list[TraceEvent]) -> list[float]: + """Seconds since the first event, per event, from the recorded timestamps. + + Monotonic by construction: an unparseable or out-of-order `ts` — these files + can be foreign or hand-written — holds the previous offset rather than + sending the replay backwards or raising. + """ + offsets: list[float] = [] + first: datetime | None = None + latest = 0.0 + for event in events: + try: + stamp = datetime.fromisoformat(event.ts) + except (TypeError, ValueError): + offsets.append(latest) + continue + if first is None: + first = stamp + try: + latest = max(latest, (stamp - first).total_seconds()) + except TypeError: # one naive stamp among aware ones + pass + offsets.append(latest) + return offsets + + +def replay_schedule( + events: list[TraceEvent], + speed: float = 1.0, + *, + max_seconds: float = REPLAY_MAX_SECONDS, + max_frames: int = REPLAY_MAX_FRAMES, +) -> list[tuple[int, float]]: + """`(events to include, seconds into the replay)` per frame, in order. + + Recorded speed divided by `speed`, then capped: whatever the multiplier, the + whole replay fits in `max_seconds`, so yesterday's 40-minute incident trace + is watchable. A nonsense multiplier (zero, negative, NaN) reads as 1.0 + rather than dividing the schedule by it. + """ + if not events: + return [] + if not speed > 0 or math.isnan(speed): + speed = 1.0 + speed = min(speed, REPLAY_MAX_SPEED) + offsets = _elapsed_seconds(events) + span = offsets[-1] + if max_seconds > 0 and span / speed > max_seconds: + speed = span / max_seconds + stride = max(1, math.ceil(len(events) / max(1, max_frames))) + indices = sorted({*range(0, len(events), stride), len(events) - 1}) + return [(index + 1, offsets[index] / speed) for index in indices] + + #: How many of the newest traces the index fully parses for run ids. The rest #: are listed by name and size only: the live root accumulates one directory #: per run forever, and re-validating every byte of the whole corpus per @@ -222,6 +455,20 @@ def scan_traces(root: Path) -> list[dict[str, Any]]: return found +def _safe_next(raw: str) -> str: + """Where sign-in may send the caller: a path inside `/live`, or nothing. + + The form carries its destination, and a form field is attacker-supplied by + definition — an absolute URL, a scheme-relative `//host` or a backslash the + browser normalises would turn the sign-in page into an open redirect. + """ + if not raw.startswith("/live") or raw.startswith("//"): + return "/live" + if any(bad in raw for bad in ("\\", "\n", "\r", "\t")): + return "/live" + return raw + + def live_router( root: str | Path, *, @@ -233,21 +480,71 @@ def live_router( root_path = Path(root) router = APIRouter(prefix="/live") - def _authorized(request: Request) -> None: - if token is None: - return - supplied = request.query_params.get("token") - header = request.headers.get("authorization", "") - if header.startswith("Bearer "): - supplied = supplied or header.removeprefix("Bearer ") + # What the sign-in cookie carries: a digest of the token, never the token. + # Always ASCII, so a non-ASCII secret survives a header round trip (cookies + # are latin-1 on the wire), and a stolen cookie is not the secret itself. + cookie_value = ( + hashlib.sha256(token.encode("utf-8")).hexdigest() if token is not None else "" + ) + + def _matches(supplied: str) -> bool: # Compared as bytes: `compare_digest` refuses `str` outside ASCII, so # comparing text turned a one-character guess into a 500 — the gate # crashing on the strangers it exists to refuse. Encoding keeps the # constant-time property, which is the reason it is here at all. - if supplied is None or not secrets.compare_digest( + return token is not None and secrets.compare_digest( supplied.encode("utf-8"), token.encode("utf-8") + ) + + def _refusal(request: Request, *, allow_query: bool = False) -> str | None: + """None when the request may proceed, otherwise why it may not. + + `allow_query` is the SSE route's exemption and nothing else's: a page + opened by hand, an index, a JSON listing all have a header or the + sign-in cookie available, and a token they put in the URL is refused + with its own reason rather than quietly accepted into the logs. + """ + if token is None: + return None + header = request.headers.get("authorization", "") + candidates = [] + if header.startswith("Bearer "): + candidates.append(header.removeprefix("Bearer ")) + if allow_query: + query = request.query_params.get("token") + if query is not None: + candidates.append(query) + cookie = request.cookies.get(LIVE_COOKIE) or "" + # Every credential presented is tried, not the first one found: a stale + # cookie must not lock out a request that also carries a good header. + if any(_matches(candidate) for candidate in candidates) or secrets.compare_digest( + cookie.encode("utf-8"), cookie_value.encode("utf-8") ): - raise HTTPException(status_code=401, detail="missing or wrong token") + return None + if not allow_query and "token" in request.query_params: + return QUERY_TOKEN_REFUSED + return "missing or wrong token" + + def _return_to(request: Request) -> str: + """Where to send the caller after signing in — minus any `?token=`. + + Echoing the refused parameter back into the form would put the secret + into the next URL, which is the whole thing being fixed. + """ + kept = [ + (key, value) + for key, value in request.query_params.multi_items() + if key != "token" + ] + query = urlencode(kept) + return _safe_next(request.url.path + (f"?{query}" if query else "")) + + def _sign_in(target: str, reason: str) -> HTMLResponse: + page = ( + SIGNIN_HTML.replace("__REASON__", html.escape(reason)) + .replace("__NEXT__", html.escape(target, quote=True)) + ) + return HTMLResponse(page, status_code=401) def _resolved(raw: str) -> str: """Validate confinement; 404 on refusal (don't map what exists outside). @@ -261,10 +558,34 @@ def _resolved(raw: str) -> str: raise HTTPException(status_code=404, detail="no such trace") from None return raw + @router.post("/auth", include_in_schema=False) + async def auth(request: Request) -> Response: + """Trade the token for a cookie, so no navigation carries it in a URL. + + A form post, not a GET: a GET would put the secret in the query string + of the very request that exists to keep it out of query strings. + """ + form = await request.form() + target = _safe_next(str(form.get("next") or "/live")) + if token is not None and not _matches(str(form.get("token") or "")): + return _sign_in(target, "missing or wrong token") + response = RedirectResponse(target, status_code=303) + if token is not None: + response.set_cookie( + LIVE_COOKIE, + cookie_value, + httponly=True, + samesite="strict", + path="/live", + secure=request.url.scheme == "https", + ) + return response + @router.get("", response_class=HTMLResponse, include_in_schema=False) def index(request: Request) -> HTMLResponse: - _authorized(request) - keep = f"&token={quote(token, safe='')}" if token else "" + refusal = _refusal(request) + if refusal: + return _sign_in(_return_to(request), refusal) rows = [] for item in scan_traces(root_path): # Trace names and run ids come from whoever wrote the files — the @@ -274,9 +595,10 @@ def index(request: Request) -> HTMLResponse: # token sits in this same DOM). runs = html.escape(", ".join(item["runs"]) or "—") shown = html.escape(item["trace"]) - href = html.escape( - f"/live/view?trace={quote(item['trace'], safe='')}{keep}" - ) + # No token in the link: the cookie authenticates the page the + # operator clicks through to, and a link is the thing that ends up + # in history and in the referrer of anything the page opens. + href = html.escape(f"/live/view?trace={quote(item['trace'], safe='')}") rows.append( f'{shown}' f"{runs}{item['size']}" @@ -286,12 +608,16 @@ def index(request: Request) -> HTMLResponse: @router.get("/api/runs") def runs(request: Request) -> dict[str, Any]: - _authorized(request) + refusal = _refusal(request) + if refusal: + raise HTTPException(status_code=401, detail=refusal) return {"root": str(root_path), "traces": scan_traces(root_path)} @router.get("/view", response_class=HTMLResponse, include_in_schema=False) def view(request: Request, trace: str = Query(...)) -> HTMLResponse: - _authorized(request) + refusal = _refusal(request) + if refusal: + return _sign_in(_return_to(request), refusal) _resolved(trace) return HTMLResponse(VIEW_HTML) @@ -300,11 +626,78 @@ async def stream( request: Request, trace: str = Query(...), run: str | None = Query(None), + replay_recorded: int = Query(0, alias="replay"), + speed: float = Query(1.0), ) -> StreamingResponse: - _authorized(request) + # The one route a token may reach by query string: `EventSource` cannot + # set a header, and the page has nothing else to authenticate with. + refusal = _refusal(request, allow_query=True) + if refusal: + raise HTTPException(status_code=401, detail=refusal) rel = _resolved(trace) path = resolve_trace(root_path, rel) + async def replay_frames(): + """The recorded run, re-emitted as the snapshots it would have sent. + + A finished trace renders instantly all-green, so the amber "running" + frame — the thing the runtime is best at showing — is unreachable + for every run that is already over. This walks the file's own + timestamps and rebuilds each intermediate snapshot from a prefix, + which is the same computation a live stream does, only sourced from + the past. + """ + from grapharc.server.app import _disconnected, _sse + + try: + events = await asyncio.to_thread(TailRecorder(path).read_events) + except Exception: + events = [] + if not events: + yield _sse("snapshot", LiveSnapshot(trace=rel).model_dump(mode="json")) + yield _sse("done", {"trace": rel, "replay": True}) + return + run_ids = list(dict.fromkeys(e.run_id for e in events)) + chosen = run if run in run_ids else run_ids[-1] + run_events = [e for e in events if e.run_id == chosen] + try: + size = path.stat().st_size + except OSError: + size = 0 + started = monotonic() + for count, offset in replay_schedule(run_events, speed): + wait = offset - (monotonic() - started) + if wait > 0: + await asyncio.sleep(wait) + if await _disconnected(request): + return + prefix = run_events[:count] + try: + snapshot = await asyncio.to_thread( + compose_snapshot, + rel, + _FrozenRecorder(path, prefix), + prefix, + chosen, + # A replay has no live file to stat: it is running by + # definition until its last frame, and every frame must + # be reproducible, so neither depends on the clock. + size=size, + quiet_for=0.0, + ) + except Exception: + yield ": snapshot unavailable\n\n" + continue + yield _sse("snapshot", snapshot.model_dump(mode="json")) + yield _sse("done", {"trace": rel, "replay": True}) + + if replay_recorded: + return StreamingResponse( + replay_frames(), + media_type="text/event-stream", + headers={"cache-control": "no-cache", "x-accel-buffering": "no"}, + ) + async def frames(): from grapharc.server.app import _disconnected, _sse @@ -379,8 +772,32 @@ async def frames(): """ +SIGNIN_HTML = """ +grapharc live · sign in + + +

grapharc live

+

__REASON__

+
+ + + +
+

The token is exchanged for a cookie scoped to /live, +so it never appears in a URL — where it would be copied into access logs, +browser history and referrer headers. Scripts can send +Authorization: Bearer TOKEN instead.

+ +""" + VIEW_HTML = """ grapharc live view +
connecting… - + +
waiting for the run to start…
+