From 6483c7b16448cfe8a6b1672558f684e7180aeb94 Mon Sep 17 00:00:00 2001
From: Fawaz <67429056+FwazB@users.noreply.github.com>
Date: Mon, 13 Jul 2026 21:18:14 -0700
Subject: [PATCH] Add protected canonical claims read endpoint for the
Observatory (#139)
* Add protected canonical claims read API
* Normalize claim loader exits in worker thread
* Allow file-gated claims reader defaults
---
diagnostics/app.py | 23 +-
diagnostics/kb_claim_routes.py | 511 +++++++++++++++++++++++++++++--
docs/canonical-claims-browser.md | 119 +++++++
tests/test_kb_claim_routes.py | 449 ++++++++++++++++++++++++++-
4 files changed, 1078 insertions(+), 24 deletions(-)
create mode 100644 docs/canonical-claims-browser.md
diff --git a/diagnostics/app.py b/diagnostics/app.py
index 47315a2..ec35046 100644
--- a/diagnostics/app.py
+++ b/diagnostics/app.py
@@ -23,10 +23,16 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "pipeline"))
from aiohttp import web
from daily_digest_routes import register_daily_digest_routes
-from kb_claim_routes import KB_CLAIM_PUBLIC_PATHS, KB_CLAIM_PUBLIC_PREFIXES, register_kb_claim_routes
+from kb_claim_routes import (
+ KB_CLAIM_GLOBAL_AUTH_PREFIXES,
+ KB_CLAIM_LIST_API_KEY,
+ KB_CLAIM_SELF_AUTH_PATHS,
+ load_claim_list_api_key,
+ register_kb_claim_routes,
+)
from kb_proposal_routes import KB_PROPOSAL_PUBLIC_PATHS, register_kb_proposal_routes
from leaderboard_routes import LEADERBOARD_PUBLIC_PATHS, register_leaderboard_routes
-from lib.search import search as kb_search, embed_query, search_qdrant
+from lib.search import embed_query, search as kb_search, search_qdrant
from response_audit_routes import RESPONSE_AUDIT_PUBLIC_PATHS, register_response_audit_routes
from review_queue_routes import register_review_queue_routes
@@ -516,13 +522,14 @@ async def auth_middleware(request, handler):
or request.path in RESPONSE_AUDIT_PUBLIC_PATHS
or request.path in LEADERBOARD_PUBLIC_PATHS
or request.path in KB_PROPOSAL_PUBLIC_PATHS
- or request.path in KB_CLAIM_PUBLIC_PATHS
- or request.path.startswith(KB_CLAIM_PUBLIC_PREFIXES)
+ or (request.method == "GET" and request.path in KB_CLAIM_SELF_AUTH_PATHS)
or request.path.startswith("/api/response-audit/")
):
return await handler(request)
expected = request.app.get("api_key")
if not expected:
+ if request.path.startswith(KB_CLAIM_GLOBAL_AUTH_PREFIXES):
+ return web.json_response({"error": "argus_auth_unconfigured"}, status=503)
# No key configured — all endpoints open (development mode)
return await handler(request)
provided = request.headers.get("X-Api-Key", "")
@@ -2338,10 +2345,16 @@ def create_app() -> web.Application:
app = web.Application(middlewares=[auth_middleware])
app["db"] = _get_db()
app["api_key"] = _load_secret(API_KEY_FILE)
+ canonical_claims_api_key = load_claim_list_api_key()
+ if canonical_claims_api_key:
+ app[KB_CLAIM_LIST_API_KEY] = canonical_claims_api_key
+ logger.info("Canonical claims API auth enabled")
+ else:
+ logger.warning("Canonical claims API auth is not configured; list endpoint fails closed")
if app["api_key"]:
logger.info("API key auth enabled (protected endpoints require X-Api-Key)")
else:
- logger.info("No API key configured — all endpoints open")
+ logger.info("No global API key configured — canonical claim routes still fail closed")
# Root redirects to /ops (legacy dashboard still at /legacy)
app.router.add_get("/", handle_root_redirect)
app.router.add_get("/prs", handle_prs_page)
diff --git a/diagnostics/kb_claim_routes.py b/diagnostics/kb_claim_routes.py
index 268c207..bbde750 100644
--- a/diagnostics/kb_claim_routes.py
+++ b/diagnostics/kb_claim_routes.py
@@ -1,16 +1,23 @@
"""Read-only canonical KB claim routes for Argus.
These routes show the Postgres-backed claim graph that Leo uses through the
-``teleo-kb`` bridge: one claim, its evidence rows, and its graph edges.
+``teleo-kb`` bridge: a protected canonical summary list plus one claim, its
+evidence rows, and its graph edges.
"""
from __future__ import annotations
import argparse
+import asyncio
+import base64
+import binascii
+import hmac
import json
import logging
import os
import re
+import shutil
+import subprocess
import sys
from datetime import datetime, timezone
from html import escape
@@ -33,13 +40,84 @@ import kb_proposal_review_packet as proposal_review # noqa: E402
logger = logging.getLogger("argus.kb_claims")
-KB_CLAIM_PUBLIC_PATHS = frozenset()
-KB_CLAIM_PUBLIC_PREFIXES = ("/kb/claims/", "/api/kb/claims/")
+KB_CLAIM_SELF_AUTH_PATHS = frozenset({"/api/kb/claims"})
+KB_CLAIM_GLOBAL_AUTH_PREFIXES = ("/api/kb/claims/", "/kb/claims/")
KB_CLAIM_LOADER_KEY = web.AppKey("kb_claim_loader", object)
+KB_CLAIM_LIST_LOADER_KEY = web.AppKey("kb_claim_list_loader", object)
+KB_CLAIM_LIST_API_KEY = web.AppKey("kb_claim_list_api_key", str)
+
+CLAIM_LIST_SCHEMA = "livingip.canonical-claims.v1"
+CLAIM_LIST_DEFAULT_LIMIT = 25
+CLAIM_LIST_MAX_LIMIT = 100
+CLAIM_LIST_MAX_RESPONSE_BYTES = 500_000
+CLAIM_LIST_MAX_QUERY_LENGTH = 200
+CLAIM_LIST_MAX_FILTER_LENGTH = 64
+CLAIM_LIST_REQUIRED_ROLE = "kb_observatory_read"
+CLAIM_LIST_DEFAULT_API_KEY_FILE = "/opt/teleo-eval/secrets/kb-observatory-api-key"
+CLAIM_LIST_DEFAULT_SECRETS_FILE = "/opt/teleo-eval/secrets/kb-observatory-read-password"
+CLAIM_LIST_HANDLER_TIMEOUT_SECONDS = 10.0
+CLAIM_LIST_PROCESS_TIMEOUT_SECONDS = 8.0
+CLAIM_LIST_CONNECT_TIMEOUT_SECONDS = 3
+CLAIM_LIST_STATEMENT_TIMEOUT_MS = 5_000
UUID_RE = re.compile(
r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$"
)
+SIMPLE_FILTER_RE = re.compile(r"^[A-Za-z0-9_.:-]+$")
+
+
+class _ClaimListLoaderExited(RuntimeError):
+ """Keep loader exits inside the worker thread's normal exception channel."""
+
+
+def load_claim_list_api_key() -> str | None:
+ """Load the Observatory-only API token from its root-managed file."""
+ path = Path(
+ os.environ.get("KB_CLAIM_BROWSER_API_KEY_FILE", CLAIM_LIST_DEFAULT_API_KEY_FILE)
+ )
+ try:
+ if not path.is_file() or path.stat().st_mode & 0o007:
+ return None
+ token = path.read_text(encoding="utf-8").strip()
+ except OSError:
+ return None
+ if (
+ len(token) < 24
+ or len(token) > 512
+ or not token.isascii()
+ or any(character.isspace() for character in token)
+ ):
+ return None
+ return token
+
+
+def _load_claim_list_password(secrets_file: str) -> str:
+ """Parse one literal ``PGPASSWORD=...`` assignment from a dedicated file."""
+ path = Path(secrets_file)
+ try:
+ if not path.is_file():
+ raise RuntimeError("canonical claim browser secrets file is unavailable")
+ if path.stat().st_mode & 0o007:
+ raise RuntimeError("canonical claim browser secrets file is world-accessible")
+ lines = path.read_text(encoding="utf-8").splitlines()
+ except OSError as exc:
+ raise RuntimeError("canonical claim browser secrets file is unavailable") from exc
+
+ password: str | None = None
+ for raw_line in lines:
+ line = raw_line.strip()
+ if not line or line.startswith("#"):
+ continue
+ key, separator, value = line.partition("=")
+ if not separator or key.strip() != "PGPASSWORD" or password is not None:
+ raise RuntimeError("canonical claim browser secrets file has invalid format")
+ password = value.strip()
+ if len(password) >= 2 and password[0] == password[-1] and password[0] in {"'", '"'}:
+ password = password[1:-1]
+
+ if not password or "\n" in password or "\r" in password:
+ raise RuntimeError("canonical claim browser secrets file has invalid format")
+ return password
def is_claim_id(value: Any) -> bool:
@@ -65,6 +143,60 @@ def short_claim_link_html(claim_id: Any) -> str:
return claim_link_html(value, label=value[:8])
+def encode_claim_cursor(sort_timestamp: str, claim_id: str) -> str:
+ payload = json.dumps(
+ {"updated_at": sort_timestamp, "id": claim_id},
+ separators=(",", ":"),
+ ).encode("utf-8")
+ return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=")
+
+
+def decode_claim_cursor(value: str) -> tuple[str, str]:
+ try:
+ padded = value + "=" * (-len(value) % 4)
+ payload = json.loads(base64.urlsafe_b64decode(padded.encode("ascii")).decode("utf-8"))
+ timestamp = str(payload["updated_at"])
+ claim_id = str(payload["id"])
+ parsed_timestamp = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
+ if parsed_timestamp.tzinfo is None:
+ raise ValueError("cursor timestamp must be timezone-aware")
+ if not is_claim_id(claim_id):
+ raise ValueError("invalid claim id")
+ return timestamp, claim_id.lower()
+ except (binascii.Error, KeyError, TypeError, ValueError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+ raise ValueError("invalid cursor") from exc
+
+
+def _parse_claim_list_request(request: web.Request) -> dict[str, Any]:
+ raw_limit = request.query.get("limit", str(CLAIM_LIST_DEFAULT_LIMIT))
+ try:
+ limit = max(1, min(int(raw_limit), CLAIM_LIST_MAX_LIMIT))
+ except ValueError as exc:
+ raise ValueError("invalid_limit") from exc
+
+ filters: dict[str, Any] = {
+ "q": request.query.get("q", "").strip(),
+ "status": request.query.get("status", "open").strip(),
+ "type": request.query.get("type", "").strip(),
+ "tag": request.query.get("tag", "").strip(),
+ "limit": limit,
+ "cursor": request.query.get("cursor", "").strip(),
+ }
+ if len(filters["q"]) > CLAIM_LIST_MAX_QUERY_LENGTH:
+ raise ValueError("query_too_long")
+ if len(filters["cursor"]) > 512:
+ raise ValueError("invalid_cursor")
+ for key in ("status", "type", "tag"):
+ value = filters[key]
+ if len(value) > CLAIM_LIST_MAX_FILTER_LENGTH or (value and not SIMPLE_FILTER_RE.match(value)):
+ raise ValueError(f"invalid_{key}")
+ if filters["cursor"]:
+ filters["cursor_values"] = decode_claim_cursor(filters["cursor"])
+ else:
+ filters["cursor_values"] = None
+ return filters
+
+
def _db_args() -> argparse.Namespace:
return argparse.Namespace(
secrets_file=os.environ.get(
@@ -81,6 +213,66 @@ def _db_args() -> argparse.Namespace:
)
+def _claim_list_db_args() -> argparse.Namespace:
+ role = os.environ.get("KB_CLAIM_BROWSER_ROLE", "").strip()
+ secrets_file = os.environ.get("KB_CLAIM_BROWSER_SECRETS_FILE", "").strip()
+ if not role and not secrets_file and Path(CLAIM_LIST_DEFAULT_SECRETS_FILE).is_file():
+ role = CLAIM_LIST_REQUIRED_ROLE
+ secrets_file = CLAIM_LIST_DEFAULT_SECRETS_FILE
+ if not role or not secrets_file:
+ raise RuntimeError("canonical claim browser read role is not configured")
+ if role != CLAIM_LIST_REQUIRED_ROLE:
+ raise RuntimeError(f"canonical claim browser requires role {CLAIM_LIST_REQUIRED_ROLE}")
+ return argparse.Namespace(
+ secrets_file=secrets_file,
+ container=os.environ.get("KB_CLAIM_BROWSER_CONTAINER", proposal_review.ap.DEFAULT_CONTAINER),
+ db=os.environ.get("KB_CLAIM_BROWSER_DB", proposal_review.ap.DEFAULT_DB),
+ host=os.environ.get("KB_CLAIM_BROWSER_HOST", proposal_review.ap.DEFAULT_HOST),
+ role=role,
+ )
+
+
+def _run_claim_list_psql(args: argparse.Namespace, sql: str, password: str) -> str:
+ """Run the bounded, read-only browser query without leaking process output."""
+ docker_binary = shutil.which("docker") or "docker"
+ command = [
+ docker_binary,
+ "exec",
+ "-e",
+ "PGPASSWORD",
+ "-e",
+ f"PGCONNECT_TIMEOUT={CLAIM_LIST_CONNECT_TIMEOUT_SECONDS}",
+ "-i",
+ args.container,
+ "psql",
+ "-U",
+ args.role,
+ "-h",
+ args.host,
+ "-d",
+ args.db,
+ "-v",
+ "ON_ERROR_STOP=1",
+ "-At",
+ "-q",
+ ]
+ try:
+ result = subprocess.run(
+ command,
+ input=sql,
+ text=True,
+ capture_output=True,
+ env={"PGPASSWORD": password, "PATH": "/usr/bin:/bin:/usr/local/bin"},
+ check=False,
+ timeout=CLAIM_LIST_PROCESS_TIMEOUT_SECONDS,
+ )
+ except subprocess.TimeoutExpired as exc:
+ raise TimeoutError("canonical claim browser database query timed out") from exc
+ if result.returncode != 0:
+ raise RuntimeError("canonical claim browser database query failed")
+ return result.stdout
+
+
def _load_claim_from_db(claim_id: str) -> dict[str, Any] | None:
password = proposal_review.ap.load_password(_db_args().secrets_file)
args = _db_args()
@@ -167,6 +359,262 @@ def _load_claim(request: web.Request, claim_id: str) -> dict[str, Any] | None:
return loader(claim_id) if loader else _load_claim_from_db(claim_id)
+def _load_claim_list_from_db(filters: dict[str, Any]) -> dict[str, Any]:
+ args = _claim_list_db_args()
+ password = _load_claim_list_password(args.secrets_file)
+ where_clauses: list[str] = []
+
+ if filters["q"]:
+ query_literal = proposal_review.ap.sql_literal(filters["q"].lower())
+ where_clauses.append(
+ "(strpos(lower(c.text), "
+ f"{query_literal}) > 0 or exists ("
+ "select 1 from unnest(coalesce(c.tags, '{}'::text[])) claim_tag "
+ f"where strpos(lower(claim_tag), {query_literal}) > 0))"
+ )
+ if filters["status"] and filters["status"].lower() != "all":
+ where_clauses.append(f"c.status::text = {proposal_review.ap.sql_literal(filters['status'])}")
+ if filters["type"]:
+ where_clauses.append(f"c.type::text = {proposal_review.ap.sql_literal(filters['type'])}")
+ if filters["tag"]:
+ where_clauses.append(
+ f"{proposal_review.ap.sql_literal(filters['tag'])} = any(coalesce(c.tags, '{{}}'::text[]))"
+ )
+
+ where_sql = f"where {' and '.join(where_clauses)}" if where_clauses else ""
+ cursor_sql = ""
+ if filters["cursor_values"]:
+ cursor_timestamp, cursor_id = filters["cursor_values"]
+ cursor_sql = (
+ "where (sort_timestamp, id) < ("
+ f"{proposal_review.ap.sql_literal(cursor_timestamp)}::timestamptz, "
+ f"{proposal_review.ap.sql_literal(cursor_id)}::uuid)"
+ )
+
+ fetch_limit = filters["limit"] + 1
+ sql = f"""
+set statement_timeout = '{CLAIM_LIST_STATEMENT_TIMEOUT_MS}ms';
+set lock_timeout = '1000ms';
+
+with filtered as (
+ select c.id,
+ c.type,
+ c.text,
+ c.status,
+ c.confidence,
+ c.tags,
+ c.superseded_by,
+ c.created_at,
+ c.updated_at,
+ coalesce(c.updated_at, c.created_at, '1970-01-01'::timestamptz) as sort_timestamp
+ from public.claims c
+ {where_sql}
+), page_rows as (
+ select *
+ from filtered
+ {cursor_sql}
+ order by sort_timestamp desc, id desc
+ limit {fetch_limit}
+), enriched_page_rows as (
+ select page_rows.*,
+ coalesce(evidence_counts.evidence_count, 0) as evidence_count,
+ coalesce(edge_counts.edge_count, 0) as edge_count
+ from page_rows
+ left join lateral (
+ select count(*) as evidence_count
+ from public.claim_evidence ce
+ where ce.claim_id = page_rows.id
+ ) evidence_counts on true
+ left join lateral (
+ select count(*) as edge_count
+ from public.claim_edges e
+ where e.from_claim = page_rows.id or e.to_claim = page_rows.id
+ ) edge_counts on true
+)
+select jsonb_build_object(
+ 'session_read_only', current_setting('transaction_read_only'),
+ 'total', (select count(*) from filtered),
+ 'rows', coalesce((
+ select jsonb_agg(
+ jsonb_build_object(
+ 'id', id::text,
+ 'type', type::text,
+ 'text', left(text, 1200),
+ 'text_truncated', length(text) > 1200,
+ 'status', status::text,
+ 'confidence', confidence,
+ 'tags', coalesce(to_jsonb(tags), '[]'::jsonb),
+ 'superseded_by', superseded_by::text,
+ 'created_at', created_at::text,
+ 'updated_at', updated_at::text,
+ 'evidence_count', evidence_count,
+ 'edge_count', edge_count,
+ '_cursor_timestamp', sort_timestamp::text
+ ) order by sort_timestamp desc, id desc
+ ) from enriched_page_rows
+ ), '[]'::jsonb)
+)::text;
+"""
+ out = _run_claim_list_psql(args, sql, password).strip()
+ result = json.loads(out) if out else {"total": 0, "rows": []}
+ if result.pop("session_read_only", None) != "on":
+ raise RuntimeError("canonical claim browser database session is not read-only")
+ return result
+
+
+def _load_claim_list_sync(request: web.Request, filters: dict[str, Any]) -> dict[str, Any]:
+ loader = request.app.get(KB_CLAIM_LIST_LOADER_KEY)
+ try:
+ return loader(filters) if loader else _load_claim_list_from_db(filters)
+ except SystemExit as exc:
+ # Python 3.11 can propagate SystemExit from asyncio.to_thread before the
+ # awaiting handler gets a chance to sanitize it. Convert only that exit
+ # signal here; KeyboardInterrupt and other BaseException types retain
+ # their normal process-level semantics.
+ raise _ClaimListLoaderExited from exc
+
+
+async def _load_claim_list(request: web.Request, filters: dict[str, Any]) -> dict[str, Any]:
+ return await asyncio.wait_for(
+ asyncio.to_thread(_load_claim_list_sync, request, filters),
+ timeout=CLAIM_LIST_HANDLER_TIMEOUT_SECONDS,
+ )
+
+
+def _sanitize_claim_summary(row: dict[str, Any]) -> tuple[dict[str, Any], str]:
+ claim_id = str(row.get("id") or "").lower()
+ if not is_claim_id(claim_id):
+ raise ValueError("invalid canonical claim row")
+ cursor_timestamp = str(row.get("_cursor_timestamp") or row.get("updated_at") or row.get("created_at") or "")
+ parsed_timestamp = datetime.fromisoformat(cursor_timestamp.replace("Z", "+00:00"))
+ if parsed_timestamp.tzinfo is None:
+ raise ValueError("invalid canonical claim timestamp")
+ tags = [str(tag)[:128] for tag in (row.get("tags") or []) if isinstance(tag, (str, int, float))][:32]
+ summary = {
+ "id": claim_id,
+ "type": str(row.get("type") or "unknown")[:64],
+ "text": str(row.get("text") or "")[:1200],
+ "text_truncated": bool(row.get("text_truncated")),
+ "status": str(row.get("status") or "unknown")[:64],
+ "confidence": row.get("confidence"),
+ "tags": tags,
+ "superseded_by": str(row["superseded_by"]) if is_claim_id(row.get("superseded_by")) else None,
+ "created_at": str(row.get("created_at") or "") or None,
+ "updated_at": str(row.get("updated_at") or "") or None,
+ "evidence_count": max(0, int(row.get("evidence_count") or 0)),
+ "edge_count": max(0, int(row.get("edge_count") or 0)),
+ }
+ return summary, cursor_timestamp
+
+
+def _claim_list_payload(raw: dict[str, Any], filters: dict[str, Any]) -> dict[str, Any]:
+ raw_rows = list(raw.get("rows") or [])
+ if len(raw_rows) > filters["limit"] + 1:
+ raise ValueError("canonical claim page exceeded its requested bound")
+
+ sanitized_rows: list[tuple[dict[str, Any], str]] = []
+ seen_ids: set[str] = set()
+ previous_key: tuple[datetime, str] | None = None
+ request_cursor_key: tuple[datetime, str] | None = None
+ if filters["cursor_values"]:
+ request_cursor_timestamp, request_cursor_id = filters["cursor_values"]
+ request_cursor_key = (
+ datetime.fromisoformat(request_cursor_timestamp.replace("Z", "+00:00")),
+ request_cursor_id,
+ )
+
+ for row in raw_rows:
+ claim, cursor_timestamp = _sanitize_claim_summary(row)
+ if claim["id"] in seen_ids:
+ raise ValueError("canonical claim page contains duplicate ids")
+ seen_ids.add(claim["id"])
+ row_key = (
+ datetime.fromisoformat(cursor_timestamp.replace("Z", "+00:00")),
+ claim["id"],
+ )
+ if request_cursor_key is not None and row_key >= request_cursor_key:
+ raise ValueError("canonical claim page crossed its request cursor")
+ if previous_key is not None and row_key >= previous_key:
+ raise ValueError("canonical claim page is not strictly ordered")
+ previous_key = row_key
+ sanitized_rows.append((claim, cursor_timestamp))
+
+ visible_rows = sanitized_rows[: filters["limit"]]
+ total = max(0, int(raw.get("total") or 0))
+ if total < len(visible_rows):
+ raise ValueError("canonical claim total is smaller than its page")
+ generated_at = datetime.now(timezone.utc).isoformat()
+
+ def build_payload(page_rows: list[tuple[dict[str, Any], str]]) -> dict[str, Any]:
+ claims = [claim for claim, _cursor_timestamp in page_rows]
+ has_more = len(page_rows) < len(sanitized_rows)
+ last_cursor_timestamp = page_rows[-1][1] if page_rows else ""
+ next_cursor = (
+ encode_claim_cursor(last_cursor_timestamp, claims[-1]["id"])
+ if has_more and claims
+ else None
+ )
+ return {
+ "schema": CLAIM_LIST_SCHEMA,
+ "generated_at": generated_at,
+ "read_only": True,
+ "source": {
+ "store": "canonical_postgres",
+ "relation": "public.claims",
+ "receipt": "server-side read-only query",
+ },
+ "filters": {
+ "q": filters["q"] or None,
+ "status": filters["status"] or None,
+ "type": filters["type"] or None,
+ "tag": filters["tag"] or None,
+ },
+ "page": {
+ "limit": filters["limit"],
+ "returned": len(claims),
+ "total": total,
+ "has_more": has_more,
+ "next_cursor": next_cursor,
+ },
+ "claims": claims,
+ }
+
+ payload = build_payload(visible_rows)
+ if len(json.dumps(payload).encode("utf-8")) <= CLAIM_LIST_MAX_RESPONSE_BYTES:
+ return payload
+
+ fitted_payload: dict[str, Any] | None = None
+ low = 1
+ high = len(visible_rows) - 1
+ while low <= high:
+ midpoint = (low + high) // 2
+ candidate = build_payload(visible_rows[:midpoint])
+ if len(json.dumps(candidate).encode("utf-8")) <= CLAIM_LIST_MAX_RESPONSE_BYTES:
+ fitted_payload = candidate
+ low = midpoint + 1
+ else:
+ high = midpoint - 1
+ if fitted_payload is None:
+ raise ValueError("canonical claim response exceeded its byte bound")
+ return fitted_payload
+
+
+def _private_json_response(payload: dict[str, Any], *, status: int = 200) -> web.Response:
+ response = web.json_response(payload, status=status)
+ response.headers["Cache-Control"] = "private, no-store, max-age=0"
+ response.headers["Pragma"] = "no-cache"
+ response.headers["Vary"] = "X-Api-Key"
+ return response
+
+
+def _private_html_response(body: str, *, status: int = 200) -> web.Response:
+ response = web.Response(text=body, content_type="text/html", status=status)
+ response.headers["Cache-Control"] = "private, no-store, max-age=0"
+ response.headers["Pragma"] = "no-cache"
+ response.headers["Vary"] = "X-Api-Key"
+ return response
+
+
def _metadata_table(claim: dict[str, Any]) -> str:
rows = [
("ID", claim_link_html(claim.get("id"))),
@@ -270,40 +718,67 @@ def render_kb_claim_page(data: dict[str, Any]) -> str:
async def handle_api_kb_claim(request: web.Request) -> web.Response:
claim_id = request.match_info["claim_id"]
if not is_claim_id(claim_id):
- return web.json_response({"error": "invalid_claim_id", "claim_id": claim_id}, status=400)
+ return _private_json_response({"error": "invalid_claim_id"}, status=400)
try:
data = _load_claim(request, claim_id)
- except BaseException as exc:
- logger.exception("KB claim API load failed")
- return web.json_response({"error": "kb_claim_load_failed", "detail": str(exc)}, status=500)
+ except (Exception, SystemExit) as exc:
+ logger.error("KB claim API load failed (%s)", type(exc).__name__)
+ return _private_json_response({"error": "kb_claim_load_failed"}, status=503)
if not data:
- return web.json_response({"error": "claim_not_found", "claim_id": claim_id}, status=404)
- return web.json_response(data)
+ return _private_json_response({"error": "claim_not_found"}, status=404)
+ return _private_json_response(data)
+
+
+async def handle_api_kb_claims(request: web.Request) -> web.Response:
+ expected_api_key = request.app.get(KB_CLAIM_LIST_API_KEY)
+ if not expected_api_key:
+ return _private_json_response({"error": "canonical_claims_auth_unconfigured"}, status=503)
+ provided_api_key = request.headers.get("X-Api-Key", "")
+ try:
+ authenticated = hmac.compare_digest(
+ provided_api_key.encode("ascii"),
+ str(expected_api_key).encode("ascii"),
+ )
+ except UnicodeEncodeError:
+ authenticated = False
+ if not authenticated:
+ return _private_json_response({"error": "unauthorized"}, status=401)
+ try:
+ filters = _parse_claim_list_request(request)
+ except ValueError as exc:
+ return _private_json_response({"error": str(exc)}, status=400)
+ try:
+ raw = await _load_claim_list(request, filters)
+ payload = _claim_list_payload(raw, filters)
+ except (Exception, SystemExit) as exc:
+ logger.error("Canonical KB claim list load failed (%s)", type(exc).__name__)
+ return _private_json_response({"error": "canonical_claims_unavailable"}, status=503)
+ return _private_json_response(payload)
async def handle_kb_claim_page(request: web.Request) -> web.Response:
claim_id = request.match_info["claim_id"]
if not is_claim_id(claim_id):
- return web.Response(text="Invalid claim id", status=400)
+ return _private_html_response("Invalid claim id", status=400)
try:
data = _load_claim(request, claim_id)
- except BaseException as exc:
- logger.exception("KB claim page load failed")
- return web.Response(
- text=render_page(
+ except (Exception, SystemExit) as exc:
+ logger.error("KB claim page load failed (%s)", type(exc).__name__)
+ return _private_html_response(
+ render_page(
"KB Claim",
"Canonical claim, evidence rows, and graph edges",
"/kb-proposals",
- f'Failed to load KB claim: {escape(str(exc))}
',
+ 'Canonical claim is temporarily unavailable.
',
),
- content_type="text/html",
status=500,
)
if not data:
- return web.Response(text="Claim not found", status=404)
- return web.Response(text=render_kb_claim_page(data), content_type="text/html")
+ return _private_html_response("Claim not found", status=404)
+ return _private_html_response(render_kb_claim_page(data))
def register_kb_claim_routes(app: web.Application) -> None:
+ app.router.add_get("/api/kb/claims", handle_api_kb_claims)
app.router.add_get("/api/kb/claims/{claim_id}", handle_api_kb_claim)
app.router.add_get("/kb/claims/{claim_id}", handle_kb_claim_page)
diff --git a/docs/canonical-claims-browser.md b/docs/canonical-claims-browser.md
new file mode 100644
index 0000000..7f36270
--- /dev/null
+++ b/docs/canonical-claims-browser.md
@@ -0,0 +1,119 @@
+# Canonical Claims Browser Read Path
+
+Status: implementation-ready, not deployed
+
+The canonical claims browser exposes a protected, read-only summary of
+`public.claims` for the LivingIP Observatory. It is deliberately separate from
+the public Markdown/Qdrant knowledge browser.
+
+## Contract
+
+`GET /api/kb/claims`
+
+- protected by a route-scoped Observatory token; the browser never receives it
+- schema: `livingip.canonical-claims.v1`
+- filters: `q`, `status`, `type`, and `tag`
+- opaque keyset cursor over `(updated_at, id)`
+- default requested page size 25; maximum 100; the server may return fewer rows
+ while preserving `has_more` and `next_cursor` to keep the serialized response
+ at or below 500,000 bytes
+- allowlisted output only: canonical UUID, bounded claim text, type, status,
+ confidence, tags, timestamps, supersession pointer, evidence count, and edge
+ count
+- response headers: `Cache-Control: private, no-store, max-age=0`
+- no mutation methods, source excerpts, storage paths, credentials, internal
+ endpoints, or embeddings
+
+Only the exact `GET /api/kb/claims` path bypasses the global Argus middleware so
+its handler can authenticate the route-scoped token. The legacy
+`/api/kb/claims/{uuid}` and `/kb/claims/{uuid}` routes require the global Argus
+key and fail closed when that key is not configured. The Observatory does not
+receive, reuse, or expose the global Argus key.
+
+Those legacy detail routes are explicitly outside
+`livingip.canonical-claims.v1`: they include source-rich evidence and still use
+the pre-existing claim-review database credential path. The Observatory adapter
+must not call them. Migrating legacy detail reads to a separate role (including
+a deliberate `public.sources` grant) is residual hardening, not a prerequisite
+for this summary-only browser.
+
+Create a separate high-entropy token in a root-managed, service-readable file:
+
+```text
+/opt/teleo-eval/secrets/kb-observatory-api-key
+```
+
+The file contains only a 24-to-512-character ASCII token with no whitespace (a
+trailing newline is accepted), and it must not be world-accessible. Set
+`KB_CLAIM_BROWSER_API_KEY_FILE` only when using a different root-managed path.
+Configure the same value as the server-only
+`OBSERVATORY_CANONICAL_API_KEY` secret in Vercel; never use a `NEXT_PUBLIC_`
+variable.
+
+## Dedicated Database Role
+
+The handler accepts either both explicit `KB_CLAIM_BROWSER_ROLE` and
+`KB_CLAIM_BROWSER_SECRETS_FILE` settings, or the protected default password file
+at `/opt/teleo-eval/secrets/kb-observatory-read-password`. The default path is
+file-gated: if the file is absent, partial environment configuration is present,
+or the file fails its permission/format checks, the endpoint fails closed. The
+only accepted role name is `kb_observatory_read`, and every response is withheld
+unless Postgres reports that the current session has
+`transaction_read_only=on`.
+
+Create a dedicated login using a password supplied out of band, then grant only
+the relations needed by the list query:
+
+```sql
+create role kb_observatory_read login password :'observatory_read_password';
+alter role kb_observatory_read set default_transaction_read_only = on;
+grant connect on database teleo to kb_observatory_read;
+grant usage on schema public to kb_observatory_read;
+grant select on public.claims, public.claim_evidence, public.claim_edges
+ to kb_observatory_read;
+```
+
+Store the password in a dedicated root-managed file readable by the Argus
+service. Its format is intentionally narrow: blank lines and `#` comments are
+allowed, followed by exactly one literal assignment (no shell expansion):
+
+```text
+PGPASSWORD='the-observatory-read-role-password'
+```
+
+No `KB_APPLY_PASSWORD`, `KB_APPLY_DB_PASSWORD`, or additional assignments are
+accepted. The file must not be world-accessible; a suitable deployment mode is
+`root:teleo 0640`. Never place the password in Git, systemd unit text, browser
+configuration, or a `NEXT_PUBLIC_` environment variable.
+
+Optional explicit Argus overrides:
+
+```text
+KB_CLAIM_BROWSER_ROLE=kb_observatory_read
+KB_CLAIM_BROWSER_SECRETS_FILE=/opt/teleo-eval/secrets/kb-observatory-read-password
+```
+
+Optional connection overrides use the same prefix:
+`KB_CLAIM_BROWSER_CONTAINER`, `KB_CLAIM_BROWSER_DB`,
+`KB_CLAIM_BROWSER_HOST`.
+
+The query has three independent bounds: a three-second Postgres connection
+timeout, a five-second Postgres statement timeout, and an eight-second process
+timeout. The aiohttp handler runs the blocking database work in a worker thread
+and stops awaiting it after ten seconds, so a slow database cannot block the
+Argus event loop.
+
+## Promotion Checks
+
+1. Run `pytest tests/test_kb_claim_routes.py -q` and Ruff on the route and test.
+2. Deploy both dedicated secret files without changing the global Argus key.
+3. Verify an unauthenticated list request returns `401`.
+4. Verify an authenticated request returns the versioned contract and no-store
+ headers.
+5. Verify every response body is at most 500,000 bytes and two cursor pages are
+ disjoint and stable, including when the first page is shortened by the byte
+ cap.
+6. From the database session, verify `current_setting('transaction_read_only')`
+ is `on` and INSERT/UPDATE/DELETE all fail.
+7. Configure the Vercel adapter with a server-only URL and API key only after
+ preview Deployment Protection is enabled.
diff --git a/tests/test_kb_claim_routes.py b/tests/test_kb_claim_routes.py
index 2fef627..cc4f6bf 100644
--- a/tests/test_kb_claim_routes.py
+++ b/tests/test_kb_claim_routes.py
@@ -5,6 +5,8 @@ from __future__ import annotations
import asyncio
import json
import sys
+import threading
+import time
from pathlib import Path
import pytest
@@ -17,11 +19,24 @@ from aiohttp.test_utils import make_mocked_request
REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "diagnostics"))
+import app as argus_app # noqa: E402
import kb_claim_routes as routes # noqa: E402
-
CLAIM_ID = "d0000000-0000-0000-0000-000000000005"
CONNECTED_ID = "e0000000-0000-0000-0000-000000000006"
+SECOND_CLAIM_ID = "c0000000-0000-0000-0000-000000000004"
+THIRD_CLAIM_ID = "b0000000-0000-0000-0000-000000000003"
+API_KEY = "test-argus-api-key"
+
+
+def _claim_list_app() -> web.Application:
+ app = web.Application()
+ app[routes.KB_CLAIM_LIST_API_KEY] = API_KEY
+ return app
+
+
+def _claim_list_request(path: str, *, app: web.Application):
+ return make_mocked_request("GET", path, app=app, headers={"X-Api-Key": API_KEY})
def _claim_data():
@@ -58,6 +73,24 @@ def _claim_data():
}
+def _claim_summary(claim_id: str, *, timestamp: str, text: str = "Canonical claim"):
+ return {
+ "id": claim_id,
+ "type": "strategy_kernel",
+ "text": text,
+ "text_truncated": False,
+ "status": "open",
+ "confidence": "0.72",
+ "tags": ["claims", "review"],
+ "created_at": timestamp,
+ "updated_at": timestamp,
+ "evidence_count": 3,
+ "edge_count": 5,
+ "_cursor_timestamp": timestamp,
+ "internal_secret": "must-not-leak",
+ }
+
+
def test_claim_link_html_links_valid_claim_ids() -> None:
html = routes.claim_link_html(CLAIM_ID)
@@ -91,3 +124,417 @@ def test_api_kb_claim_uses_loader_and_returns_claim_graph() -> None:
assert response.status == 200
assert data["claim"]["id"] == CLAIM_ID
assert data["edges"][0]["connected_id"] == CONNECTED_ID
+ assert response.headers["Cache-Control"] == "private, no-store, max-age=0"
+
+
+def test_claim_cursor_round_trip() -> None:
+ timestamp = "2026-07-13T21:12:00+00:00"
+ cursor = routes.encode_claim_cursor(timestamp, CLAIM_ID)
+
+ assert routes.decode_claim_cursor(cursor) == (timestamp, CLAIM_ID)
+
+
+def test_canonical_claim_list_database_role_fails_closed(monkeypatch, tmp_path) -> None:
+ monkeypatch.delenv("KB_CLAIM_BROWSER_ROLE", raising=False)
+ monkeypatch.delenv("KB_CLAIM_BROWSER_SECRETS_FILE", raising=False)
+ default_secrets_file = tmp_path / "missing-kb-observatory-read-password"
+ monkeypatch.setattr(routes, "CLAIM_LIST_DEFAULT_SECRETS_FILE", str(default_secrets_file))
+
+ with pytest.raises(RuntimeError, match="read role is not configured"):
+ routes._claim_list_db_args()
+
+ default_secrets_file.write_text("PGPASSWORD=not-printed\n", encoding="utf-8")
+ default_secrets_file.chmod(0o600)
+ default_args = routes._claim_list_db_args()
+
+ assert default_args.role == routes.CLAIM_LIST_REQUIRED_ROLE
+ assert default_args.secrets_file == str(default_secrets_file)
+
+ monkeypatch.setenv("KB_CLAIM_BROWSER_ROLE", "kb_apply")
+ monkeypatch.setenv("KB_CLAIM_BROWSER_SECRETS_FILE", "/run/secrets/read-password")
+ with pytest.raises(RuntimeError, match=routes.CLAIM_LIST_REQUIRED_ROLE):
+ routes._claim_list_db_args()
+
+ monkeypatch.setenv("KB_CLAIM_BROWSER_ROLE", routes.CLAIM_LIST_REQUIRED_ROLE)
+ args = routes._claim_list_db_args()
+
+ assert args.role == routes.CLAIM_LIST_REQUIRED_ROLE
+ assert args.secrets_file == "/run/secrets/read-password"
+
+
+def test_canonical_claim_list_database_session_must_be_read_only(monkeypatch, tmp_path) -> None:
+ secrets_file = tmp_path / "kb-observatory-read.env"
+ secrets_file.write_text("PGPASSWORD=not-printed\n", encoding="utf-8")
+ secrets_file.chmod(0o640)
+ monkeypatch.setenv("KB_CLAIM_BROWSER_ROLE", routes.CLAIM_LIST_REQUIRED_ROLE)
+ monkeypatch.setenv("KB_CLAIM_BROWSER_SECRETS_FILE", str(secrets_file))
+ observed_sql = {}
+
+ def run_psql(_args, sql, _password):
+ observed_sql["value"] = sql
+ return json.dumps({"session_read_only": "off", "total": 0, "rows": []})
+
+ monkeypatch.setattr(routes, "_run_claim_list_psql", run_psql)
+ filters = {
+ "q": "",
+ "status": "all",
+ "type": "",
+ "tag": "",
+ "limit": 25,
+ "cursor": "",
+ "cursor_values": None,
+ }
+
+ with pytest.raises(RuntimeError, match="session is not read-only"):
+ routes._load_claim_list_from_db(filters)
+
+ assert "current_setting('transaction_read_only')" in observed_sql["value"]
+ assert "set statement_timeout = '5000ms'" in observed_sql["value"]
+ assert observed_sql["value"].index("page_rows as") < observed_sql["value"].index(
+ "from public.claim_evidence"
+ )
+
+
+def test_canonical_claim_list_secrets_file_accepts_only_pgpassword(tmp_path) -> None:
+ secrets_file = tmp_path / "kb-observatory-read.env"
+ secrets_file.write_text("# dedicated read role\nPGPASSWORD='read-only-secret'\n", encoding="utf-8")
+ secrets_file.chmod(0o640)
+
+ assert routes._load_claim_list_password(str(secrets_file)) == "read-only-secret"
+
+ secrets_file.write_text("KB_APPLY_PASSWORD=wrong-boundary\n", encoding="utf-8")
+ with pytest.raises(RuntimeError, match="invalid format"):
+ routes._load_claim_list_password(str(secrets_file))
+
+ secrets_file.write_text("PGPASSWORD=secret\n", encoding="utf-8")
+ secrets_file.chmod(0o644)
+ with pytest.raises(RuntimeError, match="world-accessible"):
+ routes._load_claim_list_password(str(secrets_file))
+
+
+def test_canonical_claim_list_process_timeout_is_enforced_and_sanitized(monkeypatch) -> None:
+ args = routes.argparse.Namespace(
+ container="postgres",
+ db="teleo",
+ host="127.0.0.1",
+ role=routes.CLAIM_LIST_REQUIRED_ROLE,
+ )
+ observed = {}
+
+ def timeout_run(command, **kwargs):
+ observed["command"] = command
+ observed.update(kwargs)
+ raise routes.subprocess.TimeoutExpired(command, kwargs["timeout"], stderr="secret")
+
+ monkeypatch.setattr(routes.subprocess, "run", timeout_run)
+
+ with pytest.raises(TimeoutError, match="database query timed out") as exc_info:
+ routes._run_claim_list_psql(args, "select 1", "must-not-leak")
+
+ assert "secret" not in str(exc_info.value)
+ assert observed["timeout"] == routes.CLAIM_LIST_PROCESS_TIMEOUT_SECONDS
+ assert f"PGCONNECT_TIMEOUT={routes.CLAIM_LIST_CONNECT_TIMEOUT_SECONDS}" in observed[
+ "command"
+ ]
+
+
+def test_canonical_claim_list_api_key_loads_only_from_dedicated_file(
+ monkeypatch, tmp_path
+) -> None:
+ api_key_file = tmp_path / "kb-observatory-api-key"
+ api_key_file.write_text("route-scoped-observatory-key\n", encoding="utf-8")
+ api_key_file.chmod(0o640)
+ monkeypatch.setenv("KB_CLAIM_BROWSER_API_KEY_FILE", str(api_key_file))
+
+ assert routes.load_claim_list_api_key() == "route-scoped-observatory-key"
+
+ api_key_file.write_text("two tokens\n", encoding="utf-8")
+ assert routes.load_claim_list_api_key() is None
+
+ api_key_file.write_text("é" * 24, encoding="utf-8")
+ assert routes.load_claim_list_api_key() is None
+
+ api_key_file.write_text("route-scoped-observatory-key\n", encoding="utf-8")
+ api_key_file.chmod(0o644)
+ assert routes.load_claim_list_api_key() is None
+
+
+def test_canonical_claim_list_is_versioned_paginated_and_sanitized() -> None:
+ app = _claim_list_app()
+ captured_filters = {}
+
+ def loader(filters):
+ captured_filters.update(filters)
+ return {
+ "total": 1837,
+ "rows": [
+ _claim_summary(CLAIM_ID, timestamp="2026-07-13T21:12:00+00:00"),
+ _claim_summary(SECOND_CLAIM_ID, timestamp="2026-07-13T20:12:00+00:00"),
+ ],
+ }
+
+ app[routes.KB_CLAIM_LIST_LOADER_KEY] = loader
+ req = _claim_list_request(
+ "/api/kb/claims?limit=1&q=falsifiable&type=strategy_kernel&tag=review",
+ app=app,
+ )
+
+ response = asyncio.run(routes.handle_api_kb_claims(req))
+ data = json.loads(response.body.decode())
+
+ assert response.status == 200
+ assert response.headers["Cache-Control"] == "private, no-store, max-age=0"
+ assert response.headers["Vary"] == "X-Api-Key"
+ assert data["schema"] == routes.CLAIM_LIST_SCHEMA
+ assert data["read_only"] is True
+ assert data["source"]["store"] == "canonical_postgres"
+ assert data["page"] == {
+ "limit": 1,
+ "returned": 1,
+ "total": 1837,
+ "has_more": True,
+ "next_cursor": data["page"]["next_cursor"],
+ }
+ assert routes.decode_claim_cursor(data["page"]["next_cursor"])[1] == CLAIM_ID
+ assert data["claims"][0]["id"] == CLAIM_ID
+ assert "internal_secret" not in data["claims"][0]
+ assert captured_filters["q"] == "falsifiable"
+ assert captured_filters["type"] == "strategy_kernel"
+ assert captured_filters["tag"] == "review"
+
+
+def test_canonical_claim_list_caps_response_bytes_without_skipping_rows(monkeypatch) -> None:
+ app = _claim_list_app()
+ rows = [
+ _claim_summary(CLAIM_ID, timestamp="2026-07-13T21:12:00+00:00", text="🔥" * 1200),
+ _claim_summary(
+ SECOND_CLAIM_ID,
+ timestamp="2026-07-13T20:12:00+00:00",
+ text="🔥" * 1200,
+ ),
+ _claim_summary(
+ THIRD_CLAIM_ID,
+ timestamp="2026-07-13T19:12:00+00:00",
+ text="🔥" * 1200,
+ ),
+ ]
+ for row in rows:
+ row["tags"] = ["🔥" * 128] * 32
+ app[routes.KB_CLAIM_LIST_LOADER_KEY] = lambda _filters: {"total": 3, "rows": rows}
+ assert routes.CLAIM_LIST_MAX_RESPONSE_BYTES < 512_000
+ byte_limit = 100_000
+ monkeypatch.setattr(routes, "CLAIM_LIST_MAX_RESPONSE_BYTES", byte_limit)
+ request = _claim_list_request("/api/kb/claims?limit=3&status=all", app=app)
+
+ response = asyncio.run(routes.handle_api_kb_claims(request))
+ data = json.loads(response.body.decode())
+
+ assert response.status == 200
+ assert len(response.body) <= byte_limit
+ assert data["page"]["limit"] == 3
+ assert data["page"]["returned"] == 1
+ assert data["page"]["has_more"] is True
+ assert routes.decode_claim_cursor(data["page"]["next_cursor"])[1] == CLAIM_ID
+ assert [claim["id"] for claim in data["claims"]] == [CLAIM_ID]
+
+
+def test_canonical_claim_list_clamps_limit_and_rejects_bad_cursor() -> None:
+ app = _claim_list_app()
+ observed = {}
+ app[routes.KB_CLAIM_LIST_LOADER_KEY] = lambda filters: observed.update(filters) or {"total": 0, "rows": []}
+ large_req = _claim_list_request("/api/kb/claims?limit=999", app=app)
+
+ large_response = asyncio.run(routes.handle_api_kb_claims(large_req))
+ large_data = json.loads(large_response.body.decode())
+
+ assert large_response.status == 200
+ assert large_data["page"]["limit"] == routes.CLAIM_LIST_MAX_LIMIT
+ assert observed["limit"] == routes.CLAIM_LIST_MAX_LIMIT
+
+ bad_req = _claim_list_request("/api/kb/claims?cursor=not-a-cursor", app=app)
+ bad_response = asyncio.run(routes.handle_api_kb_claims(bad_req))
+ bad_data = json.loads(bad_response.body.decode())
+
+ assert bad_response.status == 400
+ assert bad_data == {"error": "invalid cursor"}
+
+ long_req = _claim_list_request(f"/api/kb/claims?cursor={'x' * 513}", app=app)
+ long_response = asyncio.run(routes.handle_api_kb_claims(long_req))
+
+ assert long_response.status == 400
+
+
+def test_canonical_claim_list_failure_is_sanitized() -> None:
+ app = _claim_list_app()
+
+ def fail(_filters):
+ raise SystemExit("database password would have appeared here")
+
+ app[routes.KB_CLAIM_LIST_LOADER_KEY] = fail
+ req = _claim_list_request("/api/kb/claims", app=app)
+
+ response = asyncio.run(routes.handle_api_kb_claims(req))
+ data = json.loads(response.body.decode())
+
+ assert response.status == 503
+ assert data == {"error": "canonical_claims_unavailable"}
+ assert b"password" not in response.body
+
+
+def test_canonical_claim_auth_boundaries_are_exact() -> None:
+ list_path = "/api/kb/claims"
+
+ assert list_path in routes.KB_CLAIM_SELF_AUTH_PATHS
+ assert f"{list_path}/{CLAIM_ID}" not in routes.KB_CLAIM_SELF_AUTH_PATHS
+ assert f"/api/kb/claims/{CLAIM_ID}".startswith(routes.KB_CLAIM_GLOBAL_AUTH_PREFIXES)
+ assert f"/kb/claims/{CLAIM_ID}".startswith(routes.KB_CLAIM_GLOBAL_AUTH_PREFIXES)
+
+
+def test_canonical_claim_list_authentication_fails_closed() -> None:
+ app_without_key = web.Application()
+ unconfigured_req = make_mocked_request("GET", "/api/kb/claims", app=app_without_key)
+ unconfigured_response = asyncio.run(routes.handle_api_kb_claims(unconfigured_req))
+
+ assert unconfigured_response.status == 503
+ assert json.loads(unconfigured_response.body.decode()) == {"error": "canonical_claims_auth_unconfigured"}
+
+ configured_app = _claim_list_app()
+ unauthorized_req = make_mocked_request("GET", "/api/kb/claims", app=configured_app)
+ unauthorized_response = asyncio.run(routes.handle_api_kb_claims(unauthorized_req))
+
+ assert unauthorized_response.status == 401
+ assert json.loads(unauthorized_response.body.decode()) == {"error": "unauthorized"}
+
+ non_ascii_request = make_mocked_request(
+ "GET",
+ "/api/kb/claims",
+ app=configured_app,
+ headers={"X-Api-Key": "é" * 24},
+ )
+ non_ascii_response = asyncio.run(routes.handle_api_kb_claims(non_ascii_request))
+
+ assert non_ascii_response.status == 401
+ assert json.loads(non_ascii_response.body.decode()) == {"error": "unauthorized"}
+
+
+def test_argus_middleware_only_bypasses_global_key_for_exact_list_path() -> None:
+ async def ok_handler(_request):
+ return web.Response(status=204)
+
+ app = web.Application()
+ list_request = make_mocked_request("GET", "/api/kb/claims", app=app)
+ detail_request = make_mocked_request(
+ "GET", f"/api/kb/claims/{CLAIM_ID}", app=app
+ )
+ html_request = make_mocked_request("GET", f"/kb/claims/{CLAIM_ID}", app=app)
+
+ list_response = asyncio.run(argus_app.auth_middleware(list_request, ok_handler))
+ detail_response = asyncio.run(argus_app.auth_middleware(detail_request, ok_handler))
+ html_response = asyncio.run(argus_app.auth_middleware(html_request, ok_handler))
+
+ assert list_response.status == 204
+ assert detail_response.status == 503
+ assert html_response.status == 503
+
+
+def test_canonical_claim_list_loader_runs_off_the_event_loop() -> None:
+ app = _claim_list_app()
+ event_loop_thread = threading.get_ident()
+ loader_thread = None
+
+ def loader(_filters):
+ nonlocal loader_thread
+ loader_thread = threading.get_ident()
+ return {"total": 0, "rows": []}
+
+ app[routes.KB_CLAIM_LIST_LOADER_KEY] = loader
+ request = _claim_list_request("/api/kb/claims", app=app)
+
+ response = asyncio.run(routes.handle_api_kb_claims(request))
+
+ assert response.status == 200
+ assert loader_thread is not None
+ assert loader_thread != event_loop_thread
+
+
+def test_canonical_claim_list_loader_timeout_is_sanitized(monkeypatch) -> None:
+ app = _claim_list_app()
+
+ def slow_loader(_filters):
+ time.sleep(0.03)
+ return {"total": 0, "rows": []}
+
+ app[routes.KB_CLAIM_LIST_LOADER_KEY] = slow_loader
+ monkeypatch.setattr(routes, "CLAIM_LIST_HANDLER_TIMEOUT_SECONDS", 0.001)
+ request = _claim_list_request("/api/kb/claims", app=app)
+
+ response = asyncio.run(routes.handle_api_kb_claims(request))
+
+ assert response.status == 503
+ assert json.loads(response.body.decode()) == {"error": "canonical_claims_unavailable"}
+
+
+def test_canonical_claim_list_rejects_duplicate_or_out_of_order_pages() -> None:
+ filters = {
+ "q": "",
+ "status": "all",
+ "type": "",
+ "tag": "",
+ "limit": 2,
+ "cursor": "",
+ "cursor_values": None,
+ }
+ newest = _claim_summary(CLAIM_ID, timestamp="2026-07-13T21:12:00+00:00")
+ older = _claim_summary(SECOND_CLAIM_ID, timestamp="2026-07-13T20:12:00+00:00")
+
+ with pytest.raises(ValueError, match="duplicate ids"):
+ routes._claim_list_payload({"total": 2, "rows": [newest, newest]}, filters)
+
+ with pytest.raises(ValueError, match="strictly ordered"):
+ routes._claim_list_payload({"total": 2, "rows": [older, newest]}, filters)
+
+
+def test_canonical_claim_list_rejects_rows_at_or_before_wrong_cursor_side() -> None:
+ cursor_timestamp = "2026-07-13T20:12:00+00:00"
+ filters = {
+ "q": "",
+ "status": "all",
+ "type": "",
+ "tag": "",
+ "limit": 1,
+ "cursor": routes.encode_claim_cursor(cursor_timestamp, SECOND_CLAIM_ID),
+ "cursor_values": (cursor_timestamp, SECOND_CLAIM_ID),
+ }
+ newer = _claim_summary(CLAIM_ID, timestamp="2026-07-13T21:12:00+00:00")
+
+ with pytest.raises(ValueError, match="crossed its request cursor"):
+ routes._claim_list_payload({"total": 2, "rows": [newer]}, filters)
+
+
+def test_legacy_claim_errors_do_not_expose_exception_text() -> None:
+ app = web.Application()
+
+ def fail(_claim_id):
+ raise RuntimeError("PGPASSWORD=must-not-leak")
+
+ app[routes.KB_CLAIM_LOADER_KEY] = fail
+ api_request = make_mocked_request(
+ "GET",
+ f"/api/kb/claims/{CLAIM_ID}",
+ app=app,
+ match_info={"claim_id": CLAIM_ID},
+ )
+ html_request = make_mocked_request(
+ "GET",
+ f"/kb/claims/{CLAIM_ID}",
+ app=app,
+ match_info={"claim_id": CLAIM_ID},
+ )
+
+ api_response = asyncio.run(routes.handle_api_kb_claim(api_request))
+ html_response = asyncio.run(routes.handle_kb_claim_page(html_request))
+
+ assert api_response.status == 503
+ assert html_response.status == 500
+ assert b"PGPASSWORD" not in api_response.body
+ assert b"PGPASSWORD" not in html_response.body