Aryan SinghandClaude Opus 5
cc3490483e
`year` was appended to the album directory name AFTER safe() had run, so it
was the one path component reaching the filesystem unchecked:
POST /api/request {"artist":"E","title":"p","year":"../../../../tmp/pwned"}
-> library/E/Singles (../../../../tmp/pwned)/p.flac
-> normalises to ../tmp/pwned)/p.flac
Any authenticated user could do it, guests included, and it wrote a real
audio file to an arbitrary path the process could reach. Existing files were
not overwritten -- tag_and_file returns early when the destination exists --
so this created new files and directories rather than clobbering.
Fixed in two layers, because the first is a property of five call sites that
must each stay correct and the second cannot be forgotten:
1. safe_year() whitelists \d{4}, applied in enqueue_job (the choke point for
all three request paths) and again in tag_and_file.
2. tag_and_file resolves the destination and refuses anything that is not
under the library root, before mkdir rather than after.
Also from the same pass:
- Cap import at 500 pasted lines / 1000 expanded tracks. URL expansion runs
BEFORE the quota check -- it has to, since a playlist URL is one line and
200 tracks -- so without a cap a paste of 10,000 album URLs was 10,000
requests to Deezer from this IP before any quota was consulted.
- Clamp search limit (1-50), jobs limit (1-1000) and activity days (1-365).
search limit fanned out one /album/{id} enrichment per distinct album.
- safe_cover() requires https and <=300 chars. img-src permits data:, so
the field was free storage for megabytes of base64 per job.
- Truncate filename components to 150 chars; a long title used to fail the
import with ENAMETOOLONG after the download had already completed.
- Treat Deezer's 403 as no-results. Its WAF rejects injection-looking
queries, which surfaced to the user as a 502 "service broken".
Add tests/: 42 security assertions and 35 authorization assertions covering
the role matrix, last-admin protection, mid-session revocation, job scoping
and OAuth CSRF. All 77 pass.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
115 lines
4.9 KiB
Python
115 lines
4.9 KiB
Python
"""Security regression suite. Run against a dev.sh instance on :5041.
|
|
|
|
Covers the classes that matter for an internet-facing request front door:
|
|
path traversal into the library, authz on every admin route, session
|
|
revocation, OAuth CSRF, injection, and resource amplification.
|
|
"""
|
|
import subprocess, sys, time, os, signal, sqlite3, json
|
|
import httpx
|
|
|
|
BASE = "http://127.0.0.1:5041"
|
|
ok = fail = 0
|
|
|
|
|
|
def check(name, cond, detail=""):
|
|
global ok, fail
|
|
if cond:
|
|
ok += 1
|
|
print(f" PASS {name}")
|
|
else:
|
|
fail += 1
|
|
print(f" FAIL {name} {detail}")
|
|
|
|
|
|
def section(t):
|
|
print(f"\n=== {t} ===")
|
|
|
|
|
|
# ── path traversal ────────────────────────────────────────────────────────
|
|
section("path traversal into the library")
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
os.environ.update(MUSE_AUTH="off", MUSE_DB="dev/muse.db",
|
|
MUSE_LIBRARY_DIR="dev/library", MUSE_DOWNLOAD_DIR="dev/downloads")
|
|
import app # noqa: E402
|
|
from pathlib import Path # noqa: E402
|
|
|
|
for probe in ["../../etc", "..", ".", "/etc/passwd", "....//....//x",
|
|
"a\x00b", "con", " . . ", "x" * 400]:
|
|
s = app.safe(probe)
|
|
check(f"safe({probe[:18]!r})", "/" not in s and "\\" not in s
|
|
and s not in ("", ".", "..") and len(s) <= app.MAX_COMPONENT, f"-> {s!r}")
|
|
|
|
for bad in ["../../tmp/x", "1993/../../x", "19933", "", "abc", None]:
|
|
y = app.safe_year(bad)
|
|
check(f"safe_year({bad!r})", y is None, f"-> {y!r}")
|
|
check("safe_year('1993')", app.safe_year("1993") == "1993")
|
|
check("safe_year(' 1993 ') trims", app.safe_year(" 1993 ") == "1993")
|
|
|
|
for bad in ["data:image/svg+xml;base64," + "A" * 5000, "javascript:alert(1)",
|
|
"http://evil/x.png", "x" * 400]:
|
|
check(f"safe_cover({bad[:22]!r})", app.safe_cover(bad) is None)
|
|
check("safe_cover(deezer)", app.safe_cover("https://e-cdns-images.dzcdn.net/x.jpg"))
|
|
|
|
# containment holds even if the field validators are bypassed
|
|
src = Path("dev/downloads/fake.flac")
|
|
src.parent.mkdir(parents=True, exist_ok=True)
|
|
src.write_bytes(b"x")
|
|
real_safe_year = app.safe_year
|
|
app.safe_year = lambda y: (str(y) if y else None)
|
|
row = {"album": None, "year": "../../../../../../tmp/muse-pwned", "artist": "E",
|
|
"title": "p", "track": None}
|
|
try:
|
|
app.tag_and_file(row, src)
|
|
check("containment blocks escape", False, "wrote outside the library")
|
|
except ValueError:
|
|
check("containment blocks escape", True)
|
|
except Exception as e:
|
|
check("containment blocks escape", False, f"wrong error {type(e).__name__}: {e}")
|
|
app.safe_year = real_safe_year
|
|
check("nothing landed in /tmp", not Path("/tmp/muse-pwned").exists())
|
|
|
|
# ── live HTTP ─────────────────────────────────────────────────────────────
|
|
c = httpx.Client(base_url=BASE, timeout=30)
|
|
|
|
section("resource amplification")
|
|
r = c.post("/api/import", json={"text": "\n".join(["a"] * 5000)})
|
|
check("import line cap", r.status_code == 413, f"got {r.status_code}")
|
|
r = c.get("/api/search", params={"q": "x", "limit": 100000})
|
|
check("search limit clamped", r.status_code == 422, f"got {r.status_code}")
|
|
r = c.get("/api/jobs", params={"limit": -1})
|
|
check("jobs limit clamped", r.status_code == 422, f"got {r.status_code}")
|
|
r = c.get("/api/admin/activity", params={"days": 99999})
|
|
check("activity days clamped", r.status_code == 422, f"got {r.status_code}")
|
|
|
|
section("injection")
|
|
for payload in ["'; DROP TABLE users;--", "' OR '1'='1", "\x00", "%00",
|
|
"Don't Stop Me Now", "Guns N' Roses"]:
|
|
r = c.get("/api/search", params={"q": payload})
|
|
check(f"search survives {payload[:16]!r}", r.status_code == 200,
|
|
f"got {r.status_code}")
|
|
r = c.post("/api/admin/users", json={"username": "'; DROP TABLE users;--",
|
|
"provider": "github", "role": "user"})
|
|
with app.db() as d:
|
|
still = d.execute("SELECT COUNT(*) n FROM users").fetchone()["n"]
|
|
check("users table intact after SQLi attempt", still > 0)
|
|
c.request("DELETE", "/api/admin/users/'; DROP TABLE users;--")
|
|
|
|
section("security headers")
|
|
r = c.get("/")
|
|
h = r.headers
|
|
check("CSP present", "content-security-policy" in h)
|
|
check("CSP has no unsafe-inline for script",
|
|
"'unsafe-inline'" not in h.get("content-security-policy", "").split("style-src")[0])
|
|
check("nosniff", h.get("x-content-type-options") == "nosniff")
|
|
check("frame deny", h.get("x-frame-options") == "DENY")
|
|
check("referrer", h.get("referrer-policy") == "no-referrer")
|
|
check("img-src is not blanket https:",
|
|
" https: " not in h.get("content-security-policy", ""))
|
|
|
|
section("health endpoint leaks nothing")
|
|
r = c.get("/healthz")
|
|
check("healthz is bare ok", r.json() == {"ok": True}, r.text[:80])
|
|
|
|
print(f"\n{ok} passed, {fail} failed")
|
|
sys.exit(1 if fail else 0)
|