"""Sign in with Discord: account resolution, session issuance, and the guards that keep the login flow from becoming an account-takeover path. Discord HTTP is mocked; reuses the throwaway accounts DB from conftest.""" import pytest from fastapi.testclient import TestClient from web import app as appmod from accounts import db from notifications import discord_link as dl V2 = "/thermograph/api/v2" PW = "supersecret123" @pytest.fixture(scope="module", autouse=True) def _tables(): db.Base.metadata.create_all(db.sync_engine) @pytest.fixture(autouse=True) def _configured(monkeypatch): monkeypatch.setattr(dl, "CLIENT_ID", "app123") monkeypatch.setattr(dl, "CLIENT_SECRET", "secret456") class _Resp: def __init__(self, code, data): self.status_code = code; self._data = data def json(self): return self._data def _mock_discord(monkeypatch, profile): """Stand in for httpx.AsyncClient: token exchange, then /users/@me -> profile.""" class _Client: def __init__(self, *a, **k): pass async def __aenter__(self): return self async def __aexit__(self, *a): return False async def post(self, url, **k): return _Resp(200, {"access_token": "tok"}) async def get(self, url, **k): return _Resp(200, profile) monkeypatch.setattr(dl.httpx, "AsyncClient", _Client) def _register(client, email): r = client.post(f"{V2}/auth/register", json={"email": email, "password": PW}) assert r.status_code in (201, 400) def _login(client, email): _register(client, email) assert client.post(f"{V2}/auth/login", data={"username": email, "password": PW}).status_code == 204 def _callback(client, state): return client.get(f"{V2}/discord/link/callback?code=abc&state={state}", follow_redirects=False) # --- start ------------------------------------------------------------------- def test_login_start_needs_no_session_and_asks_for_email(): c = TestClient(appmod.app) r = c.get(f"{V2}/discord/login/start", follow_redirects=False) assert r.status_code == 303 loc = r.headers["location"] assert loc.startswith("https://discord.com/api/oauth2/authorize") # The email scope is what makes account resolution possible at all. assert "scope=identify+email" in loc and "state=" in loc def test_login_start_while_signed_in_links_instead_of_switching_account(): """Someone already signed in wants to connect Discord, not be switched into whichever account the Discord identity happens to resolve to.""" c = TestClient(appmod.app) _login(c, "already-in@example.com") r = c.get(f"{V2}/discord/login/start", follow_redirects=False) assert r.status_code == 303 loc = r.headers["location"] assert "scope=identify&" in loc or loc.endswith("scope=identify") state = loc.split("state=")[1].split("&")[0] uid = c.get(f"{V2}/users/me").json()["id"] assert dl._verify_state(state, "link") == uid def test_login_start_when_unconfigured_bounces_back(monkeypatch): monkeypatch.setattr(dl, "CLIENT_ID", "") monkeypatch.setattr(dl, "CLIENT_SECRET", "") c = TestClient(appmod.app) r = c.get(f"{V2}/discord/login/start", follow_redirects=False) assert r.status_code == 303 and "discord=unavailable" in r.headers["location"] # --- state purposes are not interchangeable ---------------------------------- def test_state_purposes_do_not_cross_over(): login_state = dl._sign_state(None, "login") link_state = dl._sign_state("user-123", "link") # A login state carries no user id, but verifies as "" — distinct from reject. assert dl._verify_state(login_state, "login") == "" assert dl._verify_state(login_state, "link") is None assert dl._verify_state(link_state, "link") == "user-123" assert dl._verify_state(link_state, "login") is None def test_a_login_state_cannot_link_the_signed_in_account(monkeypatch): """Forced-linking guard. A login state carries no user id, so it is replayable against whoever is signed in; if the callback treated it as a link, an attacker could attach their own Discord identity to a victim's account and then sign in as them. It must complete as a plain login instead.""" _mock_discord(monkeypatch, {"id": "d-crossover", "email": "attacker@example.com", "verified": True}) c = TestClient(appmod.app) _login(c, "crossover@example.com") # A state whose purpose says login, but naming the signed-in user — the shape # a forced-link attempt would take. r = _callback(c, dl._sign_state(c.get(f"{V2}/users/me").json()["id"], "login")) # Resolved as a login into the Discord identity's own account, not as a link. assert "discord=created" in r.headers["location"] assert c.get(f"{V2}/users/me").json()["email"] == "attacker@example.com" # The account that had been signed in is untouched. victim = TestClient(appmod.app) _login(victim, "crossover@example.com") assert victim.get(f"{V2}/users/me").json()["discord_id"] is None # --- login resolves an account ----------------------------------------------- def test_login_creates_an_account_and_signs_in(monkeypatch): _mock_discord(monkeypatch, {"id": "d-new", "email": "New.User@Example.com", "verified": True}) c = TestClient(appmod.app) r = _callback(c, dl._sign_state(None, "login")) assert r.status_code == 303 and "discord=created" in r.headers["location"] me = c.get(f"{V2}/users/me") assert me.status_code == 200 body = me.json() assert body["email"] == "new.user@example.com" # normalised assert body["discord_id"] == "d-new" assert body["is_verified"] is True # Discord vouched for it assert body["discord_only"] is True assert body["discord_dm"] is True def test_login_recognises_a_previously_linked_account(monkeypatch): """The discord_id is the durable key: it resolves the account with no email.""" _mock_discord(monkeypatch, {"id": "d-returning", "email": "link-me@example.com", "verified": True}) linker = TestClient(appmod.app) _login(linker, "link-me@example.com") uid = linker.get(f"{V2}/users/me").json()["id"] assert "discord=linked" in _callback(linker, dl._sign_state(uid, "link") ).headers["location"] # A fresh client, no session, and Discord now returns no email at all. _mock_discord(monkeypatch, {"id": "d-returning"}) c = TestClient(appmod.app) r = _callback(c, dl._sign_state(None, "login")) assert r.status_code == 303 and "discord=signedin" in r.headers["location"] assert c.get(f"{V2}/users/me").json()["id"] == uid def test_login_connects_a_verified_email_to_an_existing_account(monkeypatch): """This is the 'connect the account with that' path: an existing password account picks up the Discord id and is signed in.""" c = TestClient(appmod.app) _register(c, "existing@example.com") _mock_discord(monkeypatch, {"id": "d-existing", "email": "existing@example.com", "verified": True}) fresh = TestClient(appmod.app) r = _callback(fresh, dl._sign_state(None, "login")) assert r.status_code == 303 and "discord=signedin" in r.headers["location"] body = fresh.get(f"{V2}/users/me").json() assert body["email"] == "existing@example.com" assert body["discord_id"] == "d-existing" # It kept its password, so it is not Discord-only and may unlink. assert body["discord_only"] is False def test_login_refuses_an_unverified_discord_email(monkeypatch): """Discord's verified flag is the only evidence the person owns the address. Without it, this would hand over any account whose email someone typed in.""" c = TestClient(appmod.app) _register(c, "victim@example.com") _mock_discord(monkeypatch, {"id": "d-attacker", "email": "victim@example.com", "verified": False}) fresh = TestClient(appmod.app) r = _callback(fresh, dl._sign_state(None, "login")) assert r.status_code == 303 and "discord=unverified" in r.headers["location"] assert fresh.get(f"{V2}/users/me").status_code == 401 # no session issued # And the targeted account is untouched. _login(c, "victim@example.com") assert c.get(f"{V2}/users/me").json()["discord_id"] is None def test_login_without_an_email_cannot_create_an_account(monkeypatch): _mock_discord(monkeypatch, {"id": "d-noemail"}) c = TestClient(appmod.app) r = _callback(c, dl._sign_state(None, "login")) assert r.status_code == 303 and "discord=noemail" in r.headers["location"] assert c.get(f"{V2}/users/me").status_code == 401 def test_login_will_not_move_an_account_between_discord_identities(monkeypatch): """The email matches an account that is already linked to a different Discord id — signing in would silently re-point it.""" owner = TestClient(appmod.app) _login(owner, "owned@example.com") uid = owner.get(f"{V2}/users/me").json()["id"] _mock_discord(monkeypatch, {"id": "d-owner", "email": "owned@example.com", "verified": True}) assert "discord=linked" in _callback(owner, dl._sign_state(uid, "link") ).headers["location"] # A second Discord account claiming the same address. _mock_discord(monkeypatch, {"id": "d-interloper", "email": "owned@example.com", "verified": True}) fresh = TestClient(appmod.app) r = _callback(fresh, dl._sign_state(None, "login")) assert r.status_code == 303 and "discord=mismatch" in r.headers["location"] assert fresh.get(f"{V2}/users/me").status_code == 401 assert owner.get(f"{V2}/users/me").json()["discord_id"] == "d-owner" # --- guards on the connected account ----------------------------------------- def test_linking_a_discord_account_someone_else_owns_is_refused(monkeypatch): first = TestClient(appmod.app) _login(first, "first-owner@example.com") _mock_discord(monkeypatch, {"id": "d-contested", "email": "first-owner@example.com", "verified": True}) uid = first.get(f"{V2}/users/me").json()["id"] assert "discord=linked" in _callback(first, dl._sign_state(uid, "link") ).headers["location"] # A different Thermograph account tries to claim the same Discord identity. second = TestClient(appmod.app) _login(second, "second-owner@example.com") uid2 = second.get(f"{V2}/users/me").json()["id"] r = _callback(second, dl._sign_state(uid2, "link")) assert r.status_code == 303 and "discord=taken" in r.headers["location"] assert second.get(f"{V2}/users/me").json()["discord_id"] is None def test_a_discord_only_account_cannot_unlink_itself_into_a_lockout(monkeypatch): _mock_discord(monkeypatch, {"id": "d-onlyway", "email": "onlyway@example.com", "verified": True}) c = TestClient(appmod.app) assert "discord=created" in _callback(c, dl._sign_state(None, "login") ).headers["location"] r = c.post(f"{V2}/discord/unlink") assert r.status_code == 409 assert "no password" in r.json()["detail"] assert c.get(f"{V2}/users/me").json()["discord_id"] == "d-onlyway" # Muting DMs is still fine — that is not a lockout. assert c.post(f"{V2}/discord/dm", json={"enabled": False}).status_code == 204