rfc-app/backend/app/device_trust.py

"""§6.2 / v0.11.0: trust device for 30 days (roadmap item #9).

After a successful OTC or passcode sign-in, a contributor may check
"trust this device for 30 days." The framework then issues a
server-issued opaque token, hashes it (bcrypt) for storage in the
`device_trust` table, and sets a long-lived cookie carrying the raw
token. On a subsequent visit, the cookie is presented at
`/auth/device-trust/start`; if a non-expired, non-revoked row matches,
the session is re-established without another OTC / passcode round
trip.

The shape:

  * `issue(user_id, user_agent)` — mint a fresh CSPRNG token, hash it,
    insert a row, and return the raw token + row id so the endpoint
    can set the cookie. The 30-day expiry is the only knob; the
    `revoked_at` column stays NULL.
  * `lookup(raw_token)` — walk the user's active rows (the unique
    index keys on the hash, so we read a small candidate set), check
    the bcrypt hash in constant time, drop any row whose `expires_at`
    has passed or whose `revoked_at` is non-NULL, and return the
    matched row or None. On a hit, refresh `last_seen_at`.
  * `list_for_user(user_id)` — return the active rows for the
    /settings/devices surface. Revoked + expired rows are filtered out
    so the surface only shows live trust grants.
  * `revoke(user_id, row_id)` — stamp `revoked_at` on the row. The
    next lookup refuses the cookie token (the row is dead).
  * `revoke_all(user_id)` — bulk-revoke every active row for the user.
    The /settings/devices surface's "revoke all" button calls this.

Cookie shape: `rfc_device_trust`. HttpOnly, Secure, SameSite=Lax,
Max-Age=2592000 (30 days), Path=/. The cookie value is the raw token;
server-side storage is the hash. The cookie is "essential" per the
v0.13.0 cookie-consent banner (it is part of authentication, not
analytics), so the framework sets it regardless of analytics /
other-cookies choices.

Constant-time comparison: bcrypt's `checkpw` is already constant-time
over the hash bytes. We walk the candidate set linearly with `_check`
which delegates to `bcrypt.checkpw`; no early-exit shortcut leaks
which row was the match.

The raw token never appears in a log line or an exception message;
the helpers carry the token only as a parameter and forget it after
hashing.

The cookie sits orthogonal to the §6.1 `permission_state` gate: a
revoked or pending user with a valid device-trust cookie still
re-establishes their session (the cookie identifies the user, not
their admission state), and the existing `require_contributor` /
`require_admin` dependencies in `auth.py` continue to refuse the
unrelated write surfaces.
"""
from __future__ import annotations

import logging
import secrets
from dataclasses import dataclass

import bcrypt

from . import db
from .auth import SessionUser

log = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Tunables — hard-coded in v0.11.0 (§19.2 candidate to env-ify later).
# ---------------------------------------------------------------------------

TRUST_DURATION_DAYS = 30
COOKIE_NAME = "rfc_device_trust"
COOKIE_MAX_AGE_SECONDS = TRUST_DURATION_DAYS * 24 * 60 * 60
# 256 bits of CSPRNG entropy. `secrets.token_urlsafe(32)` yields ~43
# URL-safe characters; the bcrypt hash is what's stored, so the raw
# token only ever lives in the cookie.
TOKEN_BYTES = 32
# User-Agent header values seen in the wild can be unbounded; clamp
# to a reasonable ceiling so a hostile UA doesn't bloat the row.
USER_AGENT_MAX_LENGTH = 1024


# ---------------------------------------------------------------------------
# Issue
# ---------------------------------------------------------------------------


@dataclass
class IssueOutcome:
    """The shape returned from `issue`.

    `raw_token` is the cookie value to send to the client; it never
    appears in storage. `row_id` is the surrogate key for the
    /settings/devices UI to address the row by id.
    """
    raw_token: str
    row_id: int


def _new_token() -> str:
    return secrets.token_urlsafe(TOKEN_BYTES)


