S2: fix zip-bomb on import + SVG stored-XSS on attachment download
M9 section S2 — two real security fixes in notes.py: - Zip decompression bomb (issue #1980): note import read each zip entry with a whole-entry zf.read() and no cap, so a small archive could inflate to GBs and exhaust memory/disk. Add _ImportBudget — streams entries with a per-entry (64MB) and cumulative (512MB) decompressed cap, raising _ImportTooLarge past either; reject >10k entries up front; abort → 413 with the transaction rolled back. - SVG stored-XSS (issue #1981): attachment download served anything image/* inline, so an image/svg+xml attachment could execute script in-origin — and notes are shareable (rule 47), so this hit shared-note viewers. Inline now allowlists the trusted raster types only (png/jpeg/gif/webp); svg/html/xml/etc. download. Verified py_compile + ruff. Runtime (importing a bomb, opening an SVG) is operator-verified on deploy — no Postgres CI lane. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FRgehjoz7Yv8LkUfADxACm
This commit is contained in:
+56
-11
@@ -735,7 +735,37 @@ def _keep_spec(kn: dict, keep_dir: str) -> dict:
|
||||
}
|
||||
|
||||
|
||||
def _read_import_specs(zf: zipfile.ZipFile) -> tuple[list[dict], str]:
|
||||
IMPORT_MAX_ENTRIES = 10_000
|
||||
IMPORT_MAX_ENTRY_BYTES = 64 * 1024 * 1024 # 64 MB decompressed per file
|
||||
IMPORT_MAX_TOTAL_BYTES = 512 * 1024 * 1024 # 512 MB decompressed across the whole import
|
||||
|
||||
|
||||
class _ImportTooLarge(Exception):
|
||||
"""An import zip decompressed past the byte budget (a zip bomb, or just too big)."""
|
||||
|
||||
|
||||
class _ImportBudget:
|
||||
"""Caps DECOMPRESSED bytes pulled from an import zip — per entry and cumulatively.
|
||||
|
||||
zipfile inflates into memory on read, so an archive that's tiny on disk can expand
|
||||
to gigabytes. We stream each entry and read at most the remaining budget + 1 byte,
|
||||
so an oversized (or size-lying) entry is caught mid-read instead of after it has
|
||||
already been fully inflated."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.remaining = IMPORT_MAX_TOTAL_BYTES
|
||||
|
||||
def read(self, zf: zipfile.ZipFile, name: str) -> bytes:
|
||||
cap = min(IMPORT_MAX_ENTRY_BYTES, self.remaining)
|
||||
with zf.open(name) as fh:
|
||||
data = fh.read(cap + 1)
|
||||
if len(data) > cap:
|
||||
raise _ImportTooLarge()
|
||||
self.remaining -= len(data)
|
||||
return data
|
||||
|
||||
|
||||
def _read_import_specs(zf: zipfile.ZipFile, budget: _ImportBudget) -> tuple[list[dict], str]:
|
||||
"""Detect the archive format and return (specs, source). A ThoughtSync export
|
||||
is recognized by its notes.json (app == thoughtsync); otherwise each Keep-shaped
|
||||
<note>.json is imported. Returns ([], "") when nothing importable is found."""
|
||||
@@ -743,7 +773,7 @@ def _read_import_specs(zf: zipfile.ZipFile) -> tuple[list[dict], str]:
|
||||
for name in names:
|
||||
if posixpath.basename(name) == "notes.json":
|
||||
try:
|
||||
doc = json.loads(zf.read(name))
|
||||
doc = json.loads(budget.read(zf, name))
|
||||
except (ValueError, KeyError):
|
||||
continue
|
||||
if isinstance(doc, dict) and doc.get("app") == "thoughtsync":
|
||||
@@ -756,7 +786,7 @@ def _read_import_specs(zf: zipfile.ZipFile) -> tuple[list[dict], str]:
|
||||
if not name.lower().endswith(".json") or posixpath.basename(name) == "notes.json":
|
||||
continue
|
||||
try:
|
||||
kn = json.loads(zf.read(name))
|
||||
kn = json.loads(budget.read(zf, name))
|
||||
except (ValueError, KeyError):
|
||||
continue
|
||||
if isinstance(kn, dict) and any(k in kn for k in keep_keys):
|
||||
@@ -764,14 +794,14 @@ def _read_import_specs(zf: zipfile.ZipFile) -> tuple[list[dict], str]:
|
||||
return (keep_specs, "keep") if keep_specs else ([], "")
|
||||
|
||||
|
||||
def _import_attachment(db, note: Note, zf: zipfile.ZipFile, att: dict) -> bool:
|
||||
def _import_attachment(db, note: Note, zf: zipfile.ZipFile, att: dict, budget: _ImportBudget) -> bool:
|
||||
"""Copy one attachment (any type — incl. Keep audio memos) out of the zip into
|
||||
media storage and record it, preserving its filename + hash. Returns True if written."""
|
||||
zpath = att.get("file")
|
||||
if not zpath:
|
||||
return False
|
||||
try:
|
||||
raw = zf.read(zpath)
|
||||
raw = budget.read(zf, zpath)
|
||||
except KeyError:
|
||||
return False
|
||||
filename = _safe_filename(posixpath.basename(zpath))
|
||||
@@ -796,7 +826,9 @@ def _import_attachment(db, note: Note, zf: zipfile.ZipFile, att: dict) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
async def _create_imported_note(db, owner_id, spec: dict, zf: zipfile.ZipFile, position: int) -> bool:
|
||||
async def _create_imported_note(
|
||||
db, owner_id, spec: dict, zf: zipfile.ZipFile, position: int, budget: _ImportBudget
|
||||
) -> bool:
|
||||
"""Insert one imported note plus its items/labels/attachments, reusing the same
|
||||
display-title derivation + tag/link reconciliation as create_note. Returns False
|
||||
(nothing written) when the spec is empty."""
|
||||
@@ -857,7 +889,7 @@ async def _create_imported_note(db, owner_id, spec: dict, zf: zipfile.ZipFile, p
|
||||
|
||||
for att in spec.get("attachments") or []:
|
||||
if isinstance(att, dict):
|
||||
_import_attachment(db, note, zf, att)
|
||||
_import_attachment(db, note, zf, att, budget)
|
||||
|
||||
await _rewrite_links(db, note)
|
||||
await _reconcile_tags(db, note)
|
||||
@@ -881,7 +913,16 @@ async def import_notes():
|
||||
zf = zipfile.ZipFile(io.BytesIO(raw))
|
||||
except zipfile.BadZipFile:
|
||||
return json_error("that file isn't a valid .zip archive", 400)
|
||||
specs, source = _read_import_specs(zf)
|
||||
|
||||
if len(zf.infolist()) > IMPORT_MAX_ENTRIES:
|
||||
return json_error("that archive has too many files to import", 413)
|
||||
|
||||
# Bound total decompressed bytes so a zip bomb can't exhaust memory/disk. Any entry
|
||||
# (or the cumulative total) blowing the budget aborts the whole import — nothing is
|
||||
# committed, since the session rolls back when the exception exits its block.
|
||||
budget = _ImportBudget()
|
||||
try:
|
||||
specs, source = _read_import_specs(zf, budget)
|
||||
if not specs:
|
||||
return json_error(
|
||||
"no importable notes found — expected a ThoughtSync export or a Google Keep Takeout zip", 400
|
||||
@@ -897,12 +938,14 @@ async def import_notes():
|
||||
)
|
||||
pos = int(max_pos)
|
||||
for spec in specs:
|
||||
if await _create_imported_note(db, g.user_id, spec, zf, pos + 1):
|
||||
if await _create_imported_note(db, g.user_id, spec, zf, pos + 1, budget):
|
||||
pos += 1
|
||||
imported += 1
|
||||
else:
|
||||
skipped += 1
|
||||
await db.commit()
|
||||
except _ImportTooLarge:
|
||||
return json_error("that archive is too large to import", 413)
|
||||
return jsonify({"source": source, "imported": imported, "skipped": skipped}), 201
|
||||
|
||||
|
||||
@@ -1414,8 +1457,10 @@ async def get_attachment(note_id: str, att_id: str):
|
||||
response = await send_file(str(file_path), mimetype=mime)
|
||||
response.headers["X-Content-Type-Options"] = "nosniff"
|
||||
response.headers["Cache-Control"] = "private, max-age=86400"
|
||||
# Images display inline; every other type downloads with its original name.
|
||||
disposition = "inline" if mime.startswith("image/") else "attachment"
|
||||
# Only trusted raster image types render inline; everything else — notably
|
||||
# image/svg+xml, which can carry script — downloads, so a shared note's attachment
|
||||
# can't execute script in a viewer's session (rule 47 = shares).
|
||||
disposition = "inline" if mime in ALLOWED_IMAGE_MIMES else "attachment"
|
||||
response.headers["Content-Disposition"] = f'{disposition}; filename="{_header_filename(filename)}"'
|
||||
return response
|
||||
|
||||
|
||||
Reference in New Issue
Block a user