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