"""Invites and provider binding, against a live server with a fake SMTP sink. The rule under test: an invite is bound to the provider it was issued for, so a Google invite cannot be redeemed over GitHub. That is what makes "the owner on GitHub, everyone else on Google" an actual boundary rather than a habit. """ import asyncio, email, 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", "invites.db") BASE = "http://127.0.0.1:5064" SMTP_PORT = 5065 ok = fail = 0 INBOX: list[tuple[str, str, str]] = [] # (to, subject, body) def check(name, cond, detail=""): global ok, fail if cond: ok += 1; print(f" PASS {name}") else: fail += 1; print(f" FAIL {name} {detail}") # ── a minimal SMTP server, so the test proves a real conversation ───────── async def smtp_sink(): async def handle(reader, writer): async def send(s): writer.write(s.encode() + b"\r\n"); await writer.drain() await send("220 sink") data, collecting = [], False while True: line = await reader.readline() if not line: break txt = line.decode(errors="replace").rstrip("\r\n") if collecting: if txt == ".": raw = "\r\n".join(data) msg = email.message_from_string(raw) INBOX.append((msg["To"], msg["Subject"], msg.get_payload(decode=True).decode())) data, collecting = [], False await send("250 ok") else: data.append(txt) continue up = txt.upper() if up.startswith(("HELO", "EHLO")): await send("250-sink"); await send("250 STARTTLS" if False else "250 OK") elif up.startswith(("MAIL", "RCPT")): await send("250 ok") elif up.startswith("DATA"): collecting = True; await send("354 go") elif up.startswith("QUIT"): await send("221 bye"); break else: await send("250 ok") writer.close() s = await asyncio.start_server(handle, "127.0.0.1", SMTP_PORT) async with s: await s.serve_forever() def main() -> int: 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.aryan-singh.dev", "MUSE_GITHUB_CLIENT_ID": "x", "MUSE_GITHUB_CLIENT_SECRET": "y", "MUSE_GOOGLE_CLIENT_ID": "x", "MUSE_GOOGLE_CLIENT_SECRET": "y", "MUSE_ALLOWED_USERS": "geneticglitch1", "MUSE_SMTP_HOST": "127.0.0.1", "MUSE_SMTP_PORT": str(SMTP_PORT), "MUSE_SMTP_FROM": "muse@aryan-singh.dev", "MUSE_SMTP_TLS": "none", "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", "5064"], 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"); return 1 def seed(username, role, provider="github"): tok = secrets.token_urlsafe(32) d = sqlite3.connect(DB) d.execute("INSERT OR REPLACE INTO users (username,provider,role) " "VALUES (?,?,?)", (username, provider, role)) d.execute("INSERT INTO sessions (token_hash,username,expires_at) " "VALUES (?,?,?)", (hashlib.sha256(tok.encode()).hexdigest(), username, (datetime.now(timezone.utc) + timedelta(days=1)).isoformat())) d.commit(); d.close() return httpx.Client(base_url=BASE, timeout=20, cookies={"muse_session": tok}) admin = seed("geneticglitch1", "admin") print("\n=== inviting sends mail ===") INBOX.clear() r = admin.post("/api/admin/users", json={"username": "friend@gmail.com", "provider": "google"}) check("google invite accepted", r.status_code == 200, r.text[:120]) check("reports the mail went", r.json().get("invite_sent") is True, r.text[:160]) time.sleep(1.2) check("one message delivered", len(INBOX) == 1, str(INBOX)) if INBOX: to, subj, body = INBOX[0] check("addressed to the invitee", to == "friend@gmail.com", to) check("names the inviter", "geneticglitch1" in subj, subj) check("carries the instance URL", "https://muse.aryan-singh.dev" in body, body[:120]) check("says which provider", "Google" in body, body[:200]) check("says which account", "friend@gmail.com" in body, body[:200]) check("carries NO token", "token" not in body and "?t=" not in body) print("\n=== validation ===") check("google invite must be an address", admin.post("/api/admin/users", json={"username": "notanemail", "provider": "google"}).status_code == 400) check("bad email rejected", admin.post("/api/admin/users", json={"username": "x", "provider": "github", "email": "nope"}).status_code == 400) check("duplicate rejected", admin.post("/api/admin/users", json={"username": "friend@gmail.com", "provider": "google"}).status_code == 409) print("\n=== github invite without an address ===") INBOX.clear() r = admin.post("/api/admin/users", json={"username": "octocat", "provider": "github"}) check("accepted", r.status_code == 200, r.text[:120]) check("says no mail was sent", r.json().get("invite_sent") is False) check("explains why", "no email address" in (r.json().get("invite_why") or ""), str(r.json().get("invite_why"))) time.sleep(0.8) check("nothing delivered", len(INBOX) == 0) print("\n=== resend ===") INBOX.clear() check("resend to a google invitee", admin.post("/api/admin/users/friend@gmail.com/resend").status_code == 200) time.sleep(1.2) check("delivered again", len(INBOX) == 1, str(len(INBOX))) check("resend without an address -> 409", admin.post("/api/admin/users/octocat/resend").status_code == 409) check("resend unknown user -> 404", admin.post("/api/admin/users/nobody/resend").status_code == 404) print("\n=== provider binding ===") # may_login is the single rule the callback consults, so exercising it # directly tests the boundary without standing up real OAuth. sys.path.insert(0, ROOT) os.environ["MUSE_ALLOWED_USERS"] = "geneticglitch1" import importlib import app as muse importlib.reload(muse) d = sqlite3.connect(DB); d.row_factory = sqlite3.Row goog = d.execute("SELECT * FROM users WHERE username='friend@gmail.com'").fetchone() gh = d.execute("SELECT * FROM users WHERE username='octocat'").fetchone() d.close() check("google invitee may use google", muse.may_login("friend@gmail.com", goog, "google") is True) check("google invitee may NOT use github", muse.may_login("friend@gmail.com", goog, "github") is False) check("github invitee may use github", muse.may_login("octocat", gh, "github") is True) check("github invitee may NOT use google", muse.may_login("octocat", gh, "google") is False) check("owner on the allowlist may use github", muse.may_login("geneticglitch1", None, "github") is True) check("owner on the allowlist may use google too", muse.may_login("geneticglitch1", None, "google") is True) check("a stranger is refused on either", not muse.may_login("stranger", None, "google") and not muse.may_login("stranger", None, "github")) class Row(dict): def __getitem__(self, k): return dict.__getitem__(self, k) disabled = Row(goog); disabled["disabled"] = 1 check("a disabled invitee is refused on their own provider", muse.may_login("friend@gmail.com", disabled, "google") is False) check("per-request re-check ignores provider", muse.may_login("friend@gmail.com", goog, None) is True) finally: srv.send_signal(signal.SIGINT) try: srv.wait(timeout=10) except Exception: srv.kill() return 1 if fail else 0 if __name__ == "__main__": t = None import threading loop = asyncio.new_event_loop() t = threading.Thread(target=lambda: (asyncio.set_event_loop(loop), loop.run_until_complete(smtp_sink())), daemon=True) t.start() time.sleep(0.5) code = main() print(f"\n{ok} passed, {fail} failed") sys.exit(code)