"""Authorization matrix with MUSE_AUTH=github actually enabled. Sessions are inserted straight into the DB rather than driven through OAuth, so this exercises the guard, not GitHub. Run standalone: it starts its own server on :5042 with its own database. """ import hashlib, os, secrets, signal, sqlite3, subprocess, sys, time from datetime import datetime, timedelta, timezone import httpx HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(HERE) DB = os.path.join(ROOT, "dev", "authz.db") BASE = "http://127.0.0.1:5042" 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}") for f in (DB, DB + "-wal", DB + "-shm"): if os.path.exists(f): os.remove(f) env = {**os.environ, "MUSE_AUTH": "github,google", "MUSE_DB": DB, "MUSE_BASE_URL": "https://muse.example.com", "MUSE_GITHUB_CLIENT_ID": "x", "MUSE_GITHUB_CLIENT_SECRET": "y", "MUSE_GOOGLE_CLIENT_ID": "x", "MUSE_GOOGLE_CLIENT_SECRET": "y", "MUSE_ALLOWED_USERS": "bootstrap", "MUSE_LIBRARY_DIR": os.path.join(ROOT, "dev", "library"), "MUSE_DOWNLOAD_DIR": os.path.join(ROOT, "dev", "downloads"), "MUSE_COOKIE_SECURE": "false", "SLSKD_URL": "http://127.0.0.1:1"} srv = subprocess.Popen([os.path.join(ROOT, ".venv/bin/uvicorn"), "app:app", "--host", "127.0.0.1", "--port", "5042"], cwd=ROOT, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) try: for _ in range(60): try: if httpx.get(BASE + "/healthz", timeout=1).status_code == 200: break except Exception: time.sleep(0.3) else: print("server never came up"); sys.exit(1) def seed(username, role, disabled=0): tok = secrets.token_urlsafe(32) h = hashlib.sha256(tok.encode()).hexdigest() exp = (datetime.now(timezone.utc) + timedelta(days=1)).isoformat() d = sqlite3.connect(DB) d.execute("INSERT OR REPLACE INTO users (username,provider,role,disabled) " "VALUES (?,'github',?,?)", (username, role, disabled)) d.execute("INSERT INTO sessions (token_hash,username,expires_at) VALUES (?,?,?)", (h, username, exp)) d.commit(); d.close() return httpx.Client(base_url=BASE, timeout=20, cookies={"muse_session": tok}) admin = seed("adm", "admin") user = seed("usr", "user") guest = seed("gst", "guest") anon = httpx.Client(base_url=BASE, timeout=20) print("\n=== admin routes by role ===") for path in ("/api/admin/users", "/api/admin/system", "/api/admin/activity"): check(f"admin {path}", admin.get(path).status_code == 200) check(f"user {path} -> 403", user.get(path).status_code == 403) check(f"guest {path} -> 403", guest.get(path).status_code == 403) check(f"anon {path} -> 401", anon.get(path).status_code == 401) print("\n=== mutating admin routes ===") check("user cannot invite", user.post("/api/admin/users", json={"username": "x", "provider": "github"}).status_code == 403) check("user cannot patch", user.patch("/api/admin/users/gst", json={"role": "admin"}).status_code == 403) check("user cannot delete", user.request("DELETE", "/api/admin/users/gst").status_code == 403) print("\n=== last-admin protection ===") check("demote last admin -> 409", admin.patch("/api/admin/users/adm", json={"role": "user"}).status_code == 409) check("disable last admin -> 409", admin.patch("/api/admin/users/adm", json={"disabled": True}).status_code == 409) check("delete last admin -> 409", admin.request("DELETE", "/api/admin/users/adm").status_code == 409) print("\n=== revocation takes effect mid-session ===") check("guest works before", guest.get("/api/me").json()["user"] == "gst") admin.patch("/api/admin/users/gst", json={"disabled": True}) check("guest 401 after disable", guest.get("/api/jobs").status_code == 401) print("\n=== job scoping ===") for cl, who in ((user, "usr"),): cl.post("/api/request", json={"artist": "A", "title": "T"}) admin.post("/api/request", json={"artist": "B", "title": "U"}) time.sleep(0.5) ujobs = user.get("/api/jobs").json()["jobs"] ajobs = admin.get("/api/jobs").json()["jobs"] check("user sees only own", all(j["requested_by"] == "usr" for j in ujobs), str(ujobs)) check("admin sees all", len(ajobs) > len(ujobs)) other = [j for j in ajobs if j["requested_by"] != "usr"] if other: check("cross-user retry -> 404", user.post(f"/api/retry/{other[0]['id']}").status_code == 404) print("\n=== unauthenticated surface ===") check("anon /api/search -> 401", anon.get("/api/search?q=x").status_code == 401) check("anon / redirects to login", anon.get("/", follow_redirects=False).status_code == 302) check("anon /_next bundle is NOT public", anon.get("/_next/static/x.js", follow_redirects=False).status_code == 302) check("anon /healthz is public", anon.get("/healthz").status_code == 200) print("\n=== OAuth CSRF ===") check("callback with no state -> 400", anon.get("/auth/callback?code=x", follow_redirects=False).status_code == 400) check("callback with unmatched state -> 400", anon.get("/auth/callback?code=x&state=zzz", follow_redirects=False).status_code == 400) c2 = httpx.Client(base_url=BASE, timeout=20, cookies={"muse_oauth_state": "github|realnonce"}) check("callback with wrong nonce -> 400", c2.get("/auth/callback?code=x&state=wrong", follow_redirects=False).status_code == 400) c3 = httpx.Client(base_url=BASE, timeout=20, cookies={"muse_oauth_state": "evil|n"}) check("callback with unknown provider -> 400", c3.get("/auth/callback?code=x&state=n", follow_redirects=False).status_code == 400) r = anon.get("/auth/login?provider=github", follow_redirects=False) check("login sets state cookie", "muse_oauth_state" in r.headers.get("set-cookie", "")) check("state cookie is httponly", "httponly" in r.headers.get("set-cookie", "").lower()) check("login redirects to github", r.headers.get("location", "").startswith("https://github.com/login/oauth")) check("unknown provider -> 404", anon.get("/auth/login?provider=evil", follow_redirects=False).status_code == 404) print(f"\n{ok} passed, {fail} failed") finally: srv.send_signal(signal.SIGINT) srv.wait(timeout=10) sys.exit(1 if fail else 0)