Fine login, with a fine card.

#!/usr/bin/env python3
# Copyright 2026 Petri Koistinen <petri.koistinen@iki.fi>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied. See the License for the specific language governing
# permissions and limitations under the License.

"""FINEID client-certificate guestbook CGI.

Apache authenticates the card certificate before invoking this program.
The application accepts identity only from mod_ssl's CGI environment; an
HTTP request header with a similar name becomes HTTP_* and is ignored.
"""

from __future__ import annotations

import hashlib
import hmac
import html
import os
import re
import sqlite3
import sys
import time
import urllib.parse
from contextlib import closing
from dataclasses import dataclass
from pathlib import Path


MAX_BODY_BYTES = 4096
MAX_MESSAGE_CHARS = 280
MAX_DISPLAY_NAME_CHARS = 160
DISPLAY_LIMIT = 50
CSRF_LIFETIME_SECONDS = 10 * 60
CSRF_CLOCK_SKEW_SECONDS = 30
PEUIN_CHECK_CHARACTERS = "0123456789ABCDEFHJKLMNPRSTUVWXY"
PAGE_STYLE = """
<style>
:root {
  --paper: #f3f5f7;
  --raised: #fafbfc;
  --ink: #101820;
  --muted: #536273;
  --blue: #002f6c;
  --line: #d4dde8;
  --max: 96rem;
  --sans: Inter, "Avenir Next", "Helvetica Neue", Arial, sans-serif;
  --serif: Iowan Old Style, Baskerville, Georgia, serif;
  --mono: "SF Mono", "Cascadia Code", "Liberation Mono", Menlo, monospace;
}
* {
  box-sizing: border-box;
}
body {
  margin: 0;
  background: var(--paper);
  color: var(--ink);
  font: 18px/1.58 var(--sans);
}
main {
  width: min(var(--max), calc(100% - 40px));
  margin: 0 auto;
  padding: 64px 0 80px;
}
a {
  color: var(--blue);
  text-underline-offset: 4px;
}
h1, h2 {
  font-family: var(--serif);
}
h1 {
  margin: .15em 0 .4em;
  font-size: clamp(3rem, 7vw, 6.5rem);
  line-height: 1;
  letter-spacing: 0;
}
h2 {
  margin-top: 2em;
  font-size: 2rem;
}
.eyebrow, .status, .meta, .entry footer, label, button, .button {
  font-family: var(--mono);
}
.eyebrow, .status {
  color: var(--blue);
  font-size: .8rem;
  font-weight: 700;
  letter-spacing: .12em;
  text-transform: uppercase;
}
.proof, .reason, .help {
  margin: 32px 0;
  border-left: 4px solid var(--blue);
  background: #e8eef6;
  padding: 24px 28px;
}
.help h2 {
  margin: 0 0 16px;
}
.meta, .entry footer, .empty {
  color: var(--muted);
  font-size: .85rem;
}
form.compose {
  margin: 48px 0;
}
label {
  display: block;
  margin-bottom: 8px;
  font-weight: 700;
}
textarea {
  display: block;
  width: 100%;
  min-height: 120px;
  border: 1px solid var(--line);
  border-radius: 4px;
  background: var(--raised);
  color: var(--ink);
  padding: 16px;
  font: inherit;
}
button, .button {
  display: inline-block;
  margin-top: 12px;
  border: 1px solid var(--blue);
  border-radius: 4px;
  background: var(--blue);
  color: #fff;
  padding: 12px 22px;
  font-weight: 700;
  text-decoration: none;
  cursor: pointer;
}
button:disabled {
  cursor: default;
  opacity: .6;
}
.entry {
  border-top: 1px solid var(--line);
  padding: 22px 0;
}
.entry p {
  font-size: 1.2rem;
  white-space: pre-wrap;
  overflow-wrap: anywhere;
}
.notice {
  border-left: 4px solid var(--blue);
  padding-left: 14px;
}
li {
  margin: .65em 0;
}
@media (max-width: 560px) {
  body {
    font-size: 16px;
  }
}
</style>
"""


