siftlode/backend/app/routes/diagnostics.py

"""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:
  * the whole body is capped before it is parsed, and so are the number of events and each event's
    attributes;
  * 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)

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)."""
    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 rows:
        db.execute(insert(LogEvent), rows)
        db.commit()
    return {"ok": True, "stored": len(rows)}