"""Browser login and protected review routes, enabled by owner configuration."""
from __future__ import annotations
import html
from http.cookies import CookieError, SimpleCookie
import json
import os
import secrets
import re
import sqlite3
from urllib.parse import parse_qs
from .oidc import KeyCapeLogin, LoginError, ORIGIN
from .approval_client import ApprovalEngineError
from .disposition import DispositionRefused, Verb
from .policy import PolicyError
from .provenance import HumanControlNotDischargeable
from .review import ReviewError
from .store import Conflict, EvidenceUnavailable, StoreError
from .ui import STYLES, document, error_page, review_page
FLOW_COOKIE = "__Host-infd-flow"
SESSION_COOKIE = "__Host-infd-session"
def _cookie(name: str, value: str, seconds: int) -> tuple[str, str]:
return "Set-Cookie", f"{name}={value}; Path=/; Secure; HttpOnly; SameSite=Lax; Max-Age={seconds}"
def _one(query: str, *, limit=8192, fields=16) -> dict[str, str]:
if len(query) > limit:
raise ValueError("request too large")
values = parse_qs(query, keep_blank_values=True, max_num_fields=fields, errors="strict")
if any(len(v) != 1 for v in values.values()):
raise ValueError("duplicate parameter")
return {k: v[0] for k, v in values.items()}
class App:
def __init__(self, login: KeyCapeLogin, review=None, *, readiness=lambda: False):
self.login = login
self.review, self.readiness = review, readiness
def __call__(self, environ, start_response):
path = environ.get("PATH_INFO", "/")
# Chromium sends Origin: null for a form under no-referrer. Review
# pages need same-origin so legitimate POSTs satisfy the exact-origin
# CSRF check. Authentication URLs and downloads never send a referrer.
referrer_policy = "no-referrer" if path.startswith("/auth/") or "/packet/" in path else "same-origin"
headers = [
("Cache-Control", "no-store"), ("Pragma", "no-cache"),
("Referrer-Policy", referrer_policy), ("X-Content-Type-Options", "nosniff"),
("Content-Security-Policy", "default-src 'none'; style-src 'self'; form-action 'self'; frame-ancestors 'none'; base-uri 'none'"),
("Strict-Transport-Security", "max-age=31536000"),
]
try:
cookies = SimpleCookie()
cookies.load(environ.get("HTTP_COOKIE", ""))
sid = cookies[SESSION_COOKIE].value if SESSION_COOKIE in cookies else ""
browser = cookies[FLOW_COOKIE].value if FLOW_COOKIE in cookies else ""
if not sid.isascii() or not browser.isascii():
raise ValueError("invalid cookie")
status, body, content_type, extra = self.dispatch(environ, sid, browser)
headers.extend(extra)
except LoginError:
status, body, content_type = 401, "Sign-in could not be completed. Start a new sign-in.", "text/plain"
headers.append(_cookie(FLOW_COOKIE, "", 0))
except (ValueError, CookieError, UnicodeError):
status, body, content_type = 400, "Invalid request.", "text/plain"
except ReviewError as exc:
messages = {
"session_expired": "Your session expired. Sign in again to continue.",
"invalid_form": "This form could not be verified. Reopen the review before submitting again.",
"policy_denied": "The permission service refused this review action.",
"wrong_recipient": "This review is addressed to another person.",
"stale_presentation": "The memo changed. Open its current version before taking an action.",
"binding_changed": "The approval no longer matches this memo. A revised review is required.",
"renderer_changed": "This memo names an earlier review interface. A new memo version is required before showing it here.",
"unsupported_review_profile": "This review interface supports English organizational approvals. The memo requires a different review profile.",
"existing_entry_unlinked": "An approval entry already exists, but this review cannot prove its original presentation. Operator recovery is required; no new entry was submitted.",
"act_unavailable": "This approval is no longer open for a new entry.",
}
status, body, content_type = exc.status, error_page(messages.get(exc.code,
"The review could not complete its required checks. Reload after the service is available.")), "text/html"
except DispositionRefused as exc:
status, body, content_type = 409, error_page({
"G_ACK": "Record all required highlight acknowledgments before accepting or declining.",
"G_REASONS": "Choose a reason when returning this memo.",
"G_PRES": "The memo changed. Open its current version before taking an action.",
}.get(exc.guard, "This action is not available for the current review.")), "text/html"
except HumanControlNotDischargeable:
status, body, content_type = 403, error_page("A verified human session is required."), "text/html"
except EvidenceUnavailable:
status, body, content_type = 404, error_page("The requested review or its evidence is unavailable."), "text/html"
except Conflict:
status, body, content_type = 409, error_page("The review changed or already has a submission. Reload the original record."), "text/html"
except (PolicyError, ApprovalEngineError, StoreError, sqlite3.Error):
status, body, content_type = 503, error_page("The required service or evidence store is unavailable. A submission may be unresolved; reopen the original review before taking another action."), "text/html"
raw = body if isinstance(body, bytes) else body.encode()
headers += [("Content-Type", content_type + ("; charset=utf-8" if content_type != "application/octet-stream" else "")), ("Content-Length", str(len(raw)))]
start_response(f"{status} {'OK' if status < 400 else 'ERROR'}", headers)
return [raw]
def dispatch(self, environ, sid, browser):
method, path = environ.get("REQUEST_METHOD", "GET"), environ.get("PATH_INFO", "/")
if method == "GET" and path == "/healthz":
return 200, '{"status":"ok"}', "application/json", []
if method == "GET" and path == "/readyz":
if self.review is not None and self.readiness():
return 200, '{"status":"ready"}', "application/json", []
return 503, json.dumps({"status": "incomplete", "reason": "approval_path_not_connected"}), "application/json", []
if method == "GET" and path == "/assets/review.css":
return 200, STYLES, "text/css", []
if method == "GET" and path == "/auth/start":
url, browser = self.login.start()
return 303, "", "text/plain", [("Location", url), _cookie(FLOW_COOKIE, browser, 300)]
if method == "GET" and path == "/auth/callback":
params = _one(environ.get("QUERY_STRING", ""))
# Even an issuer refusal consumes state. Never echo error_description.
new_sid = self.login.finish(params.get("state", ""), browser,
"" if "error" in params else params.get("code", ""))
self.login.logout(sid)
return 303, "", "text/plain", [("Location", "/"), _cookie(FLOW_COOKIE, "", 0),
_cookie(SESSION_COOKIE, new_sid, 900)]
session = self.login.session(sid)
if self.review is not None and (path == "/review" or path.startswith("/presentations/")):
if session is None:
raise ReviewError(401, "session_expired")
if method == "GET" and path == "/review":
params = _one(environ.get("QUERY_STRING", ""))
if set(params) != {"memo_id"} or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,255}", params["memo_id"]):
raise ValueError("invalid memo id")
page = self.review.open(session, params["memo_id"])
return 200, review_page(page, session), "text/html", []
match = re.fullmatch(r"/presentations/(pres-[a-f0-9-]{36})(?:/(ack|act|packet/([0-9]{1,3})))?", path)
if not match:
return 404, "Not found.", "text/plain", []
presentation_id, route, index = match.groups()
if method == "GET" and (route is None or index is not None):
page = self.review.load(session, presentation_id)
if index is not None:
number = int(index)
if number >= len(page.memo.packet):
return 404, "Not found.", "text/plain", []
item = page.memo.packet[number]
return 200, page.documents[item.item_id], "application/octet-stream", [
("Content-Disposition", f'attachment; filename="review-document-{number + 1}.bin"')]
return 200, review_page(page, session), "text/html", []
if method == "POST" and route in ("ack", "act"):
params = self._form(environ, session)
if route == "ack":
if any(k != "csrf" and not re.fullmatch(r"h[0-9]{1,3}", k) for k in params):
raise ValueError("unexpected acknowledgment field")
self.review.acknowledge(session, presentation_id, [v for k,v in params.items() if k != "csrf"])
target = presentation_id
else:
if set(params) - {"csrf", "verb", "operation_id", "reason", "note"}:
raise ValueError("unexpected action field")
if not re.fullmatch(r"[a-f0-9-]{36}", params.get("operation_id", "")):
raise ValueError("invalid operation id")
if params.get("verb") == "accept" and not self.readiness():
raise ReviewError(503, "audit_delivery_unavailable")
target = self.review.act(session, presentation_id, Verb(params.get("verb")),
operation_id=params["operation_id"], reasons=(params["reason"],) if params.get("reason") else (),
note=params.get("note") or None)
return 303, "", "text/plain", [("Location", "/presentations/" + target)]
return 405, "Method not allowed.", "text/plain", [("Allow", "POST" if route in ("ack", "act") else "GET")]
if method == "POST" and path == "/auth/logout":
if environ.get("HTTP_ORIGIN") != ORIGIN or not session:
return 403, "Invalid sign-out request.", "text/plain", []
length = int(environ.get("CONTENT_LENGTH") or 0)
if not 0 < length <= 2048 or environ.get("CONTENT_TYPE", "").split(";")[0] != "application/x-www-form-urlencoded":
raise ValueError("invalid body")
raw = environ["wsgi.input"].read(length)
if len(raw) != length:
raise ValueError("truncated body")
csrf = _one(raw.decode()).get("csrf", "")
if not csrf.isascii() or not secrets.compare_digest(csrf, session.csrf):
return 403, "Invalid sign-out request.", "text/plain", []
self.login.logout(sid)
return 303, "", "text/plain", [("Location", "/"), _cookie(SESSION_COOKIE, "", 0)]
if method == "GET" and path == "/":
if session:
content = (f"Signed in as {html.escape(session.subject)}.
"
+ ('Open a decision review
Enter the memo identifier supplied with your review request.
'
'' if self.review is not None else
'Decision review is being prepared. No approval has been recorded.
') +
'')
else:
content = 'Sign in with KeyCape
'
return 200, document("Informed Decision", content, session.subject if session else None), "text/html", []
return 404, "Not found.", "text/plain", []
def _form(self, environ, session):
if environ.get("HTTP_ORIGIN") != ORIGIN:
raise ReviewError(403, "invalid_form")
length = int(environ.get("CONTENT_LENGTH") or 0)
if not 0 < length <= 65536 or environ.get("CONTENT_TYPE", "").split(";")[0] != "application/x-www-form-urlencoded":
raise ValueError("invalid body")
raw = environ["wsgi.input"].read(length)
if len(raw) != length:
raise ValueError("truncated body")
params = _one(raw.decode(), limit=65536, fields=256)
csrf = params.get("csrf", "")
if not csrf.isascii() or not secrets.compare_digest(csrf, session.csrf):
raise ReviewError(403, "invalid_form")
return params
def main():
from waitress import serve
host = os.environ.get("INFD_LISTEN_HOST", "127.0.0.1")
if host not in ("127.0.0.1", "0.0.0.0"):
raise ValueError("unsupported listener address")
# Waitress does not log request targets; a proxy must also omit callback
# query strings and cookies. No debug traceback middleware belongs here.
login = KeyCapeLogin(os.environ["INFD_KEYCAPE_ISSUER"])
runtime = None
if os.environ.get("INFD_REVIEW_CONFIG"):
from .runtime import Runtime
runtime = Runtime.from_file(os.environ["INFD_REVIEW_CONFIG"])
runtime.pump.start()
app = App(login, runtime.controller if runtime else None,
readiness=runtime.pump.ready if runtime else lambda: False)
try:
serve(app, host=host, port=8080, threads=4)
finally:
if runtime:
runtime.pump.stop()