guestbook.py
"""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
from cryptography import x509
from cryptography.x509.oid import NameOID
MAX_MESSAGE_CHARS = 280
MAX_FORM_BYTES = 4096
DISPLAY_LIMIT = 50
DELETION_QUORUM = 2
CSRF_LIFETIME_SECONDS = 10 * 60
DB_PATH = Path("/home/guestbook/data/guestbook.sqlite3")
KEY_PATH = Path("/home/guestbook/guestbook.key")
PAGE_STYLE = """
<style>
body {
margin: 0;
background: #f3f5f7;
color: #101820;
font: 18px/1.5 system-ui, sans-serif;
}
main {
max-width: 96rem;
margin: 0 auto;
padding: 3rem 1.25rem;
}
a {
color: #002f6c;
}
h1, h2 {
font-family: serif;
}
h1 {
font-size: clamp(3rem, 7vw, 6rem);
line-height: 1.05;
}
.reason, .help, .notice {
border-left: .25rem solid #002f6c;
padding-left: 1rem;
}
.compose {
margin: 3rem 0;
}
label {
display: block;
}
textarea {
box-sizing: border-box;
display: block;
width: 100%;
min-height: 7rem;
border: 1px solid #b8c0c8;
padding: 1rem;
font: inherit;
}
button, .button {
display: inline-block;
border: 0;
background: #002f6c;
color: #fff;
padding: .75rem 1.25rem;
font: inherit;
font-weight: 700;
text-decoration: none;
}
.entry {
border-top: 1px solid #b8c0c8;
padding: 1.25rem 0;
}
.entry p {
white-space: pre-wrap;
overflow-wrap: anywhere;
}
.entry footer {
font-size: .9rem;
}
</style>
"""
@dataclass(frozen=True)
class Entry:
"""One public guestbook entry."""
entry_id: int
given_names: str
surname: str
message: str
created_at: int
deletion_votes: int
viewer_voted: bool
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 extract_peuin_from_environment(environ: dict[str, str]) -> str:
"""Return serialNumber from the certificate verified by mod_ssl."""
structured = environ.get("SSL_CLIENT_S_DN_serialNumber", "")
if structured:
return structured.upper()
pem = environ.get("SSL_CLIENT_CERT", "")
try:
certificate = x509.load_pem_x509_certificate(pem.encode("ascii"))
attributes = certificate.subject.get_attributes_for_oid(NameOID.SERIAL_NUMBER)
except (UnicodeError, ValueError):
attributes = []
if len(attributes) != 1:
raise RequestError(
"403 Forbidden",
"The authentication certificate has no PEUIN.",
card_help=True,
)
return attributes[0].value.upper()
def verified_peuin(environ: dict[str, str]) -> str:
"""Return PEUIN only when mod_ssl reports a verified client chain."""
if environ.get("SSL_CLIENT_VERIFY") != "SUCCESS":
raise RequestError(
"403 Forbidden",
"Card login did not complete. Choose the authentication certificate and try again.",
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_names(environ: dict[str, str]) -> tuple[str, str]:
"""Return the structured certificate name fields."""
given_names = " ".join(environ.get("SSL_CLIENT_S_DN_G", "").split())
surname = " ".join(environ.get("SSL_CLIENT_S_DN_S", "").split())
if not given_names or not surname:
raise RequestError(
"403 Forbidden",
"The authentication certificate has no usable holder name.",
card_help=True,
)
return given_names, surname
def display_name(given_names: str, surname: str) -> str:
"""Format stored certificate name fields for public display."""
return f"{emrtd_name(given_names)} {emrtd_name(surname)}"
def load_key(path: Path) -> bytes:
"""Load the server-only form-token key."""
return path.read_bytes()
def csrf_token(key: bytes, peuin: str, now: int) -> str:
"""Create a short-lived identity-bound CSRF token."""
issued = str(now)
mac = hmac.new(key, f"{peuin}\0{issued}".encode(), hashlib.sha256).hexdigest()
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."""
try:
issued = int(supplied.split(".", 1)[0])
except ValueError:
raise RequestError("403 Forbidden", "The form expired. Reload and try again.")
expected = csrf_token(key, peuin, issued)
if now - issued > CSRF_LIFETIME_SECONDS or not hmac.compare_digest(supplied, expected):
raise RequestError("403 Forbidden", "The form expired. Reload and try again.")
def connect_db(path: Path) -> sqlite3.Connection:
"""Open and initialize the durable SQLite store."""
db = sqlite3.connect(path)
db.execute("PRAGMA foreign_keys = ON")
db.execute(
"""
CREATE TABLE IF NOT EXISTS entries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
author_peuin TEXT NOT NULL,
given_names TEXT NOT NULL,
surname 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)"
)
db.execute(
"""
CREATE TABLE IF NOT EXISTS deletion_votes (
entry_id INTEGER NOT NULL REFERENCES entries(id) ON DELETE CASCADE,
voter_peuin TEXT NOT NULL,
PRIMARY KEY (entry_id, voter_peuin)
)
"""
)
return db
def recent_entries(db: sqlite3.Connection, viewer_peuin: str) -> list[Entry]:
"""Fetch recent public entries."""
rows = db.execute(
"SELECT entries.id, entries.given_names, entries.surname, entries.message, "
"entries.created_at, COUNT(deletion_votes.voter_peuin), "
"COUNT(CASE WHEN deletion_votes.voter_peuin = ? THEN 1 END) "
"FROM entries LEFT JOIN deletion_votes "
"ON deletion_votes.entry_id = entries.id "
"GROUP BY entries.id "
"ORDER BY entries.created_at DESC, entries.id DESC LIMIT ?",
(viewer_peuin, DISPLAY_LIMIT),
).fetchall()
return [Entry(*row[:-1], viewer_voted=bool(row[-1])) for row in rows]
def read_form(environ: dict[str, str], stream: object) -> dict[str, list[str]]:
"""Read the guestbook form."""
try:
length = int(environ.get("CONTENT_LENGTH", "0"))
except ValueError as error:
raise RequestError("400 Bad Request", "The form is invalid.") from error
if not 0 <= length <= MAX_FORM_BYTES:
raise RequestError("400 Bad Request", "The form is invalid.")
body = stream.read(length)
try:
return urllib.parse.parse_qs(body.decode(), keep_blank_values=True)
except UnicodeDecodeError as error:
raise RequestError("400 Bad Request", "The form is invalid.") from error
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,
given_names: str,
surname: str,
form: dict[str, list[str]],
now: int,
) -> None:
"""Validate and insert one public message with its certificate 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(author_peuin, given_names, surname, message, created_at) "
"VALUES (?, ?, ?, ?, ?)",
(peuin, given_names, surname, message, now),
)
def request_deletion(
db: sqlite3.Connection,
key: bytes,
peuin: str,
form: dict[str, list[str]],
now: int,
) -> tuple[bool, bool]:
"""Delete an entry after two distinct card holders agree."""
require_csrf(key, peuin, single_field(form, "csrf"), now)
try:
entry_id = int(single_field(form, "entry_id"))
except ValueError as error:
raise RequestError("400 Bad Request", "The deletion request is invalid.") from error
if entry_id < 1:
raise RequestError("400 Bad Request", "The deletion request is invalid.")
with db:
added = db.execute(
"INSERT OR IGNORE INTO deletion_votes(entry_id, voter_peuin) VALUES (?, ?)",
(entry_id, peuin),
).rowcount == 1
votes = db.execute(
"SELECT COUNT(*) FROM deletion_votes WHERE entry_id = ?", (entry_id,)
).fetchone()[0]
deleted = votes >= DELETION_QUORUM
if deleted:
db.execute("DELETE FROM entries WHERE id = ?", (entry_id,))
return added, deleted
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))
if entry.viewer_voted:
moderation = f"Deletion requested ({entry.deletion_votes}/{DELETION_QUORUM})"
else:
moderation = (
'<form class="moderation" method="post" action="/guestbook">'
'<input type="hidden" name="action" value="request-deletion">'
f'<input type="hidden" name="entry_id" value="{entry.entry_id}">'
f'<input type="hidden" name="csrf" value="{token}">'
f'<button type="submit">Vote for deletion '
f'({entry.deletion_votes}/{DELETION_QUORUM})</button></form>'
)
cards.append(
"<article class=entry>"
f"<p>{html.escape(entry.message)}</p>"
f"<footer>{html.escape(display_name(entry.given_names, entry.surname))} -- {timestamp}</footer>"
f"{moderation}"
"</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>
{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 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)
given_names, surname = certificate_names(environ)
public_name = display_name(given_names, surname)
given_name = emrtd_name(given_names.split()[0])
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, given_names, surname, form, now)
return "303 See Other", "", "/guestbook?posted=1"
if action == "request-deletion":
added, deleted = request_deletion(db, key, peuin, form, now)
if deleted:
return "303 See Other", "", "/guestbook?deleted=1"
if added:
return "303 See Other", "", "/guestbook?deletion-requested=1"
return "303 See Other", "", "/guestbook?already-requested=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.",
"deleted=1": "Entry deleted after two card holders agreed.",
"deletion-requested=1": "Deletion requested. One more card holder must agree.",
"already-requested=1": "You already requested deletion of that entry.",
}
notice = notices.get(environ.get("QUERY_STRING"), "")
return (
"200 OK",
page(
peuin,
public_name,
given_name,
key,
recent_entries(db, peuin),
now,
notice,
),
None,
)
def main() -> int:
"""CGI entry point."""
environ = dict(os.environ)
try:
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())