siftlode/backend/app/apitokens.py

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

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 or_, select, update
from sqlalchemy.orm import Session

from app import debugsession, logs
from app.auth import current_user
from app.db import get_db
from app.models import ApiToken, ApiTokenGrant, User
from app.ratelimit import RateLimiter, client_key
from app.refusal import refused
from app.security import hash_token

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)


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:]
    if any(c not in _ALPHABET for c in body):
        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