Ebbe BaßandClaude Sonnet 5.5
94c1d39c9d
Page at /email asks for the new address and the current password (shared failure counter with the password page, five requests per hour) and mails a one-hour confirmation link to the new address. Confirming writes the address and mails the old address a 24-hour undo link; undoing restores it, ends every session of the user and notifies the Admin (ADMIN_NOTIFY_EMAIL). Link pages show on GET and apply on POST; tokens are stored only as hashes, single use, and are blanked from every log line. New table owned by profile, German text, tests; review findings logged in the plan. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FocPJ75XPma7TcuYEMy8zm
72 lines
2.5 KiB
Python
72 lines
2.5 KiB
Python
"""JSON logs on stdout with a request id (AD-18)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from contextvars import ContextVar
|
|
from datetime import UTC, datetime
|
|
|
|
request_id_var: ContextVar[str] = ContextVar("request_id", default="-")
|
|
|
|
# Attributes set by logging itself; everything else passed via `extra` is emitted.
|
|
_STANDARD = set(logging.LogRecord("", 0, "", 0, "", (), None).__dict__) | {"message", "asctime"}
|
|
|
|
|
|
# One-time links carry their token in the path (AD-9); a log line must never hold it.
|
|
_LINK_TOKEN = re.compile(r"(/email/(?:confirm|undo)/)[^/\s\"']+")
|
|
|
|
|
|
def redact(text: str) -> str:
|
|
return _LINK_TOKEN.sub(r"\1<redacted>", text)
|
|
|
|
|
|
class RedactLinkTokens(logging.Filter):
|
|
"""Blank the token of a one-time link in a record's message."""
|
|
|
|
def filter(self, record: logging.LogRecord) -> bool:
|
|
message = record.getMessage()
|
|
redacted = redact(message)
|
|
if redacted != message:
|
|
record.msg, record.args = redacted, ()
|
|
return True
|
|
|
|
|
|
class JsonFormatter(logging.Formatter):
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
payload: dict[str, object] = {
|
|
"ts": datetime.fromtimestamp(record.created, UTC).isoformat(timespec="milliseconds"),
|
|
"level": record.levelname,
|
|
"logger": record.name,
|
|
"message": record.getMessage(),
|
|
"request_id": getattr(record, "request_id", None) or request_id_var.get(),
|
|
}
|
|
for key, value in record.__dict__.items():
|
|
if key in _STANDARD or key in ("request_id", "request"):
|
|
continue
|
|
if value is None or isinstance(value, str | int | float | bool):
|
|
payload[key] = value
|
|
if record.exc_info:
|
|
payload["exception"] = self.formatException(record.exc_info)
|
|
return json.dumps(payload, default=str)
|
|
|
|
|
|
def build_logging(level: str) -> dict:
|
|
return {
|
|
"version": 1,
|
|
"disable_existing_loggers": False,
|
|
"formatters": {"json": {"()": "portal.config.logging.JsonFormatter"}},
|
|
"filters": {"redact_link_tokens": {"()": "portal.config.logging.RedactLinkTokens"}},
|
|
"handlers": {
|
|
"stdout": {
|
|
"class": "logging.StreamHandler",
|
|
"stream": "ext://sys.stdout",
|
|
"formatter": "json",
|
|
"filters": ["redact_link_tokens"],
|
|
}
|
|
},
|
|
"loggers": {"django.request": {"level": "ERROR"}},
|
|
"root": {"handlers": ["stdout"], "level": level},
|
|
}
|