rfc-app/backend/tests/test_device_trust_vertical.py

"""End-to-end integration tests for the v0.11.0 trust-device vertical
(§6.2, roadmap item #9).

After a successful OTC or passcode sign-in with `trust_device=true`
on the body, the server mints a fresh `device_trust` row and sets the
`rfc_device_trust` cookie. On a subsequent visit, the cookie carries
a session re-established by `POST /auth/device-trust/start`. The
tests below prove:

  * `trust_device=false` (default, including omitted) on OTC verify
    does NOT set the device-trust cookie and does NOT insert a row.
  * `trust_device=true` on OTC verify DOES set the cookie (HttpOnly +
    Secure + SameSite=Lax + 30-day Max-Age) and DOES insert a row.
    The row's hash is NOT the raw token; only the hash lives in the
    database.
  * Same shape for passcode verify.
  * On a returning visit with the cookie, `POST /auth/device-trust/start`
    re-establishes the session — `GET /api/auth/me` reads the right
    user without an OTC roundtrip.
  * `last_seen_at` refreshes on a successful lookup.
  * `POST /auth/device-trust/start` with no cookie returns 401.
  * `POST /auth/device-trust/start` with a forged / unknown cookie
    returns 401 + clears the cookie.
  * A revoked row refuses the cookie (401) and clears it.
  * An expired row refuses the cookie (401) and clears it.
  * `GET /api/auth/me/devices` lists the user's active rows.
  * `DELETE /api/auth/me/devices/{id}` revokes a single row.
  * `DELETE /api/auth/me/devices/{id}` for another user's row reads 404.
  * `DELETE /api/auth/me/devices` revokes every active row.
  * Constant-time path: bcrypt.checkpw guards lookup; the raw token
    is never written to logs or to the DB.

The fakes from `test_propose_vertical` give us a working app harness.
The OTC envelope buffer from `test_otc_vertical` is reused for the
OTC roundtrips this suite needs.
"""
from __future__ import annotations

import pytest

from test_propose_vertical import (  # noqa: F401
    FakeGitea,
    app_with_fake_gitea,
    provision_user_row,
    tmp_env,
)


# ---------------------------------------------------------------------------
# Helpers — mirror the OTC suite's outbound-buffer helpers.
# ---------------------------------------------------------------------------


COOKIE_NAME = "rfc_device_trust"

# The device-trust cookie is set with Secure=True, which httpx (the
# TestClient's underlying transport) will only return on an https
# scheme. We use a `base_url="https://testserver"` so the cookie
# roundtrips faithfully — that mirrors how production deployments
# serve the framework (per the v0.11.0 upgrade-step requiring HTTPS).
HTTPS_BASE = "https://testserver"


def _reset_outbound():
    from app import email as email_mod
    email_mod.reset_sent_envelopes()


def _outbound_otc_codes(to_address: str | None = None) -> list[str]:
    from app import email as email_mod
    out = []
    for env in email_mod.sent_envelopes():
        if env.get("kind") != "otc":
            continue
        if to_address is not None and env["to"] != to_address:
            continue
        for line in env["body"].splitlines():
            tok = line.strip()
            if tok.isdigit() and len(tok) == 6:
                out.append(tok)
                break
    return out


def _sign_in_via_otc(client, email: str, *, trust_device: bool = False) -> None:
    r = client.post("/auth/otc/request", json={"email": email})
    assert r.status_code == 200, r.text
    code = _outbound_otc_codes(email)[-1]
    body = {"email": email, "code": code, "trust_device": trust_device}
    r = client.post("/auth/otc/verify", json=body)
    assert r.status_code == 200, r.text


def _device_rows_for_email(email: str) -> list[dict]:
    from app import db
    rows = db.conn().execute(
        """
        SELECT dt.*
          FROM device_trust dt
          JOIN users u ON u.id = dt.user_id
         WHERE u.email = ? COLLATE NOCASE
         ORDER BY dt.id
        """,
        (email,),
    ).fetchall()
    return [dict(r) for r in rows]


# ---------------------------------------------------------------------------
# trust_device flag controls cookie issuance
# ---------------------------------------------------------------------------


