peter
cca3315836
E2EE direct messages, the chat dock, the Messages module, the message WebSocket and its realtime registry, and the messages / message_keys tables (migration 0097; the downgrade restores the empty schema). Prod held two welcome notes, no message between people and one key — never used (owner decision, E34). WebSocket now has no owner module: the lint bans it everywhere. Privacy policy updated: it no longer promises E2EE for a feature that is gone.
437 lines
17 KiB
Python
437 lines
17 KiB
Python
"""E22 S2 — the one logging configuration, its redaction, and the request line that replaced
|
|
uvicorn's access log.
|
|
|
|
Redaction is the part worth guarding hardest: it is applied at the HANDLER, so these cases are the
|
|
only thing standing between a future call site and a token in a log we attach to a Gitea issue.
|
|
The request-line cases pin what uvicorn's access log got wrong — the query string (`?account=1`,
|
|
`?token=…`) was printed on every request, and nothing correlated two lines of the same request.
|
|
|
|
Pure logic + an in-process ASGI app: no database.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import logs
|
|
from app.config import LogLevel, choices_of
|
|
|
|
|
|
class _Sink(logging.Handler):
|
|
"""Captures records AND their formatted JSON, with the same filter/formatter the real stdout
|
|
handler uses — so a test sees exactly the bytes a deployment would write."""
|
|
|
|
def __init__(self):
|
|
super().__init__(level=logging.NOTSET)
|
|
self.lines: list[str] = []
|
|
self.records: list[logging.LogRecord] = []
|
|
self.setFormatter(logs.JsonFormatter())
|
|
self.addFilter(logs.ContextFilter())
|
|
|
|
def emit(self, record):
|
|
self.lines.append(self.format(record))
|
|
self.records.append(record)
|
|
|
|
def objects(self) -> list[dict]:
|
|
return [json.loads(line) for line in self.lines]
|
|
|
|
|
|
@pytest.fixture
|
|
def sink():
|
|
"""Attached to `siftlode` (not root) so a `configure_logging` call inside a test — which
|
|
replaces the ROOT handlers — can't silently detach it mid-test."""
|
|
logger = logging.getLogger("siftlode")
|
|
handler = _Sink()
|
|
previous = logger.level
|
|
logger.addHandler(handler)
|
|
logger.setLevel(logs.TRACE)
|
|
try:
|
|
yield handler
|
|
finally:
|
|
logger.removeHandler(handler)
|
|
logger.setLevel(previous)
|
|
|
|
|
|
# --- redaction ------------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"text, must_not_contain",
|
|
[
|
|
("user signed in: someone@example.com", "someone@example.com"),
|
|
("callback https://app/auth/callback?code=4/0AY&state=xyz", "4/0AY"),
|
|
("mail link https://app/#verify=abc123def", "abc123def"),
|
|
("share https://app/s?token=SHARETOKEN", "SHARETOKEN"),
|
|
("reset https://app/#reset=RESETTOKEN", "RESETTOKEN"),
|
|
],
|
|
)
|
|
def test_free_text_secrets_never_reach_the_line(text, must_not_contain):
|
|
assert must_not_contain not in logs.redact_text(text)
|
|
|
|
|
|
def test_a_key_that_names_a_secret_is_dropped_whole():
|
|
# Not truncated to a prefix: a prefix is still a leak and is useless for triage anyway.
|
|
out = logs.redact(
|
|
{
|
|
"password": "hunter2",
|
|
"current_password": "hunter2",
|
|
"plex_server_token": "abc",
|
|
"Cookie": "session=xyz",
|
|
"authorization": "Bearer abc",
|
|
"nested": {"refresh_token": "rt", "keep": "visible"},
|
|
"list": [{"api_key": "k"}],
|
|
"count": 7,
|
|
}
|
|
)
|
|
assert out["password"] == logs.REDACTED
|
|
assert out["current_password"] == logs.REDACTED
|
|
assert out["plex_server_token"] == logs.REDACTED
|
|
assert out["Cookie"] == logs.REDACTED
|
|
assert out["authorization"] == logs.REDACTED
|
|
assert out["nested"]["refresh_token"] == logs.REDACTED
|
|
assert out["nested"]["keep"] == "visible" # only the named keys go
|
|
assert out["list"][0]["api_key"] == logs.REDACTED
|
|
assert out["count"] == 7
|
|
|
|
|
|
def test_an_email_in_an_attribute_value_is_scrubbed_too():
|
|
assert logs.redact({"to": "a.b+tag@mail.example.org"})["to"] == logs.REDACTED_EMAIL
|
|
|
|
|
|
# The SAME cases frontend/src/lib/log.contract.test.ts runs against the browser's copy of redact():
|
|
# the one thing that keeps the two logs' idea of "safe to keep" from drifting apart.
|
|
_REDACTION_CASES = json.loads(
|
|
(Path(__file__).parent / "fixtures" / "log_redaction_cases.json").read_text(encoding="utf-8")
|
|
)["cases"]
|
|
|
|
|
|
@pytest.mark.parametrize("case", _REDACTION_CASES, ids=[c["name"] for c in _REDACTION_CASES])
|
|
def test_the_shared_redaction_cases(case):
|
|
assert logs.redact(case["value"], key=case["key"]) == case["expected"]
|
|
|
|
|
|
def test_redaction_runs_on_the_message_the_attrs_and_the_traceback(sink):
|
|
log = logging.getLogger("siftlode.test")
|
|
try:
|
|
raise RuntimeError("boom for someone@example.com")
|
|
except RuntimeError:
|
|
log.exception(
|
|
"failed for %s",
|
|
"someone@example.com",
|
|
extra=logs.event("test.failed", password="hunter2", url="https://x/?token=T0K3N"),
|
|
)
|
|
line = sink.lines[-1]
|
|
assert "someone@example.com" not in line
|
|
assert "hunter2" not in line
|
|
assert "T0K3N" not in line
|
|
payload = json.loads(line)
|
|
assert payload["attrs"]["password"] == logs.REDACTED
|
|
assert "RuntimeError" in payload["exc"] # the diagnosis survives the scrubbing
|
|
|
|
|
|
# --- event shape ----------------------------------------------------------------------------
|
|
|
|
|
|
def test_every_line_carries_the_agreed_fields(sink):
|
|
logging.getLogger("siftlode.test").info("hello", extra=logs.event("test.hello", n=1))
|
|
payload = sink.objects()[-1]
|
|
for field in ("ts", "level", "sev", "source", "logger", "msg", "event", "version"):
|
|
assert field in payload, field
|
|
assert payload["level"] == "INFO"
|
|
assert payload["sev"] == 9
|
|
assert payload["event"] == "test.hello"
|
|
assert payload["attrs"] == {"n": 1}
|
|
|
|
|
|
def test_a_plain_log_call_still_produces_a_well_formed_line(sink):
|
|
"""An event code is optional — 125 existing call sites have none yet (E22.5 adds them), and they
|
|
must not be worse off than before this module."""
|
|
logging.getLogger("siftlode.test").warning("just a message")
|
|
payload = sink.objects()[-1]
|
|
assert payload["msg"] == "just a message"
|
|
assert payload["level"] == "WARNING"
|
|
assert "event" not in payload
|
|
|
|
|
|
def test_a_broken_format_call_reports_itself_instead_of_raising():
|
|
"""A record is formatted directly rather than logged: a genuinely broken call ALSO breaks
|
|
pytest's own capture handler (stdlib formatting raises there too), which would fail the test for
|
|
the wrong reason. What is under test is our formatter's refusal to raise."""
|
|
record = logging.LogRecord(
|
|
"siftlode.test", logging.INFO, __file__, 0, "two args %s %s", ("only-one",), None
|
|
)
|
|
payload = json.loads(logs.JsonFormatter().format(record))
|
|
assert "log formatting failed" in payload["msg"]
|
|
|
|
|
|
def test_bound_correlation_appears_and_is_released(sink):
|
|
log = logging.getLogger("siftlode.test")
|
|
with logs.bind(job_id=42):
|
|
log.info("inside")
|
|
log.info("outside")
|
|
assert sink.objects()[-2]["job_id"] == 42
|
|
assert "job_id" not in sink.objects()[-1]
|
|
|
|
|
|
def test_every_selectable_level_resolves_to_a_real_logging_level():
|
|
# The picker's options come from `LogLevel`; each must be a level `logging` knows, TRACE included.
|
|
assert [logs.resolve_level(n) for n in choices_of(LogLevel)] == [
|
|
logs.TRACE,
|
|
logging.DEBUG,
|
|
logging.INFO,
|
|
logging.WARNING,
|
|
logging.ERROR,
|
|
]
|
|
with pytest.raises(ValueError):
|
|
logs.resolve_level("nonsense")
|
|
|
|
|
|
def test_configure_logging_silences_the_uvicorn_access_log_and_keeps_ours_propagating():
|
|
logs.configure_logging("api")
|
|
assert logging.getLogger("uvicorn.access").getEffectiveLevel() >= logging.CRITICAL
|
|
# propagate stays on so every line reaches the ONE root handler — and so pytest's caplog works.
|
|
assert logging.getLogger("siftlode.anything").propagate is True
|
|
assert len(logging.getLogger().handlers) == 1
|
|
|
|
|
|
# --- the request line -----------------------------------------------------------------------
|
|
|
|
|
|
def _client() -> TestClient:
|
|
app = FastAPI()
|
|
app.middleware("http")(logs.log_requests)
|
|
|
|
@app.get("/api/videos/{video_id}")
|
|
def _video(video_id: str, q: str | None = None):
|
|
return {"ok": video_id}
|
|
|
|
@app.get("/healthz")
|
|
def _health():
|
|
return {"ok": True}
|
|
|
|
@app.get("/api/admin/logs/facets")
|
|
def _log_console(fail: bool = False):
|
|
if fail:
|
|
raise HTTPException(status_code=400, detail={"code": "bad_cursor"})
|
|
return {"ok": True}
|
|
|
|
@app.get("/api/me/notifications/unread_count")
|
|
def _unread():
|
|
return {"count": 0}
|
|
|
|
@app.post("/api/me/notifications/unread_count")
|
|
def _unread_write():
|
|
return {"ok": True}
|
|
|
|
@app.get("/boom")
|
|
def _boom():
|
|
raise RuntimeError("kaboom")
|
|
|
|
return TestClient(app, raise_server_exceptions=False)
|
|
|
|
|
|
def test_the_request_line_has_the_route_template_and_no_query_string(sink):
|
|
logging.getLogger("siftlode.http").setLevel(logs.TRACE)
|
|
r = _client().get("/api/videos/abc123?q=cats&token=SECRET")
|
|
assert r.status_code == 200
|
|
payload = sink.objects()[-1]
|
|
assert payload["event"] == "http.request"
|
|
# The TEMPLATE, so every video request groups into one queryable path...
|
|
assert payload["attrs"]["path"] == "/api/videos/{video_id}"
|
|
# ...and the query string — uvicorn's access log printed it, `?account=1` and all — is gone.
|
|
assert "SECRET" not in sink.lines[-1]
|
|
assert "cats" not in sink.lines[-1]
|
|
assert payload["attrs"]["status"] == 200
|
|
assert isinstance(payload["attrs"]["duration_ms"], float)
|
|
|
|
|
|
def test_anonymous_request_lines_are_capped_per_client_for_storage_but_still_logged(sink):
|
|
# A scanner's 404s are WARNINGs, and WARNINGs are stored. Past the cap they stay on stdout but are
|
|
# marked not-for-storage; a signed-in request is never capped. No address appears in any line.
|
|
logs._anonymous_lines.reset()
|
|
client = _client()
|
|
for _ in range(logs.ANONYMOUS_LINES_PER_MINUTE + 3):
|
|
client.get("/wp-login.php")
|
|
lines = sink.objects()[-(logs.ANONYMOUS_LINES_PER_MINUTE + 3) :]
|
|
assert len(lines) == logs.ANONYMOUS_LINES_PER_MINUTE + 3 # stdout keeps every one
|
|
records = sink.records[-(logs.ANONYMOUS_LINES_PER_MINUTE + 3) :]
|
|
stored = [getattr(r, logs.PERSIST, True) for r in records]
|
|
assert stored.count(False) == 3 and all(stored[: logs.ANONYMOUS_LINES_PER_MINUTE])
|
|
assert "testclient" not in "".join(sink.lines) # the client key never reaches the line
|
|
logs._anonymous_lines.reset()
|
|
|
|
|
|
def test_a_signed_in_request_is_never_capped_and_names_its_account(sink):
|
|
logs._anonymous_lines.reset()
|
|
app = FastAPI()
|
|
app.middleware("http")(logs.log_requests)
|
|
|
|
@app.get("/api/me")
|
|
def _me(request: Request):
|
|
request.state.user_id = 42 # what the auth dependency does
|
|
return {}
|
|
|
|
client = TestClient(app)
|
|
for _ in range(logs.ANONYMOUS_LINES_PER_MINUTE + 3):
|
|
assert client.get("/api/me").status_code == 200
|
|
records = sink.records[-(logs.ANONYMOUS_LINES_PER_MINUTE + 3) :]
|
|
assert all(getattr(r, logs.PERSIST, True) for r in records)
|
|
assert sink.objects()[-1]["user_id"] == 42 and "user_id" not in sink.objects()[-1]["attrs"]
|
|
|
|
|
|
def test_the_request_id_is_echoed_and_shared_by_every_line_of_that_request(sink):
|
|
client = _client()
|
|
r = client.get("/api/videos/abc123")
|
|
rid = r.headers["X-Request-Id"]
|
|
assert rid and sink.objects()[-1]["request_id"] == rid
|
|
|
|
|
|
def test_the_route_template_goes_back_to_the_browser_but_never_a_raw_path(sink):
|
|
# The browser logs its own copy of each request, grouped by this template (frontend lib/http.ts).
|
|
client = _client()
|
|
assert client.get("/api/videos/SeCrEt-VaLuE").headers["X-Route"] == "/api/videos/{video_id}"
|
|
# Unmatched: the only template there is the raw path, which may carry a value — so none is sent.
|
|
assert "X-Route" not in client.get("/api/share/SeCrEt-VaLuE").headers
|
|
|
|
|
|
def test_a_client_supplied_id_is_reused_only_when_it_is_well_formed(sink):
|
|
client = _client()
|
|
good = client.get("/api/videos/x", headers={"X-Request-Id": "web-0123456789ab"})
|
|
assert good.headers["X-Request-Id"] == "web-0123456789ab"
|
|
# An unbounded header would be a way to write arbitrary text into the log; short/odd ones are
|
|
# replaced with our own id rather than trusted.
|
|
for bogus in ("short", "has spaces and \n newline", "x" * 200):
|
|
r = client.get("/api/videos/x", headers={"X-Request-Id": bogus})
|
|
assert r.headers["X-Request-Id"] != bogus
|
|
|
|
|
|
def test_the_healthcheck_is_logged_at_debug_so_it_cannot_drown_the_log(sink):
|
|
http = logging.getLogger("siftlode.http")
|
|
http.setLevel(logs.TRACE)
|
|
_client().get("/healthz")
|
|
assert sink.objects()[-1]["level"] == "DEBUG"
|
|
# ...but it is still logged when something is wrong with it: the floor drops, nothing is hidden.
|
|
http.setLevel(logging.INFO)
|
|
before = len(sink.lines)
|
|
_client().get("/healthz")
|
|
assert len(sink.lines) == before
|
|
|
|
|
|
def test_the_log_consoles_own_reads_are_debug_but_its_failures_are_not(sink):
|
|
http = logging.getLogger("siftlode.http")
|
|
http.setLevel(logs.TRACE)
|
|
client = _client()
|
|
assert client.get("/api/admin/logs/facets").status_code == 200
|
|
assert sink.objects()[-1]["level"] == "DEBUG"
|
|
assert client.get("/api/admin/logs/facets?fail=1").status_code == 400
|
|
assert sink.objects()[-1]["level"] == "WARNING"
|
|
|
|
|
|
def test_a_polled_read_is_debug_a_write_to_the_same_route_is_not(sink):
|
|
# One open tab polled the unread counts, /api/me and the download index every 30 s: at INFO
|
|
# that was ~95% of the stored log on prod.
|
|
http = logging.getLogger("siftlode.http")
|
|
http.setLevel(logs.TRACE)
|
|
client = _client()
|
|
r = client.get("/api/me/notifications/unread_count")
|
|
assert sink.objects()[-1]["level"] == "DEBUG"
|
|
# ...and says so, for the browser's copy of the line (frontend lib/http.ts).
|
|
assert r.headers["X-Log-Quiet"] == "1"
|
|
r = client.post("/api/me/notifications/unread_count")
|
|
assert sink.objects()[-1]["level"] == "INFO"
|
|
assert "X-Log-Quiet" not in r.headers
|
|
assert client.get("/api/videos/abc").status_code == 200
|
|
assert sink.objects()[-1]["level"] == "INFO"
|
|
|
|
|
|
def test_every_polled_route_is_a_route_the_app_has():
|
|
# The list names route TEMPLATES; one renamed on the router side would silently fall back to
|
|
# INFO, so the list is held to the app's own routes.
|
|
from app.main import app
|
|
|
|
# The schema, not `app.routes`: FastAPI 0.141 keeps an included router behind a wrapper there.
|
|
gets = {path for path, ops in app.openapi()["paths"].items() if "get" in ops}
|
|
assert logs._POLLED_ROUTES - gets == set()
|
|
|
|
|
|
def test_a_failing_request_is_logged_by_status_and_an_unhandled_error_is_logged_at_all(sink):
|
|
client = _client()
|
|
assert client.get("/api/videos/missing/deeper").status_code == 404 # 404 from the router
|
|
assert sink.objects()[-1]["level"] == "WARNING"
|
|
client.get("/boom")
|
|
payload = sink.objects()[-1]
|
|
# The request that most needs a line is the one that raised: uvicorn's access log wrote it, but
|
|
# our middleware sees the exception instead of a response, so it has to log it explicitly.
|
|
assert payload["level"] == "ERROR"
|
|
assert payload["event"] == "http.request"
|
|
assert "RuntimeError" in payload["exc"]
|
|
|
|
|
|
# --- a stack whose messages are data (`logs.scrubbed_stack`) -----------------------------------
|
|
def _upper(exc: BaseException) -> str:
|
|
return str(exc).upper()
|
|
|
|
|
|
def _raised(make):
|
|
try:
|
|
make()
|
|
except Exception as exc: # noqa: BLE001 — whatever the case raised is the case
|
|
return exc
|
|
raise AssertionError("the case did not raise")
|
|
|
|
|
|
def test_a_raise_from_is_printed_as_its_direct_cause():
|
|
def make():
|
|
try:
|
|
raise KeyError("inner")
|
|
except KeyError as inner:
|
|
raise ValueError("outer") from inner
|
|
|
|
stack = logs.scrubbed_stack(_raised(make), _upper)["scrubbed_stack"]
|
|
assert "The above exception was the direct cause" in stack
|
|
assert "During handling" not in stack
|
|
assert "'INNER'" in stack and "OUTER" in stack
|
|
|
|
|
|
def test_a_raise_from_none_hides_the_link_its_author_hid():
|
|
def make():
|
|
try:
|
|
raise KeyError("hidden")
|
|
except KeyError:
|
|
raise ValueError("shown") from None
|
|
|
|
stack = logs.scrubbed_stack(_raised(make), _upper)["scrubbed_stack"]
|
|
assert "HIDDEN" not in stack and "SHOWN" in stack
|
|
|
|
|
|
def test_a_chain_that_loops_back_ends():
|
|
a, b = ValueError("a"), ValueError("b")
|
|
a.__context__, b.__context__ = b, a
|
|
assert "A" in logs.scrubbed_stack(a, _upper)["scrubbed_stack"]
|
|
|
|
|
|
def test_a_message_that_cannot_be_rendered_does_not_fail_the_report():
|
|
def refuse(exc):
|
|
raise RuntimeError("no")
|
|
|
|
assert (
|
|
logs.scrubbed_stack(ValueError("x"), refuse)["scrubbed_stack"]
|
|
== "ValueError: <unprintable>"
|
|
)
|
|
|
|
|
|
def test_the_text_format_carries_the_scrubbed_stack_too():
|
|
title = "a private title" # a variable: a frame prints its source line, which is code
|
|
exc = ValueError(title)
|
|
record = logging.LogRecord("siftlode.x", logging.WARNING, __file__, 1, "failed", None, None)
|
|
for key, value in logs.scrubbed_stack(exc, lambda e: "…").items():
|
|
setattr(record, key, value)
|
|
|
|
line = logs.TextFormatter().format(record)
|
|
assert "ValueError: …" in line and "private" not in line
|
|
assert "scrubbed_stack" not in line # the stack, not an attribute
|