"""Personal access tokens for AI agent / external tool authentication. Revision ID: 069 Revises: 068 Create Date: 2026-04-27 Adds ``auth_personal_access_tokens`` so an authenticated employee can issue long-lived bearer credentials that AI clients (Claude Desktop / Cursor / a custom MCP client) use to call the Helix API on the user's behalf. The token grants exactly the same permissions as the issuing user — there is no separate scope flag, by explicit product decision (see DECISIONS.md "PAT-only auth for AI agents"). Storage: * ``token_hash`` — SHA-256 of the full secret. The plaintext is never persisted; it is shown ONCE to the user on creation and otherwise unrecoverable. Compromising the DB does not leak the live token. * ``token_prefix`` — first 12 chars of the plaintext (``helix_pat_`` + 2 random chars), kept verbatim so the user can recognise their own tokens in the UI without revealing the secret. Indexed for the ``Authorization`` header lookup fast path. * ``expires_at`` — NULL means "never expires"; user-selectable on creation (30 / 90 / 365 / never). The MCP middleware rejects expired rows even if they are not yet revoked. * ``revoked_at`` — soft-deletion timestamp; once set, the row is permanently invalid. Audit history stays intact for forensics. * ``last_used_at`` / ``last_used_ip`` / ``last_used_user_agent`` — updated on every successful PAT-authenticated request so the user can spot a stolen token (unfamiliar IP / UA) on the management page. Per-user limit (max 4 active tokens) is enforced in the service layer, not the schema, so admins can override on a case-by-case basis without a migration. Idempotent. Downgrade drops the table. """ from __future__ import annotations import sqlalchemy as sa from alembic import op revision = "069" down_revision = "068" branch_labels = None depends_on = None def _table_exists(bind, table: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t" ), {"t": table}, ).scalar() is not None def _index_exists(bind, table: str, idx: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.STATISTICS " "WHERE TABLE_SCHEMA = DATABASE() " "AND TABLE_NAME = :t AND INDEX_NAME = :i" ), {"t": table, "i": idx}, ).scalar() is not None def _scheduled_job_exists(bind, name: str) -> bool: return bind.execute( sa.text("SELECT 1 FROM scheduled_jobs WHERE name = :n"), {"n": name}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() if not _table_exists(bind, "auth_personal_access_tokens"): op.create_table( "auth_personal_access_tokens", sa.Column("id", sa.String(36), primary_key=True), sa.Column("user_id", sa.String(36), nullable=False), sa.Column("name", sa.String(120), nullable=False), # 64 hex chars = SHA-256 of the plaintext secret. sa.Column("token_hash", sa.String(64), nullable=False), # The first 12 chars of the plaintext: ``helix_pat_`` + 2 random. # Shown in the UI so the user can identify which token is which # without ever recovering the full secret. sa.Column("token_prefix", sa.String(16), nullable=False), sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True), sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True), sa.Column("revoked_by_user_id", sa.String(36), nullable=True), sa.Column("last_used_at", sa.DateTime(timezone=True), nullable=True), sa.Column("last_used_ip", sa.String(64), nullable=True), sa.Column("last_used_user_agent", sa.String(512), nullable=True), sa.Column( "expiry_notified_at", sa.DateTime(timezone=True), nullable=True, comment="Set once the 7-day-before-expiry inbox warning has been sent.", ), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("created_ip", sa.String(64), nullable=True), sa.Column("created_user_agent", sa.String(512), nullable=True), ) # Unique index on token_hash — every secret hashes to one row. if not _index_exists( bind, "auth_personal_access_tokens", "uq_auth_pat_token_hash", ): op.create_index( "uq_auth_pat_token_hash", "auth_personal_access_tokens", ["token_hash"], unique=True, ) # Hot path: list a user's active tokens (filter on revoked_at IS NULL). if not _index_exists( bind, "auth_personal_access_tokens", "ix_auth_pat_user_active", ): op.create_index( "ix_auth_pat_user_active", "auth_personal_access_tokens", ["user_id", "revoked_at"], ) # Expiry-warning scheduled job scans rows whose expires_at falls # within the next 7 days and where expiry_notified_at IS NULL. if not _index_exists( bind, "auth_personal_access_tokens", "ix_auth_pat_expires_notif", ): op.create_index( "ix_auth_pat_expires_notif", "auth_personal_access_tokens", ["expires_at", "expiry_notified_at"], ) # FKs declared separately so a TiDB type-mismatch on the parent # ``identity_persons.id`` (CHAR(32) vs VARCHAR(36)) doesn't take # the table create down with it. Same pattern as 068 / 052. for name, col, ref_table, ref_col, on_delete in [ ( "fk_auth_pat_user", "user_id", "identity_persons", "id", "CASCADE", ), ( "fk_auth_pat_revoked_by", "revoked_by_user_id", "identity_persons", "id", "SET NULL", ), ]: try: op.create_foreign_key( name, "auth_personal_access_tokens", ref_table, [col], [ref_col], ondelete=on_delete, ) except Exception: # noqa: BLE001 # Already exists OR rejected by engine. Service-layer # integrity checks still hold; the FK is defence-in-depth. pass # ── Seed the daily PAT expiry-warning scheduled job ── # Runs at 09:00 Asia/Shanghai every day; scans tokens whose # ``expires_at`` falls within the next 7 days and pushes one inbox # notification to the owner. Configured via ``warn_days`` in # ``config`` if a different window is needed later. if not _scheduled_job_exists(bind, "auth.pat_expiry_warning"): op.execute( sa.text( "INSERT INTO scheduled_jobs " "(id, name, display_name, description, handler_key, " " trigger_type, interval_seconds, cron_expr, timezone, " " enabled, config, run_count, failure_count, " " created_at, updated_at) " "VALUES (UUID(), :name, :display_name, :description, :name, " " 'cron', NULL, '0 9 * * *', 'Asia/Shanghai', 1, " " CAST('{\"warn_days\": 7}' AS JSON), 0, 0, " " NOW(), NOW())" ).bindparams( name="auth.pat_expiry_warning", display_name="API Token 即将到期提醒", description=( "每日 09:00 扫 auth_personal_access_tokens,对距离过期 ≤ 7 天且未撤销、" "且尚未发过提醒的 token 推一条 inbox 通知给持有人。" ), ) ) def downgrade() -> None: bind = op.get_bind() op.execute( sa.text( "DELETE FROM scheduled_jobs WHERE name = :n" ).bindparams(n="auth.pat_expiry_warning") ) if _table_exists(bind, "auth_personal_access_tokens"): op.drop_table("auth_personal_access_tokens")