feat(pat): add personal access token storage and settings

A personal_access_tokens table (migration 0032) holds scoped user-level API
credentials. Only the SHA-256 of the secret is stored, like device session
tokens. Lookups exclude revoked and expired tokens and the tokens of
deactivated users. PAT_* settings cover the feature switch, default and
maximum lifetime, the operator opt-in for non-expiring tokens and the
per-user cap.

.env-template

@@ -101,3 +101,11 @@ MICROSOFT_AUTHORITY=https://{tenantId}.ciamlogin.com/{tenantId}
 # pair with OIDC_USER_ID_CLAIM=email so SCIM userName matches the OIDC user id)
 # SCIM_ENABLED=false
 # SCIM_TOKEN=<long random bearer token presented by the IdP's SCIM client>
+
+# Personal access tokens (scoped API tokens for CLI and CI/CD; Settings → Access Tokens).
+# Available with AUTH_TYPE=oidc or unset.
+# PAT_ENABLED=true
+# PAT_DEFAULT_LIFETIME_DAYS=90
+# PAT_MAX_LIFETIME_DAYS=365
+# PAT_ALLOW_NON_EXPIRING=false
+# PAT_MAX_PER_USER=25

docs/content/Deploying/Settings-Reference.mdx

@@ -132,6 +132,36 @@ Type `str`, default unset.
 
 Bearer token for IdP SCIM clients (required when SCIM is enabled).
 
+### `PAT_ENABLED`
+
+Type `bool`, default `true`.
+
+Allow users to create personal access tokens. Tokens are only issued under AUTH_TYPE=oidc or unset (None); simple_jwt and session_jwt have no stable user identity to bind a token to.
+
+### `PAT_DEFAULT_LIFETIME_DAYS`
+
+Type `int`, default `90`, must be `> 0`.
+
+Lifetime of a personal access token created without an explicit expiry.
+
+### `PAT_MAX_LIFETIME_DAYS`
+
+Type `int`, default `365`, must be `> 0`.
+
+Longest lifetime a user may request for a personal access token.
+
+### `PAT_ALLOW_NON_EXPIRING`
+
+Type `bool`, default `false`.
+
+Let users create personal access tokens that never expire. Off by default.
+
+### `PAT_MAX_PER_USER`
+
+Type `int`, default `25`, must be `> 0`.
+
+Maximum number of live personal access tokens per user.
+
 
 ## LLM providers
 

docsgpt/alembic/versions/0032_personal_access_tokens.py

@@ -0,0 +1,72 @@
+"""0032 personal access tokens — scoped, user-level API credentials.
+
+A personal access token (PAT) authenticates its owner against the management
+API for CLI and CI/CD use. Only the SHA-256 of the secret is stored, mirroring
+``devices.token_hash``: the plaintext is shown once at creation and a database
+leak cannot reconstruct it. ``token_prefix`` keeps the first characters so a
+user can tell their tokens apart in the UI.
+
+``scopes`` is the server-side grant list (never read from the credential
+itself). ``resource_filter`` optionally narrows a resource family to specific
+ids, e.g. ``{"agents": ["<uuid>"]}``; an absent family is unrestricted within
+the token's scopes. ``expires_at`` is NULL only when the operator allows
+non-expiring tokens.
+
+``user_id`` is the auth ``sub``; no FK or trigger, mirroring ``devices`` and
+``user_roles`` so a token row never blocks user deletion.
+
+Revision ID: 0032_personal_access_tokens
+Revises: 0031_token_usage_cache_tokens
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+
+revision: str = "0032_personal_access_tokens"
+down_revision: Union[str, None] = "0031_token_usage_cache_tokens"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE IF NOT EXISTS personal_access_tokens (
+            id              UUID        PRIMARY KEY DEFAULT gen_random_uuid(),
+            user_id         TEXT        NOT NULL,
+            name            TEXT        NOT NULL,
+            token_hash      TEXT        NOT NULL,
+            token_prefix    TEXT        NOT NULL,
+            scopes          TEXT[]      NOT NULL DEFAULT '{}',
+            resource_filter JSONB       NOT NULL DEFAULT '{}'::jsonb,
+            status          TEXT        NOT NULL DEFAULT 'active'
+                                        CHECK (status IN ('active', 'revoked')),
+            expires_at      TIMESTAMPTZ,
+            last_used_at    TIMESTAMPTZ,
+            last_used_ip    TEXT,
+            created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
+            revoked_at      TIMESTAMPTZ,
+            revoke_reason   TEXT
+        );
+        """
+    )
+    # Looked up on every PAT-authenticated request.
+    op.execute(
+        "CREATE UNIQUE INDEX IF NOT EXISTS personal_access_tokens_hash_uidx "
+        "ON personal_access_tokens(token_hash);"
+    )
+    # Names are unique among a user's live tokens; a revoked name can be reused.
+    op.execute(
+        "CREATE UNIQUE INDEX IF NOT EXISTS personal_access_tokens_user_name_uidx "
+        "ON personal_access_tokens(user_id, name) WHERE status = 'active';"
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS personal_access_tokens_user_idx "
+        "ON personal_access_tokens(user_id, created_at DESC);"
+    )
+
+
+def downgrade() -> None:
+    op.execute("DROP TABLE IF EXISTS personal_access_tokens;")

