muse/tests/test_authz.py

"""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)