"""Machine access to the Files library: API tokens, and what each one may do where. A user makes a token for a machine of theirs — their hub, a script — and it reaches their library as `Authorization: Bearer sfl_…`. Only on `/api/files`: the routes there take `files_caller` instead of `current_user`, and nothing else does (a test holds the route table to that), so a token can never manage the account, the tokens, or anything outside the library. And only in the folders its grants name, with the rights they give: list, read, write, delete. Every route asks the one question, `FilesCaller.may(folder_id, right)`; a signed-in user may everything in their own folders. The token is `sfl_` + 32 random bytes in base62 + a CRC32 of them: a mistyped or cut-off token is refused without a database read, and a secret scanner can tell one by its shape. Only its SHA-256 is kept (`security.hash_token`) — the user sees it once, when it is made — with its first characters to tell tokens apart by. It is looked up by that hash on every request: no signed claims, so a deleted token stops working at once. A token about to run out warns its user once, a week before (`warn_expiring`), so a machine does not lose its key unannounced. """ import secrets import zlib from collections.abc import Mapping from dataclasses import dataclass from datetime import datetime, timedelta, timezone from enum import StrEnum from fastapi import Depends, Request from sqlalchemy import delete, or_, select, update from sqlalchemy.orm import Session from app import audit, debugsession, logs from app.audit import AuditAction from app.auth import current_user from app.db import get_db from app.models import ApiToken, ApiTokenGrant, User from app.notifications import create_notification from app.ratelimit import RateLimiter, client_key from app.refusal import refused from app.security import hash_token from app.utils import iso log = logs.get_logger(__name__) PREFIX = "sfl_" _ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" _BODY_BYTES = 32 _BODY_LEN = 43 # 62**43 > 2**256: every 32-byte value fits _CHECK_LEN = 6 # 62**6 > 2**32: every CRC32 fits TOKEN_LEN = len(PREFIX) + _BODY_LEN + _CHECK_LEN # What a token is shown by once it is made: enough to tell a user's tokens apart, far too little to # narrow a guess (8 of its 43 random characters). SHOWN_LEN = len(PREFIX) + 8 # A machine that runs away (a loop in a script) is held back per token; a caller trying tokens is # held back per client, counting only the refusals. _calls = RateLimiter(600, 60) _bad = RateLimiter(20, 300) # How stale `last_used_at` may get before a request writes it again: a busy script costs one write a # minute, not one per request. _TOUCH_EVERY = timedelta(minutes=1) # How long before a token runs out its user is told (`warn_expiring`); the token list marks it # "expires soon" for the same span. EXPIRY_WARNING = timedelta(days=7) class Right(StrEnum): LIST = "list" READ = "read" WRITE = "write" DELETE = "delete" # Each right's column on a grant. A dict over the enum, so a new right without a column fails here. GRANT_COLUMNS = { Right.LIST: "can_list", Right.READ: "can_read", Right.WRITE: "can_write", Right.DELETE: "can_delete", } def _b62(value: int, width: int) -> str: out = [] for _ in range(width): value, digit = divmod(value, 62) out.append(_ALPHABET[digit]) return "".join(reversed(out)) def _check(body: str) -> str: return _b62(zlib.crc32(body.encode()), _CHECK_LEN) def mint() -> str: """A new token, in full. The caller stores its hash and shows it to the user, once.""" body = _b62(int.from_bytes(secrets.token_bytes(_BODY_BYTES)), _BODY_LEN) return PREFIX + body + _check(body) def well_formed(raw: str) -> bool: """Whether `raw` has a token's shape and its checksum adds up.""" if len(raw) != TOKEN_LEN or not raw.startswith(PREFIX): return False body, check = raw[len(PREFIX) : -_CHECK_LEN], raw[-_CHECK_LEN:] # The check too: compare_digest refuses (TypeError, a 500) a str with anything non-ASCII in it. if any(c not in _ALPHABET for c in body + check): return False return secrets.compare_digest(_check(body), check) def grant_rights(grant: ApiTokenGrant) -> frozenset[Right]: return frozenset(r for r, col in GRANT_COLUMNS.items() if getattr(grant, col)) @dataclass(frozen=True) class FilesCaller: """Who is asking the Files library, and where they may do what. `grants` is None for a signed-in user (everything in their own folders); for a token, folder id → its rights there.""" user: User token_id: int | None = None token_prefix: str | None = None grants: Mapping[int, frozenset[Right]] | None = None def may(self, folder_id: int | None, right: Right) -> bool: if self.grants is None: return True return folder_id is not None and right in self.grants.get(folder_id, frozenset()) def folders_with(self, right: Right) -> set[int] | None: """The folders `right` is granted in; None = any of the user's own.""" if self.grants is None: return None return {f for f, rights in self.grants.items() if right in rights} @property def via(self) -> str | None: """How an audit row says a token did it, None when the user did.""" return None if self.token_prefix is None else f"api token {self.token_prefix}…" def bearer_token(request: Request) -> str | None: """The token of an `Authorization: Bearer` header; None when the request carries none (another scheme counts as none — it is not ours to judge).""" scheme, _, value = request.headers.get("authorization", "").partition(" ") if scheme.lower() != "bearer": return None return value.strip() def session_caller(request: Request, db: Session = Depends(get_db)) -> User | None: """The signed-in user, or None when the request brings a token instead: a token wins over a cookie, so a machine's request is judged by its grants alone, whatever else it carries.""" if bearer_token(request) is not None: return None return current_user(request, db) def files_caller( request: Request, db: Session = Depends(get_db), user: User | None = Depends(session_caller), ) -> FilesCaller: """The `/api/files` caller: a signed-in user, or a valid token's user with its grants.""" if user is not None: return FilesCaller(user=user) return _token_caller(request, db, bearer_token(request) or "") def _refuse_token(key: str, reason: str) -> Exception: _bad.allow(key) log.info( "a request brought an API token that does not hold", extra=logs.event("apitoken.refused", reason=reason), ) return refused(401, "not_authenticated") def _token_caller(request: Request, db: Session, raw: str) -> FilesCaller: key = client_key(request) if _bad.exhausted(key): raise refused(429, "api_token_rate_limited") if not well_formed(raw): raise _refuse_token(key, "malformed") token = db.execute( select(ApiToken).where(ApiToken.token_hash == hash_token(raw)) ).scalar_one_or_none() if token is None: raise _refuse_token(key, "unknown") now = datetime.now(timezone.utc) if token.expires_at is not None and token.expires_at <= now: raise _refuse_token(key, "expired") user = db.get(User, token.user_id) if user is None or not user.is_active or user.is_suspended: raise _refuse_token(key, "account") if not _calls.allow(str(token.id)): raise refused(429, "api_token_rate_limited") grants = { g.folder_id: grant_rights(g) for g in db.execute(select(ApiTokenGrant).where(ApiTokenGrant.token_id == token.id)) .scalars() .all() } caller = FilesCaller(user=user, token_id=token.id, token_prefix=token.prefix, grants=grants) db.execute( update(ApiToken) .where( ApiToken.id == token.id, or_(ApiToken.last_used_at.is_(None), ApiToken.last_used_at < now - _TOUCH_EVERY), ) .values(last_used_at=now) .execution_options(synchronize_session=False) ) # Committed here, ahead of the route: a read-only route never commits, and a write route must # not carry this along into a rollback of its own. db.commit() logs.set_user(user.id, *debugsession.level_for(user, None)) request.state.user_id = user.id return caller def revoke_all(db: Session, user_id: int) -> int: """Revoke every token of `user_id`, each with its audit row (NOT committed). A token is a credential like a session, so whatever ends every session of an account — a password reset or change, "log out everywhere", a revoked untrusted password — ends its tokens too: one minted from a stolen session must not outlive the owner's answer to that theft. Returns how many.""" gone = ( db.execute(delete(ApiToken).where(ApiToken.user_id == user_id).returning(ApiToken.id)) .scalars() .all() ) for token_id in gone: audit.record( db, user_id, AuditAction.API_TOKEN_REVOKE, target_type="api_token", target_id=token_id, summary="revoked with every session of the account", ) return len(gone) def expiring(token: ApiToken, now: datetime) -> bool: """Whether `token` runs out within the warning span (and has not yet).""" return token.expires_at is not None and now < token.expires_at <= now + EXPIRY_WARNING def warn_expiring(db: Session) -> int: """Tell each user whose token runs out within `EXPIRY_WARNING` — once per token: the mark and the notice commit together, and a token made again (a new row) warns afresh. Returns how many.""" now = datetime.now(timezone.utc) hits = db.execute( update(ApiToken) .where( ApiToken.expiry_warned.is_(False), ApiToken.expires_at.is_not(None), ApiToken.expires_at > now, ApiToken.expires_at <= now + EXPIRY_WARNING, ) .values(expiry_warned=True) .returning(ApiToken.id, ApiToken.user_id, ApiToken.name, ApiToken.expires_at) .execution_options(synchronize_session=False) ).all() for token_id, user_id, name, expires_at in hits: create_notification( db, user_id, "api_token_expiring", title=f"API token {name} expires soon", # English fallback; the client names it from data data={"token_id": token_id, "name": name, "expires_at": iso(expires_at)}, commit=False, ) db.commit() return len(hits)