"""PAT lifecycle service — generate, verify, list, revoke. Token format: ``helix_pat_`` + 32 base32 chars (160 bits of entropy). The first 12 chars (prefix incl. underscore) are stored verbatim so the UI can list tokens by prefix without revealing the secret. The full secret is hashed with SHA-256 and the hash is what we look up on every authenticated request. Why SHA-256 and not bcrypt/argon2: PATs have 160 bits of entropy straight out of ``secrets.token_bytes(20)`` — there is no offline dictionary attack to slow down. We need lookup by hash to be O(1) on every API call (millions/day target), and any password-grade KDF would make that prohibitively expensive. This is the same trade-off GitHub / GitLab / 1Password make for their PATs. """ from __future__ import annotations import base64 import hashlib import logging import secrets from datetime import UTC, datetime from uuid import UUID from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.auth.tokens.models import PersonalAccessToken from app.core.exceptions import ConflictError, NotFoundError, ValidationError logger = logging.getLogger(__name__) #: Plaintext token format: ``helix_pat_<32 base32 chars>``. PAT_PREFIX = "helix_pat_" #: Bytes of randomness for the secret portion. 20 bytes = 160 bits, #: encoded as 32 base32 chars (no padding) — comfortably above NIST #: 128-bit minimum and matches GitHub PAT entropy. _SECRET_BYTES = 20 #: Length of the prefix kept on the row for UI display: ``helix_pat_`` + #: first 2 chars of the secret. Two extra chars give us ~32^2 = 1024 #: distinct buckets, plenty to differentiate at-a-glance. PREFIX_DISPLAY_LEN = len(PAT_PREFIX) + 2 #: Per-user cap on active (non-revoked) tokens. Soft limit — IT Admin #: can override by directly revoking on a user's behalf if a higher #: cap is genuinely needed. The cap is here to keep blast radius #: manageable (a stolen laptop with one PAT is bad; with twenty is #: catastrophic). MAX_TOKENS_PER_USER = 4 #: Allowed expiry presets (days). NULL means "never expires"; we treat #: that as a separate UI option (no number). ALLOWED_EXPIRY_DAYS: frozenset[int] = frozenset({30, 90, 365}) def _hash_token(plaintext: str) -> str: """SHA-256 the plaintext to its hex digest. Matches the column width.""" return hashlib.sha256(plaintext.encode("utf-8")).hexdigest() def _as_aware_utc(dt: datetime | None) -> datetime | None: """Coerce a possibly-naive datetime to a UTC-aware one. MySQL's ``DATETIME`` column type doesn't store a timezone, so the asyncmy driver hands rows back as naive ``datetime`` objects even when the SQLAlchemy column was declared ``DateTime(timezone=True)``. Comparing those naive values against ``datetime.now(UTC)`` raises ``TypeError: can't compare offset-naive and offset-aware datetimes``. Helix writes always use UTC, so coercing reads to UTC-aware is correct (not a fallback / not lossy). """ if dt is None: return None if dt.tzinfo is None: return dt.replace(tzinfo=UTC) return dt def _generate_plaintext() -> tuple[str, str]: """Return (plaintext, prefix). Plaintext is the full secret to send to the caller; prefix is the safe-to-store identifier. """ raw = secrets.token_bytes(_SECRET_BYTES) # Lowercase base32 without padding — URL-safe, easy to copy from a # terminal, and the same alphabet across platforms. body = base64.b32encode(raw).decode("ascii").rstrip("=").lower() plaintext = f"{PAT_PREFIX}{body}" return plaintext, plaintext[:PREFIX_DISPLAY_LEN] class PersonalAccessTokenService: def __init__(self, db: AsyncSession) -> None: self.db = db async def list_for_user(self, user_id: UUID) -> list[PersonalAccessToken]: """Return the user's tokens (active + revoked + expired) ordered by created_at desc. Revoked rows are kept so the user can see the history of "yes I did revoke that one". """ result = await self.db.execute( select(PersonalAccessToken) .where(PersonalAccessToken.user_id == user_id) .order_by(PersonalAccessToken.created_at.desc()) ) return list(result.scalars().all()) async def count_active_for_user(self, user_id: UUID) -> int: """Count tokens that count against ``MAX_TOKENS_PER_USER`` — i.e. neither revoked nor expired. Expired rows are kept in the table for forensics but they do NOT consume a slot, so a user whose all-four tokens lapsed can still create a new one without going through admin support. """ from sqlalchemy import func, or_ now = datetime.now(UTC) result = await self.db.execute( select(func.count()) .select_from(PersonalAccessToken) .where( PersonalAccessToken.user_id == user_id, PersonalAccessToken.revoked_at.is_(None), or_( PersonalAccessToken.expires_at.is_(None), PersonalAccessToken.expires_at > now, ), ) ) return int(result.scalar() or 0) async def create( self, *, user_id: UUID, name: str, expires_in_days: int | None, created_ip: str | None = None, created_user_agent: str | None = None, ) -> tuple[PersonalAccessToken, str]: """Mint a new PAT for ``user_id`` and return (row, plaintext). The plaintext is the only time the caller will ever see the secret — the row only retains its hash. Validation: * ``name`` 1..120 chars after strip; rejected if empty. * ``expires_in_days`` must be one of ``ALLOWED_EXPIRY_DAYS`` or None (= never). * Active-token count enforced before insert; over the cap → 409 with a hint to revoke an existing token. * Hash collision (1 in 2^256, basically impossible) maps to 409 so the UI can prompt "try again". Caller must commit the session. """ clean_name = (name or "").strip() if not clean_name: raise ValidationError("Token name is required.") if len(clean_name) > 120: raise ValidationError("Token name must be 120 characters or fewer.") if expires_in_days is not None and expires_in_days not in ALLOWED_EXPIRY_DAYS: raise ValidationError( f"expires_in_days must be one of {sorted(ALLOWED_EXPIRY_DAYS)} or null." ) active_count = await self.count_active_for_user(user_id) if active_count >= MAX_TOKENS_PER_USER: raise ConflictError( f"Maximum of {MAX_TOKENS_PER_USER} active tokens per user. " "Revoke an existing token before creating a new one." ) plaintext, prefix = _generate_plaintext() token_hash = _hash_token(plaintext) # SHA-256 collisions are not realistic but cheaper to detect # than to debug at 3am if they ever happen — surface as 409. existing = await self.db.execute( select(PersonalAccessToken).where( PersonalAccessToken.token_hash == token_hash, ) ) if existing.scalar_one_or_none() is not None: raise ConflictError("Token generation collision; please retry.") expires_at: datetime | None = None if expires_in_days is not None: from datetime import timedelta expires_at = datetime.now(UTC) + timedelta(days=expires_in_days) row = PersonalAccessToken( user_id=user_id, name=clean_name, token_hash=token_hash, token_prefix=prefix, expires_at=expires_at, created_ip=created_ip, created_user_agent=(created_user_agent or "")[:512] or None, ) self.db.add(row) await self.db.flush() return row, plaintext async def revoke( self, *, token_id: UUID, actor_user_id: UUID, is_admin: bool = False, ) -> PersonalAccessToken: """Mark a PAT revoked. Only the owner (or an admin) may revoke. Idempotent: revoking an already-revoked or expired row returns the row unchanged so a duplicate UI click reads as success. """ row = await self.db.get(PersonalAccessToken, token_id) # Collapse "no such token" and "not your token" into an identical # NotFoundError for non-admin callers, so an authenticated attacker # POSTing random UUIDs can't distinguish a live token id (403) from a # non-existent one (404) — a token-existence oracle. Mirrors the # api-key /reveal 404→403 collapse. An admin still resolves any row. if row is None or (row.user_id != actor_user_id and not is_admin): raise NotFoundError("PersonalAccessToken", str(token_id)) if row.revoked_at is not None: return row row.revoked_at = datetime.now(UTC) row.revoked_by_user_id = actor_user_id await self.db.flush() return row async def authenticate( self, plaintext: str, *, client_ip: str | None = None, user_agent: str | None = None, ) -> PersonalAccessToken | None: """Resolve a plaintext PAT to its row, updating last-used metadata. Returns ``None`` for any authentication failure (unknown / revoked / expired) so the caller can return a single uniform 401 without leaking which case occurred. Updates ``last_used_*`` fields only on success — failed lookups don't touch the row, so an attacker can't probe for valid prefixes by side-effect. The bookkeeping update (``last_used_at`` / ``last_used_ip`` / ``last_used_user_agent``) is committed in a *separate* short-lived session so it persists even when the request that triggered the authentication is a read-only endpoint that never commits its own transaction. This means a stolen-token investigation always has accurate "last used" data. """ if not plaintext or not plaintext.startswith(PAT_PREFIX): return None token_hash = _hash_token(plaintext) result = await self.db.execute( select(PersonalAccessToken).where( PersonalAccessToken.token_hash == token_hash, ) ) row = result.scalar_one_or_none() if row is None: return None if row.revoked_at is not None: return None expires = _as_aware_utc(row.expires_at) if expires is not None and expires <= datetime.now(UTC): return None # Update last-used fields and commit in an independent session # so the bookkeeping survives even if the request transaction # rolls back. We refresh the in-memory row's fields too so the # caller sees the same values that were persisted. from app.core.database import async_session_factory now = datetime.now(UTC) ip_clipped = client_ip[:64] if client_ip else None ua_clipped = user_agent[:512] if user_agent else None try: async with async_session_factory() as bookkeeping: fresh = await bookkeeping.get(PersonalAccessToken, row.id) if fresh is not None: fresh.last_used_at = now if ip_clipped: fresh.last_used_ip = ip_clipped if ua_clipped: fresh.last_used_user_agent = ua_clipped await bookkeeping.commit() except Exception as exc: # noqa: BLE001 — fail-open on bookkeeping # Persisting last-used data is best-effort: a DB blip here # must NOT cause a 401 on an otherwise-valid request. logger.warning("PAT bookkeeping update failed: %s", exc) # Reflect the values on the returned row so callers can read # them back without a refetch. row.last_used_at = now if ip_clipped: row.last_used_ip = ip_clipped if ua_clipped: row.last_used_user_agent = ua_clipped return row async def get_for_user( self, *, token_id: UUID, user_id: UUID, ) -> PersonalAccessToken | None: row = await self.db.get(PersonalAccessToken, token_id) if row is None or row.user_id != user_id: return None return row def is_active(row: PersonalAccessToken) -> bool: """A token is 'active' iff not revoked AND not expired.""" if row.revoked_at is not None: return False expires = _as_aware_utc(row.expires_at) if expires is not None and expires <= datetime.now(UTC): return False return True __all__ = [ "ALLOWED_EXPIRY_DAYS", "MAX_TOKENS_PER_USER", "PAT_PREFIX", "PersonalAccessTokenService", "is_active", ]