siftlode/backend/app/routes/diagnostics.py

"""The browser's log, sent while its user has a raised diagnostics level.

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 — where an open debug
    session the request names counts as consent to ITS level, and only its own events keep
    its tag;
  * 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 debugsession, diagnostics, logs, logsink
from app.auth import current_user
from app.db import get_db, release_read_connection
from app.models import DebugSession, LogEvent, User
from app.ratelimit import RateLimiter
from app.routes._common import capped_body

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}$")
    # The debug session the event was captured in; kept only when it is the one this request
    # names in `X-Debug-Session` and the auth dependency found it the caller's own and open.
    debug_session_id: str | None = Field(default=None, pattern=r"^[0-9a-f]{16}$")
    version: str | None = Field(default=None, max_length=64)


class ClientEventsIn(BaseModel):
    events: list[ClientEvent] = Field(max_length=EVENTS_MAX)


async def _parsed_events(request: Request, db: Session = Depends(get_db)) -> 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"})
    # `current_user` resolved above on a SELECT, so this session is holding a pooled connection.
    # Give it back before awaiting the body — the sink is a background beacon and its sender may be
    # on a link far slower than the 512 KiB suggests. See `db.release_read_connection`.
    release_read_connection(db)
    raw = await capped_body(request, BODY_MAX_BYTES)
    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,
    session: DebugSession | None,
) -> 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,
            "debug_session_id": (
                session.id if session and item.debug_session_id == session.id else None
            ),
            "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.
    request: Request,
    user: User = Depends(current_user),
    body: ClientEventsIn = Depends(_parsed_events),
    db: Session = Depends(get_db),
) -> dict:
    # A debug session the request names raises the level for its own events: starting one is
    # the consent to that level for its duration, whatever the user's own level is.
    session = debugsession.from_header(db, user, request.headers.get(debugsession.HEADER))
    level, _ = debugsession.level_for(user, session)
    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, session)
        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,
        }
    )