stackpilot/backend/services/api_token_service.py

"""Long-lived API tokens for scripts and CI.

A session token is the wrong credential for automation: it expires in an hour,
it is minted by typing a password, and revoking it means signing every one of
that person's devices out. So a CI job gets its own credential, which can be
revoked on its own, is capped to read-only if that is all it needs, and shows up
in the audit log as itself.

**Only a hash is stored.** Unlike a registry password — which has to be handed
back to the registry, so it is encrypted and recoverable — a token is only ever
compared against. It is shown once at creation and cannot be recovered, which is
the difference between leaking the database and leaking everything it protects.

The hash is a plain SHA-256 and deliberately not bcrypt. Bcrypt is slow on
purpose, to make guessing low-entropy human passwords expensive; a token is 256
bits of ``secrets`` output, where guessing is not the threat and the cost would
instead land on every single API request.
"""
from __future__ import annotations

import hashlib
import secrets
from datetime import datetime, timedelta, timezone
from typing import Optional

from sqlmodel import Session, select

from models.api_token import ApiToken
from models.user import User

#: Marks a StackPilot token at a glance — in a log, in a CI settings page, or to
#: a secret scanner. It is also how the auth dependency tells a token from a JWT
#: without trying to decode it.
PREFIX = "sp_"

#: How stale last_used_at may get before a request writes it again. Without a
#: floor this would be a database write on every single API call.
_TOUCH_INTERVAL = timedelta(minutes=5)


def looks_like_token(value: str) -> bool:
    return (value or "").startswith(PREFIX)


def _hash(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def _aware(value: Optional[datetime]) -> Optional[datetime]:
    """SQLite hands back naive datetimes; compare them as UTC."""
    if value is None:
        return None
    return value if value.tzinfo else value.replace(tzinfo=timezone.utc)


def is_expired(row: ApiToken, now: Optional[datetime] = None) -> bool:
    expires = _aware(row.expires_at)
    if expires is None:
        return False
    return expires <= (now or datetime.now(timezone.utc))


def mint(
    session: Session,
    *,
    name: str,
    user: User,
    scope: str = "read",
    expires_in_days: Optional[int] = None,
) -> tuple[ApiToken, str]:
    """Create a token. Returns the row and the secret, which is shown once."""
    token = PREFIX + secrets.token_urlsafe(32)
    expires_at = (
        datetime.now(timezone.utc) + timedelta(days=expires_in_days)
        if expires_in_days
        else None
    )
    row = ApiToken(
        name=name,
        prefix=token[: len(PREFIX) + 8],
        token_hash=_hash(token),
        scope=scope if scope in ("read", "admin") else "read",
        user_id=user.id,
        expires_at=expires_at,
    )
    session.add(row)
    session.commit()
    session.refresh(row)
    return row, token


def resolve(session: Session, token: str) -> Optional[tuple[ApiToken, User]]:
    """The token row and its owner, or None if it cannot be used.

    None covers every reason equally — unknown, expired, owner disabled — so a
    caller cannot learn which by watching the responses.
    """
    if not looks_like_token(token):
        return None
    prefix = token[: len(PREFIX) + 8]
    row = session.exec(select(ApiToken).where(ApiToken.prefix == prefix)).first()
    if not row:
        return None
    # Constant-time, so a wrong token cannot be narrowed down by timing.
    if not secrets.compare_digest(row.token_hash, _hash(token)):
        return None
    if is_expired(row):
        return None
    user = session.get(User, row.user_id)
    if not user or not user.is_active:
        return None
    return row, user


def effective_role(row: ApiToken, user: User) -> str:
    """What this token may do, which is never more than its owner may.

    A token keeps working when its owner is demoted, but drops to read-only with
    them — the alternative is an admin token outliving the admin.
    """
    if row.scope == "admin" and user.role == "admin":
        return "admin"
    return "user"


def touch(session: Session, row: ApiToken) -> None:
    """Record that the token was used, at most once every few minutes."""
    now = datetime.now(timezone.utc)
    last = _aware(row.last_used_at)
    if last and now - last < _TOUCH_INTERVAL:
        return
    row.last_used_at = now
    session.add(row)
    session.commit()