def test_otc_verify_without_trust_device_does_not_issue_cookie(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app, base_url=HTTPS_BASE) as client:
        _reset_outbound()
        _sign_in_via_otc(client, "alice@example.com", trust_device=False)
        # No cookie set on the response.
        assert COOKIE_NAME not in {c.name for c in client.cookies.jar}
        # No row inserted.
        assert _device_rows_for_email("alice@example.com") == []


def test_otc_verify_with_trust_device_issues_cookie_and_row(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app, base_url=HTTPS_BASE) as client:
        _reset_outbound()
        r = client.post("/auth/otc/request", json={"email": "alice@example.com"})
        assert r.status_code == 200, r.text
        code = _outbound_otc_codes("alice@example.com")[-1]
        r = client.post(
            "/auth/otc/verify",
            json={"email": "alice@example.com", "code": code, "trust_device": True},
            headers={"User-Agent": "Mozilla/5.0 (TestBrowser)"},
        )
        assert r.status_code == 200, r.text

        # Cookie present on the response.
        set_cookie = r.headers.get("set-cookie", "")
        assert COOKIE_NAME in set_cookie
        # Cookie attribute set asserts the spec'd shape. Starlette emits
        # the attribute names case-insensitively (`samesite=lax`,
        # `httponly`); we normalize when asserting.
        lower = set_cookie.lower()
        assert "httponly" in lower
        assert "secure" in lower
        assert "samesite=lax" in lower
        assert "max-age=" in lower

        # Row inserted; hash is not the raw token.
        rows = _device_rows_for_email("alice@example.com")
        assert len(rows) == 1
        row = rows[0]
        assert row["revoked_at"] is None
        assert row["user_agent"] == "Mozilla/5.0 (TestBrowser)"
        cookie_token = client.cookies.get(COOKIE_NAME)
        assert cookie_token
        assert cookie_token != row["device_token_hash"]
        # bcrypt hash shape (starts with $2)
        assert row["device_token_hash"].startswith("$2")


def test_passcode_verify_with_trust_device_issues_cookie(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app, base_url=HTTPS_BASE) as client:
        _reset_outbound()
        _sign_in_via_otc(client, "alice@example.com")

        # Set a passcode.
        r = client.post("/auth/passcode/set", json={"passcode": "secret123"})
        assert r.status_code == 200, r.text

        # Sign out so the passcode verify path is the active sign-in.
        client.cookies.clear()

        # Passcode verify with trust_device=true issues a row.
        r = client.post(
            "/auth/passcode/verify",
            json={"email": "alice@example.com", "passcode": "secret123", "trust_device": True},
            headers={"User-Agent": "Test/Phone"},
        )
        assert r.status_code == 200, r.text
        set_cookie = r.headers.get("set-cookie", "")
        assert COOKIE_NAME in set_cookie

        rows = _device_rows_for_email("alice@example.com")
        assert len(rows) == 1
        assert rows[0]["user_agent"] == "Test/Phone"


# ---------------------------------------------------------------------------
# /auth/device-trust/start
# ---------------------------------------------------------------------------


def test_device_trust_start_with_no_cookie_returns_401(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app, base_url=HTTPS_BASE) as client:
        r = client.post("/auth/device-trust/start")
        assert r.status_code == 401


def test_device_trust_start_with_valid_cookie_establishes_session(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app, base_url=HTTPS_BASE) as client:
        _reset_outbound()
        # Trust the device.
        r = client.post("/auth/otc/request", json={"email": "alice@example.com"})
        assert r.status_code == 200, r.text
        code = _outbound_otc_codes("alice@example.com")[-1]
        r = client.post(
            "/auth/otc/verify",
            json={"email": "alice@example.com", "code": code, "trust_device": True},
        )
        assert r.status_code == 200, r.text
        trust_cookie = client.cookies.get(COOKIE_NAME)
        assert trust_cookie

        # Clear the session cookie so only the device-trust cookie is in
        # play. We keep `rfc_device_trust` and drop `rfc_session`.
        for cookie in list(client.cookies.jar):
            if cookie.name != COOKIE_NAME:
                client.cookies.jar.clear(cookie.domain, cookie.path, cookie.name)

        # The session cookie is gone — /api/auth/me reads anonymous.
        me = client.get("/api/auth/me").json()
        assert me["authenticated"] is False

        # Hit the trust-start endpoint; the cookie re-establishes the session.
        r = client.post("/auth/device-trust/start")
        assert r.status_code == 200, r.text
        assert r.json()["user"]["email"] == "alice@example.com"

        # /api/auth/me now reads authenticated.
        me = client.get("/api/auth/me").json()
        assert me["authenticated"] is True
        assert me["user"]["email"] == "alice@example.com"


