informed-decision/informed_decision/web.py

"""Browser login shell. Protected memo rendering/binding is not wired yet."""

from __future__ import annotations

import html
from http.cookies import CookieError, SimpleCookie
import json
import os
import secrets
from urllib.parse import parse_qs

from .oidc import KeyCapeLogin, LoginError, ORIGIN

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) -> dict[str, str]:
    if len(query) > 8192:
        raise ValueError("request too large")
    values = parse_qs(query, keep_blank_values=True, max_num_fields=16)
    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):
        self.login = login

    def __call__(self, environ, start_response):
        headers = [
            ("Cache-Control", "no-store"), ("Pragma", "no-cache"),
            ("Referrer-Policy", "no-referrer"), ("X-Content-Type-Options", "nosniff"),
            ("Content-Security-Policy", "default-src 'none'; 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"
        raw = body.encode()
        headers += [("Content-Type", content_type + "; charset=utf-8"), ("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":
            return 503, json.dumps({"status": "incomplete", "reason": "approval_path_not_connected"}), "application/json", []
        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 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"<p>Signed in as {html.escape(session.subject)}.</p>"
                    "<p>Decision review is being prepared. No approval has been recorded.</p>"
                    '<form method="post" action="/auth/logout">'
                    f'<input type="hidden" name="csrf" value="{session.csrf}">'
                    '<button type="submit">Sign out</button></form>')
            else:
                content = '<p><a href="/auth/start">Sign in with KeyCape</a></p>'
            return 200, ('<!doctype html><html lang="en"><meta charset="utf-8">'
                '<meta name="viewport" content="width=device-width, initial-scale=1">'
                '<title>Informed Decision</title><body><h1>Informed Decision</h1>'
                + content + '</body></html>'), "text/html", []
        return 404, "Not found.", "text/plain", []


def main():
    from waitress import serve
    # Waitress does not log request targets; a proxy must also omit callback
    # query strings and cookies. No debug traceback middleware belongs here.
    app = App(KeyCapeLogin(os.environ["INFD_KEYCAPE_ISSUER"]))
    serve(app, host="127.0.0.1", port=8080, threads=4)