@dataclass(frozen=True)
class Entry:
    """One public guestbook entry."""

    entry_id: int
    public_name: str
    message: str
    created_at: int


class RequestError(Exception):
    """Expected request rejection with an HTTP status."""

    def __init__(self, status: str, message: str, *, card_help: bool = False) -> None:
        super().__init__(message)
        self.status = status
        self.message = message
        self.card_help = card_help


def valid_peuin(value: str) -> bool:
    """Check the statutory eight digits and modulo-31 check character."""

    if not re.fullmatch(r"[0-9]{8}[0-9A-Z]", value):
        return False
    expected = PEUIN_CHECK_CHARACTERS[int(value[:8]) % 31]
    return hmac.compare_digest(value[-1], expected)


def extract_peuin_from_environment(environ: dict[str, str]) -> str:
    """Require mod_ssl's unambiguous structured serialNumber component."""

    structured = environ.get("SSL_CLIENT_S_DN_serialNumber", "")
    if not structured or environ.get("SSL_CLIENT_S_DN_serialNumber_1", ""):
        raise RequestError(
            "403 Forbidden",
            "The authentication certificate has no unique PEUIN.",
            card_help=True,
        )
    peuin = structured.upper()
    if not valid_peuin(peuin):
        raise RequestError(
            "403 Forbidden",
            "The authentication certificate contains a malformed PEUIN.",
            card_help=True,
        )
    return peuin


def verified_peuin(environ: dict[str, str]) -> str:
    """Return PEUIN only when mod_ssl reports a verified client chain."""

    if environ.get("HTTPS", "").lower() != "on":
        raise RequestError("403 Forbidden", "HTTPS is required.", card_help=True)
    verification = environ.get("SSL_CLIENT_VERIFY", "")
    if not hmac.compare_digest(verification, "SUCCESS"):
        if verification in ("", "NONE"):
            message = (
                "No authentication certificate was presented by the browser. "
                "The card was not used for this connection."
            )
        else:
            message = (
                "The browser presented a certificate, but the server could not "
                "verify it as a current FINEID authentication certificate."
            )
        raise RequestError("403 Forbidden", message, card_help=True)
    return extract_peuin_from_environment(environ)


def emrtd_name(name: str) -> str:
    """Render uppercase Finnish certificate spelling as a human name."""

    native = name.upper()
    replacements = (
        ("AA", "\u00c5"),
        ("AE", "\u00c4"),
        ("OE", "\u00d6"),
    )
    for encoded, character in replacements:
        native = native.replace(encoded, character)
    return re.sub(
        r"[^\W\d_]+",
        lambda match: match.group(0).capitalize(),
        native,
    )


def certificate_display_name(environ: dict[str, str], peuin: str) -> str:
    """Return a given-name-first public name from structured DN components."""

    given_names = " ".join(environ.get("SSL_CLIENT_S_DN_G", "").split())
    surname = " ".join(environ.get("SSL_CLIENT_S_DN_S", "").split())
    duplicated = environ.get("SSL_CLIENT_S_DN_G_1", "") or environ.get(
        "SSL_CLIENT_S_DN_S_1", ""
    )
    components = (given_names, surname)
    valid = (
        not duplicated
        and all(components)
        and sum(map(len, components)) + 1 <= MAX_DISPLAY_NAME_CHARS
        and all(
            character.isalpha() or character in " -'"
            for component in components
            for character in component
        )
    )
    if not valid:
        raise RequestError(
            "403 Forbidden",
            "The authentication certificate has no usable holder name.",
            card_help=True,
        )
    public_name = f"{emrtd_name(given_names)} {emrtd_name(surname)}"
    if peuin.upper() in public_name.upper():
        raise RequestError(
            "403 Forbidden",
            "The authentication certificate holder name exposes PEUIN.",
            card_help=True,
        )
    return public_name