def _hash(token: str) -> str:
    return bcrypt.hashpw(token.encode("utf-8"), bcrypt.gensalt()).decode("ascii")


def _check(token: str, token_hash: str) -> bool:
    try:
        return bcrypt.checkpw(token.encode("utf-8"), token_hash.encode("ascii"))
    except (ValueError, TypeError):
        return False


def _trim_user_agent(ua: str) -> str:
    ua = (ua or "").strip()
    if len(ua) > USER_AGENT_MAX_LENGTH:
        return ua[:USER_AGENT_MAX_LENGTH]
    return ua


def issue(user_id: int, user_agent: str) -> IssueOutcome:
    """Mint a fresh device-trust token + row for `user_id`.

    The row's expiry is set 30 days in the future. The hash, not the
    raw token, lands in the database. The caller (the endpoint) sets
    the cookie with the raw token returned here.
    """
    raw = _new_token()
    h = _hash(raw)
    ua = _trim_user_agent(user_agent)
    cur = db.conn().execute(
        f"""
        INSERT INTO device_trust (user_id, device_token_hash, expires_at, user_agent)
        VALUES (?, ?, datetime('now', '+{TRUST_DURATION_DAYS} days'), ?)
        """,
        (user_id, h, ua),
    )
    row_id = cur.lastrowid
    return IssueOutcome(raw_token=raw, row_id=row_id)


# ---------------------------------------------------------------------------
# Lookup
# ---------------------------------------------------------------------------


@dataclass
class LookupOutcome:
    """The result of `lookup`.

    `user` is populated only on a hit. `reason` distinguishes the
    failure modes so the endpoint can decide whether to clear the
    cookie ('expired', 'revoked', 'unknown') or just refuse ('invalid').
    """
    ok: bool
    user: SessionUser | None
    reason: str  # 'ok' | 'invalid' | 'unknown' | 'expired' | 'revoked'
    row_id: int | None = None


def lookup(raw_token: str) -> LookupOutcome:
    """Resolve a presented cookie token to a user.

    A hit refreshes `last_seen_at` on the matched row. A miss returns
    a reason so the endpoint can clear the stale cookie if the row
    was revoked or expired (vs. simply unknown, which probably means
    the cookie was forged or the row was wiped by a /settings/devices
    revoke from another browser).
    """
    raw = (raw_token or "").strip()
    if not raw:
        return LookupOutcome(ok=False, user=None, reason="invalid")

    # The unique index on `device_token_hash` would let us SELECT by
    # hash if bcrypt were a stable hash, but bcrypt incorporates a
    # per-row salt — equal tokens produce different hashes. We walk
    # the candidate set instead. In practice the set is small (a
    # human has a handful of trusted devices) and bcrypt is cheap on
    # the order of milliseconds; the walk is bounded by the user's
    # active device count.
    #
    # We don't pre-filter by `revoked_at IS NULL` here so that a
    # token presented for a recently-revoked row produces a
    # 'revoked' outcome (the endpoint surfaces a different shape).
    # Same for expired: we let the walk hit and classify after.
    rows = db.conn().execute(
        """
        SELECT id, user_id, device_token_hash, expires_at, revoked_at
          FROM device_trust
         ORDER BY id DESC
        """,
    ).fetchall()

    matched = None
    for row in rows:
        if _check(raw, row["device_token_hash"]):
            matched = row
            break

    if matched is None:
        return LookupOutcome(ok=False, user=None, reason="unknown")

    if matched["revoked_at"] is not None:
        return LookupOutcome(ok=False, user=None, reason="revoked", row_id=matched["id"])

    expired = db.conn().execute(
        "SELECT datetime(?) < datetime('now') AS expired",
        (matched["expires_at"],),
    ).fetchone()["expired"]
    if expired:
        return LookupOutcome(ok=False, user=None, reason="expired", row_id=matched["id"])

    # Refresh last-seen so the /settings/devices surface can show the
    # user when each device was last active. This is the only write
    # the lookup path does on the hot read.
    db.conn().execute(
        "UPDATE device_trust SET last_seen_at = datetime('now') WHERE id = ?",
        (matched["id"],),
    )
    user_row = db.conn().execute(
        """
        SELECT id, gitea_id, gitea_login, email, display_name, avatar_url, role, permission_state
          FROM users
         WHERE id = ?
        """,
        (matched["user_id"],),
    ).fetchone()
    if user_row is None:
        # The user row was deleted but the device_trust row hadn't
        # cascaded yet (shouldn't happen under the FK ON DELETE
        # CASCADE — be defensive anyway). Treat as 'unknown' so the
        # endpoint clears the cookie.
        return LookupOutcome(ok=False, user=None, reason="unknown", row_id=matched["id"])

    # Also stamp last_seen_at on the user row so the user's overall
    # activity stamp keeps pace with cookie-only sign-ins.
    db.conn().execute(
        "UPDATE users SET last_seen_at = datetime('now') WHERE id = ?",
        (matched["user_id"],),
    )

    return LookupOutcome(
        ok=True,
        user=SessionUser(
            user_id=user_row["id"],
            gitea_id=user_row["gitea_id"] or 0,
            gitea_login=user_row["gitea_login"] or "",
            display_name=user_row["display_name"],
            email=user_row["email"] or "",
            avatar_url=user_row["avatar_url"] or "",
            role=user_row["role"],
            permission_state=user_row["permission_state"] or "granted",
        ),
        reason="ok",
        row_id=matched["id"],
    )


