Release 0.11.0: trust device for 30 days
This commit is contained in:
Ben Stull
@@ -0,0 +1,351 @@
|
||||
"""§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
|
||||
Reference in New Issue
Block a user