docsgpt/core/settings/auth.py

@@ -85,6 +85,28 @@ class AuthSettings(SettingsGroup):
         default=None, description="Bearer token for IdP SCIM clients (required when SCIM is enabled)."
     )
 
+    # Personal access tokens: scoped user-level API credentials for CLI and CI/CD use.
+    PAT_ENABLED: bool = Field(
+        default=True,
+        description=(
+            "Allow users to create personal access tokens. Tokens are only issued under AUTH_TYPE=oidc or "
+            "unset (None); simple_jwt and session_jwt have no stable user identity to bind a token to."
+        ),
+    )
+    PAT_DEFAULT_LIFETIME_DAYS: int = Field(
+        default=90, gt=0, description="Lifetime of a personal access token created without an explicit expiry."
+    )
+    PAT_MAX_LIFETIME_DAYS: int = Field(
+        default=365, gt=0, description="Longest lifetime a user may request for a personal access token."
+    )
+    PAT_ALLOW_NON_EXPIRING: bool = Field(
+        default=False,
+        description="Let users create personal access tokens that never expire. Off by default.",
+    )
+    PAT_MAX_PER_USER: int = Field(
+        default=25, gt=0, description="Maximum number of live personal access tokens per user."
+    )
+
     @field_validator("AUTH_TYPE", mode="before")
     @classmethod
     def _normalize_auth_type(cls, v):
@@ -99,4 +121,6 @@ class AuthSettings(SettingsGroup):
                 raise ValueError(f"AUTH_TYPE=oidc requires settings: {', '.join(missing)}")
         if self.SCIM_ENABLED and not self.SCIM_TOKEN:
             raise ValueError("SCIM_ENABLED requires settings: SCIM_TOKEN")
+        if self.PAT_DEFAULT_LIFETIME_DAYS > self.PAT_MAX_LIFETIME_DAYS:
+            raise ValueError("PAT_DEFAULT_LIFETIME_DAYS must not exceed PAT_MAX_LIFETIME_DAYS")
         return self

docsgpt/storage/db/models.py

