"""Sign in with an external provider: account resolution, session issuance, and the guards that keep the login flow from becoming an account-takeover path. Parametrised over every configured provider, so Discord and Google are held to the same rules rather than one being tested and the other assumed. Provider 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, oauth V2 = "/thermograph/api/v2" PW = "supersecret123" # (provider, callback path, profile builder). Discord's callback is grandfathered # to the URL registered in its developer portal; anything newer uses the canonical # /oauth/{provider}/callback. PROVIDERS = [ ("discord", f"{V2}/discord/link/callback", lambda sub, email, ver: {"id": sub, "email": email, "verified": ver}), ("google", f"{V2}/oauth/google/callback", lambda sub, email, ver: {"sub": sub, "email": email, "email_verified": ver}), ] IDS = [p[0] for p in PROVIDERS] @pytest.fixture(scope="module", autouse=True) def _tables(): db.Base.metadata.create_all(db.sync_engine) @pytest.fixture(autouse=True) def _configured(monkeypatch): for name in oauth.PROVIDERS: monkeypatch.setitem(oauth.CREDENTIALS, name, (f"{name}-id", f"{name}-secret")) @pytest.fixture(params=PROVIDERS, ids=IDS) def prov(request): """A provider under test: .name, .callback, .profile(sub, email, verified).""" name, callback, builder = request.param class P: pass p = P() p.name, p.callback, p.profile = name, callback, builder return p class _Resp: def __init__(self, code, data): self.status_code = code; self._data = data def json(self): return self._data def _mock(monkeypatch, profile): """Stand in for httpx.AsyncClient: token exchange, then userinfo -> 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(oauth.httpx, "AsyncClient", _Client) def _register(client, email): assert client.post(f"{V2}/auth/register", json={"email": email, "password": PW}).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 _cb(client, prov, state): return client.get(f"{prov.callback}?code=abc&state={state}", follow_redirects=False) def _me(client): return client.get(f"{V2}/users/me").json() # --- config ------------------------------------------------------------------- def test_config_lists_every_provider_and_its_enabled_state(monkeypatch): c = TestClient(appmod.app) monkeypatch.setitem(oauth.CREDENTIALS, "google", ("", "")) by_name = {p["name"]: p for p in c.get(f"{V2}/oauth/config").json()["providers"]} assert by_name["discord"]["enabled"] is True assert by_name["google"]["enabled"] is False # half-configured is inert assert by_name["google"]["label"] == "Google" def test_unknown_provider_is_404(): c = TestClient(appmod.app) assert c.get(f"{V2}/oauth/gitlab/login/start", follow_redirects=False).status_code == 404 # --- start -------------------------------------------------------------------- def test_login_start_needs_no_session_and_asks_for_an_email_scope(prov): c = TestClient(appmod.app) r = c.get(f"{V2}/oauth/{prov.name}/login/start", follow_redirects=False) assert r.status_code == 303 loc = r.headers["location"] assert loc.startswith(oauth.PROVIDERS[prov.name].authorize_url) # Whatever the provider calls it, the login scope must be able to yield an # address — without one there is no account to resolve or create. assert "email" in loc and "state=" in loc def test_login_start_while_signed_in_links_instead_of_switching_account(prov): c = TestClient(appmod.app) _login(c, f"already-in-{prov.name}@example.com") r = c.get(f"{V2}/oauth/{prov.name}/login/start", follow_redirects=False) state = r.headers["location"].split("state=")[1].split("&")[0] assert oauth._verify_state(state, "link", prov.name) == _me(c)["id"] def test_start_when_unconfigured_bounces_back(prov, monkeypatch): monkeypatch.setitem(oauth.CREDENTIALS, prov.name, ("", "")) c = TestClient(appmod.app) r = c.get(f"{V2}/oauth/{prov.name}/login/start", follow_redirects=False) assert r.status_code == 303 and "oauth=unavailable" in r.headers["location"] # --- signed state ------------------------------------------------------------- def test_state_purposes_and_providers_do_not_cross_over(): login = oauth._sign_state(None, "login", "google") link = oauth._sign_state("user-123", "link", "google") # A login state carries no user id but verifies as "" — distinct from reject. assert oauth._verify_state(login, "login", "google") == "" assert oauth._verify_state(login, "link", "google") is None assert oauth._verify_state(link, "link", "google") == "user-123" assert oauth._verify_state(link, "login", "google") is None # And a state minted for one provider is worthless at another's callback. assert oauth._verify_state(link, "link", "discord") is None assert oauth._verify_state(login, "login", "discord") is None def test_state_rejects_tampering_and_expiry(monkeypatch): s = oauth._sign_state("u", "link", "google") payload, mac = s.split(".", 1) assert oauth._verify_state(f"{payload}.{'0' * len(mac)}", "link", "google") is None assert oauth._verify_state(payload, "link", "google") is None assert oauth._verify_state("garbage", "link", "google") is None monkeypatch.setattr(oauth, "_STATE_TTL", -1) assert oauth._verify_state(oauth._sign_state("u", "link", "google"), "link", "google") is None def test_state_is_secret_dependent(monkeypatch): s = oauth._sign_state("u", "link", "google") monkeypatch.setattr(oauth, "SECRET", oauth.SECRET + "-different") assert oauth._verify_state(s, "link", "google") is None def test_a_login_state_cannot_link_the_signed_in_account(prov, 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 identity to a victim's account and then sign in as them. It must complete as a plain login instead.""" attacker = f"attacker-{prov.name}@example.com" _mock(monkeypatch, prov.profile(f"{prov.name}-cross", attacker, True)) c = TestClient(appmod.app) _login(c, f"crossover-{prov.name}@example.com") r = _cb(c, prov, oauth._sign_state(_me(c)["id"], "login", prov.name)) assert "oauth=created" in r.headers["location"] assert _me(c)["email"] == attacker # The account that had been signed in is untouched. victim = TestClient(appmod.app) _login(victim, f"crossover-{prov.name}@example.com") assert _me(victim)["oauth_providers"] == [] # --- login resolves an account ------------------------------------------------ def test_login_creates_an_account_and_signs_in(prov, monkeypatch): # Mixed case on purpose: the address must be normalised before it is matched or # stored, or the same person gets two accounts depending on how they typed it. _mock(monkeypatch, prov.profile(f"{prov.name}-new", f"New.User.{prov.name}@Example.com", True)) c = TestClient(appmod.app) r = _cb(c, prov, oauth._sign_state(None, "login", prov.name)) assert r.status_code == 303 and "oauth=created" in r.headers["location"] me = _me(c) assert me["email"] == f"new.user.{prov.name}@example.com" # normalised assert me["is_verified"] is True # the provider vouched assert me["oauth_only"] is True assert me["oauth_providers"] == [prov.name] def test_login_recognises_a_previously_linked_account(prov, monkeypatch): """The subject is the durable key: it resolves the account with no email, which is what keeps a login working after someone changes their address upstream.""" email = f"link-me-{prov.name}@example.com" _mock(monkeypatch, prov.profile(f"{prov.name}-ret", email, True)) linker = TestClient(appmod.app) _login(linker, email) uid = _me(linker)["id"] assert "oauth=linked" in _cb( linker, prov, oauth._sign_state(uid, "link", prov.name)).headers["location"] # A fresh client, no session, and the provider now returns no email at all. _mock(monkeypatch, prov.profile(f"{prov.name}-ret", "", False)) c = TestClient(appmod.app) r = _cb(c, prov, oauth._sign_state(None, "login", prov.name)) assert r.status_code == 303 and "oauth=signedin" in r.headers["location"] assert _me(c)["id"] == uid def test_login_connects_a_verified_email_to_an_existing_account(prov, monkeypatch): """The 'connect the account' path: an existing password account picks up the identity and is signed in.""" email = f"existing-{prov.name}@example.com" c = TestClient(appmod.app) _register(c, email) _mock(monkeypatch, prov.profile(f"{prov.name}-exist", email, True)) fresh = TestClient(appmod.app) r = _cb(fresh, prov, oauth._sign_state(None, "login", prov.name)) assert r.status_code == 303 and "oauth=signedin" in r.headers["location"] me = _me(fresh) assert me["email"] == email and me["oauth_providers"] == [prov.name] # It kept its password, so it is not OAuth-only and may unlink. assert me["oauth_only"] is False def test_login_refuses_an_unverified_email(prov, monkeypatch): """The provider's verified flag is the only evidence the person owns the address. Without it, this would hand over any account whose email was typed in.""" email = f"victim-{prov.name}@example.com" c = TestClient(appmod.app) _register(c, email) _mock(monkeypatch, prov.profile(f"{prov.name}-attacker", email, False)) fresh = TestClient(appmod.app) r = _cb(fresh, prov, oauth._sign_state(None, "login", prov.name)) assert r.status_code == 303 and "oauth=unverified" in r.headers["location"] assert fresh.get(f"{V2}/users/me").status_code == 401 # no session issued _login(c, email) assert _me(c)["oauth_providers"] == [] # target untouched def test_login_without_an_email_cannot_create_an_account(prov, monkeypatch): _mock(monkeypatch, prov.profile(f"{prov.name}-noemail", "", True)) c = TestClient(appmod.app) r = _cb(c, prov, oauth._sign_state(None, "login", prov.name)) assert r.status_code == 303 and "oauth=noemail" in r.headers["location"] assert c.get(f"{V2}/users/me").status_code == 401 def test_login_will_not_move_an_account_between_identities(prov, monkeypatch): email = f"owned-{prov.name}@example.com" owner = TestClient(appmod.app) _login(owner, email) uid = _me(owner)["id"] _mock(monkeypatch, prov.profile(f"{prov.name}-owner", email, True)) assert "oauth=linked" in _cb( owner, prov, oauth._sign_state(uid, "link", prov.name)).headers["location"] # A second provider account claiming the same address. _mock(monkeypatch, prov.profile(f"{prov.name}-interloper", email, True)) fresh = TestClient(appmod.app) r = _cb(fresh, prov, oauth._sign_state(None, "login", prov.name)) assert r.status_code == 303 and "oauth=mismatch" in r.headers["location"] assert fresh.get(f"{V2}/users/me").status_code == 401 def test_linking_an_identity_someone_else_owns_is_refused(prov, monkeypatch): first = TestClient(appmod.app) _login(first, f"first-owner-{prov.name}@example.com") _mock(monkeypatch, prov.profile(f"{prov.name}-contested", f"first-owner-{prov.name}@example.com", True)) assert "oauth=linked" in _cb( first, prov, oauth._sign_state(_me(first)["id"], "link", prov.name)).headers["location"] second = TestClient(appmod.app) _login(second, f"second-owner-{prov.name}@example.com") r = _cb(second, prov, oauth._sign_state(_me(second)["id"], "link", prov.name)) assert r.status_code == 303 and "oauth=taken" in r.headers["location"] assert _me(second)["oauth_providers"] == [] def test_relinking_the_same_identity_is_idempotent(prov, monkeypatch): email = f"relink-{prov.name}@example.com" c = TestClient(appmod.app) _login(c, email) _mock(monkeypatch, prov.profile(f"{prov.name}-relink", email, True)) uid = _me(c)["id"] for _ in range(2): r = _cb(c, prov, oauth._sign_state(uid, "link", prov.name)) assert "oauth=linked" in r.headers["location"] assert _me(c)["oauth_providers"] == [prov.name] # not duplicated # --- unlink guards ------------------------------------------------------------ def test_last_provider_cannot_be_unlinked_into_a_lockout(prov, monkeypatch): _mock(monkeypatch, prov.profile(f"{prov.name}-onlyway", f"onlyway-{prov.name}@example.com", True)) c = TestClient(appmod.app) assert "oauth=created" in _cb( c, prov, oauth._sign_state(None, "login", prov.name)).headers["location"] r = c.post(f"{V2}/oauth/{prov.name}/unlink") assert r.status_code == 409 and "no password" in r.json()["detail"] assert _me(c)["oauth_providers"] == [prov.name] def test_an_oauth_only_account_may_unlink_once_a_second_provider_is_linked(monkeypatch): """The lockout rule is about the *last* credential, not about Discord. With two providers linked, dropping one still leaves a way in.""" _mock(monkeypatch, {"sub": "g-two", "email": "two@example.com", "email_verified": True}) c = TestClient(appmod.app) assert "oauth=created" in _cb( c, PROV_GOOGLE, oauth._sign_state(None, "login", "google")).headers["location"] uid = _me(c)["id"] # Now connect Discord to the same, still password-less, account. _mock(monkeypatch, {"id": "d-two", "email": "two@example.com", "verified": True}) assert "oauth=linked" in _cb( c, PROV_DISCORD, oauth._sign_state(uid, "link", "discord")).headers["location"] assert _me(c)["oauth_providers"] == ["discord", "google"] # Either one may now go. assert c.post(f"{V2}/oauth/google/unlink").status_code == 204 assert _me(c)["oauth_providers"] == ["discord"] # But the survivor is once again the only way in. assert c.post(f"{V2}/oauth/discord/unlink").status_code == 409 def test_unlink_requires_auth(prov): c = TestClient(appmod.app) assert c.post(f"{V2}/oauth/{prov.name}/unlink").status_code == 401 def test_link_start_requires_auth(prov): c = TestClient(appmod.app) assert c.get(f"{V2}/oauth/{prov.name}/link/start", follow_redirects=False).status_code == 401 # Bare provider handles for the cross-provider test above, which needs both at once. class _P: def __init__(self, name, callback): self.name, self.callback = name, callback PROV_GOOGLE = _P("google", f"{V2}/oauth/google/callback") PROV_DISCORD = _P("discord", f"{V2}/discord/link/callback")