def certificate_given_name(environ: dict[str, str]) -> str:
    """Return the first given name from mod_ssl's structured field."""

    given_names = environ.get("SSL_CLIENT_S_DN_G", "")
    candidate = given_names.split()[0] if given_names else ""
    valid = candidate and all(
        character.isalpha() or character in "-'" for character in candidate
    )
    if not valid:
        raise RequestError(
            "403 Forbidden",
            "The authentication certificate has no usable given name.",
            card_help=True,
        )
    return emrtd_name(candidate)


def load_key(path: Path) -> bytes:
    """Load the server-only key used for identity-bound CSRF tokens."""

    key = path.read_bytes()
    if len(key) < 32:
        raise RuntimeError("guestbook key must contain at least 32 bytes")
    return key


def keyed_token(key: bytes, purpose: bytes, peuin: str) -> bytes:
    """Domain-separated HMAC over a PEUIN."""

    return hmac.new(key, purpose + b"\0" + peuin.encode("ascii"), hashlib.sha256).digest()


def csrf_token(key: bytes, peuin: str, now: int) -> str:
    """Create a short-lived identity-bound CSRF token."""

    issued = str(now)
    payload = f"{peuin}\0{issued}"
    mac = keyed_token(key, b"csrf-v2", payload).hex()
    return f"{issued}.{mac}"


def require_csrf(key: bytes, peuin: str, supplied: str, now: int) -> None:
    """Reject malformed, expired, future, or incorrectly bound CSRF tokens."""

    issued_text, separator, _mac = supplied.partition(".")
    if not separator or not issued_text.isascii() or not issued_text.isdigit():
        raise RequestError("403 Forbidden", "The form expired. Reload and try again.")
    issued = int(issued_text)
    age = now - issued
    expected = csrf_token(key, peuin, issued)
    valid_age = -CSRF_CLOCK_SKEW_SECONDS <= age <= CSRF_LIFETIME_SECONDS
    if not valid_age or not hmac.compare_digest(supplied, expected):
        raise RequestError("403 Forbidden", "The form expired. Reload and try again.")


def remove_stored_identities(db: sqlite3.Connection) -> None:
    """Preserve messages while removing legacy identity columns and votes."""

    columns = {row[1] for row in db.execute("PRAGMA table_info(entries)")}
    if not ({"peuin", "account_id"} & columns):
        return
    db.execute("BEGIN IMMEDIATE")
    try:
        db.execute(
            "CREATE TABLE entries_without_identity ("
            "id INTEGER PRIMARY KEY AUTOINCREMENT, "
            "public_id TEXT NOT NULL, "
            "message TEXT NOT NULL, "
            "created_at INTEGER NOT NULL)"
        )
        db.execute(
            "INSERT INTO entries_without_identity(id, public_id, message, created_at) "
            "SELECT id, public_id, message, created_at FROM entries"
        )
        db.execute("DROP TABLE IF EXISTS deletion_votes")
        db.execute("DROP TABLE entries")
        db.execute("ALTER TABLE entries_without_identity RENAME TO entries")
        db.commit()
    except Exception:
        db.rollback()
        raise


def connect_db(path: Path) -> sqlite3.Connection:
    """Open and initialize the durable SQLite store."""

    db = sqlite3.connect(path, timeout=5.0)
    db.execute("PRAGMA busy_timeout = 5000")
    db.execute("PRAGMA journal_mode = WAL")
    db.execute("PRAGMA foreign_keys = ON")
    remove_stored_identities(db)
    db.execute(
        """
        CREATE TABLE IF NOT EXISTS entries (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            public_id TEXT NOT NULL,
            message TEXT NOT NULL,
            created_at INTEGER NOT NULL
        )
        """
    )
    db.execute(
        "CREATE INDEX IF NOT EXISTS entries_created_at_idx "
        "ON entries(created_at DESC)"
    )
    return db


def recent_entries(db: sqlite3.Connection) -> list[Entry]:
    """Fetch recent public entries."""

    rows = db.execute(
        "SELECT id, public_id, message, created_at FROM entries "
        "ORDER BY entries.created_at DESC, entries.id DESC LIMIT ?",
        (DISPLAY_LIMIT,),
    ).fetchall()
    return [Entry(*row) for row in rows]