@@ -1079,3 +1079,44 @@ device_auto_approve_patterns_table = Table(
     Column("created_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
     UniqueConstraint("device_id", "user_id", "pattern", name="device_auto_approve_uidx"),
 )
+
+# --- Personal access tokens (migration 0032) --------------------------------
+# Scoped user-level API credentials. Only the SHA-256 of the secret is stored.
+
+personal_access_tokens_table = Table(
+    "personal_access_tokens",
+    metadata,
+    Column("id", UUID(as_uuid=True), primary_key=True, server_default=func.gen_random_uuid()),
+    Column("user_id", Text, nullable=False),
+    Column("name", Text, nullable=False),
+    Column("token_hash", Text, nullable=False),
+    Column("token_prefix", Text, nullable=False),
+    Column("scopes", ARRAY(Text), nullable=False, server_default="{}"),
+    Column("resource_filter", JSONB, nullable=False, server_default=text("'{}'::jsonb")),
+    Column("status", Text, nullable=False, server_default="active"),
+    Column("expires_at", DateTime(timezone=True)),
+    Column("last_used_at", DateTime(timezone=True)),
+    Column("last_used_ip", Text),
+    Column("created_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
+    Column("revoked_at", DateTime(timezone=True)),
+    Column("revoke_reason", Text),
+    CheckConstraint("status IN ('active', 'revoked')", name="personal_access_tokens_status_check"),
+)
+
+Index(
+    "personal_access_tokens_hash_uidx",
+    personal_access_tokens_table.c.token_hash,
+    unique=True,
+)
+Index(
+    "personal_access_tokens_user_name_uidx",
+    personal_access_tokens_table.c.user_id,
+    personal_access_tokens_table.c.name,
+    unique=True,
+    postgresql_where=personal_access_tokens_table.c.status == "active",
+)
+Index(
+    "personal_access_tokens_user_idx",
+    personal_access_tokens_table.c.user_id,
+    personal_access_tokens_table.c.created_at.desc(),
+)

docsgpt/storage/db/repositories/personal_access_tokens.py

@@ -0,0 +1,156 @@
+"""Repository for the ``personal_access_tokens`` table."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import Connection, text
+
+from docsgpt.storage.db.base_repository import row_to_dict
+
+
+# token_hash never leaves the repository except through find_active_by_hash.
+_PUBLIC_COLUMNS = (
+    "id, user_id, name, token_prefix, scopes, resource_filter, status, "
+    "expires_at, last_used_at, last_used_ip, created_at, revoked_at, revoke_reason"
+)
+
+
+class PersonalAccessTokensRepository:
+    """CRUD for personal access tokens. Callers hash the secret; only the hash is stored."""
+
+    def __init__(self, conn: Connection) -> None:
+        self._conn = conn
+
+    def create(
+        self,
+        user_id: str,
+        name: str,
+        *,
+        token_hash: str,
+        token_prefix: str,
+        scopes: list[str],
+        resource_filter: Optional[dict] = None,
+        expires_at: Optional[datetime] = None,
+    ) -> dict:
+        row = self._conn.execute(
+            text(
+                f"""
+                INSERT INTO personal_access_tokens (
+                    user_id, name, token_hash, token_prefix, scopes,
+                    resource_filter, expires_at
+                ) VALUES (
+                    :user_id, :name, :token_hash, :token_prefix, :scopes,
+                    CAST(:resource_filter AS jsonb), :expires_at
+                ) RETURNING {_PUBLIC_COLUMNS}
+                """
+            ),
+            {
+                "user_id": user_id,
+                "name": name,
+                "token_hash": token_hash,
+                "token_prefix": token_prefix,
+                "scopes": list(scopes),
+                "resource_filter": json.dumps(resource_filter or {}),
+                "expires_at": expires_at,
+            },
+        ).fetchone()
+        return row_to_dict(row)
+
+    def get(self, token_id: str, user_id: Optional[str] = None) -> Optional[dict]:
+        sql = f"SELECT {_PUBLIC_COLUMNS} FROM personal_access_tokens WHERE id = CAST(:id AS uuid)"
+        params: dict = {"id": token_id}
+        if user_id is not None:
+            sql += " AND user_id = :user_id"
+            params["user_id"] = user_id
+        row = self._conn.execute(text(sql), params).fetchone()
+        return row_to_dict(row) if row is not None else None
+
+    def list_for_user(self, user_id: str, *, include_revoked: bool = False) -> list[dict]:
+        sql = f"SELECT {_PUBLIC_COLUMNS} FROM personal_access_tokens WHERE user_id = :user_id"
+        if not include_revoked:
+            sql += " AND status = 'active'"
+        sql += " ORDER BY created_at DESC"
+        result = self._conn.execute(text(sql), {"user_id": user_id})
+        return [row_to_dict(r) for r in result.fetchall()]
+
+    def count_active(self, user_id: str) -> int:
+        """Live tokens only: revoked and expired rows don't count against the per-user cap."""
+        return self._conn.execute(
+            text(
+                "SELECT count(*) FROM personal_access_tokens "
+                "WHERE user_id = :user_id AND status = 'active' "
+                "AND (expires_at IS NULL OR expires_at > now())"
+            ),
+            {"user_id": user_id},
+        ).scalar_one()
+
+    def name_in_use(self, user_id: str, name: str) -> bool:
+        return (
+            self._conn.execute(
+                text(
+                    "SELECT 1 FROM personal_access_tokens "
+                    "WHERE user_id = :user_id AND name = :name AND status = 'active' LIMIT 1"
+                ),
+                {"user_id": user_id, "name": name},
+            ).fetchone()
+            is not None
+        )
+
+    def find_active_by_hash(self, token_hash: str) -> Optional[dict]:
+        """Resolve the credential on each request.
+
+        Revoked and expired tokens never match, and neither do the tokens of a
+        deactivated user (admin or SCIM), so deactivation needs no token sweep
+        and reactivation restores them.
+        """
+        row = self._conn.execute(
+            text(
+                f"SELECT {_PUBLIC_COLUMNS} FROM personal_access_tokens pat "
+                "WHERE token_hash = :token_hash AND status = 'active' "
+                "AND (expires_at IS NULL OR expires_at > now()) "
+                "AND NOT EXISTS (SELECT 1 FROM users u "
+                "WHERE u.user_id = pat.user_id AND u.active = false) "
+                "LIMIT 1"
+            ),
+            {"token_hash": token_hash},
+        ).fetchone()
+        return row_to_dict(row) if row is not None else None
+
+    def touch_last_used(self, token_id: str, ip: Optional[str], *, min_interval_seconds: int = 60) -> None:
+        """Record use, at most once per ``min_interval_seconds`` so hot tokens don't write per request."""
+        self._conn.execute(
+            text(
+                "UPDATE personal_access_tokens "
+                "SET last_used_at = now(), last_used_ip = :ip "
+                "WHERE id = CAST(:id AS uuid) AND (last_used_at IS NULL "
+                "OR last_used_at <= now() - make_interval(secs => :min_interval))"
+            ),
+            {"id": token_id, "ip": ip, "min_interval": min_interval_seconds},
+        )
+
+    def revoke(self, token_id: str, user_id: Optional[str] = None, *, reason: str = "user_revoked") -> bool:
+        """Revoke one token. ``user_id=None`` is the admin path (any owner)."""
+        sql = (
+            "UPDATE personal_access_tokens "
+            "SET status = 'revoked', revoked_at = now(), revoke_reason = :reason "
+            "WHERE id = CAST(:id AS uuid) AND status = 'active'"
+        )
+        params: dict = {"id": token_id, "reason": reason}
+        if user_id is not None:
+            sql += " AND user_id = :user_id"
+            params["user_id"] = user_id
+        return self._conn.execute(text(sql), params).rowcount > 0
+
+    def revoke_all_for_user(self, user_id: str, *, reason: str = "admin_revoked") -> int:
+        result = self._conn.execute(
+            text(
+                "UPDATE personal_access_tokens "
+                "SET status = 'revoked', revoked_at = now(), revoke_reason = :reason "
+                "WHERE user_id = :user_id AND status = 'active'"
+            ),
+            {"user_id": user_id, "reason": reason},
+        )
+        return result.rowcount

tests/storage/db/repositories/test_personal_access_tokens.py

@@ -0,0 +1,163 @@
+"""Tests for PersonalAccessTokensRepository against a real Postgres."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import text
+
+from docsgpt.storage.db.repositories.personal_access_tokens import (
+    PersonalAccessTokensRepository,
+)
+
+
+def _create(repo, user_id="u1", name="ci", token_hash="h1", **kwargs):
+    kwargs.setdefault("scopes", ["agents:read"])
+    return repo.create(
+        user_id, name, token_hash=token_hash, token_prefix="dgpt_pat_abc123", **kwargs
+    )
+
+
+class TestCreateAndRead:
+    def test_create_returns_public_columns_only(self, pg_conn):
+        row = _create(
+            PersonalAccessTokensRepository(pg_conn),
+            resource_filter={"agents": ["00000000-0000-0000-0000-000000000001"]},
+        )
+        assert "token_hash" not in row
+        assert row["status"] == "active"
+        assert row["scopes"] == ["agents:read"]
+        assert row["resource_filter"] == {"agents": ["00000000-0000-0000-0000-000000000001"]}
+        assert row["expires_at"] is None
+
+    def test_get_is_owner_scoped(self, pg_conn):
+        repo = PersonalAccessTokensRepository(pg_conn)
+        row = _create(repo)
+        assert repo.get(str(row["id"]), "u1")["name"] == "ci"
+        assert repo.get(str(row["id"]), "someone-else") is None
+        assert repo.get(str(row["id"]))["user_id"] == "u1"
+
+    def test_list_hides_revoked_by_default(self, pg_conn):
+        repo = PersonalAccessTokensRepository(pg_conn)
+        kept = _create(repo, name="kept", token_hash="h1")
+        gone = _create(repo, name="gone", token_hash="h2")
+        repo.revoke(str(gone["id"]), "u1")
+        assert [r["id"] for r in repo.list_for_user("u1")] == [kept["id"]]
+        assert len(repo.list_for_user("u1", include_revoked=True)) == 2
+        assert repo.list_for_user("u2") == []
+
+
+class TestUniqueness:
+    def test_duplicate_active_name_rejected(self, pg_conn):
+        from sqlalchemy.exc import IntegrityError
+
+        repo = PersonalAccessTokensRepository(pg_conn)
+        _create(repo, token_hash="h1")
+        with pytest.raises(IntegrityError), pg_conn.begin_nested():
+            _create(repo, token_hash="h2")
+
+    def test_revoked_name_can_be_reused(self, pg_conn):
+        repo = PersonalAccessTokensRepository(pg_conn)
+        first = _create(repo, token_hash="h1")
+        repo.revoke(str(first["id"]), "u1")
+        assert not repo.name_in_use("u1", "ci")
+        assert _create(repo, token_hash="h2")["name"] == "ci"
+
+    def test_same_name_for_other_user_is_fine(self, pg_conn):
+        repo = PersonalAccessTokensRepository(pg_conn)
+        _create(repo, user_id="u1", token_hash="h1")
+        _create(repo, user_id="u2", token_hash="h2")
+        assert repo.name_in_use("u1", "ci") and repo.name_in_use("u2", "ci")
+
+
+class TestFindActiveByHash:
+    def test_finds_live_token(self, pg_conn):
+        repo = PersonalAccessTokensRepository(pg_conn)
+        _create(repo)
+        found = repo.find_active_by_hash("h1")
+        assert found["user_id"] == "u1"
+        assert "token_hash" not in found
+
+    def test_unknown_hash(self, pg_conn):
+        assert PersonalAccessTokensRepository(pg_conn).find_active_by_hash("nope") is None
+
+    def test_revoked_token_never_matches(self, pg_conn):
+        repo = PersonalAccessTokensRepository(pg_conn)
+        row = _create(repo)
+        repo.revoke(str(row["id"]), "u1")
+        assert repo.find_active_by_hash("h1") is None
+
+    def test_expired_token_never_matches(self, pg_conn):
+        repo = PersonalAccessTokensRepository(pg_conn)
+        _create(repo, expires_at=datetime.now(timezone.utc) - timedelta(seconds=1))
+        assert repo.find_active_by_hash("h1") is None
+
+    def test_future_expiry_matches(self, pg_conn):
+        repo = PersonalAccessTokensRepository(pg_conn)
+        _create(repo, expires_at=datetime.now(timezone.utc) + timedelta(days=1))
+        assert repo.find_active_by_hash("h1") is not None
+
+    def test_deactivated_user_token_never_matches(self, pg_conn):
+        repo = PersonalAccessTokensRepository(pg_conn)
+        _create(repo)
+        pg_conn.execute(
+            text("INSERT INTO users (user_id, active) VALUES ('u1', false)")
+        )
+        assert repo.find_active_by_hash("h1") is None
+        pg_conn.execute(text("UPDATE users SET active = true WHERE user_id = 'u1'"))
+        assert repo.find_active_by_hash("h1") is not None
+
+
+class TestRevoke:
+    def test_revoke_is_owner_scoped(self, pg_conn):
+        repo = PersonalAccessTokensRepository(pg_conn)
+        row = _create(repo)
+        assert repo.revoke(str(row["id"]), "someone-else") is False
+        assert repo.revoke(str(row["id"]), "u1") is True
+        assert repo.revoke(str(row["id"]), "u1") is False
+        stored = repo.get(str(row["id"]))
+        assert stored["status"] == "revoked"
+        assert stored["revoke_reason"] == "user_revoked"
+        assert stored["revoked_at"] is not None
+
+    def test_admin_revoke_needs_no_owner(self, pg_conn):
+        repo = PersonalAccessTokensRepository(pg_conn)
+        row = _create(repo)
+        assert repo.revoke(str(row["id"]), reason="admin_revoked") is True
+        assert repo.get(str(row["id"]))["revoke_reason"] == "admin_revoked"
+
+    def test_revoke_all_for_user(self, pg_conn):
+        repo = PersonalAccessTokensRepository(pg_conn)
+        _create(repo, name="a", token_hash="h1")
+        _create(repo, name="b", token_hash="h2")
+        _create(repo, user_id="u2", token_hash="h3")
+        assert repo.revoke_all_for_user("u1") == 2
+        assert repo.list_for_user("u1") == []
+        assert len(repo.list_for_user("u2")) == 1
+
+
+class TestCountAndUsage:
+    def test_count_active_ignores_revoked_and_expired(self, pg_conn):
+        repo = PersonalAccessTokensRepository(pg_conn)
+        _create(repo, name="live", token_hash="h1")
+        _create(
+            repo,
+            name="expired",
+            token_hash="h2",
+            expires_at=datetime.now(timezone.utc) - timedelta(days=1),
+        )
+        revoked = _create(repo, name="revoked", token_hash="h3")
+        repo.revoke(str(revoked["id"]), "u1")
+        assert repo.count_active("u1") == 1
+
+    def test_touch_last_used_is_throttled(self, pg_conn):
+        repo = PersonalAccessTokensRepository(pg_conn)
+        row = _create(repo)
+        repo.touch_last_used(str(row["id"]), "10.0.0.1")
+        first = repo.get(str(row["id"]))
+        assert first["last_used_ip"] == "10.0.0.1"
+        repo.touch_last_used(str(row["id"]), "10.0.0.2")
+        assert repo.get(str(row["id"]))["last_used_ip"] == "10.0.0.1"
+        repo.touch_last_used(str(row["id"]), "10.0.0.3", min_interval_seconds=0)
+        assert repo.get(str(row["id"]))["last_used_ip"] == "10.0.0.3"