"""Service-token lifecycle — mint, authenticate, revoke. Token format: ``helix_svc_`` + 32 base32 chars (160 bits of entropy), hashed with SHA-256 for storage — identical security posture to PATs (see ``app.auth.tokens.service`` for the SHA-256-vs-KDF rationale). The distinct prefix is deliberate: a leaked credential is immediately classifiable (person vs system) in logs, and the PAT-only MCP endpoint rejects ``helix_svc_`` bearers outright. """ from __future__ import annotations import base64 import hashlib import logging import secrets from datetime import UTC, datetime, timedelta from uuid import UUID from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.core.exceptions import ConflictError, NotFoundError, ValidationError from app.identity.consumers.models import ( CONSUMER_STATUS_ACTIVE, IdentityConsumer, IdentityConsumerToken, ) logger = logging.getLogger(__name__) SVC_TOKEN_PREFIX = "helix_svc_" _SECRET_BYTES = 20 PREFIX_DISPLAY_LEN = len(SVC_TOKEN_PREFIX) + 2 #: Expiry presets (days) for service tokens. Wider than PATs because a #: machine credential's rotation cadence is a policy choice, not a #: personal-hygiene one; NULL (never) stays allowed for consumers whose #: rotation is driven by the admin console instead of a clock. ALLOWED_EXPIRY_DAYS: frozenset[int] = frozenset({30, 90, 180, 365}) #: Cap of live (non-revoked, non-expired) tokens per consumer. Two is #: enough for an overlap rotation; the cap exists so a forgotten #: rotation script can't accumulate credentials. MAX_ACTIVE_TOKENS_PER_CONSUMER = 3 def hash_token(plaintext: str) -> str: return hashlib.sha256(plaintext.encode("utf-8")).hexdigest() def _generate_plaintext() -> tuple[str, str]: raw = secrets.token_bytes(_SECRET_BYTES) body = base64.b32encode(raw).decode("ascii").rstrip("=").lower() plaintext = f"{SVC_TOKEN_PREFIX}{body}" return plaintext, plaintext[:PREFIX_DISPLAY_LEN] def _as_aware_utc(dt: datetime | None) -> datetime | None: """MySQL DATETIME round-trips naive; Helix writes are always UTC.""" if dt is None: return None if dt.tzinfo is None: return dt.replace(tzinfo=UTC) return dt def token_is_active(row: IdentityConsumerToken) -> bool: if row.revoked_at is not None: return False expires = _as_aware_utc(row.expires_at) return expires is None or expires > datetime.now(UTC) class ConsumerTokenService: def __init__(self, db: AsyncSession) -> None: self.db = db async def mint( self, *, consumer: IdentityConsumer, name: str, expires_in_days: int | None, created_by: UUID | None, ) -> tuple[IdentityConsumerToken, str]: """Mint a token for ``consumer`` and return ``(row, plaintext)``. The plaintext is shown exactly once — only its hash persists. Caller commits 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 = [t for t in consumer.tokens if token_is_active(t)] if len(active) >= MAX_ACTIVE_TOKENS_PER_CONSUMER: raise ConflictError( f"Maximum of {MAX_ACTIVE_TOKENS_PER_CONSUMER} active tokens per " "consumer. Revoke an old token before minting a new one." ) plaintext, prefix = _generate_plaintext() token_hash = hash_token(plaintext) existing = await self.db.execute( select(IdentityConsumerToken.id).where( IdentityConsumerToken.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: expires_at = datetime.now(UTC) + timedelta(days=expires_in_days) row = IdentityConsumerToken( consumer_id=consumer.id, name=clean_name, token_hash=token_hash, token_prefix=prefix, expires_at=expires_at, created_by=created_by, ) self.db.add(row) await self.db.flush() return row, plaintext async def revoke( self, *, consumer_id: UUID, token_id: UUID, actor_user_id: UUID, ) -> IdentityConsumerToken: """Idempotent revoke — a duplicate click reads as success.""" row = await self.db.get(IdentityConsumerToken, token_id) if row is None or row.consumer_id != consumer_id: raise NotFoundError("IdentityConsumerToken", 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, ) -> tuple[IdentityConsumer, IdentityConsumerToken] | None: """Resolve a plaintext service token to ``(consumer, token)``. Returns ``None`` for every failure mode (unknown / revoked / expired token, disabled / deleted consumer) so the caller can answer one uniform 401 without leaking which check failed. Does NOT write anything — the auth dependency persists last-used + usage bookkeeping in its own short session so a read-only request that never commits still leaves a trail. """ if not plaintext or not plaintext.startswith(SVC_TOKEN_PREFIX): return None token_hash = hash_token(plaintext) result = await self.db.execute( select(IdentityConsumerToken) .where(IdentityConsumerToken.token_hash == token_hash) .options( selectinload(IdentityConsumerToken.consumer) .selectinload(IdentityConsumer.tokens) ) ) row = result.scalar_one_or_none() if row is None or not token_is_active(row): return None consumer = row.consumer if ( consumer is None or consumer.deleted_at is not None or consumer.status != CONSUMER_STATUS_ACTIVE ): return None return consumer, row __all__ = [ "ALLOWED_EXPIRY_DAYS", "MAX_ACTIVE_TOKENS_PER_CONSUMER", "PREFIX_DISPLAY_LEN", "SVC_TOKEN_PREFIX", "ConsumerTokenService", "hash_token", "token_is_active", ]