thermograph/backend/tests/notifications/test_discord_login.py

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