def read_form(environ: dict[str, str], stream: object) -> dict[str, list[str]]:
    """Read a tightly bounded URL-encoded POST body."""

    content_type = environ.get("CONTENT_TYPE", "").split(";", 1)[0].strip().lower()
    if content_type != "application/x-www-form-urlencoded":
        raise RequestError("415 Unsupported Media Type", "Use the guestbook form.")
    try:
        length = int(environ.get("CONTENT_LENGTH", "0"))
    except ValueError as error:
        raise RequestError("400 Bad Request", "Invalid request length.") from error
    if length < 0 or length > MAX_BODY_BYTES:
        raise RequestError("413 Content Too Large", "The form submission is too large.")
    body = stream.read(length)
    if not isinstance(body, bytes) or len(body) != length:
        raise RequestError("400 Bad Request", "The form submission was incomplete.")
    try:
        text = body.decode("utf-8", errors="strict")
    except UnicodeDecodeError as error:
        raise RequestError("400 Bad Request", "The form must be UTF-8.") from error
    return urllib.parse.parse_qs(text, keep_blank_values=True, strict_parsing=False)


def single_field(form: dict[str, list[str]], name: str) -> str:
    """Require exactly one value for a form field."""

    values = form.get(name, [])
    if len(values) != 1:
        raise RequestError("400 Bad Request", "The form is invalid.")
    return values[0]


def add_entry(
    db: sqlite3.Connection,
    key: bytes,
    peuin: str,
    public_name: str,
    form: dict[str, list[str]],
    now: int,
) -> None:
    """Validate and insert one public message without storing identity."""

    supplied_csrf = single_field(form, "csrf")
    require_csrf(key, peuin, supplied_csrf, now)
    message = single_field(form, "message").strip()
    if not message:
        raise RequestError("400 Bad Request", "Write a message first.")
    if len(message) > MAX_MESSAGE_CHARS:
        raise RequestError(
            "400 Bad Request",
            f"Messages may contain at most {MAX_MESSAGE_CHARS} characters.",
        )
    with db:
        db.execute(
            "INSERT INTO entries(public_id, message, created_at) VALUES (?, ?, ?)",
            (public_name, message, now),
        )


def page(
    peuin: str,
    public_name: str,
    given_name: str,
    key: bytes,
    entries: list[Entry],
    now: int,
    notice: str = "",
) -> str:
    """Render the authenticated guestbook without disclosing any PEUIN."""

    token = csrf_token(key, peuin, now)
    cards = []
    for entry in entries:
        timestamp = time.strftime("%Y-%m-%d %H:%M UTC", time.gmtime(entry.created_at))
        cards.append(
            "<article class=entry>"
            f"<p>{html.escape(entry.message)}</p>"
            f"<footer>{html.escape(entry.public_name)} -- {timestamp}</footer>"
            "</article>"
        )
    entry_html = "".join(cards) or "<p class=empty>No messages yet. Be the first.</p>"
    notice_html = f"<p class=notice>{html.escape(notice)}</p>" if notice else ""
    return f"""<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Card-authenticated guestbook -- ReFineID</title>
{PAGE_STYLE}
</head>
<body><main>
<h1>Hi {html.escape(given_name)}!</h1>
<p class="meta">Logging in creates no guestbook record. Publishing stores only your name, message, and publication time -- never PEUIN (SATU), a derived identifier, or PIN1.</p>
{notice_html}
<form class="compose" method="post" action="/guestbook">
<input type="hidden" name="action" value="post">
<input type="hidden" name="csrf" value="{token}">
<label for="message">Sign the guestbook</label>
<textarea id="message" name="message" maxlength="{MAX_MESSAGE_CHARS}" required></textarea>
<button type="submit">Publish as {html.escape(public_name)}</button>
</form>
<section><h2>Recent messages</h2>{entry_html}</section>
<p><a href="https://www.refineid.fi/demo/">Read how this login works</a></p>
<p><a class="button" href="https://www.refineid.fi/">Log out</a></p>
</main></body></html>"""