def test_device_trust_start_refreshes_last_seen_at(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app, base_url=HTTPS_BASE) as client:
        _reset_outbound()
        r = client.post("/auth/otc/request", json={"email": "alice@example.com"})
        assert r.status_code == 200, r.text
        code = _outbound_otc_codes("alice@example.com")[-1]
        r = client.post(
            "/auth/otc/verify",
            json={"email": "alice@example.com", "code": code, "trust_device": True},
        )
        assert r.status_code == 200, r.text

        # Force the existing row's last_seen_at into the past so we can
        # assert the refresh moved it forward.
        db.conn().execute(
            """
            UPDATE device_trust
               SET last_seen_at = datetime('now', '-7 days')
            """
        )

        # Hit the start endpoint.
        r = client.post("/auth/device-trust/start")
        assert r.status_code == 200, r.text

        # last_seen_at is now recent (within the last minute).
        row = db.conn().execute(
            "SELECT last_seen_at, datetime('now') >= datetime(last_seen_at, '-1 minute') AS fresh FROM device_trust LIMIT 1"
        ).fetchone()
        assert row["fresh"] == 1


def test_device_trust_start_with_revoked_row_refuses_and_clears(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app, base_url=HTTPS_BASE) as client:
        _reset_outbound()
        r = client.post("/auth/otc/request", json={"email": "alice@example.com"})
        assert r.status_code == 200, r.text
        code = _outbound_otc_codes("alice@example.com")[-1]
        r = client.post(
            "/auth/otc/verify",
            json={"email": "alice@example.com", "code": code, "trust_device": True},
        )
        assert r.status_code == 200, r.text

        # Revoke the row out-of-band.
        db.conn().execute(
            "UPDATE device_trust SET revoked_at = datetime('now')"
        )

        # Now the start endpoint refuses + clears the cookie.
        r = client.post("/auth/device-trust/start")
        assert r.status_code == 401
        # The cookie is cleared via a Set-Cookie header with Max-Age=0
        # (Starlette's `delete_cookie` shape).
        set_cookie = r.headers.get("set-cookie", "")
        assert COOKIE_NAME in set_cookie
        assert "Max-Age=0" in set_cookie or 'expires=Thu, 01 Jan 1970' in set_cookie.lower().replace("expires=thu", "expires=Thu")


def test_device_trust_start_with_expired_row_refuses_and_clears(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app, base_url=HTTPS_BASE) as client:
        _reset_outbound()
        r = client.post("/auth/otc/request", json={"email": "alice@example.com"})
        assert r.status_code == 200, r.text
        code = _outbound_otc_codes("alice@example.com")[-1]
        r = client.post(
            "/auth/otc/verify",
            json={"email": "alice@example.com", "code": code, "trust_device": True},
        )
        assert r.status_code == 200, r.text

        # Backdate the expiry into the past.
        db.conn().execute(
            "UPDATE device_trust SET expires_at = datetime('now', '-1 day')"
        )

        r = client.post("/auth/device-trust/start")
        assert r.status_code == 401
        set_cookie = r.headers.get("set-cookie", "")
        assert COOKIE_NAME in set_cookie


def test_device_trust_start_with_forged_cookie_refuses_and_clears(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app, base_url=HTTPS_BASE) as client:
        # No real row; just paste a cookie value.
        client.cookies.set(COOKIE_NAME, "definitely-not-a-real-token-value-xxx")
        r = client.post("/auth/device-trust/start")
        assert r.status_code == 401
        set_cookie = r.headers.get("set-cookie", "")
        assert COOKIE_NAME in set_cookie


# ---------------------------------------------------------------------------
# /api/auth/me/devices — list + revoke
# ---------------------------------------------------------------------------


def test_list_devices_requires_session(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app, base_url=HTTPS_BASE) as client:
        r = client.get("/api/auth/me/devices")
        assert r.status_code == 401


