"""ORM model for ``auth_oauth_clients`` (migration 325). A row represents an OAuth 2.0 *public* client dynamically registered via RFC 7591 — typically a Claude / Cursor MCP connector. Public clients hold no secret; PKCE (RFC 7636, S256) is the proof-of-possession that binds the authorization code to the originating client. We persist clients (rather than keeping them in Redis) because a connector registers once and then reuses its ``client_id`` across weeks of token refreshes — the redirect-URI allowlist must survive process restarts to keep the open-redirect gate honest. """ from __future__ import annotations import uuid from datetime import datetime from sqlalchemy import JSON, DateTime, String, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column from sqlalchemy.sql import func from app.core.database import Base from app.core.sql_types import GUID as Uuid class OAuthClient(Base): __tablename__ = "auth_oauth_clients" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column( Uuid, primary_key=True, default=uuid.uuid4, ) # Public, client-facing identifier (``helix_mcp_``). Unique; # this is what the connector echoes on every authorize / token call. client_id: Mapped[str] = mapped_column( String(64), nullable=False, unique=True, ) # Human label from the DCR request (``client_name``) — shown on the # consent screen so the user knows which app is asking. Best-effort. client_name: Mapped[str | None] = mapped_column(String(255)) # Exact-match redirect-URI allowlist (JSON array of strings). The # authorize + token endpoints reject any redirect_uri not in this set. redirect_uris: Mapped[list] = mapped_column(JSON, nullable=False) # Echoed-back DCR metadata, kept for audit / introspection. grant_types: Mapped[list | None] = mapped_column(JSON) response_types: Mapped[list | None] = mapped_column(JSON) scope: Mapped[str | None] = mapped_column(String(255)) token_endpoint_auth_method: Mapped[str] = mapped_column( String(32), nullable=False, server_default="none", ) # Soft kill-switch — a disabled client fails authorize + token. disabled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) created_ip: Mapped[str | None] = mapped_column(String(64)) created_user_agent: Mapped[str | None] = mapped_column(String(512)) class OAuthGrant(Base): """A durable record that ``user_id`` authorized ``client_id`` to act as them over MCP. One row per (user, client) — re-authorizing the same connector refreshes ``granted_at`` and clears any revocation. This is the source of truth for the governance surfaces (admin "connected apps", employee "my connections") and for revocation: a row with ``revoked_at`` set, or whose client is disabled, is treated as immediate logout by both the MCP middleware and the refresh grant. """ __tablename__ = "auth_oauth_grants" __mapper_args__ = {"eager_defaults": True} # One row per (user, client). Mirrors migration 327's named unique # constraint — declared on the model too so the ORM-built schema # (tests) enforces it, and so the upsert in ``record_grant`` has a # conflict target to land its ON DUPLICATE KEY UPDATE on. __table_args__ = ( UniqueConstraint("user_id", "client_id", name="uq_oauth_grant_user_client"), ) id: Mapped[uuid.UUID] = mapped_column( Uuid, primary_key=True, default=uuid.uuid4, ) user_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False, index=True) client_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True) granted_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) revoked_by: Mapped[str | None] = mapped_column(String(16)) __all__ = ["OAuthClient", "OAuthGrant"]