diff --git a/backend/app/auth.py b/backend/app/auth.py index 26384175..2dae04ba 100644 --- a/backend/app/auth.py +++ b/backend/app/auth.py @@ -248,7 +248,10 @@ def purge_user(db: Session, user: User, background: BackgroundTasks) -> None: # Raw delete so Postgres applies ON DELETE CASCADE / SET NULL on every dependent table. db.execute(delete(User).where(User.id == user_id)) # Stored log events name the account by id with no foreign key to cascade from (models.LogEvent), - # so they go explicitly — they are request history, i.e. personal data. + # so they go explicitly — they are request history, i.e. personal data. Forgotten FIRST: this very + # request's access line, a queued row, a later background task would otherwise put rows back + # under the id after this DELETE (logs.forget_user). + logs.forget_user(user_id) db.execute(delete(LogEvent).where(LogEvent.user_id == user_id)) # Every canonically-equal invite goes, not just the literal spelling: is_allowed matches on the # canonical form, so leaving a sibling row behind would keep granting access to a mailbox whose diff --git a/backend/app/logs.py b/backend/app/logs.py index 117a6add..135e35ee 100644 --- a/backend/app/logs.py +++ b/backend/app/logs.py @@ -40,6 +40,7 @@ import time import uuid from datetime import datetime, timezone from typing import Any +from urllib.parse import parse_qsl from app.config import settings from app.ratelimit import RateLimiter, client_key @@ -204,6 +205,14 @@ def redact_text(text: str) -> str: return _EMAIL_RE.sub(REDACTED_EMAIL, scrubbed) +def redact_query(query: str) -> str: + """Scrub a raw query string. DECODED first: the text rules match what was typed, and a browser + sends it percent-encoded — `someone%40example.com` is an address `_EMAIL_RE` cannot see. Each + parameter also goes through the key rules, so `?new_password=` goes as `redact` would drop it.""" + pairs = parse_qsl(query, keep_blank_values=True) + return "&".join(redact_text(f"{key}={redact(value, key=key)}") for key, value in pairs) + + def redact(value: Any, *, key: str = "") -> Any: """Recursively scrub a structure destined for the log. @@ -457,13 +466,57 @@ def _apply_root_level(force: bool = False) -> None: if _base_level == logging.NOTSET: return # not configured: the root keeps whatever the process set up now = datetime.now(timezone.utc) + # The level is SET under the lock too, not just computed: two threads that each computed a + # different answer would otherwise apply them in either order, and the older one could win — + # the root back at the base while an elevation is still active. with _elevations_lock: for user_id in [u for u, (_, until) in _elevations.items() if until <= now]: del _elevations[user_id] lowest = min([_base_level, *(lvl for lvl, _ in _elevations.values())]) - if force or lowest != _root_applied: - logging.getLogger().setLevel(lowest) - _root_applied = lowest + if force or lowest != _root_applied: + logging.getLogger().setLevel(lowest) + _root_applied = lowest + + +# --- erased accounts ------------------------------------------------------------------------ +# `auth.purge_user` deletes an account's `log_events` rows, but a line about that account can still +# be on its way: the erasing request's own access line is logged AFTER the handler committed, a queued +# row waits for the sink's writer, a background task logs later. Those rows are stored without the +# account (`logsink` asks here). Remembered for an hour in this process — a line later than that has +# no request left to come from, and the retention job is the backstop for any other process. +ERASED_MEMORY_SECONDS = 3600.0 +_erased: dict[int, float] = {} +_erased_lock = threading.Lock() +# Held by whoever checks rows against `is_erased` AND writes them (`erasure_guard`). `forget_user` +# takes it too, so an erasure waits for a write that already passed the check to commit — and the +# purge's DELETE, which runs after `forget_user`, then removes those rows as well. Without it a row +# checked a moment before the erasure could be inserted a moment after the DELETE. +_erasure_guard = threading.Lock() + + +def erasure_guard() -> threading.Lock: + """Hold this around "drop erased accounts from these rows, insert them, commit".""" + return _erasure_guard + + +def forget_user(user_id: int) -> None: + """Mark `user_id` erased: no line logged about it from now on is stored under its id. Call BEFORE + deleting its rows (see `_erasure_guard`).""" + now = time.monotonic() + with _erasure_guard, _erased_lock: + for stale in [u for u, at in _erased.items() if now - at > ERASED_MEMORY_SECONDS]: + del _erased[stale] + _erased[user_id] = now + with _elevations_lock: + _elevations.pop(user_id, None) + + +def is_erased(user_id: int | None) -> bool: + if user_id is None: + return False + with _erased_lock: + at = _erased.get(user_id) + return at is not None and time.monotonic() - at <= ERASED_MEMORY_SECONDS def current_source() -> str: @@ -537,10 +590,10 @@ async def log_requests(request, call_next): "status": status, "duration_ms": duration_ms, } - # Teljes (TRACE, the user's or the whole process's): the query string too — redacted, and in - # attrs only, never in the message. + # Teljes (TRACE) the user consented to: the query string too — redacted, and in attrs only, never + # in the message. if request.url.query and _traced(scope): - attrs["query"] = redact_text(request.url.query) + attrs["query"] = redact_query(request.url.query) _request_log.log( level, "%s %s %s", @@ -563,10 +616,10 @@ async def log_requests(request, call_next): def _traced(scope: _RequestScope) -> bool: - """Whether this request is logged at TRACE — its user's consent, or the process's base level.""" - if scope.level is not None and scope.level <= TRACE: - return True - return _base_level != logging.NOTSET and _base_level <= TRACE + """Whether this request's query string is logged: only under its OWN user's Teljes consent. An + admin setting the whole process to TRACE gets more detail, not every user's search terms — the + privacy policy promises those are kept only for a user who agreed to it.""" + return scope.level is not None and scope.level <= TRACE # Anonymous request lines kept per client per minute. Signed-in traffic is not capped: it is a known diff --git a/backend/app/logsink.py b/backend/app/logsink.py index fb6d2e38..d1b5db4b 100644 --- a/backend/app/logsink.py +++ b/backend/app/logsink.py @@ -78,21 +78,47 @@ def should_store(record: logging.LogRecord) -> bool: return getattr(record, "event", None) == "http.request" and record.levelno >= logging.INFO +def _without_nul(value: Any) -> Any: + """`value` with every NUL character replaced, in keys and nested values alike. Postgres refuses + NUL in text and `\\u0000` in JSONB, and a refused row fails the WHOLE batch it rides in — every + user's rows with it. Anything logged can carry one (`GET /assets/%00` logs the decoded path).""" + if isinstance(value, str): + return value.replace("\x00", "�") + if isinstance(value, dict): + return {_without_nul(k): _without_nul(v) for k, v in value.items()} + if isinstance(value, list): + return [_without_nul(v) for v in value] + return value + + def to_row(fields: dict[str, Any]) -> dict[str, Any]: """`logs.record_fields` output → a `log_events` row: known columns only, text clipped to its - column, attrs made JSON-safe (anything exotic becomes its `str`, as on stdout).""" + column and free of NUL, attrs made JSON-safe (anything exotic becomes its `str`, as on stdout).""" row = { column: fields.get(column) for column in LogEvent.__table__.columns.keys() if column != "id" } for column, limit in _LIMITS.items(): value = row.get(column) - if isinstance(value, str) and len(value) > limit: - row[column] = value[: limit - 1] + "…" + if isinstance(value, str): + value = _without_nul(value) + row[column] = value[: limit - 1] + "…" if len(value) > limit else value if row["attrs"] is not None: - row["attrs"] = json.loads(json.dumps(row["attrs"], ensure_ascii=False, default=str)) + row["attrs"] = _without_nul( + json.loads(json.dumps(row["attrs"], ensure_ascii=False, default=str)) + ) return row +def without_erased(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + """The rows, with the account taken off any whose account was erased meanwhile + (`logs.forget_user`). Checked right before the INSERT, under `logs.erasure_guard()` held until + the commit: a row built before the erasure can still be waiting for its batch.""" + for row in rows: + if logs.is_erased(row.get("user_id")): + row["user_id"] = None + return rows + + def _notice(code: str, msg: str, **attrs: Any) -> dict[str, Any]: """A row the sink writes about ITSELF. Built directly, not logged: logging it would route it back through this sink (and the writer thread's own records are deliberately ignored).""" @@ -220,8 +246,8 @@ class LogSink(logging.Handler): if not rows: return 0 try: - with self._bind.begin() as conn: - conn.execute(insert(LogEvent), rows) + with logs.erasure_guard(), self._bind.begin() as conn: + conn.execute(insert(LogEvent), without_erased(rows)) except Exception as exc: # noqa: BLE001 — the database is down or refused the batch print( f"log sink: could not store {len(rows)} log events: {logs.redact_text(str(exc))[:300]}", diff --git a/backend/app/models.py b/backend/app/models.py index 7878ffdb..173d4b8a 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -557,8 +557,9 @@ class LogEvent(Base): `user_id` has NO foreign key on purpose: the writer runs behind the event, so a line about an account can arrive after that account was purged, and a constraint would make it fail the whole - batch it travels in. `auth.purge_user` deletes an account's rows explicitly instead, and the - retention job (`log_retention_days`) bounds whatever a late write leaves behind.""" + batch it travels in. `auth.purge_user` deletes an account's rows explicitly instead, after + `logs.forget_user` makes every later write in this process store such a line without the account; + the retention job (`log_retention_days`) bounds whatever another process writes late.""" __tablename__ = "log_events" __table_args__ = (Index("ix_log_events_user_ts", "user_id", "ts"),) diff --git a/backend/app/ratelimit.py b/backend/app/ratelimit.py index 8a7d6ae1..2e9f3325 100644 --- a/backend/app/ratelimit.py +++ b/backend/app/ratelimit.py @@ -17,10 +17,14 @@ from app.config import settings class RateLimiter: - def __init__(self, max_events: int, window_seconds: float): + """At most `max_events` attempts per `window_seconds` per key — or, when each attempt passes a + `cost`, at most that much total cost (a byte budget counts bytes, not calls).""" + + def __init__(self, max_events: float, window_seconds: float): self.max_events = max_events self.window = window_seconds - self._hits: dict[str, list[float]] = {} + # key → (monotonic time, cost) of every counted attempt still inside the window + self._hits: dict[str, list[tuple[float, float]]] = {} self._lock = threading.Lock() def reset(self) -> None: @@ -36,27 +40,27 @@ class RateLimiter: count.""" cutoff = time.monotonic() - self.window with self._lock: - return sum(1 for t in self._hits.get(key, []) if t > cutoff) >= self.max_events + return sum(c for t, c in self._hits.get(key, []) if t > cutoff) >= self.max_events - def allow(self, key: str) -> bool: - """Record an attempt for ``key`` and return whether it is within the limit. + def allow(self, key: str, cost: float = 1) -> bool: + """Record an attempt costing ``cost`` for ``key`` and return whether it is within the limit. - True -> under the cap (the attempt is counted). - False -> the cap for the current window is already reached (attempt NOT counted, so a + True -> within the cap (the attempt is counted). + False -> the attempt would pass the cap for the current window (attempt NOT counted, so a blocked caller can't keep pushing the window forward).""" now = time.monotonic() cutoff = now - self.window with self._lock: - hits = [t for t in self._hits.get(key, []) if t > cutoff] - if len(hits) >= self.max_events: + hits = [(t, c) for t, c in self._hits.get(key, []) if t > cutoff] + if sum(c for _, c in hits) + cost > self.max_events: self._hits[key] = hits return False - hits.append(now) + hits.append((now, cost)) self._hits[key] = hits # Opportunistic cleanup so abandoned keys don't accumulate unboundedly. if len(self._hits) > 4096: for k in list(self._hits): - fresh = [t for t in self._hits[k] if t > cutoff] + fresh = [(t, c) for t, c in self._hits[k] if t > cutoff] if fresh: self._hits[k] = fresh else: diff --git a/backend/app/routes/diagnostics.py b/backend/app/routes/diagnostics.py index 277f8392..1f9b5028 100644 --- a/backend/app/routes/diagnostics.py +++ b/backend/app/routes/diagnostics.py @@ -7,8 +7,9 @@ the body), and the request id each event carries, which is what joins a browser line of the same request. Nothing from the body is trusted as-is: - * the whole body is capped before it is parsed, and so are the number of events and each event's - attributes; + * only a JSON Content-Type is read, and the whole body is capped before it is parsed, and so are + the number of events and each event's attributes; + * the pace is capped per account, and so is what an account may store per hour; * everything is redacted AGAIN with the server's own rules — the client redacts at capture, but a modified client would not; * an event below the consented level is dropped (a Részletes consent does not store what only @@ -40,6 +41,13 @@ ATTRS_MAX_BYTES = 16 * 1024 # The client flushes every few seconds and when its buffer fills; this is room for a burst, not a # pace anything legitimate keeps up for a minute. _batches = RateLimiter(max_events=30, window_seconds=60) +# What one account may STORE per hour. The pace cap alone still lets a self-consented account write +# ~15 MiB a minute into the one shared database, kept for the whole retention window; a real session +# at Teljes stores a few hundred kB an hour. Past the budget a batch is accepted and not stored (a +# refusal would only make the client resend it), and one notice per hour says so. +STORED_BYTES_PER_HOUR = 5 * 1024 * 1024 +_stored_bytes = RateLimiter(max_events=STORED_BYTES_PER_HOUR, window_seconds=3600) +_budget_notices = RateLimiter(max_events=1, window_seconds=3600) ClientLevel = Literal["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] @@ -73,7 +81,14 @@ async def _capped_body(request: Request) -> bytes: async def _parsed_events(request: Request) -> ClientEventsIn: """The body as the model — parsed only after the size cap, which is why this is a dependency and - not a plain body parameter (FastAPI reads a declared body in full before anything else runs).""" + not a plain body parameter (FastAPI reads a declared body in full before anything else runs). + + JSON only, by its Content-Type, as a declared body would be. Parsing any body as JSON would let a + CORS-simple `text/plain` form post from a same-site page write events into a signed-in user's + log; the app's own client always sends `application/json`, the unload beacon included.""" + media_type = request.headers.get("content-type", "").split(";")[0].strip().lower() + if media_type != "application/json" and not media_type.endswith("+json"): + raise HTTPException(status_code=415, detail={"code": "json_required"}) raw = await _capped_body(request) try: return ClientEventsIn.model_validate_json(raw) @@ -129,7 +144,33 @@ def ingest_events( for item in body.events if logs.SEVERITY[item.level] >= floor ] - if rows: - db.execute(insert(LogEvent), rows) + if not rows: + return {"ok": True, "stored": 0} + size = sum(len(json.dumps(row, ensure_ascii=False, default=str).encode()) for row in rows) + stored = len(rows) + if not _stored_bytes.allow(str(user.id), size): + stored = 0 + if not _budget_notices.allow(str(user.id)): + return {"ok": True, "stored": 0} + rows = [_budget_notice(user, now)] + # Under the erasure guard until the commit: an account erased meanwhile gets no rows back. + with logs.erasure_guard(): + db.execute(insert(LogEvent), logsink.without_erased(rows)) db.commit() - return {"ok": True, "stored": len(rows)} + return {"ok": True, "stored": stored} + + +def _budget_notice(user: User, now: datetime) -> dict[str, Any]: + return logsink.to_row( + { + "ts": now, + "level": "WARNING", + "sev": logs.SEVERITY["WARNING"], + "source": "web", + "logger": "web", + "event": "diagnostics.budget_spent", + "msg": "The browser log passed its hourly storage budget; the rest of the hour is not kept", + "attrs": {"budget_bytes": STORED_BYTES_PER_HOUR}, + "user_id": user.id, + } + ) diff --git a/backend/tests/test_diagnostics.py b/backend/tests/test_diagnostics.py index 78f9bd95..5027ed46 100644 --- a/backend/tests/test_diagnostics.py +++ b/backend/tests/test_diagnostics.py @@ -13,6 +13,7 @@ Three parts: import json import logging +import threading from datetime import datetime, timedelta, timezone import pytest @@ -20,7 +21,8 @@ from fastapi import Depends, FastAPI from fastapi.testclient import TestClient from sqlalchemy import select -from app import diagnostics, logs, logsink +from app import diagnostics, logs, logsink, sysconfig +from app.config import settings from app.models import LogEvent, User from app.routes import diagnostics as diagnostics_routes @@ -235,6 +237,38 @@ def test_the_pace_is_capped_per_user(elevated): assert codes[:30] == [200] * 30 and codes[30] == 429 +def test_what_an_account_stores_per_hour_is_capped_and_said_once(elevated, monkeypatch): + user, client, rows = elevated("DEBUG") + monkeypatch.setattr( + diagnostics_routes, "_stored_bytes", diagnostics_routes.RateLimiter(3_000, 3600) + ) + batch = {"events": [_event(msg="m" * 1_000)]} + stored = [client.post("/api/diagnostics/events", json=batch).json()["stored"] for _ in range(4)] + assert stored == [1, 1, 0, 0] + events = [row.event for row in rows()] + # Two batches, then ONE notice for the hour, not one per refused batch. + assert events == ["nav.view", "nav.view", "diagnostics.budget_spent"] + assert rows()[-1].user_id == user.id + + +def test_only_a_json_body_is_read(elevated): + _user, client, rows = elevated("DEBUG") + # What a same-site `
` can send without a preflight. + r = client.post( + "/api/diagnostics/events", + content=json.dumps({"events": [_event()]}), + headers={"content-type": "text/plain"}, + ) + assert r.status_code == 415 and r.json()["detail"]["code"] == "json_required" + assert rows() == [] + ok = client.post( + "/api/diagnostics/events", + content=json.dumps({"events": [_event()]}), + headers={"content-type": "application/json; charset=utf-8"}, + ) + assert ok.json()["stored"] == 1 + + # --- the server side of an elevation --------------------------------------------------------- @@ -330,6 +364,82 @@ def test_the_query_string_is_logged_only_for_a_traced_request_and_redacted(gated assert "q=lofi" in lines[1].attrs["query"] and "SECRET" not in lines[1].attrs["query"] +def test_an_address_typed_into_a_search_is_redacted_though_the_browser_encodes_it(gated): + _app().get("/api/thing/7?level=TRACE&q=someone%40example.com&new_password=hunter2") + (line,) = [r for r in gated.records if getattr(r, "event", None) == "http.request"] + assert line.attrs["query"] == "level=TRACE&q=[email]&new_password=[redacted]" + + +def test_an_admin_level_of_trace_does_not_log_other_users_query_strings(gated): + logs.set_level("TRACE") + _app().get("/api/thing/7", params={"q": "something private"}) + (line,) = [r for r in gated.records if getattr(r, "event", None) == "http.request"] + assert "query" not in line.attrs + + +def test_the_admin_level_applies_to_the_running_process_when_set_and_when_reset( + gated, db, make_user, as_user +): + admin = as_user(make_user(role="admin")) + r = admin.patch("/api/admin/config/log_level", json={"value": "DEBUG"}) + assert r.status_code == 200, r.text + assert logs.current_level() == "DEBUG" + assert admin.delete("/api/admin/config/log_level").status_code == 200 + assert logs.current_level() == settings.log_level + + +def test_the_stored_admin_level_is_applied_at_boot(gated, db, monkeypatch): + import asyncio + + import app.main as app_main + + sysconfig.set_value(db, "log_level", "DEBUG") + db.commit() + monkeypatch.setattr(app_main.state, "is_configured", lambda db: True) + monkeypatch.setattr(app_main, "start_scheduler", lambda: None) + monkeypatch.setattr(app_main, "shutdown_scheduler", lambda: None) + monkeypatch.setattr(app_main.plex_subtitles, "shutdown", lambda: None) + + async def _no_sweep(): + return None + + monkeypatch.setattr(app_main, "_sweep_hls_orphans", _no_sweep) + levels: list[str] = [] + + async def _cycle(): + async with app_main.lifespan(app_main.app): + levels.append(logs.current_level()) + + asyncio.run(_cycle()) + assert levels == ["DEBUG"] + + +def test_the_root_level_is_applied_in_the_order_the_elevations_changed(gated, monkeypatch): + """Two requests change the elevations at once: A's user leaves (the root may go back to INFO), + B's user arrives (it must go to DEBUG). B's change is the later one, so DEBUG must be what stays + — even when A is paused between working out its level and applying it.""" + root = logging.getLogger() + logs._track_elevation(1, logging.DEBUG, datetime.now(timezone.utc) + timedelta(hours=1)) + real_set = root.setLevel + other: list[threading.Thread] = [] + + def paused_set(level): + if level == logging.INFO and not other: + b = threading.Thread( + target=logs._track_elevation, + args=(2, logging.DEBUG, datetime.now(timezone.utc) + timedelta(hours=1)), + ) + other.append(b) + b.start() + b.join(0.3) # B runs to completion here unless A's level change holds it off + real_set(level) + + monkeypatch.setattr(root, "setLevel", paused_set) + logs._track_elevation(1, None, None) # A + other[0].join(5) + assert root.level == logging.DEBUG and 2 in logs._elevations + + def test_a_real_sign_in_carries_the_consented_level_into_the_process(gated, client, db, make_user): user = _signed_in(client, make_user) assert client.put("/api/me/diagnostics", json={"level": "DEBUG"}).status_code == 200 diff --git a/backend/tests/test_logs.py b/backend/tests/test_logs.py index 854e10c8..ae8b883e 100644 --- a/backend/tests/test_logs.py +++ b/backend/tests/test_logs.py @@ -102,7 +102,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.redaction.test.ts runs against the browser's copy of redact(): +# 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") diff --git a/backend/tests/test_logsink.py b/backend/tests/test_logsink.py index 7dc13ddc..06f96dec 100644 --- a/backend/tests/test_logsink.py +++ b/backend/tests/test_logsink.py @@ -12,7 +12,7 @@ from datetime import datetime, timedelta, timezone import pytest from fastapi import BackgroundTasks -from sqlalchemy import create_engine, select +from sqlalchemy import create_engine, func, select from app import logs, logsink, scheduler from app.auth import purge_user @@ -173,3 +173,75 @@ def test_deleting_an_account_deletes_its_stored_log_events(db, make_user): db.commit() purge_user(db, gone, BackgroundTasks()) assert [r.user_id for r in db.execute(select(LogEvent)).scalars()] == [kept.id] + + +def test_an_erased_accounts_late_lines_are_stored_without_it(stored, db, make_user): + """The DELETE in purge_user cannot reach a row still on its way: one queued before the erasure, + or the erasing request's own access line, logged after the handler committed.""" + _sink, log, rows = stored + user = make_user() + try: + log.warning("queued before the erasure", extra={"user_id": user.id}) + purge_user(db, user, BackgroundTasks()) + log.info( + "DELETE /api/me/account 200", extra={**logs.event("http.request"), "user_id": user.id} + ) + assert [(r.msg, r.user_id) for r in rows()] == [ + ("queued before the erasure", None), + ("DELETE /api/me/account 200", None), + ] + finally: + logs._erased.clear() + + +def test_a_nul_character_costs_nothing_but_itself(stored): + """Postgres refuses NUL in text and JSONB, and a refused row fails its whole batch — every other + user's rows with it. `GET /assets/%00` logs exactly that.""" + _sink, log, rows = stored + log.warning("before") + log.warning("GET /assets/\x00 404", extra=logs.event("http.request", path="/assets/\x00")) + log.warning("after") + before, nul, after = rows() + assert (before.msg, after.msg) == ("before", "after") + assert nul.msg == "GET /assets/� 404" and nul.attrs["path"] == "/assets/�" + + +def test_the_writer_drains_what_was_queued_when_it_is_stopped(db): + sink = logsink.LogSink(engine) + log = logging.getLogger("siftlode.sinktest.drain") + log.addHandler(sink) + try: + for i in range(logsink.BATCH_MAX + 5): + log.warning(f"line {i}") # distinct templates: no flood limiter involved + finally: + log.removeHandler(sink) + sink._stopping.set() + sink._run() # stopped before its first wait: everything goes through the drain + assert ( + db.execute(select(func.count()).select_from(LogEvent)).scalar_one() == logsink.BATCH_MAX + 5 + ) + + +def test_the_running_sink_keeps_the_base_level_for_everyone_the_elevation_is_not_for( + db, monkeypatch +): + """`logsink.start()` as production runs it. One user's elevation lowers the ROOT logger, so an + ordinary request line below the base level reaches every handler — the sink's own level gate is + what keeps it out of the table, as stdout's keeps it off the terminal.""" + root = logging.getLogger() + saved = (logs._base_level, dict(logs._elevations), logs._root_applied, root.level) + monkeypatch.setattr(logsink.settings, "log_db_sink", True) + logs.set_level("WARNING") + logs._track_elevation(1, logging.DEBUG, datetime.now(timezone.utc) + timedelta(hours=1)) + log = logging.getLogger("siftlode.sinktest.live") + try: + logsink.start() + log.info("GET /api/feed 200", extra={**logs.event("http.request"), "user_id": 2}) + log.warning("a warning") + finally: + logsink.stop() + logs._base_level, logs._root_applied = saved[0], saved[2] + logs._elevations.clear() + logs._elevations.update(saved[1]) + root.setLevel(saved[3]) + assert [r.msg for r in db.execute(select(LogEvent)).scalars()] == ["a warning"]