peter
b2eaa9e00b
Round 2 of the E22 review, on what round 1 changed. `redact_query` decodes a parameter before the text rules see it, which is what lets `%40` be recognised as an address — but `_SECRET_PARAM_RE` was written for the raw, encoded form, where a value holds no space, `&` or quote. On decoded text its value class stops at the first one, so `?state=a%20b` kept its tail and `?code=%20abc` was not matched at all. The parameter-NAME rule now decides, on the key, before any text rule runs. `forget_user` dropped the account's level elevation without recomputing the root level, so a process whose only elevated account had just deleted itself kept building TRACE records to throw away, until some later request happened to recompute it. Every other path that drops an elevation recomputes. The sink's writer took the erasure lock and THEN waited for a pooled connection, while every request path already holds a connection when it reaches that lock — so under pool pressure the writer held a process-global lock against the very requests that could hand a connection back. Same order on both sides now; the lock still spans the check, the insert and the commit. The storage-budget notice said the rest of the hour was not kept, and then smaller batches kept landing underneath it: the window slides and a refused batch is not charged to it. Says what it does. Tests: the route's half of the erasure rule had no test — only the sink's did — and a source-level contract test now covers the third insert nobody has written yet. `logs._erased` is process-global with an hour's memory while the DB fixture hands serial ids back out from 1, so it is swept before each test like the rate limiters are.
180 lines
8.0 KiB
Python
180 lines
8.0 KiB
Python
"""The browser's log, sent while its user has a raised diagnostics level (E22 S5).
|
|
|
|
The client keeps its log in memory (frontend `lib/log.ts`). Only after the user consented to Részletes
|
|
or Teljes in Settings → Diagnosztika does it stream that log here, in batches, and each batch lands in
|
|
`log_events` beside the server's own rows: `source="web"`, the account from the SESSION (never from
|
|
the body), and the request id each event carries, which is what joins a browser line to the server
|
|
line of the same request.
|
|
|
|
Nothing from the body is trusted as-is:
|
|
* 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
|
|
Teljes may carry), and so is the whole batch once the level has lapsed;
|
|
* the timestamp is clamped to the consent window, so a row cannot dodge the retention job by
|
|
claiming to come from the future.
|
|
"""
|
|
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any, Literal
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from pydantic import BaseModel, Field, ValidationError
|
|
from sqlalchemy import insert
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import diagnostics, logs, logsink
|
|
from app.auth import current_user
|
|
from app.db import get_db
|
|
from app.models import LogEvent, User
|
|
from app.ratelimit import RateLimiter
|
|
|
|
router = APIRouter(prefix="/api/diagnostics", tags=["diagnostics"])
|
|
|
|
BODY_MAX_BYTES = 512 * 1024
|
|
EVENTS_MAX = 200
|
|
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"]
|
|
|
|
|
|
class ClientEvent(BaseModel):
|
|
ts: datetime
|
|
level: ClientLevel
|
|
event: str = Field(max_length=64, pattern=r"^[a-z][a-z0-9_]*(\.[a-z0-9_]+)*$")
|
|
msg: str = Field(max_length=8_000)
|
|
attrs: dict[str, Any] | None = None
|
|
request_id: str | None = Field(default=None, pattern=r"^[A-Za-z0-9._-]{8,64}$")
|
|
tab_id: str | None = Field(default=None, pattern=r"^[A-Za-z0-9_-]{1,32}$")
|
|
version: str | None = Field(default=None, max_length=64)
|
|
|
|
|
|
class ClientEventsIn(BaseModel):
|
|
events: list[ClientEvent] = Field(max_length=EVENTS_MAX)
|
|
|
|
|
|
async def _capped_body(request: Request) -> bytes:
|
|
"""The request body, refused the moment it passes BODY_MAX_BYTES while still arriving. Counted,
|
|
not taken from Content-Length: a declared length can be absent (chunked) or wrong, and counting
|
|
stops reading at the same point a declared-length check would have."""
|
|
body = bytearray()
|
|
async for chunk in request.stream():
|
|
body += chunk
|
|
if len(body) > BODY_MAX_BYTES:
|
|
raise HTTPException(status_code=413, detail={"code": "too_large"})
|
|
return bytes(body)
|
|
|
|
|
|
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).
|
|
|
|
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)
|
|
except ValidationError as exc:
|
|
raise HTTPException(status_code=422, detail=json.loads(exc.json())) from exc
|
|
|
|
|
|
def _row(item: ClientEvent, user: User, now: datetime, window_start: datetime) -> dict[str, Any]:
|
|
attrs = logs.redact(item.attrs) if item.attrs else None
|
|
if attrs is not None:
|
|
size = len(json.dumps(attrs, ensure_ascii=False, default=str).encode())
|
|
if size > ATTRS_MAX_BYTES:
|
|
attrs = {"attrs_dropped_bytes": size}
|
|
if item.tab_id:
|
|
attrs = {**(attrs or {}), "tab_id": item.tab_id}
|
|
ts = item.ts if item.ts.tzinfo else item.ts.replace(tzinfo=timezone.utc)
|
|
return logsink.to_row(
|
|
{
|
|
"ts": min(max(ts, window_start), now),
|
|
"level": item.level,
|
|
"sev": logs.SEVERITY[item.level],
|
|
"source": "web",
|
|
"logger": "web",
|
|
"event": item.event,
|
|
"msg": logs.redact_text(item.msg),
|
|
"attrs": attrs,
|
|
"request_id": item.request_id,
|
|
"user_id": user.id,
|
|
"version": item.version,
|
|
}
|
|
)
|
|
|
|
|
|
@router.post("/events")
|
|
def ingest_events(
|
|
# The account first: dependencies resolve in order, so an anonymous caller is refused before a
|
|
# byte of its body is read.
|
|
user: User = Depends(current_user),
|
|
body: ClientEventsIn = Depends(_parsed_events),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
level = diagnostics.effective_level(user)
|
|
if level == diagnostics.BASE:
|
|
# Lapsed or withdrawn: the client stops streaming when it sees this.
|
|
raise HTTPException(status_code=409, detail={"code": "diagnostics_off"})
|
|
if not _batches.allow(str(user.id)):
|
|
raise HTTPException(status_code=429, detail={"code": "rate_limited"})
|
|
now = datetime.now(timezone.utc)
|
|
window_start = now - timedelta(hours=diagnostics.ELEVATION_HOURS)
|
|
floor = logs.SEVERITY[level]
|
|
rows = [
|
|
_row(item, user, now, window_start)
|
|
for item in body.events
|
|
if logs.SEVERITY[item.level] >= floor
|
|
]
|
|
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": 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",
|
|
# What the budget does, exactly: it is a SLIDING hour, and a refused batch is not
|
|
# charged to it, so smaller batches keep landing while there is room for them. Saying
|
|
# "the rest of the hour is not kept" would be a row contradicting the rows beneath it.
|
|
"msg": "The browser log passed its hourly storage budget; a batch bigger than what is left of it is not kept",
|
|
"attrs": {"budget_bytes": STORED_BYTES_PER_HOUR},
|
|
"user_id": user.id,
|
|
}
|
|
)
|