# ---------------------------------------------------------------------------
# List / revoke (for the /settings/devices surface)
# ---------------------------------------------------------------------------


@dataclass
class DeviceRow:
    """The shape the /settings/devices endpoint returns.

    Note the absence of `device_token_hash` — the hash is structurally
    private, and the surface has no use for it.
    """
    id: int
    created_at: str
    expires_at: str
    last_seen_at: str
    user_agent: str


def list_for_user(user_id: int) -> list[DeviceRow]:
    """Active device-trust rows for the user, freshest first.

    Filters out revoked rows and rows whose expiry has passed; the
    surface only shows live trust grants. A user wondering "which
    devices are signed in" gets the answer that matches what the
    framework would actually accept on a presented cookie.
    """
    rows = db.conn().execute(
        """
        SELECT id, created_at, expires_at, last_seen_at, user_agent
          FROM device_trust
         WHERE user_id = ?
           AND revoked_at IS NULL
           AND datetime(expires_at) > datetime('now')
         ORDER BY last_seen_at DESC, id DESC
        """,
        (user_id,),
    ).fetchall()
    return [
        DeviceRow(
            id=row["id"],
            created_at=row["created_at"],
            expires_at=row["expires_at"],
            last_seen_at=row["last_seen_at"],
            user_agent=row["user_agent"] or "",
        )
        for row in rows
    ]


def revoke(user_id: int, row_id: int) -> bool:
    """Revoke a single device-trust row for the given user.

    Returns True iff a row was matched (still active, belongs to the
    user). The user-id scope is enforced in SQL so a hostile client
    cannot revoke another user's row by guessing ids.
    """
    cur = db.conn().execute(
        """
        UPDATE device_trust
           SET revoked_at = datetime('now')
         WHERE id = ?
           AND user_id = ?
           AND revoked_at IS NULL
        """,
        (row_id, user_id),
    )
    return cur.rowcount > 0


def revoke_all(user_id: int) -> int:
    """Revoke every active device-trust row for the user. Returns the
    count of rows touched.

    The /settings/devices "revoke all" button calls this. The user's
    current request stays authenticated via its session cookie; the
    device-trust cookie on the current device is also revoked, but
    the session middleware's `rfc_session` cookie keeps the request
    flow alive until the user signs out or the session cookie
    expires.
    """
    cur = db.conn().execute(
        """
        UPDATE device_trust
           SET revoked_at = datetime('now')
         WHERE user_id = ?
           AND revoked_at IS NULL
        """,
        (user_id,),
    )
    return cur.rowcount