muse/tests/test_invites.py

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