fix(uploads): make the drift guards fail for the reasons they claim
Deep-review round 3, targeted at the surfaces round 2's fix commit changed. Five of the eight findings are the same shape: the guards written to stop half-done sweeps were themselves half-done. test_upload.py's display-field guard walked only `ast.Assign`, so the one writer with an unbounded REQUEST HEADER behind it — `spool_upload`'s `display_name=display_text(...)` keyword — was invisible to the guard meant to protect it. It also stopped at `app/routes/`, missing the four writers in `downloads/` that actually construct the row, and asserted `not offenders`, which enforces nothing once a refactor moves the writers out of its view. It now walks assignments AND keywords across `app/`, and holds exact set equality against a registry of all twelve write sites with each one's value. test_body_limit.py's release guard used `any(... ast.walk(fn))`, which accepts a `release_read_connection` that runs AFTER the await it exists to precede — satisfying the letter of the rule while pinning a connection for exactly the transfer it was supposed to free. Position is checked now, and a late release is its own, louder failure. The registry keyed sites by file basename while scanning recursively, and `admin.py`, `discover.py` and `playlists.py` each exist both at the top level and under `plex/` — the same entry-collapse the registry was widened to close. Keyed by relative path now. And all three guards recognised a body read only by the `capped_` naming convention, so a route spelling `await request.body()` inline declared no ceiling, released no connection, and was invisible to every one of them. `release_read_connection`'s precondition claimed the caller "must have NOTHING PENDING, and that is checked rather than described" while checking only UNFLUSHED ORM work. Measured: after `add`+`flush` all three collections are empty and the rollback loses the row, and Core DML never enters them at all — both of them this codebase's canonical write shapes. It now also asks the transaction directly via `pg_current_xact_id_if_assigned()`. `errors.tooLarge` was reworded to "try a smaller file", but that key answers every route's body cap: the issue composer's over-size JSON draft got advice about a file it does not have. Back to payload-neutral; the file-specific sentence keeps its own key. The upload dialog's new danger border outranked `dragOver`, so a user replacing an over-cap file got no sign the drop zone was live — and with a file picked the label shows its name rather than "drop it here", so the border was the only feedback left. Four further findings were refuted 3/3 and are not fixed: `aria-invalid` on the drop zone (measured in Chromium — the AX node does carry it on a native button), a bidi-override spoof through `display_text` (the served filename already strips those categories, and the proposed filter would have broken every ZWJ emoji), and `capped_body` not draining (the screenshot route's edge ceiling equals its own cap, so that refusal is unreachable, and the log sink is sent by beacon with nobody reading the answer).
This commit is contained in:
+20
-1
@@ -1,6 +1,6 @@
|
||||
from collections.abc import Iterator
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy import create_engine, text
|
||||
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
|
||||
|
||||
from app.config import settings
|
||||
@@ -54,6 +54,16 @@ def release_read_connection(db: Session) -> None:
|
||||
shared with every dependency that ran before, so a rollback here would silently discard a write
|
||||
one of them staged — `service.enqueue` already learned that lesson the expensive way and uses a
|
||||
SAVEPOINT for it. CLAUDE.md: a rule stated in a comment must also be enforced in code.
|
||||
|
||||
TWO checks, because the obvious one is blind to the shapes this codebase actually writes in.
|
||||
`db.new`/`deleted`/`dirty` hold only UNFLUSHED unit-of-work: the moment a helper calls
|
||||
`db.flush()` — CLAUDE.md's canonical split is "the handler commits; helpers flush" — the object
|
||||
is persistent with no unflushed history and all three sets are empty again, and Core DML
|
||||
(`db.execute(update(...))`) never enters them at all. Measured: after `add`+`flush` the sets are
|
||||
empty, the old check passed, and the rollback lost the row. So the transaction is asked
|
||||
directly. Postgres assigns a transaction id on the first WRITE and never for a read, so
|
||||
`pg_current_xact_id_if_assigned()` is NULL for exactly the state this function is safe in — and
|
||||
it sees a raw `text("UPDATE …")` too, which no session-level bookkeeping would.
|
||||
"""
|
||||
if db.new or db.deleted or db.dirty:
|
||||
raise RuntimeError(
|
||||
@@ -61,4 +71,13 @@ def release_read_connection(db: Session) -> None:
|
||||
"staged a write, which ending the transaction would discard. Commit it first, or do "
|
||||
"it after the await."
|
||||
)
|
||||
if (
|
||||
db.in_transaction()
|
||||
and db.execute(text("SELECT pg_current_xact_id_if_assigned()")).scalar() is not None
|
||||
):
|
||||
raise RuntimeError(
|
||||
"release_read_connection() on a transaction that has already WRITTEN: something ahead "
|
||||
"of this dependency flushed a row or ran an UPDATE/INSERT/DELETE, which ending the "
|
||||
"transaction would discard. Commit it first, or do it after the await."
|
||||
)
|
||||
db.rollback()
|
||||
|
||||
@@ -212,7 +212,12 @@ def test_the_default_ceiling_clears_the_largest_body_a_default_route_takes():
|
||||
# BOTH helpers, not just `capped_body`: P1 added `capped_stream_to` as its sibling and the registry
|
||||
# did not follow, so the upload route — the one this whole invariant was written for — was the one
|
||||
# site the drift guard could not see (deep-review round 1). The names come from `_common.py` itself
|
||||
# below, so a THIRD sibling cannot be added without this list noticing either.
|
||||
# below, so a THIRD sibling cannot be added without this list noticing either — and a route that
|
||||
# skips the helpers altogether and awaits `request.body()` lands here as `<no ceiling declared>`
|
||||
# rather than vanishing (round 3).
|
||||
#
|
||||
# The module is the path relative to `app/routes/`, so a site under `plex/` is spelled
|
||||
# `plex/admin.py` and cannot collide with the top-level `admin.py`.
|
||||
CAPPED_BODY_SITES = {
|
||||
("diagnostics.py", "_parsed_events", "capped_body", "BODY_MAX_BYTES"),
|
||||
("debug_sessions.py", "_image_body", "capped_body", "debugsession.SCREENSHOT_MAX_BYTES"),
|
||||
@@ -242,6 +247,42 @@ def test_the_registry_covers_every_body_reading_helper():
|
||||
)
|
||||
|
||||
|
||||
# The ways a route can read its own body WITHOUT one of `_common.py`'s helpers. Every `Request`
|
||||
# parameter under `app/routes/` is named `request`, so the base name is an exact filter and not a
|
||||
# guess — the only other `.json()` in the package is `exc.json()` on a ValidationError.
|
||||
#
|
||||
# Without this set the three guards below know only the `capped_*` naming convention, which means
|
||||
# they are blind to the shape `_common.py` itself is written in. A route spelling `await
|
||||
# request.body()` inline declares no ceiling, hands no connection back, and would be invisible to
|
||||
# all three — reading as "no new sites" rather than as a failure, which is the exact way a drift
|
||||
# guard dies quietly (deep-review round 3).
|
||||
RAW_BODY_READS = {"body", "stream", "form", "json"}
|
||||
|
||||
|
||||
def _call_name(node: ast.Call) -> str | None:
|
||||
if isinstance(node.func, ast.Attribute):
|
||||
return node.func.attr
|
||||
return getattr(node.func, "id", None)
|
||||
|
||||
|
||||
def _body_read_calls(fn: ast.AST, readers: set[str]) -> list[tuple[ast.Call, str]]:
|
||||
"""Every call inside `fn` that reads the request body, as (node, name) in source order."""
|
||||
out: list[tuple[ast.Call, str]] = []
|
||||
for node in ast.walk(fn):
|
||||
if not isinstance(node, ast.Call):
|
||||
continue
|
||||
if (
|
||||
isinstance(node.func, ast.Attribute)
|
||||
and isinstance(node.func.value, ast.Name)
|
||||
and node.func.value.id == "request"
|
||||
and node.func.attr in RAW_BODY_READS
|
||||
):
|
||||
out.append((node, f"request.{node.func.attr}"))
|
||||
elif _call_name(node) in readers:
|
||||
out.append((node, _call_name(node) or ""))
|
||||
return sorted(out, key=lambda pair: (pair[0].lineno, pair[0].col_offset))
|
||||
|
||||
|
||||
def _body_reading_functions() -> dict[tuple[str, str], dict]:
|
||||
"""Every function under `app/routes/` that reads a request body itself, keyed by
|
||||
(module, function).
|
||||
@@ -254,55 +295,57 @@ def _body_reading_functions() -> dict[tuple[str, str], dict]:
|
||||
Keyed by the ENCLOSING FUNCTION, not just the module: a set keyed by module and limit
|
||||
expression collapses two call sites that spell their limit the same way, which is exactly what
|
||||
a second route copied from the first would do.
|
||||
|
||||
The module half of the key is the path RELATIVE to `app/routes/`, not the basename: the scan
|
||||
recurses, and `admin.py`, `discover.py` and `playlists.py` each exist both at the top level and
|
||||
under `plex/`. Keying by basename would let the second of a colliding pair overwrite the first
|
||||
and take its `releases` flag out of the comparison with it — the same entry-collapse this
|
||||
registry was widened to close.
|
||||
|
||||
`_common.py` is skipped: it is where the helpers live, so its own `request.stream()` loops are
|
||||
the implementation of the rule rather than a violation of it.
|
||||
"""
|
||||
readers = _body_readers()
|
||||
routes_dir = pathlib.Path(__file__).resolve().parents[1] / "app" / "routes"
|
||||
found: dict[tuple[str, str], dict] = {}
|
||||
for py in sorted(routes_dir.rglob("*.py")):
|
||||
if py.name == "_common.py":
|
||||
continue
|
||||
module = py.relative_to(routes_dir).as_posix()
|
||||
tree = ast.parse(py.read_text("utf-8"), filename=str(py))
|
||||
for fn in ast.walk(tree):
|
||||
if not isinstance(fn, (ast.AsyncFunctionDef, ast.FunctionDef)):
|
||||
continue
|
||||
calls = [
|
||||
node
|
||||
for node in ast.walk(fn)
|
||||
if isinstance(node, ast.Call)
|
||||
and (
|
||||
node.func.attr
|
||||
if isinstance(node.func, ast.Attribute)
|
||||
else getattr(node.func, "id", None)
|
||||
)
|
||||
in readers
|
||||
]
|
||||
calls = _body_read_calls(fn, readers)
|
||||
if not calls:
|
||||
continue
|
||||
names = [
|
||||
node.func.attr
|
||||
if isinstance(node.func, ast.Attribute)
|
||||
else getattr(node.func, "id", None)
|
||||
for node in calls
|
||||
]
|
||||
names = [name for _, name in calls]
|
||||
limits = []
|
||||
for node, name in zip(calls, names):
|
||||
for node, name in zip([n for n, _ in calls], names):
|
||||
if name.startswith("request."):
|
||||
limits.append("<no ceiling declared>")
|
||||
continue
|
||||
limit = next((kw.value for kw in node.keywords if kw.arg == "limit"), None)
|
||||
if limit is None:
|
||||
pos = 2 if name == "capped_stream_to" else 1
|
||||
limit = node.args[pos] if len(node.args) > pos else None
|
||||
limits.append(ast.unparse(limit) if limit is not None else "<none>")
|
||||
releases = any(
|
||||
isinstance(node, ast.Call)
|
||||
and (
|
||||
node.func.attr
|
||||
if isinstance(node.func, ast.Attribute)
|
||||
else getattr(node.func, "id", None)
|
||||
)
|
||||
== "release_read_connection"
|
||||
# POSITION, not presence. `any(...)` over the function accepted a release that ran
|
||||
# AFTER the await it exists to precede — the natural spelling for someone reading the
|
||||
# name as "release it when you're done" — while pinning a connection for the whole
|
||||
# transfer. The first body read is what the release has to beat.
|
||||
releases = [
|
||||
node
|
||||
for node in ast.walk(fn)
|
||||
)
|
||||
found[(py.name, fn.name)] = {
|
||||
if isinstance(node, ast.Call) and _call_name(node) == "release_read_connection"
|
||||
]
|
||||
first_read = calls[0][0].lineno
|
||||
found[(module, fn.name)] = {
|
||||
"helpers": names,
|
||||
"limits": limits,
|
||||
"releases": releases,
|
||||
"releases": any(node.lineno < first_read for node in releases),
|
||||
"releases_late": bool(releases)
|
||||
and not any(node.lineno < first_read for node in releases),
|
||||
}
|
||||
return found
|
||||
|
||||
@@ -356,6 +399,43 @@ def test_release_read_connection_refuses_to_discard_pending_work(db):
|
||||
db.rollback() # the staged row is this test's own
|
||||
|
||||
|
||||
def test_release_read_connection_refuses_to_discard_a_FLUSHED_write(db):
|
||||
"""The half the unit-of-work check cannot see.
|
||||
|
||||
`db.new`/`dirty`/`deleted` empty out the moment a helper flushes, and CLAUDE.md's canonical
|
||||
split is "the handler commits; helpers flush" — so the single most likely future violator is
|
||||
the one shape the obvious check is blind to. Measured before the fix: all three sets empty, the
|
||||
guard silent, the row gone after the rollback.
|
||||
"""
|
||||
from app.db import release_read_connection
|
||||
|
||||
db.add(SystemConfig(key="probe_flushed_write", value="1"))
|
||||
db.flush()
|
||||
assert not (db.new or db.dirty or db.deleted), (
|
||||
"the flush did not clear the unit-of-work, so this test would pass for the wrong reason"
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="already WRITTEN"):
|
||||
release_read_connection(db)
|
||||
db.rollback()
|
||||
|
||||
|
||||
def test_release_read_connection_refuses_to_discard_a_core_dml_write(db):
|
||||
"""The other half: `db.execute(update(...))` never enters the unit-of-work at all, and
|
||||
`downloads/service.py` writes that way on purpose (a conditional `UPDATE … WHERE status=…` is
|
||||
the honest-states pattern)."""
|
||||
from sqlalchemy import insert
|
||||
|
||||
from app.db import release_read_connection
|
||||
|
||||
db.execute(insert(SystemConfig).values(key="probe_core_dml", value="1"))
|
||||
assert not (db.new or db.dirty or db.deleted), (
|
||||
"Core DML entered the unit-of-work, so this test would pass for the wrong reason"
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="already WRITTEN"):
|
||||
release_read_connection(db)
|
||||
db.rollback()
|
||||
|
||||
|
||||
def test_every_body_reading_site_hands_its_pooled_connection_back_first():
|
||||
"""Reading the body yourself is what makes a request SLOW inside its own database session.
|
||||
|
||||
@@ -367,15 +447,26 @@ def test_every_body_reading_site_hands_its_pooled_connection_back_first():
|
||||
|
||||
Held generically because the three sites arrived one at a time and only the newest was fixed:
|
||||
a fourth body-reading route gets the same requirement automatically.
|
||||
|
||||
BEFORE the read, not merely somewhere in the function. A release that runs after the await
|
||||
satisfies the letter of the rule and none of its purpose — the connection was held for exactly
|
||||
the span the rule exists to protect — so `releases_late` is reported as its own, louder case.
|
||||
"""
|
||||
offenders = [
|
||||
sites = _body_reading_functions()
|
||||
late = [f"{module}:{fn}" for (module, fn), site in sites.items() if site["releases_late"]]
|
||||
missing = [
|
||||
f"{module}:{fn}"
|
||||
for (module, fn), site in _body_reading_functions().items()
|
||||
if not site["releases"]
|
||||
for (module, fn), site in sites.items()
|
||||
if not site["releases"] and not site["releases_late"]
|
||||
]
|
||||
assert not offenders, (
|
||||
assert not late, (
|
||||
"a route calls db.release_read_connection AFTER it reads the body, which holds the pooled "
|
||||
f"connection for exactly the transfer it was supposed to free: {late}. Move the call above "
|
||||
"the await."
|
||||
)
|
||||
assert not missing, (
|
||||
"a route reads its own request body while still inside its read transaction, pinning a "
|
||||
f"pooled connection for the whole transfer: {offenders}. Call "
|
||||
f"pooled connection for the whole transfer: {missing}. Call "
|
||||
"db.release_read_connection(db) before the await — see routes/downloads.py:spool_upload."
|
||||
)
|
||||
|
||||
|
||||
@@ -480,34 +480,112 @@ def test_a_long_display_name_is_truncated_on_EVERY_route_that_takes_one(
|
||||
assert patched.json()["display_uploader"] == bound
|
||||
|
||||
|
||||
def test_no_route_assigns_a_display_field_without_the_shared_helper():
|
||||
"""The drift guard for the test above: it holds five routes, this holds the rule.
|
||||
def _display_write_sites() -> dict[tuple[str, str], set[str]]:
|
||||
"""Every place under `backend/app/` that puts a value into a display field, as
|
||||
(module relative to `app/`, enclosing function) -> the set of value expressions written there.
|
||||
|
||||
Parsed, not grepped. Every `display_name = ...` / `display_uploader = ...` assignment under
|
||||
`app/routes/` takes a string a user typed and puts it on its way to a varchar(255) column, so
|
||||
every one of them must go through `display_text`. A sixth route added tomorrow fails here
|
||||
instead of 500ing in production.
|
||||
TWO SHAPES, because the first version of this guard walked only `ast.Assign` and was therefore
|
||||
blind to the shape the upload route itself uses — `SpooledUpload(..., display_name=...)` is an
|
||||
`ast.keyword`, not an assignment, so the one site with an unbounded REQUEST HEADER behind it was
|
||||
the site the guard could not see (deep-review round 3).
|
||||
|
||||
ANNOTATED assignments are excluded on purpose: `display_name: Mapped[str | None] =
|
||||
mapped_column(...)` in `models.py` and `display_name: str | None = None` in a Pydantic model are
|
||||
declarations of the field, not writes into it.
|
||||
|
||||
The whole of `app/`, not just `app/routes/`: a job row is constructed in `downloads/service.py`
|
||||
and `downloads/upload.py`, so a walk stopping at the HTTP boundary misses the writers that
|
||||
actually reach the column.
|
||||
"""
|
||||
routes_dir = pathlib.Path(__file__).resolve().parents[1] / "app" / "routes"
|
||||
offenders: list[str] = []
|
||||
for py in sorted(routes_dir.rglob("*.py")):
|
||||
app_dir = pathlib.Path(__file__).resolve().parents[1] / "app"
|
||||
found: dict[tuple[str, str], set[str]] = {}
|
||||
for py in sorted(app_dir.rglob("*.py")):
|
||||
module = py.relative_to(app_dir).as_posix()
|
||||
tree = ast.parse(py.read_text("utf-8"), filename=str(py))
|
||||
enclosing: dict[int, str] = {}
|
||||
for fn in ast.walk(tree):
|
||||
if isinstance(fn, (ast.AsyncFunctionDef, ast.FunctionDef)):
|
||||
for node in ast.walk(fn):
|
||||
enclosing.setdefault(id(node), fn.name)
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.Assign):
|
||||
if isinstance(node, ast.Assign):
|
||||
names = {
|
||||
t.id if isinstance(t, ast.Name) else t.attr
|
||||
for t in node.targets
|
||||
if isinstance(t, (ast.Name, ast.Attribute))
|
||||
}
|
||||
value = node.value
|
||||
elif isinstance(node, ast.keyword):
|
||||
names = {node.arg} if node.arg else set()
|
||||
value = node.value
|
||||
else:
|
||||
continue
|
||||
names = {
|
||||
t.id if isinstance(t, ast.Name) else t.attr
|
||||
for t in node.targets
|
||||
if isinstance(t, (ast.Name, ast.Attribute))
|
||||
}
|
||||
if not names & {"display_name", "display_uploader"}:
|
||||
continue
|
||||
value = ast.unparse(node.value)
|
||||
if "display_text" not in value:
|
||||
offenders.append(f"{py.name}:{node.lineno}: {value}")
|
||||
assert not offenders, (
|
||||
"a display field is assigned without models.display_text, so a name longer "
|
||||
f"than the column is a 500 rather than a truncation: {offenders}"
|
||||
key = (module, enclosing.get(id(node), "<module>"))
|
||||
found.setdefault(key, set()).add(ast.unparse(value))
|
||||
return found
|
||||
|
||||
|
||||
# Every write into a display field, with the expression each one writes. A varchar(255) column with
|
||||
# a request header behind it turns a too-long name into a psycopg DataError — a 500 AFTER the whole
|
||||
# file has transferred — so each of these has to be bounded, and each one here has been read.
|
||||
#
|
||||
# Bounded by `display_text` unless noted:
|
||||
# auth.get_or_create_demo_user a literal.
|
||||
# service.redownload CLEARS the field so the fresh metadata wins — `None` is the
|
||||
# one value that needs no bound, and a clear is still a write.
|
||||
# service.enqueue / enqueue_edit `display_name` is already bounded by its caller; the fallback
|
||||
# slices the YouTube title to the same width.
|
||||
# downloads.upload_download passes through `spool_upload`'s value, bounded there.
|
||||
#
|
||||
# Held as EXACT EQUALITY rather than as a list of offenders. The offender form could only fail
|
||||
# upward: refactoring the writers into a shape the walker did not recognise left it asserting
|
||||
# `not []` and enforcing nothing, which is how a drift guard dies quietly (round 3). This one fails
|
||||
# when a site appears, when one disappears, and when one changes what it writes.
|
||||
DISPLAY_WRITE_SITES = {
|
||||
("auth.py", "get_or_create_demo_user"): {"'Demo'"},
|
||||
("downloads/service.py", "redownload"): {"None"},
|
||||
("downloads/service.py", "enqueue"): {
|
||||
"display_name or (asset.title[:DISPLAY_TEXT_MAX] "
|
||||
"if asset.status == 'ready' and asset.title else None)"
|
||||
},
|
||||
("downloads/service.py", "enqueue_edit"): {"display_name or asset.title[:DISPLAY_TEXT_MAX]"},
|
||||
("downloads/upload.py", "ingest_upload"): {"display_text(display_name)"},
|
||||
("routes/auth.py", "callback"): {"display_text(userinfo.get('name'))"},
|
||||
("routes/auth.py", "_complete_link"): {"display_text(userinfo.get('name'))"},
|
||||
("routes/downloads.py", "enqueue_download"): {"display_text(body.display_name)"},
|
||||
("routes/downloads.py", "enqueue_edit"): {"display_text(body.display_name)"},
|
||||
("routes/downloads.py", "update_download_meta"): {
|
||||
"display_text(body.display_name)",
|
||||
"display_text(body.display_uploader)",
|
||||
},
|
||||
("routes/downloads.py", "upload_download"): {"upload.display_name"},
|
||||
("routes/downloads.py", "spool_upload"): {
|
||||
"display_text(_header_text(request, UPLOAD_DISPLAY_NAME_HEADER))"
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_no_code_path_assigns_a_display_field_without_the_shared_helper():
|
||||
"""The drift guard for the test above: it holds five routes, this holds the rule.
|
||||
|
||||
Parsed, not grepped. Every value that reaches a display field takes a string a user typed (or
|
||||
YouTube supplied) and puts it on its way to a varchar(255) column, so every one of them must be
|
||||
bounded. A writer added tomorrow fails here instead of 500ing in production.
|
||||
"""
|
||||
found = _display_write_sites()
|
||||
moved = set(found) ^ set(DISPLAY_WRITE_SITES)
|
||||
changed = {
|
||||
key: (found[key], DISPLAY_WRITE_SITES[key])
|
||||
for key in set(found) & set(DISPLAY_WRITE_SITES)
|
||||
if found[key] != DISPLAY_WRITE_SITES[key]
|
||||
}
|
||||
assert found == DISPLAY_WRITE_SITES, (
|
||||
"a display-field write appeared, moved or changed shape. Every value reaching "
|
||||
"display_name/display_uploader must be bounded to models.DISPLAY_TEXT_MAX — normally by "
|
||||
"models.display_text — or a name longer than the column is a 500 rather than a truncation. "
|
||||
f"Sites that appeared or vanished: {moved}. Sites that changed: {changed}."
|
||||
)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user