def test_list_devices_returns_active_rows_only(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app, base_url=HTTPS_BASE) as client:
        _reset_outbound()
        _sign_in_via_otc(client, "alice@example.com", trust_device=True)

        # Add a second trusted device by re-running the verify flow.
        # OTC has a per-email cooldown, so drop the cooldown rather
        # than waiting it out.
        db.conn().execute("UPDATE otc_codes SET consumed_at = datetime('now', '-1 hour'), created_at = datetime('now', '-1 hour')")
        r = client.post("/auth/otc/request", json={"email": "alice@example.com"})
        assert r.status_code == 200, r.text
        code = _outbound_otc_codes("alice@example.com")[-1]
        r = client.post(
            "/auth/otc/verify",
            json={"email": "alice@example.com", "code": code, "trust_device": True},
            headers={"User-Agent": "Test/Tablet"},
        )
        assert r.status_code == 200, r.text

        # Revoke one row directly.
        db.conn().execute(
            "UPDATE device_trust SET revoked_at = datetime('now') WHERE id = 1"
        )

        # /api/auth/me/devices returns only the un-revoked one.
        r = client.get("/api/auth/me/devices")
        assert r.status_code == 200, r.text
        items = r.json()["items"]
        assert len(items) == 1
        assert items[0]["user_agent"] == "Test/Tablet"


def test_revoke_single_device_kills_the_row(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app, base_url=HTTPS_BASE) as client:
        _reset_outbound()
        _sign_in_via_otc(client, "alice@example.com", trust_device=True)

        r = client.get("/api/auth/me/devices")
        assert r.status_code == 200, r.text
        items = r.json()["items"]
        assert len(items) == 1
        device_id = items[0]["id"]

        # Revoke it.
        r = client.delete(f"/api/auth/me/devices/{device_id}")
        assert r.status_code == 200, r.text

        # List is empty.
        r = client.get("/api/auth/me/devices")
        assert r.json()["items"] == []

        # The row in the table has revoked_at populated.
        row = db.conn().execute(
            "SELECT revoked_at FROM device_trust WHERE id = ?", (device_id,)
        ).fetchone()
        assert row["revoked_at"] is not None


def test_revoke_other_users_device_reads_404(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app, base_url=HTTPS_BASE) as client:
        _reset_outbound()
        # Alice trusts a device.
        _sign_in_via_otc(client, "alice@example.com", trust_device=True)
        alice_device_id = client.get("/api/auth/me/devices").json()["items"][0]["id"]

        # Bob signs in (without a trusted device of his own).
        client.cookies.clear()
        db.conn().execute("UPDATE otc_codes SET consumed_at = datetime('now', '-1 hour'), created_at = datetime('now', '-1 hour')")
        _sign_in_via_otc(client, "bob@example.com", trust_device=False)

        # Bob tries to revoke Alice's row by id.
        r = client.delete(f"/api/auth/me/devices/{alice_device_id}")
        assert r.status_code == 404

        # Alice's row is still active.
        row = db.conn().execute(
            "SELECT revoked_at FROM device_trust WHERE id = ?", (alice_device_id,)
        ).fetchone()
        assert row["revoked_at"] is None


def test_revoke_all_devices_kills_every_active_row(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app, base_url=HTTPS_BASE) as client:
        _reset_outbound()
        _sign_in_via_otc(client, "alice@example.com", trust_device=True)

        # Add a second device.
        db.conn().execute("UPDATE otc_codes SET consumed_at = datetime('now', '-1 hour'), created_at = datetime('now', '-1 hour')")
        r = client.post("/auth/otc/request", json={"email": "alice@example.com"})
        assert r.status_code == 200, r.text
        code = _outbound_otc_codes("alice@example.com")[-1]
        r = client.post(
            "/auth/otc/verify",
            json={"email": "alice@example.com", "code": code, "trust_device": True},
        )
        assert r.status_code == 200, r.text

        # Two active rows.
        assert len(client.get("/api/auth/me/devices").json()["items"]) == 2

        # Revoke all.
        r = client.delete("/api/auth/me/devices")
        assert r.status_code == 200, r.text
        assert r.json()["revoked"] == 2

        # List is empty.
        assert client.get("/api/auth/me/devices").json()["items"] == []