def emit(status: str, body: str, *, location: str | None = None) -> None:
    """Emit a CGI response with strict security headers."""

    print(f"Status: {status}")
    if location is not None:
        print(f"Location: {location}")
    print("Content-Type: text/html; charset=utf-8")
    print("Cache-Control: no-store, max-age=0")
    print("Pragma: no-cache")
    print("X-Content-Type-Options: nosniff")
    print("X-Frame-Options: DENY")
    print("Referrer-Policy: no-referrer")
    print("Content-Security-Policy: default-src 'none'; style-src 'unsafe-inline'; form-action 'self'; base-uri 'none'; frame-ancestors 'none'")
    print()
    print(body)


def error_page(error: RequestError) -> str:
    """Render a non-sensitive rejection page with card-login recovery help."""

    if error.card_help:
        heading = "Card login did not complete"
        recovery = """
<section class="help">
<h2>What to check</h2>
<ol>
<li>Connect the reader and insert the card before reloading this page.</li>
<li>In Firefox, enable the ReFineID PKCS #11 security module.</li>
<li>Select the authentication certificate when the browser asks.</li>
<li>Do not keep guessing PIN1. Check its retry counter in ReFineID Card Manager.</li>
</ol>
</section>"""
    else:
        heading = "Request not accepted"
        recovery = ""
    return f"""<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{html.escape(heading)} -- ReFineID</title>
{PAGE_STYLE}
</head>
<body>
<main>
<p class="status">{html.escape(error.status)}</p>
<h1>{html.escape(heading)}</h1>
<p class="reason">{html.escape(error.message)}</p>
{recovery}
<p><a class="button" href="https://www.refineid.fi/demo/">Open the card-login guide</a></p>
</main></body></html>"""


def required_path(environ: dict[str, str], name: str) -> Path:
    """Require an explicit deployment path; never open a shadow data store."""

    value = environ.get(name, "")
    if not value:
        raise RuntimeError(f"required environment variable {name} is unset")
    return Path(value)


def application(
    environ: dict[str, str],
    stream: object,
    db_path: Path,
    key_path: Path,
    now: int,
) -> tuple[str, str, str | None]:
    """Handle one request; separated for deterministic tests."""

    peuin = verified_peuin(environ)
    public_name = certificate_display_name(environ, peuin)
    given_name = certificate_given_name(environ)
    key = load_key(key_path)
    method = environ.get("REQUEST_METHOD", "GET").upper()
    with closing(connect_db(db_path)) as db:
        if method == "POST":
            form = read_form(environ, stream)
            action = single_field(form, "action")
            if action == "post":
                add_entry(db, key, peuin, public_name, form, now)
                return "303 See Other", "", "/guestbook?posted=1"
            raise RequestError("400 Bad Request", "The form action is invalid.")
        if method != "GET":
            raise RequestError("405 Method Not Allowed", "Use GET or POST.")
        notices = {"posted=1": "Message published."}
        notice = notices.get(environ.get("QUERY_STRING"), "")
        return (
            "200 OK",
            page(
                peuin,
                public_name,
                given_name,
                key,
                recent_entries(db),
                now,
                notice,
            ),
            None,
        )


def main() -> int:
    """CGI entry point."""

    # Keep messages owner-only even though the database contains no PEUIN.
    os.umask(0o077)
    environ = dict(os.environ)
    try:
        db_path = required_path(environ, "REFINEID_GUESTBOOK_DB")
        key_path = required_path(environ, "REFINEID_GUESTBOOK_KEY")
        status, body, location = application(
            environ,
            sys.stdin.buffer,
            db_path,
            key_path,
            int(time.time()),
        )
        emit(status, body, location=location)
        return 0
    except RequestError as error:
        emit(error.status, error_page(error))
        return 0
    except Exception as error:
        print(f"guestbook: internal {type(error).__name__}", file=sys.stderr)
        emit(
            "500 Internal Server Error",
            "<!doctype html><html lang=en><meta charset=utf-8>"
            "<title>Server error</title><h1>The guestbook is temporarily unavailable.</h1>",
        )
        return 1


if __name__ == "__main__":
    raise SystemExit(main())