peter
ad6ccbac05
Tokens, notifications, quota, debug sessions and the reset-link burn filtered their owner inline. Ten of the fifteen filters had no test that noticed one going missing; each now has a test where another account's rows must not count, show or change.
489 lines
21 KiB
Python
489 lines
21 KiB
Python
"""Debug sessions: Start → reproduce → Stop, and the evidence a bug report is built from.
|
|
|
|
While a session is open its user's browser names it in `X-Debug-Session` on every request. The auth
|
|
dependency accepts the header only for the caller's OWN open session (anything else is ignored, never
|
|
an error — a stale tab must not break the app) and then logs that request at the session's level with
|
|
every line tagged by its id; the browser tags its own events the same way. The issue is later built
|
|
from exactly the rows carrying the tag, so what the review list shows is what gets sent.
|
|
|
|
Starting a session IS the consent for its level, for its duration: the start dialog carries the
|
|
consent text. It needs no change to the user's own diagnostics level, and none is made — when the
|
|
session ends the level in force is simply the user's own again.
|
|
|
|
A session is capped at `MAX_MINUTES`, set by the server at Start. An open one past its cap is over
|
|
whether or not anything has closed it yet: it reads as over at once, a new Start closes it, and the
|
|
hourly log job sweeps the rest. Sessions go with the logs they point at, after `log_retention_days`.
|
|
|
|
Core layer: imports `models`, `diagnostics`.
|
|
"""
|
|
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
from enum import StrEnum
|
|
from typing import Literal, get_args
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import and_, delete, func, not_, select, update
|
|
from sqlalchemy.orm import Session, defer
|
|
|
|
from app.userscope import owned_by
|
|
from app import diagnostics
|
|
from app.models import DebugScreenshot, DebugSession, LogEvent, User
|
|
|
|
HEADER = "x-debug-session"
|
|
MAX_MINUTES = diagnostics.SESSION_MINUTES
|
|
SCREENSHOT_MAX_BYTES = 10 * 1024 * 1024
|
|
SCREENSHOTS_MAX = 8
|
|
# What one user may keep in screenshots across ALL their sessions (until retention takes them): the
|
|
# per-session cap alone would let Start → 8 images → Stop repeat without end.
|
|
SCREENSHOT_USER_MAX_BYTES = 100 * 1024 * 1024
|
|
# The screenshot types kept, and the extension each is exported and sent under — one table, so an
|
|
# accepted type always has a name.
|
|
IMAGE_TYPES = {"image/png": "png", "image/jpeg": "jpg", "image/webp": "webp"}
|
|
# What a screenshot is called in the zip and on the issue. `issues/bundle.py` builds the names from
|
|
# this and `IMAGE_TYPES`; this module and `issues/gitea.py` read them back to tell a screenshot's
|
|
# link from a log's. One definition, because three copies of a name is how the next rename breaks
|
|
# `images_pending` into silence.
|
|
SCREENSHOT_PREFIX = "screenshot-"
|
|
# Rows an issue can carry. A 60-minute session at Teljes is a few thousand rows; past this the
|
|
# bundle says how many it left out rather than growing without bound.
|
|
BUNDLE_MAX = 5_000
|
|
|
|
SessionLevel = Literal["DEBUG", "TRACE"]
|
|
SESSION_LEVELS: tuple[str, ...] = get_args(SessionLevel)
|
|
# Detail order of the diagnostics levels: a session raises the level in force, never lowers it.
|
|
_DETAIL = {"INFO": 0, "DEBUG": 1, "TRACE": 2}
|
|
_ID_LEN = 16
|
|
|
|
|
|
class EndReason(StrEnum):
|
|
STOPPED = "stopped"
|
|
EXPIRED = "expired"
|
|
|
|
|
|
def is_live(session: DebugSession, now: datetime | None = None) -> bool:
|
|
"""Open and within its cap — THE liveness rule: an open session past its cap is over."""
|
|
return session.ended_at is None and session.until > (now or datetime.now(timezone.utc))
|
|
|
|
|
|
def open_session(db: Session, user: User) -> DebugSession | None:
|
|
"""The user's live session, or None. Read-only: one past its cap is over and reads as None;
|
|
`start` and the hourly sweep are what close it."""
|
|
session = _unclosed(db, user)
|
|
if session is None or not is_live(session, datetime.now(timezone.utc)):
|
|
return None
|
|
return session
|
|
|
|
|
|
def _unclosed(db: Session, user: User) -> DebugSession | None:
|
|
return db.scalar(
|
|
select(DebugSession).where(
|
|
owned_by(DebugSession.user_id, user), DebugSession.ended_at.is_(None)
|
|
)
|
|
)
|
|
|
|
|
|
def start(db: Session, user: User, level: SessionLevel) -> DebugSession:
|
|
"""Open a session, or return the live one — a second tab pressing Start joins the recording
|
|
rather than ending it. One past its cap is closed first: the one-open-per-user index would
|
|
refuse the new row beside it. Flushes; the caller commits. Two concurrent Starts can still both
|
|
get here: the index refuses the second, and the caller joins the one that won."""
|
|
now = datetime.now(timezone.utc)
|
|
current = _unclosed(db, user)
|
|
if current is not None:
|
|
if is_live(current, now):
|
|
return current
|
|
end(current, EndReason.EXPIRED)
|
|
db.flush()
|
|
session = DebugSession(
|
|
id=secrets.token_hex(_ID_LEN // 2),
|
|
user_id=user.id,
|
|
level=level,
|
|
started_at=now,
|
|
until=now + timedelta(minutes=MAX_MINUTES),
|
|
)
|
|
db.add(session)
|
|
db.flush()
|
|
return session
|
|
|
|
|
|
def stop(session: DebugSession) -> None:
|
|
"""The user's Stop. Past its cap the session had already ended on its own, and a late Stop must
|
|
not rename that."""
|
|
end(session, EndReason.STOPPED if is_live(session) else EndReason.EXPIRED)
|
|
|
|
|
|
def end(session: DebugSession, reason: EndReason) -> None:
|
|
"""Close a session. An expired one is dated at its cap, not at whenever it was noticed."""
|
|
if session.ended_at is not None:
|
|
return
|
|
session.ended_at = (
|
|
session.until
|
|
if reason is EndReason.EXPIRED
|
|
else min(datetime.now(timezone.utc), session.until)
|
|
)
|
|
session.ended_reason = reason
|
|
|
|
|
|
def from_header(db: Session, user: User, value: str | None) -> DebugSession | None:
|
|
"""The session a request names, if it is the caller's own and still live; else None."""
|
|
if not value or len(value) != _ID_LEN or not all(c in "0123456789abcdef" for c in value):
|
|
return None
|
|
session = db.get(DebugSession, value)
|
|
if (
|
|
session is None
|
|
or session.user_id != user.id
|
|
or not is_live(session, datetime.now(timezone.utc))
|
|
):
|
|
return None
|
|
return session
|
|
|
|
|
|
def level_for(user: User, session: DebugSession | None) -> tuple[str, datetime | None]:
|
|
"""The diagnostics level (and its expiry) a request is logged at: the user's own, raised to the
|
|
session's while one is named — whichever records more.
|
|
|
|
A named session also caps the expiry at its own: this pair is what the process tracks under the
|
|
session (app/logs.py), and an expiry outliving the recording would hold the root logger down
|
|
after it ended."""
|
|
own = diagnostics.effective_level(user)
|
|
until = user.diagnostics_until if own != diagnostics.BASE else None
|
|
if session is None:
|
|
return own, until
|
|
if _DETAIL[session.level] <= _DETAIL[own]:
|
|
return own, min(until, session.until) if until else session.until
|
|
return session.level, session.until
|
|
|
|
|
|
def lock(db: Session, session: DebugSession) -> None:
|
|
"""Re-read the session under a row lock, held until the caller commits. Everything that changes
|
|
what an issue is made of — the draft, the screenshots, the send itself — takes it first: a send
|
|
in flight sees none of them change under it, and each of them sees a send that committed."""
|
|
db.refresh(session, with_for_update=True)
|
|
|
|
|
|
def save_draft(session: DebugSession, draft: dict) -> None:
|
|
"""Keep the composer's draft (the caller validated, locked and commits). Shape: `title`, `body`
|
|
(the rendered markdown an export or an issue carries), `excluded` (log row ids taken out of the
|
|
evidence) and `form` (the composer's own fields, to reopen it as it was). A delivery's progress
|
|
is not the composer's to replace, and is kept."""
|
|
progress = (session.draft or {}).get("delivery")
|
|
session.draft = {**draft, "delivery": progress} if progress else draft
|
|
|
|
|
|
# --- a delivery's progress -----------------------------------------------------------------
|
|
# Written by app/issues/gitea.py into the draft; read HERE, so the send's decision and what the
|
|
# composer shows are one reading of it.
|
|
def delivery(session: DebugSession) -> dict:
|
|
return dict((session.draft or {}).get("delivery") or {})
|
|
|
|
|
|
def not_uploaded(session: DebugSession) -> list[str]:
|
|
"""The files a send that stopped half-way has still to upload — by name, as the issue carries
|
|
them. What the composer names to the user."""
|
|
return list(delivery(session).get("pending") or [])
|
|
|
|
|
|
def images_pending(session: DebugSession) -> bool:
|
|
"""Whether the issue's body still has to be edited to show the screenshots it already carries
|
|
(every upload went, the edit that shows them did not)."""
|
|
state = delivery(session)
|
|
shown = state.get("images_shown")
|
|
return not shown and any(n.startswith(SCREENSHOT_PREFIX) for n in state.get("links") or {})
|
|
|
|
|
|
def unfinished(session: DebugSession) -> bool:
|
|
"""Whether a send owes the issue anything — the retry's question."""
|
|
return bool(not_uploaded(session)) or images_pending(session)
|
|
|
|
|
|
# A send claims the session while it talks to the tracker, instead of holding the row's lock across
|
|
# those calls: a Stop, an autosave or a screenshot would queue behind a lock for the whole send.
|
|
# They see the claim and are refused at once instead. A claim not renewed for this long was left by
|
|
# a request that died mid-send (the process restarted), and the next send takes it over — which is
|
|
# why a live send RENEWS it every time it reports progress (`set_delivery`). A claim is a lease, and
|
|
# a send slower than the lease must not lose it while it is still running.
|
|
SEND_CLAIM_SECONDS = 600
|
|
|
|
|
|
def _claim(session: DebugSession) -> dict:
|
|
claim = delivery(session).get("sending")
|
|
return claim if isinstance(claim, dict) else {}
|
|
|
|
|
|
def _stamp(token: str) -> dict:
|
|
return {"at": datetime.now(timezone.utc).isoformat(), "by": token}
|
|
|
|
|
|
def sending(session: DebugSession) -> bool:
|
|
"""Whether a send is under way right now (a live claim)."""
|
|
at = _claim(session).get("at")
|
|
if not at:
|
|
return False
|
|
return (datetime.now(timezone.utc) - datetime.fromisoformat(at)).total_seconds() < (
|
|
SEND_CLAIM_SECONDS
|
|
)
|
|
|
|
|
|
def claim_send(session: DebugSession) -> str:
|
|
"""Mark the session as being sent, and return the token that OWNS this claim. The caller holds
|
|
the lock and commits, which is what makes the claim visible to everyone else."""
|
|
token = uuid4().hex
|
|
_write_delivery(session, {**delivery(session), "sending": _stamp(token)})
|
|
return token
|
|
|
|
|
|
def release_send(session: DebugSession, token: str) -> None:
|
|
"""Give back our OWN claim, and only ours: one that aged out and was taken over belongs to the
|
|
send holding it now, and handing that one back would let a third send in beside it."""
|
|
if _claim(session).get("by") != token:
|
|
return
|
|
_write_delivery(session, {k: v for k, v in delivery(session).items() if k != "sending"})
|
|
|
|
|
|
class SendClaimLost(Exception):
|
|
"""This send's claim is no longer on the row: it aged out and another send took it over. The
|
|
send must stop — what it uploads from here it can no longer record, so a retry would send it
|
|
again — and it must not write, because the row now holds the other send's progress."""
|
|
|
|
|
|
def claim_owner(db: Session, session: DebugSession) -> str | None:
|
|
"""Who holds the row's claim RIGHT NOW, read from the ROW and not from the object.
|
|
|
|
`expire_on_commit` is off (app/db.py), so a checkpoint does not re-read: a send that stalled
|
|
past its lease still carries its own dead token in memory long after another send took over.
|
|
A targeted scalar, not `lock`/`db.refresh` — a refresh here would discard the `issue_url` that
|
|
`gitea.deliver` assigns one line before its first progress report, and that link is the one
|
|
thing the send must not lose."""
|
|
return db.scalar(
|
|
select(DebugSession.draft["delivery"]["sending"]["by"].as_string()).where(
|
|
DebugSession.id == session.id
|
|
)
|
|
)
|
|
|
|
|
|
def set_delivery(db: Session, session: DebugSession, state: dict, token: str) -> None:
|
|
"""Write a delivery's progress (app/issues/gitea.py keeps it; this module reads it), and renew
|
|
the claim while doing it — a send still reporting progress is still alive, so one slower than
|
|
the lease cannot expire while it is still running.
|
|
|
|
The claim is not the caller's to carry: `gitea.deliver` threads ONE snapshot of this dict
|
|
through a whole send, so a progress report that carried the claim wrote a stale one back at
|
|
every step. It is stamped from `token` here instead, and only after the ROW confirms the claim
|
|
is still ours. Without that check the write is a full-column overwrite with no WHERE: a send
|
|
that lost its lease came back and stomped the take-over send's committed links, then put its
|
|
own dead token back on an idle session. `release_send` has always made
|
|
the same check; this is the other half of it."""
|
|
if claim_owner(db, session) != token:
|
|
raise SendClaimLost(session.id)
|
|
state = {k: v for k, v in state.items() if k != "sending"}
|
|
state["sending"] = _stamp(token)
|
|
_write_delivery(session, state)
|
|
|
|
|
|
def _write_delivery(session: DebugSession, state: dict) -> None:
|
|
# A new dict, so the JSONB column sees the change.
|
|
session.draft = {**(session.draft or {}), "delivery": state}
|
|
|
|
|
|
def serialize(session: DebugSession | None, screenshots: int = 0) -> dict | None:
|
|
if session is None:
|
|
return None
|
|
return {
|
|
"id": session.id,
|
|
"level": session.level,
|
|
"started_at": session.started_at.isoformat(),
|
|
"until": session.until.isoformat(),
|
|
"ended_at": session.ended_at.isoformat() if session.ended_at else None,
|
|
"ended_reason": session.ended_reason,
|
|
"issue_url": session.issue_url,
|
|
# A send that stopped half-way: the files it has still to upload, and whether the issue's
|
|
# body has still to be edited to show the screenshots it already carries. Both are true of a
|
|
# send that is RUNNING too — every step commits its progress — so `sending` goes with them:
|
|
# without it a reload or a second tab renders a live send as a failed one, and offers a
|
|
# retry that is certain to be refused.
|
|
"pending_files": not_uploaded(session),
|
|
"images_pending": images_pending(session),
|
|
"sending": sending(session),
|
|
"screenshots": screenshots,
|
|
"draft": session.draft,
|
|
}
|
|
|
|
|
|
def view(db: Session, session: DebugSession) -> dict:
|
|
"""The session as the API serves it — one shape wherever it is served."""
|
|
return serialize(session, screenshot_count(db, session)) or {}
|
|
|
|
|
|
def screenshot_count(db: Session, session: DebugSession) -> int:
|
|
return (
|
|
db.scalar(
|
|
select(func.count())
|
|
.select_from(DebugScreenshot)
|
|
.where(DebugScreenshot.session_id == session.id)
|
|
)
|
|
or 0
|
|
)
|
|
|
|
|
|
# --- screenshots ---------------------------------------------------------------------------
|
|
# By content, not by the declared type: what is stored is served back to a browser as an image.
|
|
_SIGNATURES = (
|
|
(b"\x89PNG\r\n\x1a\n", "image/png"),
|
|
(b"\xff\xd8\xff", "image/jpeg"),
|
|
)
|
|
|
|
|
|
def sniff(data: bytes) -> str | None:
|
|
"""The image type `data` really is — one of IMAGE_TYPES — or None."""
|
|
for magic, mime in _SIGNATURES:
|
|
if data.startswith(magic):
|
|
return mime
|
|
if len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP":
|
|
return "image/webp"
|
|
return None
|
|
|
|
|
|
def lock_user(db: Session, user_id: int) -> None:
|
|
"""Hold this user's own lock for the rest of the transaction. A session's row lock bounds what
|
|
THAT session may keep; the byte cap below is the user's across all of them, and a lock on one
|
|
session cannot serialize a check that reads them all — two uploads to two sessions both read the
|
|
total before either commits, and both pass."""
|
|
db.execute(select(func.pg_advisory_xact_lock(user_id)))
|
|
|
|
|
|
def stored_screenshot_bytes(db: Session, user_id: int) -> int:
|
|
return (
|
|
db.scalar(
|
|
select(func.coalesce(func.sum(DebugScreenshot.size), 0))
|
|
.join(DebugSession, DebugSession.id == DebugScreenshot.session_id)
|
|
.where(owned_by(DebugSession.user_id, user_id))
|
|
)
|
|
or 0
|
|
)
|
|
|
|
|
|
class ScreenshotRefused(Exception):
|
|
"""A screenshot that cannot be kept; `code` is the refusal's machine word."""
|
|
|
|
def __init__(self, code: str) -> None:
|
|
super().__init__(code)
|
|
self.code = code
|
|
|
|
|
|
def add_screenshot(db: Session, session: DebugSession, data: bytes) -> DebugScreenshot:
|
|
"""Keep one screenshot. Locks the user and the session, and flushes; the caller commits.
|
|
Refused while a send is under way or once the issue was sent (it can no longer carry it), past
|
|
the per-session count or the user's storage, or when it is not an image we can show."""
|
|
if not data:
|
|
raise ScreenshotRefused("empty")
|
|
mime = sniff(data)
|
|
if mime is None:
|
|
raise ScreenshotRefused("not_an_image")
|
|
# Under both locks: two uploads at once cannot both pass the count (the session's), nor both
|
|
# pass the user's byte cap from two different sessions (the user's). Always in this order.
|
|
lock_user(db, session.user_id)
|
|
lock(db, session)
|
|
if sending(session):
|
|
raise ScreenshotRefused("sending")
|
|
if session.issue_url is not None:
|
|
raise ScreenshotRefused("already_sent")
|
|
if screenshot_count(db, session) >= SCREENSHOTS_MAX:
|
|
raise ScreenshotRefused("too_many")
|
|
if stored_screenshot_bytes(db, session.user_id) + len(data) > SCREENSHOT_USER_MAX_BYTES:
|
|
raise ScreenshotRefused("storage_full")
|
|
shot = DebugScreenshot(
|
|
session_id=session.id,
|
|
created_at=datetime.now(timezone.utc),
|
|
mime=mime,
|
|
size=len(data),
|
|
data=data,
|
|
)
|
|
db.add(shot)
|
|
db.flush()
|
|
return shot
|
|
|
|
|
|
def remove_screenshot(db: Session, session: DebugSession, shot: DebugScreenshot) -> None:
|
|
"""Delete one screenshot, under the lock. Refused while a send is under way or unfinished: a
|
|
delivery names the screenshots by their order, and removing one would renumber what it still
|
|
has to send. Once the issue has everything, this only drops OUR copy — the issue keeps its
|
|
own. The caller commits."""
|
|
lock(db, session)
|
|
if sending(session):
|
|
raise ScreenshotRefused("sending")
|
|
if unfinished(session):
|
|
raise ScreenshotRefused("already_sent")
|
|
db.delete(shot)
|
|
db.flush()
|
|
|
|
|
|
def screenshots(db: Session, session: DebugSession) -> list[DebugScreenshot]:
|
|
"""The session's screenshots, oldest first, WITHOUT their bytes loaded."""
|
|
return list(
|
|
db.scalars(
|
|
select(DebugScreenshot)
|
|
.where(DebugScreenshot.session_id == session.id)
|
|
.options(defer(DebugScreenshot.data))
|
|
.order_by(DebugScreenshot.id)
|
|
)
|
|
)
|
|
|
|
|
|
# --- the evidence --------------------------------------------------------------------------
|
|
# The browser's own log upload (`POST` here, every few seconds while recording) is how the evidence
|
|
# arrives, not part of it: its server lines would fill the bundle with the recording's plumbing. The
|
|
# browser leaves its copies out for the same reason (frontend lib/diagnostics `isOwnRequest`).
|
|
INGEST_PATH = "/api/diagnostics/events"
|
|
|
|
|
|
def _evidence(session: DebugSession):
|
|
upload_line = and_(
|
|
LogEvent.event == "http.request", LogEvent.attrs["path"].astext == INGEST_PATH
|
|
)
|
|
# `IS TRUE`, not the bare condition: for a row with no event or no attrs it is NULL, and
|
|
# `NOT NULL` would drop exactly the rows that are evidence.
|
|
return (LogEvent.debug_session_id == session.id, not_(upload_line.is_(True)))
|
|
|
|
|
|
def bundle(
|
|
db: Session, session: DebugSession, *, drop_excluded: bool = False
|
|
) -> tuple[list[LogEvent], int]:
|
|
"""The session's log rows, browser and server interleaved in time, capped at BUNDLE_MAX — and
|
|
how many rows it has in all, so a cut bundle can say so. `drop_excluded` leaves out the rows
|
|
the user took out in the composer: what an export or an issue carries."""
|
|
where = list(_evidence(session))
|
|
excluded = (session.draft or {}).get("excluded") or [] if drop_excluded else []
|
|
if excluded:
|
|
where.append(LogEvent.id.not_in(excluded))
|
|
rows = list(
|
|
db.scalars(
|
|
select(LogEvent).where(*where).order_by(LogEvent.ts, LogEvent.id).limit(BUNDLE_MAX)
|
|
)
|
|
)
|
|
total = len(rows)
|
|
if total == BUNDLE_MAX:
|
|
total = db.scalar(select(func.count()).select_from(LogEvent).where(*where)) or total
|
|
return rows, total
|
|
|
|
|
|
# --- housekeeping (the hourly log job) -----------------------------------------------------
|
|
def sweep(db: Session) -> int:
|
|
"""Stage closing every open session past its cap; returns how many."""
|
|
now = datetime.now(timezone.utc)
|
|
result = db.execute(
|
|
update(DebugSession)
|
|
.where(DebugSession.ended_at.is_(None), DebugSession.until <= now)
|
|
.values(ended_at=DebugSession.until, ended_reason=EndReason.EXPIRED.value)
|
|
.execution_options(synchronize_session=False)
|
|
)
|
|
return result.rowcount or 0
|
|
|
|
|
|
def gc(db: Session, retention_days: int) -> int:
|
|
"""Stage deleting sessions (and by cascade their screenshots) started before the log retention
|
|
window — the rows they would bundle are gone by then too. Returns how many."""
|
|
cutoff = datetime.now(timezone.utc) - timedelta(days=max(1, retention_days))
|
|
result = db.execute(delete(DebugSession).where(DebugSession.started_at < cutoff))
|
|
return result.rowcount or 0
|