Merge pull request #2812 from arc53-machine/feat/personal-access-tokens

feat: personal access tokens (scoped API tokens for CLI and CI/CD)

tests/api/test_admin_dashboard.py

@@ -178,13 +178,20 @@ class TestUserLifecycle:
 
     def test_force_logout(self, client):
         events = Mock()
-        with _admin(AuthEventsRepository=Mock(return_value=events)), patch(
-            "docsgpt.api.admin.routes.denylist"
-        ) as dl:
+        tokens = Mock()
+        tokens.revoke_all_for_user.return_value = ["t1", "t2"]
+        with _admin(
+            AuthEventsRepository=Mock(return_value=events),
+            PersonalAccessTokensRepository=Mock(return_value=tokens),
+        ), patch("docsgpt.api.admin.routes.denylist") as dl:
             dl.deny_user.return_value = True
             resp = client.post("/api/admin/users/bob/revoke-sessions")
         assert resp.status_code == 200
         dl.deny_user.assert_called_once_with("bob")
+        # A forced logout also revokes the user's API tokens, one audit event each.
+        tokens.revoke_all_for_user.assert_called_once_with("bob", reason="admin_sessions_revoked")
+        recorded = [call.args[1] for call in events.insert.call_args_list]
+        assert recorded == ["pat_revoked", "pat_revoked", "admin_sessions_revoked"]
 
     def test_user_detail(self, client):
         users = Mock()

tests/api/test_agent_portability.py

@@ -822,3 +822,72 @@ def test_api_tool_without_actions_imports_with_warning(pg_conn, monkeypatch):
     agent = AgentsRepository(pg_conn).get(result["agent_id"], user)
     row = UserToolsRepository(pg_conn).get_any(agent["tools"][0], user)
     assert row["config"]["actions"] == {}
+
+
+# --- resource-restricted personal access tokens ------------------------------
+
+
+def _pat_request(app, resource_filter):
+    from flask import request
+
+    ctx = app.test_request_context("/api/import_agent", method="POST")
+    ctx.push()
+    request.decoded_token = {
+        "sub": "u_pat_import",
+        "auth_method": "pat",
+        "scopes": ["agents:read", "agents:write"],
+        "resource_filter": resource_filter,
+    }
+    return ctx
+
+
+@pytest.fixture
+def flask_ctx_app():
+    from flask import Flask
+
+    return Flask(__name__)
+
+
+def test_restricted_token_may_update_only_its_agents(pg_conn, flask_ctx_app):
+    from docsgpt.api.user.agents.portability import _restricted_token_denial
+
+    user = "u_pat_import"
+    allowed = _make_agent(pg_conn, user, slug="allowed")
+    other = _make_agent(pg_conn, user, slug="other")
+
+    ctx = _pat_request(flask_ctx_app, {"agents": [str(allowed["id"])]})
+    try:
+        assert _restricted_token_denial(pg_conn, user, _doc(_slug="allowed")) is None
+        assert "only update" in _restricted_token_denial(pg_conn, user, _doc(_slug="other"))
+        assert "only update" in _restricted_token_denial(
+            pg_conn, user, {"metadata": {"id": str(other["id"])}}
+        )
+        # A slug that matches nothing would create a new agent: never for a restricted token.
+        assert "only update" in _restricted_token_denial(pg_conn, user, _doc(_slug="brand-new"))
+    finally:
+        ctx.pop()
+
+
+@pytest.mark.parametrize("family", ["sources", "prompts", "tools", "workflows"])
+def test_token_restricted_on_referenced_families_cannot_import(pg_conn, flask_ctx_app, family):
+    from docsgpt.api.user.agents.portability import _restricted_token_denial
+
+    ctx = _pat_request(flask_ctx_app, {family: ["aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"]})
+    try:
+        assert family in _restricted_token_denial(pg_conn, "u_pat_import", _doc())
+    finally:
+        ctx.pop()
+
+
+def test_unrestricted_token_and_sessions_import_freely(pg_conn, flask_ctx_app):
+    from flask import request
+
+    from docsgpt.api.user.agents.portability import _restricted_token_denial
+
+    ctx = _pat_request(flask_ctx_app, {})
+    try:
+        assert _restricted_token_denial(pg_conn, "u_pat_import", _doc()) is None
+        request.decoded_token = {"sub": "u_pat_import"}
+        assert _restricted_token_denial(pg_conn, "u_pat_import", _doc()) is None
+    finally:
+        ctx.pop()

tests/api/test_asgi_auth.py

@@ -114,3 +114,69 @@ def test_json_error_shape():
     response = asgi_auth.json_error("Forbidden", 403)
     assert response.status_code == 403
     assert json.loads(response.body) == {"success": False, "message": "Forbidden"}
+
+
+_PAT_CLAIMS = {
+    "sub": "alice",
+    "auth_method": "pat",
+    "pat_id": "t1",
+    "scopes": ["chat:run"],
+    "resource_filter": {},
+}
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+class TestPersonalAccessTokens:
+    async def test_routes_reject_tokens_unless_they_name_a_scope(self):
+        with patch.object(asgi_auth, "handle_auth", return_value=dict(_PAT_CLAIMS)):
+            decoded, error = await asgi_auth.authenticate(_request())
+        assert decoded is None
+        assert error.status_code == 403
+        assert json.loads(error.body)["error"] == "insufficient_scope"
+
+    async def test_token_with_the_route_scope_passes(self):
+        with patch.object(asgi_auth, "handle_auth", return_value=dict(_PAT_CLAIMS)):
+            decoded, error = await asgi_auth.authenticate(_request(), pat_scope="chat:run")
+        assert error is None
+        assert decoded["sub"] == "alice"
+
+    async def test_token_without_the_route_scope_is_refused(self):
+        claims = dict(_PAT_CLAIMS, scopes=["agents:read"])
+        with patch.object(asgi_auth, "handle_auth", return_value=claims):
+            decoded, error = await asgi_auth.authenticate(_request(), pat_scope="chat:run")
+        assert decoded is None
+        assert error.status_code == 403
+
+    async def test_session_denylist_is_not_consulted_for_tokens(self, monkeypatch):
+        monkeypatch.setattr(asgi_auth.settings, "AUTH_TYPE", "oidc")
+        with patch.object(asgi_auth, "handle_auth", return_value=dict(_PAT_CLAIMS)), patch.object(
+            asgi_auth, "oidc_session_denied", return_value=True
+        ) as denied:
+            decoded, error = await asgi_auth.authenticate(_request(), pat_scope="chat:run")
+        denied.assert_not_called()
+        assert error is None and decoded is not None
+
+    async def test_restricted_token_is_refused_even_with_the_scope(self):
+        claims = dict(_PAT_CLAIMS, resource_filter={"agents": ["a1"]})
+        with patch.object(asgi_auth, "handle_auth", return_value=claims):
+            decoded, error = await asgi_auth.authenticate(_request(), pat_scope="chat:run")
+        assert decoded is None
+        assert error.status_code == 403
+        assert json.loads(error.body)["error"] == "resource_not_allowed"
+
+    async def test_any_of_several_scopes_admits_the_token(self):
+        claims = dict(_PAT_CLAIMS, scopes=["conversations:read"])
+        with patch.object(asgi_auth, "handle_auth", return_value=claims):
+            decoded, error = await asgi_auth.authenticate(
+                _request(), pat_scope=("conversations:read", "chat:run")
+            )
+        assert error is None and decoded["sub"] == "alice"
+
+    async def test_message_events_accepts_the_same_scopes_as_message_tail(self):
+        from docsgpt.api import async_sse
+        from docsgpt.api.pat import rules
+
+        tail = rules.RULES[("/api/messages/<string:message_id>/tail", "GET")]
+        assert tail.scopes == rules.MESSAGE_REPLAY_SCOPES
+        assert async_sse.MESSAGE_REPLAY_SCOPES is rules.MESSAGE_REPLAY_SCOPES

tests/api/test_pat_routes.py

@@ -0,0 +1,393 @@
+"""Endpoint tests for personal access token management (/api/user/tokens, admin)."""
+
+from __future__ import annotations
+
+import json
+from contextlib import contextmanager
+from unittest.mock import patch
+
+import pytest
+from sqlalchemy import text
+
+from docsgpt.api.pat import routes as pat_routes
+from docsgpt.api.pat import tokens as pat_tokens
+
+AGENT_A = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
+
+
+@pytest.fixture
+def client():
+    from docsgpt.app import app
+
+    app.config["TESTING"] = True
+    return app.test_client()
+
+
+@pytest.fixture(autouse=True)
+def _policy(monkeypatch):
+    monkeypatch.setattr(pat_tokens.settings, "AUTH_TYPE", "oidc")
+    monkeypatch.setattr(pat_tokens.settings, "PAT_ENABLED", True)
+    monkeypatch.setattr(pat_tokens.settings, "PAT_DEFAULT_LIFETIME_DAYS", 90)
+    monkeypatch.setattr(pat_tokens.settings, "PAT_MAX_LIFETIME_DAYS", 365)
+    monkeypatch.setattr(pat_tokens.settings, "PAT_ALLOW_NON_EXPIRING", False)
+    monkeypatch.setattr(pat_tokens.settings, "PAT_MAX_PER_USER", 25)
+
+
+@pytest.fixture
+def db(pg_conn):
+    """Route every session the token code opens onto the test's rolled-back connection."""
+
+    @contextmanager
+    def _yield_conn():
+        yield pg_conn
+
+    with patch.object(pat_routes, "db_session", _yield_conn), patch.object(
+        pat_routes, "db_readonly", _yield_conn
+    ), patch.object(pat_tokens, "db_session", _yield_conn), patch.object(
+        pat_tokens, "db_readonly", _yield_conn
+    ):
+        yield pg_conn
+
+
+@contextmanager
+def _session(sub="alice", roles=("user",)):
+    with patch("docsgpt.app.handle_auth", return_value={"sub": sub}), patch(
+        "docsgpt.app.resolve_roles", return_value=list(roles)
+    ), patch("docsgpt.app.oidc_session_denied", return_value=False):
+        yield
+
+
+def _create(client, **body):
+    body.setdefault("name", "ci")
+    body.setdefault("scopes", ["agents:write"])
+    with _session():
+        return client.post("/api/user/tokens", json=body)
+
+
+class TestCreate:
+    def test_returns_the_secret_once_and_stores_only_its_hash(self, client, db):
+        response = _create(client)
+        assert response.status_code == 201
+        body = json.loads(response.data)
+        token = body["token"]
+        assert token.startswith("dgpt_pat_")
+        public = body["personal_access_token"]
+        assert public["token_prefix"] == token[:15]
+        assert "token" not in public and "token_hash" not in public
+        stored = db.execute(text("SELECT token_hash FROM personal_access_tokens")).scalar_one()
+        assert stored == pat_tokens.hash_token(token)
+        assert token not in stored
+
+        with _session():
+            listed = json.loads(client.get("/api/user/tokens").data)
+        assert [t["name"] for t in listed["tokens"]] == ["ci"]
+        assert token not in json.dumps(listed)
+
+    def test_default_expiry_applies(self, client, db):
+        body = json.loads(_create(client).data)
+        assert body["personal_access_token"]["expires_at"] is not None
+
+    def test_non_expiring_needs_operator_opt_in(self, client, db, monkeypatch):
+        assert _create(client, expires_in_days=0).status_code == 400
+        monkeypatch.setattr(pat_tokens.settings, "PAT_ALLOW_NON_EXPIRING", True)
+        response = _create(client, expires_in_days=0)
+        assert response.status_code == 201
+        assert json.loads(response.data)["personal_access_token"]["expires_at"] is None
+
+    def test_lifetime_cap(self, client, db):
+        assert _create(client, expires_in_days=366).status_code == 400
+
+    def test_resource_filter_is_stored(self, client, db):
+        response = _create(client, resource_filter={"agents": [AGENT_A]})
+        assert json.loads(response.data)["personal_access_token"]["resource_filter"] == {
+            "agents": [AGENT_A]
+        }
+
+    @pytest.mark.parametrize(
+        "body",
+        [
+            {"name": ""},
+            {"name": "x" * 101},
+            {"scopes": []},
+            {"scopes": ["admin:all"]},
+            {"resource_filter": {"agents": ["nope"]}},
+            {"resource_filter": {"sources": [AGENT_A]}},
+            {"expires_in_days": "soon"},
+        ],
+    )
+    def test_validation(self, client, db, body):
+        assert _create(client, **body).status_code == 400
+
+    @pytest.mark.parametrize("body", [[1], "text", 5])
+    def test_non_object_body_is_a_client_error(self, client, db, body):
+        with _session():
+            assert client.post("/api/user/tokens", json=body).status_code == 400
+
+    def test_duplicate_name_conflicts(self, client, db):
+        assert _create(client).status_code == 201
+        assert _create(client).status_code == 409
+
+    def test_expired_token_does_not_reserve_its_name(self, client, db):
+        assert _create(client).status_code == 201
+        db.execute(text("UPDATE personal_access_tokens SET expires_at = now() - interval '1 day'"))
+        assert _create(client).status_code == 201
+        rows = db.execute(
+            text("SELECT status, revoke_reason FROM personal_access_tokens ORDER BY created_at")
+        ).all()
+        assert [tuple(r) for r in rows] == [("revoked", "expired"), ("active", None)]
+
+    def test_per_user_cap(self, client, db, monkeypatch):
+        monkeypatch.setattr(pat_tokens.settings, "PAT_MAX_PER_USER", 1)
+        assert _create(client, name="one").status_code == 201
+        assert _create(client, name="two").status_code == 409
+
+    @pytest.mark.parametrize("auth_type", ["simple_jwt", "session_jwt"])
+    def test_unavailable_without_a_stable_identity(self, client, db, monkeypatch, auth_type):
+        monkeypatch.setattr(pat_tokens.settings, "AUTH_TYPE", auth_type)
+        assert _create(client).status_code == 403
+
+    def test_disabled_by_operator(self, client, db, monkeypatch):
+        monkeypatch.setattr(pat_tokens.settings, "PAT_ENABLED", False)
+        assert _create(client).status_code == 403
+        with _session():
+            assert json.loads(client.get("/api/user/tokens").data)["policy"]["enabled"] is False
+
+    def test_requires_a_session(self, client, db):
+        with patch("docsgpt.app.handle_auth", return_value=None):
+            assert client.post("/api/user/tokens", json={"name": "x", "scopes": ["agents:read"]}).status_code == 401
+
+    def test_audited(self, client, db):
+        _create(client)
+        event, metadata = db.execute(
+            text("SELECT event, metadata FROM auth_events WHERE user_id = 'alice'")
+        ).one()
+        assert event == "pat_created"
+        assert metadata["scopes"] == ["agents:write"]
+        assert "dgpt_pat_" not in json.dumps(metadata)
+
+
+class TestExpiredStatus:
+    def test_expired_token_is_reported_as_expired_not_active(self, client, db):
+        _create(client)
+        with _session():
+            assert json.loads(client.get("/api/user/tokens").data)["tokens"][0]["status"] == "active"
+        db.execute(text("UPDATE personal_access_tokens SET expires_at = now() - interval '1 day'"))
+        with _session():
+            assert json.loads(client.get("/api/user/tokens").data)["tokens"][0]["status"] == "expired"
+
+
+class TestList:
+    def test_includes_scope_catalog_and_policy(self, client, db):
+        with _session():
+            body = json.loads(client.get("/api/user/tokens").data)
+        assert {s["name"] for s in body["scopes"]} == set(pat_tokens.SCOPES)
+        assert body["policy"] == {
+            "enabled": True,
+            "default_lifetime_days": 90,
+            "max_lifetime_days": 365,
+            "allow_non_expiring": False,
+            "max_per_user": 25,
+            "filterable_families": list(pat_tokens.FILTERABLE_FAMILIES),
+        }
+
+    def test_is_owner_scoped(self, client, db):
+        _create(client)
+        with _session(sub="bob"):
+            assert json.loads(client.get("/api/user/tokens").data)["tokens"] == []
+
+
+class TestEndToEnd:
+    def test_created_token_authenticates_and_revocation_is_immediate(self, client, db):
+        created = json.loads(_create(client, scopes=["prompts:read"]).data)
+        headers = {"Authorization": f"Bearer {created['token']}"}
+
+        me = client.get("/api/user/me", headers=headers)
+        assert me.status_code == 200
+        body = json.loads(me.data)
+        assert body["user_id"] == "alice"
+        assert body["token"]["scopes"] == ["prompts:read"]
+
+        # Scoped: cannot list agents, cannot manage tokens.
+        assert client.get("/api/get_agents", headers=headers).status_code == 403
+        assert client.get("/api/user/tokens", headers=headers).status_code == 403
+        assert client.post("/api/user/tokens", headers=headers, json={}).status_code == 403
+
+        with _session():
+            token_id = created["personal_access_token"]["id"]
+            assert client.delete(f"/api/user/tokens/{token_id}").status_code == 200
+        assert client.get("/api/user/me", headers=headers).status_code == 401
+
+    def test_tampered_token_is_rejected(self, client, db):
+        token = json.loads(_create(client).data)["token"]
+        response = client.get("/api/user/me", headers={"Authorization": f"Bearer {token}x"})
+        assert response.status_code == 401
+
+
+class TestRegenerate:
+    def _regen(self, client, token_id, **body):
+        with _session():
+            return client.post(f"/api/user/tokens/{token_id}/regenerate", json=body)
+
+    def test_new_secret_works_old_one_stops_and_the_rest_is_kept(self, client, db):
+        created = json.loads(
+            _create(client, scopes=["prompts:read"], resource_filter={"prompts": [AGENT_A]}).data
+        )
+        old, token_id = created["token"], created["personal_access_token"]["id"]
+        response = self._regen(client, token_id)
+        assert response.status_code == 200
+        body = json.loads(response.data)
+        new, public = body["token"], body["personal_access_token"]
+        assert new.startswith("dgpt_pat_") and new != old
+        assert public["id"] == token_id and public["name"] == "ci"
+        assert public["scopes"] == ["prompts:read"]
+        assert public["resource_filter"] == {"prompts": [AGENT_A]}
+        assert public["token_prefix"] == new[:15]
+        assert public["regenerated_at"] is not None
+        assert public["last_used_at"] is None
+
+        assert client.get("/api/user/me", headers={"Authorization": f"Bearer {old}"}).status_code == 401
+        me = client.get("/api/user/me", headers={"Authorization": f"Bearer {new}"})
+        assert me.status_code == 200
+        assert json.loads(me.data)["token"]["id"] == token_id
+        stored = db.execute(text("SELECT token_hash FROM personal_access_tokens")).scalar_one()
+        assert stored == pat_tokens.hash_token(new)
+
+    def test_expiry_is_reset_to_the_original_lifetime(self, client, db):
+        token_id = json.loads(_create(client, expires_in_days=30).data)["personal_access_token"]["id"]
+        # 20 days in: 10 days left.
+        db.execute(
+            text(
+                "UPDATE personal_access_tokens SET created_at = now() - interval '20 days', "
+                "expires_at = now() + interval '10 days'"
+            )
+        )
+        self._regen(client, token_id)
+        days_left = db.execute(
+            text("SELECT extract(epoch FROM expires_at - now()) / 86400 FROM personal_access_tokens")
+        ).scalar_one()
+        assert 29.9 < float(days_left) < 30.1
+
+    def test_explicit_lifetime_is_honoured_and_capped(self, client, db):
+        token_id = json.loads(_create(client).data)["personal_access_token"]["id"]
+        assert self._regen(client, token_id, expires_in_days=7).status_code == 200
+        days_left = db.execute(
+            text("SELECT extract(epoch FROM expires_at - now()) / 86400 FROM personal_access_tokens")
+        ).scalar_one()
+        assert 6.9 < float(days_left) < 7.1
+        assert self._regen(client, token_id, expires_in_days=366).status_code == 400
+        assert self._regen(client, token_id, expires_in_days=0).status_code == 400
+
+    def test_an_expired_token_can_be_renewed(self, client, db):
+        created = json.loads(_create(client, expires_in_days=30).data)
+        db.execute(
+            text(
+                "UPDATE personal_access_tokens SET created_at = now() - interval '31 days', "
+                "expires_at = now() - interval '1 day'"
+            )
+        )
+        old_headers = {"Authorization": f"Bearer {created['token']}"}
+        assert client.get("/api/user/me", headers=old_headers).status_code == 401
+        body = json.loads(self._regen(client, created["personal_access_token"]["id"]).data)
+        assert body["personal_access_token"]["status"] == "active"
+        headers = {"Authorization": f"Bearer {body['token']}"}
+        assert client.get("/api/user/me", headers=headers).status_code == 200
+
+    def test_non_expiring_token_stays_non_expiring_only_while_the_operator_allows_it(
+        self, client, db, monkeypatch
+    ):
+        monkeypatch.setattr(pat_tokens.settings, "PAT_ALLOW_NON_EXPIRING", True)
+        token_id = json.loads(_create(client, expires_in_days=0).data)["personal_access_token"]["id"]
+        assert json.loads(self._regen(client, token_id).data)["personal_access_token"]["expires_at"] is None
+        monkeypatch.setattr(pat_tokens.settings, "PAT_ALLOW_NON_EXPIRING", False)
+        renewed = json.loads(self._regen(client, token_id).data)["personal_access_token"]
+        assert renewed["expires_at"] is not None  # falls back to the default lifetime
+
+    def test_revoked_foreign_unknown_and_malformed_tokens_are_not_found(self, client, db):
+        token_id = json.loads(_create(client).data)["personal_access_token"]["id"]
+        with _session(sub="bob"):
+            assert client.post(f"/api/user/tokens/{token_id}/regenerate").status_code == 404
+        with _session():
+            client.delete(f"/api/user/tokens/{token_id}")
+        assert self._regen(client, token_id).status_code == 404
+        assert self._regen(client, AGENT_A).status_code == 404
+        assert self._regen(client, f"urn:uuid:{AGENT_A}").status_code == 404
+
+    def test_a_token_cannot_regenerate_itself_or_others(self, client, db):
+        created = json.loads(_create(client, scopes=list(pat_tokens.SCOPES)).data)
+        headers = {"Authorization": f"Bearer {created['token']}"}
+        token_id = created["personal_access_token"]["id"]
+        response = client.post(f"/api/user/tokens/{token_id}/regenerate", headers=headers)
+        assert response.status_code == 403
+        assert json.loads(response.data)["error"] == "not_available_to_tokens"
+
+    def test_requires_a_session_and_an_object_body(self, client, db):
+        token_id = json.loads(_create(client).data)["personal_access_token"]["id"]
+        with patch("docsgpt.app.handle_auth", return_value=None):
+            assert client.post(f"/api/user/tokens/{token_id}/regenerate").status_code == 401
+        with _session():
+            assert client.post(f"/api/user/tokens/{token_id}/regenerate", json=[1]).status_code == 400
+
+    def test_audited_without_the_secret(self, client, db):
+        token_id = json.loads(_create(client).data)["personal_access_token"]["id"]
+        self._regen(client, token_id)
+        metadata = db.execute(
+            text("SELECT metadata FROM auth_events WHERE event = 'pat_regenerated'")
+        ).scalar_one()
+        assert metadata["token_id"] == token_id
+        assert "dgpt_pat_" not in json.dumps(metadata)
+
+
+class TestRevoke:
+    def test_cannot_revoke_someone_elses_token(self, client, db):
+        token_id = json.loads(_create(client).data)["personal_access_token"]["id"]
+        with _session(sub="bob"):
+            assert client.delete(f"/api/user/tokens/{token_id}").status_code == 404
+
+    @pytest.mark.parametrize(
+        "token_id",
+        [AGENT_A, "not-a-uuid", f"urn:uuid:{AGENT_A}", "{" + AGENT_A + "}", AGENT_A.replace("-", "")],
+    )
+    def test_unknown_and_malformed_ids(self, client, db, token_id):
+        # uuid.UUID() accepts urn:/braced/bare-hex spellings that Postgres rejects; none may reach the cast.
+        with _session():
+            assert client.delete(f"/api/user/tokens/{token_id}").status_code == 404
+        with _session(sub="root", roles=("admin", "user")):
+            assert client.delete(f"/api/admin/tokens/{token_id}").status_code == 404
+
+
+class TestAdmin:
+    def test_requires_admin(self, client, db):
+        with _session():
+            assert client.get("/api/admin/users/alice/tokens").status_code == 403
+            assert client.delete(f"/api/admin/tokens/{AGENT_A}").status_code == 403
+
+    def test_admin_can_list_and_revoke_any_token(self, client, db):
+        created = json.loads(_create(client).data)
+        token_id = created["personal_access_token"]["id"]
+        with _session(sub="root", roles=("admin", "user")):
+            listed = json.loads(client.get("/api/admin/users/alice/tokens").data)
+            assert [t["id"] for t in listed["tokens"]] == [token_id]
+            assert client.delete(f"/api/admin/tokens/{token_id}").status_code == 200
+            assert client.delete(f"/api/admin/tokens/{token_id}").status_code == 404
+        headers = {"Authorization": f"Bearer {created['token']}"}
+        assert client.get("/api/user/me", headers=headers).status_code == 401
+
+    def test_revoke_sessions_also_revokes_tokens(self, client, db):
+        from docsgpt.api.admin import routes as admin_routes
+
+        @contextmanager
+        def _yield_conn():
+            yield db
+
+        created = json.loads(_create(client).data)
+        with _session(sub="root", roles=("admin", "user")), patch.object(
+            admin_routes, "db_session", _yield_conn
+        ), patch.object(admin_routes.denylist, "deny_user", return_value=True):
+            assert client.post("/api/admin/users/alice/revoke-sessions").status_code == 200
+        headers = {"Authorization": f"Bearer {created['token']}"}
+        assert client.get("/api/user/me", headers=headers).status_code == 401
+        events = db.execute(
+            text("SELECT metadata FROM auth_events WHERE user_id = 'alice' AND event = 'pat_revoked'")
+        ).all()
+        assert [e[0]["token_id"] for e in events] == [created["personal_access_token"]["id"]]
+        assert events[0][0]["via"] == "admin_sessions_revoked"

tests/api/test_pat_rules.py

@@ -0,0 +1,448 @@
+"""The PAT rule table: route classification and enforcement at the Flask chokepoint."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import patch
+
+import pytest
+
+from docsgpt.api.pat import rules
+from docsgpt.api.pat.tokens import SCOPES, expand_scopes
+
+AGENT_A = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
+AGENT_B = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"
+SOURCE_A = "cccccccc-cccc-cccc-cccc-cccccccccccc"
+SOURCE_B = "dddddddd-dddd-dddd-dddd-dddddddddddd"
+
+
+@pytest.fixture(scope="module")
+def flask_app():
+    from docsgpt.app import app
+
+    app.config["TESTING"] = True
+    return app
+
+
+@pytest.fixture
+def client(flask_app):
+    return flask_app.test_client()
+
+
+def _claims(scopes, resource_filter=None):
+    return {
+        "sub": "alice",
+        "auth_method": "pat",
+        "pat_id": "t1",
+        "pat_name": "ci",
+        "scopes": sorted(expand_scopes(scopes)),
+        "resource_filter": resource_filter or {},
+    }
+
+
+def _call(client, method, path, claims, **kwargs):
+    """Send a request as a PAT; returns the response, short-circuiting the view."""
+    with patch("docsgpt.app.handle_auth", return_value=claims):
+        return client.open(path, method=method, **kwargs)
+
+
+def _denied(response):
+    """The rule table refused (as opposed to the view answering 4xx itself)."""
+    if response.status_code != 403:
+        return None
+    return (json.loads(response.data) or {}).get("error")
+
+
+@pytest.mark.unit
+class TestClassification:
+    def test_every_route_is_classified(self, flask_app):
+        unclassified = []
+        for url_rule in flask_app.url_map.iter_rules():
+            for method in sorted(url_rule.methods - {"HEAD", "OPTIONS"}):
+                if (url_rule.rule, method) in rules.RULES:
+                    continue
+                if rules.is_denied(url_rule.rule, method):
+                    continue
+                unclassified.append(f"{method} {url_rule.rule}")
+        assert not unclassified, (
+            "New routes must be given a scope in docsgpt/api/pat/rules.py RULES, "
+            f"or listed in DENIED: {unclassified}"
+        )
+
+    def test_no_stale_rules(self, flask_app):
+        registered = {
+            (r.rule, m) for r in flask_app.url_map.iter_rules() for m in r.methods
+        }
+        assert not [key for key in rules.RULES if key not in registered]
+        known_rules = {r.rule for r in flask_app.url_map.iter_rules()}
+        assert not [rule for rule in rules.DENIED if rule not in known_rules]
+
+    def test_a_route_is_never_both_allowed_and_denied(self):
+        assert not [key for key in rules.RULES if rules.is_denied(*key)]
+
+    def test_rules_only_name_real_scopes(self):
+        for key, rule in rules.RULES.items():
+            for scope in rule.scopes:
+                assert scope in SCOPES, (key, scope)
+
+    @pytest.mark.parametrize(
+        "rule,method",
+        [
+            ("/api/user/tokens", "POST"),
+            ("/api/user/tokens", "GET"),
+            ("/api/user/tokens/<string:token_id>", "DELETE"),
+            ("/api/user/tokens/<string:token_id>/regenerate", "POST"),
+            ("/api/admin/users", "GET"),
+            ("/api/admin/tokens/<string:token_id>", "DELETE"),
+            ("/api/generate_token", "GET"),
+            ("/api/devices/pairings", "POST"),
+            ("/api/connectors/auth", "GET"),
+            ("/api/mcp_server/callback", "GET"),
+        ],
+    )
+    def test_sensitive_routes_are_never_token_reachable(self, rule, method):
+        assert (rule, method) not in rules.RULES
+        assert rules.is_denied(rule, method)
+
+
+@pytest.mark.unit
+class TestScopeEnforcement:
+    def test_unlisted_route_is_refused(self, client):
+        response = _call(client, "GET", "/api/user/tokens", _claims(list(SCOPES)))
+        assert _denied(response) == "not_available_to_tokens"
+
+    def test_admin_routes_are_refused_even_with_every_scope(self, client):
+        response = _call(client, "GET", "/api/admin/users", _claims(list(SCOPES)))
+        assert _denied(response) == "not_available_to_tokens"
+
+    def test_unknown_path_and_wrong_method_keep_their_own_status(self, client):
+        claims = _claims(list(SCOPES))
+        assert _call(client, "GET", "/api/no_such_route", claims).status_code == 404
+        assert _call(client, "DELETE", "/api/get_agents", claims).status_code == 405
+
+    def test_missing_scope_is_refused_and_names_the_scope(self, client):
+        response = _call(client, "GET", "/api/get_agents", _claims(["sources:read"]))
+        assert _denied(response) == "insufficient_scope"
+        assert json.loads(response.data)["required_scope"] == "agents:read"
+
+    def test_read_scope_cannot_write(self, client):
+        response = _call(
+            client, "DELETE", f"/api/delete_agent?id={AGENT_A}", _claims(["agents:read"])
+        )
+        assert _denied(response) == "insufficient_scope"
+
+    def test_write_scope_can_read(self, client):
+        response = _call(client, "GET", "/api/get_agents", _claims(["agents:write"]))
+        assert _denied(response) is None
+
+    def test_agent_key_regeneration_needs_its_own_scope(self, client):
+        response = _call(
+            client, "POST", f"/api/regenerate_agent_key/{AGENT_A}", _claims(["agents:write"])
+        )
+        assert _denied(response) == "insufficient_scope"
+
+    def test_any_token_can_identify_itself(self, client):
+        response = _call(client, "GET", "/api/user/me", _claims(["prompts:read"]))
+        assert response.status_code == 200
+        body = json.loads(response.data)
+        assert body["auth_method"] == "pat"
+        assert body["token"]["scopes"] == ["prompts:read"]
+        assert body["roles"] == ["user"]
+
+    def test_token_never_carries_admin(self, client):
+        with patch("docsgpt.app.resolve_roles", return_value=["admin", "user"]) as resolver:
+            response = _call(client, "GET", "/api/user/me", _claims(["agents:read"]))
+        resolver.assert_not_called()
+        assert json.loads(response.data)["roles"] == ["user"]
+
+    def test_session_callers_bypass_the_table(self, client):
+        with patch("docsgpt.app.handle_auth", return_value={"sub": "alice"}), patch(
+            "docsgpt.app.resolve_roles", return_value=["user"]
+        ), patch("docsgpt.app.authorize_pat") as authorize:
+            client.get("/api/user/me")
+        authorize.assert_not_called()
+
+    def test_invalid_token_is_401(self, client):
+        with patch(
+            "docsgpt.app.handle_auth",
+            return_value={"error": "invalid_token", "message": "Authentication error: invalid token"},
+        ):
+            assert client.get("/api/get_agents").status_code == 401
+
+
+@pytest.mark.unit
+class TestResourceRestrictions:
+    def _restricted(self, scopes, **families):
+        return _claims(scopes, resource_filter=families)
+
+    def test_allowed_id_passes_and_other_id_is_refused(self, client):
+        claims = self._restricted(["agents:read"], agents=[AGENT_A])
+        assert _denied(_call(client, "GET", f"/api/get_agent?id={AGENT_A}", claims)) is None
+        assert (
+            _denied(_call(client, "GET", f"/api/get_agent?id={AGENT_B}", claims))
+            == "resource_not_allowed"
+        )
+
+    def test_id_comparison_ignores_case(self, client):
+        claims = self._restricted(["agents:read"], agents=[AGENT_A])
+        assert _denied(_call(client, "GET", f"/api/get_agent?id={AGENT_A.upper()}", claims)) is None
+
+    def test_view_arg_ids(self, client):
+        claims = self._restricted(["agents:write"], agents=[AGENT_A])
+        ok = _call(client, "PUT", f"/api/update_agent/{AGENT_A}", claims, json={"name": "x"})
+        bad = _call(client, "PUT", f"/api/update_agent/{AGENT_B}", claims, json={"name": "x"})
+        assert _denied(ok) is None
+        assert _denied(bad) == "resource_not_allowed"
+
+    def test_json_body_ids_including_lists(self, client):
+        claims = self._restricted(["agents:write"], agents=[AGENT_A])
+        ok = _call(client, "POST", "/api/agents/folders/bulk_move", claims, json={"agent_ids": [AGENT_A]})
+        bad = _call(
+            client, "POST", "/api/agents/folders/bulk_move", claims, json={"agent_ids": [AGENT_A, AGENT_B]}
+        )
+        assert _denied(ok) is None
+        assert _denied(bad) == "resource_not_allowed"
+
+    def test_missing_id_is_refused_for_a_restricted_token(self, client):
+        claims = self._restricted(["agents:read"], agents=[AGENT_A])
+        assert _denied(_call(client, "GET", "/api/get_agent", claims)) == "resource_not_allowed"
+
+    def test_restricted_token_cannot_create(self, client):
+        claims = self._restricted(["agents:write"], agents=[AGENT_A])
+        response = _call(client, "POST", "/api/create_agent", claims, json={"name": "new"})
+        assert _denied(response) == "resource_not_allowed"
+
+    def test_unrestricted_family_is_untouched(self, client):
+        claims = self._restricted(["agents:write", "prompts:write"], agents=[AGENT_A])
+        response = _call(client, "POST", "/api/create_prompt", claims, json={})
+        assert _denied(response) is None
+
+    def test_cross_family_references_are_checked(self, client):
+        claims = self._restricted(["agents:write", "sources:read"], sources=[SOURCE_A])
+        ok = _call(client, "PUT", f"/api/update_agent/{AGENT_A}", claims, json={"source": SOURCE_A})
+        bad = _call(client, "PUT", f"/api/update_agent/{AGENT_A}", claims, json={"sources": [SOURCE_A, SOURCE_B]})
+        form = _call(
+            client, "PUT", f"/api/update_agent/{AGENT_A}", claims,
+            data={"sources": json.dumps([SOURCE_B])},
+        )
+        assert _denied(ok) is None
+        assert _denied(bad) == "resource_not_allowed"
+        assert _denied(form) == "resource_not_allowed"
+
+    def test_routes_hanging_off_a_restricted_family_are_blocked(self, client):
+        claims = self._restricted(["agents:read", "schedules:read", "analytics:read"], agents=[AGENT_A])
+        assert _denied(_call(client, "GET", "/api/schedules/s1", claims)) == "resource_not_allowed"
+        assert _denied(_call(client, "POST", "/api/get_token_analytics", claims, json={})) == "resource_not_allowed"
+        assert _denied(_call(client, "GET", f"/api/agents/{AGENT_A}/schedules", claims)) is None
+        assert _denied(_call(client, "GET", f"/api/agents/{AGENT_B}/schedules", claims)) == "resource_not_allowed"
+
+    @pytest.mark.parametrize("family", ["agents", "sources", "prompts", "tools", "workflows"])
+    def test_message_tail_is_closed_to_any_restricted_token(self, client, family):
+        claims = _claims(["chat:run"], {family: [AGENT_A]})
+        assert _denied(_call(client, "GET", "/api/messages/m1/tail", claims)) == "resource_not_allowed"
+        assert _denied(_call(client, "GET", "/api/messages/m1/tail", _claims(["chat:run"]))) is None
+
+    def test_sql_paged_listing_is_closed_to_restricted_tokens(self, client):
+        claims = self._restricted(["sources:read"], sources=[SOURCE_A])
+        assert _denied(_call(client, "GET", "/api/sources/paginated", claims)) == "resource_not_allowed"
+
+
+WORKFLOW_A = "eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee"
+TOOL_A = "ffffffff-ffff-ffff-ffff-ffffffffffff"
+
+
+@pytest.mark.unit
+class TestRelationshipsBeyondIds:
+    """Rows whose content or parent the table cannot see are closed to restricted tokens."""
+
+    @pytest.mark.parametrize("family", ["sources", "tools", "prompts"])
+    def test_workflow_writes_are_closed_to_tokens_restricted_on_what_a_graph_can_name(self, client, family):
+        claims = _claims(["workflows:write"], {family: [SOURCE_A]})
+        assert _denied(_call(client, "POST", "/api/workflows", claims, json={})) == "resource_not_allowed"
+        assert (
+            _denied(_call(client, "PUT", f"/api/workflows/{WORKFLOW_A}", claims, json={}))
+            == "resource_not_allowed"
+        )
+        assert _denied(_call(client, "GET", f"/api/workflows/{WORKFLOW_A}", claims)) is None
+
+    def test_workflow_restricted_token_can_still_edit_its_workflows(self, client):
+        claims = _claims(["workflows:write"], {"workflows": [WORKFLOW_A]})
+        assert _denied(_call(client, "PUT", f"/api/workflows/{WORKFLOW_A}", claims, json={})) is None
+
+    @pytest.mark.parametrize("family", ["sources", "tools", "prompts"])
+    def test_agent_cannot_be_pointed_at_a_workflow_by_a_token_restricted_on_its_contents(self, client, family):
+        claims = _claims(["agents:write"], {family: [SOURCE_A]})
+        for body in ({"workflow": WORKFLOW_A}, {"workflow": {"id": WORKFLOW_A}}):
+            response = _call(client, "PUT", f"/api/update_agent/{AGENT_A}", claims, json=body)
+            assert _denied(response) == "resource_not_allowed"
+        form = _call(client, "PUT", f"/api/update_agent/{AGENT_A}", claims, data={"workflow": WORKFLOW_A})
+        assert _denied(form) == "resource_not_allowed"
+        assert _denied(_call(client, "PUT", f"/api/update_agent/{AGENT_A}", claims, json={"name": "x"})) is None
+
+    def test_workflow_allowlist_lets_the_agent_use_those_workflows_only(self, client):
+        claims = _claims(["agents:write"], {"sources": [SOURCE_A], "workflows": [WORKFLOW_A]})
+        ok = _call(client, "PUT", f"/api/update_agent/{AGENT_A}", claims, json={"workflow": WORKFLOW_A})
+        bad = _call(client, "PUT", f"/api/update_agent/{AGENT_A}", claims, json={"workflow": SOURCE_B})
+        assert _denied(ok) is None
+        assert _denied(bad) == "resource_not_allowed"
+
+    @pytest.mark.parametrize("family", ["sources", "prompts", "tools", "workflows"])
+    def test_schedules_are_closed_to_tokens_restricted_on_anything_but_agents(self, client, family):
+        claims = _claims(["schedules:write"], {family: [SOURCE_A]})
+        for method, path in (
+            ("GET", f"/api/agents/{AGENT_A}/schedules"),
+            ("POST", f"/api/agents/{AGENT_A}/schedules"),
+            ("GET", "/api/schedules/s1"),
+            ("POST", "/api/schedules/s1/run"),
+            ("GET", "/api/schedules/s1/runs"),
+        ):
+            assert _denied(_call(client, method, path, claims, json={})) == "resource_not_allowed", path
+
+    @pytest.mark.parametrize("family", ["agents", "sources", "prompts", "tools", "workflows"])
+    def test_conversations_and_analytics_are_closed_to_every_restricted_token(self, client, family):
+        claims = _claims(["conversations:write", "analytics:read"], {family: [SOURCE_A]})
+        for method, path in (
+            ("GET", "/api/get_conversations"),
+            ("GET", "/api/get_single_conversation?id=c1"),
+            ("GET", "/api/search_conversations?q=x"),
+            ("POST", "/api/delete_conversation"),
+            ("POST", "/api/feedback"),
+            ("POST", "/api/get_message_analytics"),
+            ("POST", "/api/get_user_logs"),
+        ):
+            assert _denied(_call(client, method, path, claims, json={})) == "resource_not_allowed", path
+
+
+@pytest.mark.unit
+class TestAgentKeyVisibility:
+    def _request(self, flask_app, claims):
+        from flask import request
+
+        ctx = flask_app.test_request_context("/")
+        ctx.push()
+        request.decoded_token = claims
+        return ctx, request
+
+    def test_sessions_and_tokens_with_the_keys_scope_see_the_key(self, flask_app):
+        for claims in ({"sub": "alice"}, _claims(["agents:write", "agents:keys"])):
+            ctx, request = self._request(flask_app, claims)
+            try:
+                assert rules.may_see_agent_keys(request) is True
+            finally:
+                ctx.pop()
+
+    def test_token_without_the_keys_scope_does_not(self, flask_app):
+        ctx, request = self._request(flask_app, _claims(["agents:write"]))
+        try:
+            assert rules.may_see_agent_keys(request) is False
+        finally:
+            ctx.pop()
+
+    def test_mask(self):
+        assert rules.mask_agent_key("12345678-aaaa-bbbb-cccc-1234567890ab") == "1234...90ab"
+        assert rules.mask_agent_key("") == "" and rules.mask_agent_key(None) == ""
+
+
+@pytest.mark.unit
+class TestChatRestrictions:
+    def _chat(self, client, claims, body):
+        return _denied(_call(client, "POST", "/api/answer", claims, json=body))
+
+    def test_unrestricted_token_can_chat_any_way(self, client):
+        claims = _claims(["chat:run"])
+        assert self._chat(client, claims, {"question": "hi", "api_key": "k"}) is None
+        assert self._chat(client, claims, {"question": "hi", "agent_id": AGENT_B}) is None
+
+    def test_agent_restricted_token_must_name_an_allowed_agent(self, client):
+        claims = _claims(["chat:run"], {"agents": [AGENT_A]})
+        assert self._chat(client, claims, {"question": "hi", "agent_id": AGENT_A}) is None
+        assert self._chat(client, claims, {"question": "hi", "agent_id": AGENT_B}) == "resource_not_allowed"
+        assert self._chat(client, claims, {"question": "hi"}) == "resource_not_allowed"
+        assert self._chat(client, claims, {"question": "hi", "api_key": "k"}) == "resource_not_allowed"
+
+    def test_restricted_token_cannot_run_an_inline_workflow(self, client):
+        claims = _claims(["chat:run"], {"agents": [AGENT_A]})
+        body = {"question": "hi", "agent_id": AGENT_A, "workflow": {"nodes": []}}
+        assert self._chat(client, claims, body) == "resource_not_allowed"
+
+    def test_source_restricted_token_cannot_reach_other_sources_through_an_agent(self, client):
+        claims = _claims(["chat:run"], {"sources": [SOURCE_A]})
+        assert self._chat(client, claims, {"question": "hi", "active_docs": SOURCE_A}) is None
+        assert self._chat(client, claims, {"question": "hi", "active_docs": [SOURCE_A, SOURCE_B]}) == "resource_not_allowed"
+        assert self._chat(client, claims, {"question": "hi", "agent_id": AGENT_A}) == "resource_not_allowed"
+
+    @pytest.mark.parametrize("extra", [{}, {"agents": [AGENT_A]}])
+    def test_tools_restricted_token_cannot_chat_at_all(self, client, extra):
+        claims = _claims(["chat:run"], {"tools": [TOOL_A], **extra})
+        assert self._chat(client, claims, {"question": "hi"}) == "resource_not_allowed"
+        assert self._chat(client, claims, {"question": "hi", "agent_id": AGENT_A}) == "resource_not_allowed"
+
+    @pytest.mark.parametrize("family", ["prompts", "workflows"])
+    def test_agentless_chat_needs_an_agent_restriction_unless_only_sources_are_restricted(self, client, family):
+        claims = _claims(["chat:run"], {family: [SOURCE_A]})
+        assert self._chat(client, claims, {"question": "hi"}) == "resource_not_allowed"
+        assert self._chat(client, claims, {"question": "hi", "agent_id": AGENT_A}) == "resource_not_allowed"
+
+    def test_conversation_must_belong_to_the_agent_being_run(self, client):
+        claims = _claims(["chat:run"], {"agents": [AGENT_A]})
+        body = {"question": "hi", "agent_id": AGENT_A, "conversation_id": "c1"}
+        with patch.object(rules, "_conversation_agent_id", return_value=(True, AGENT_A.upper())) as lookup:
+            assert self._chat(client, claims, body) is None
+        lookup.assert_called_once_with("c1", "alice")
+        for result in ((True, AGENT_B), (True, ""), (False, "")):
+            with patch.object(rules, "_conversation_agent_id", return_value=result):
+                assert self._chat(client, claims, body) == "resource_not_allowed", result
+
+    def test_conversation_resume_with_tool_actions_is_held_to_the_same_rule(self, client):
+        claims = _claims(["chat:run"], {"agents": [AGENT_A]})
+        body = {"agent_id": AGENT_A, "conversation_id": "c1", "tool_actions": [{"call_id": "x"}]}
+        with patch.object(rules, "_conversation_agent_id", return_value=(True, AGENT_B)):
+            assert self._chat(client, claims, body) == "resource_not_allowed"
+
+    def test_agentless_token_cannot_continue_an_agent_conversation(self, client):
+        claims = _claims(["chat:run"], {"sources": [SOURCE_A]})
+        body = {"question": "hi", "active_docs": SOURCE_A, "conversation_id": "c1"}
+        with patch.object(rules, "_conversation_agent_id", return_value=(True, AGENT_B)):
+            assert self._chat(client, claims, body) == "resource_not_allowed"
+        with patch.object(rules, "_conversation_agent_id", return_value=(True, "")):
+            assert self._chat(client, claims, body) is None
+
+    def test_unrestricted_token_never_pays_for_the_conversation_lookup(self, client):
+        with patch.object(rules, "_conversation_agent_id") as lookup:
+            self._chat(client, _claims(["chat:run"]), {"question": "hi", "conversation_id": "c1"})
+        lookup.assert_not_called()
+
+    def test_conversation_lookup_fails_closed(self):
+        with patch("docsgpt.storage.db.session.db_readonly", side_effect=RuntimeError("db down")):
+            assert rules._conversation_agent_id("c1", "alice") == (False, "")
+        assert rules._conversation_agent_id("c1", None) == (False, "")
+
+    def test_retrieval_test_honours_the_source_allowlist(self, client):
+        claims = _claims(["chat:run"], {"sources": [SOURCE_A]})
+        ok = _call(client, "POST", f"/api/sources/{SOURCE_A}/search", claims, json={"query": "q"})
+        bad = _call(client, "POST", f"/api/sources/{SOURCE_B}/search", claims, json={"query": "q"})
+        assert _denied(ok) is None
+        assert _denied(bad) == "resource_not_allowed"
+
+
+@pytest.mark.unit
+class TestListingFilter:
+    def test_unrestricted_and_session_callers_see_everything(self, flask_app):
+        from flask import request
+
+        items = [{"id": AGENT_A}, {"id": AGENT_B}]
+        with flask_app.test_request_context("/"):
+            request.decoded_token = {"sub": "alice"}
+            assert rules.filter_listing(request, "agents", items) == items
+            request.decoded_token = _claims(["agents:read"])
+            assert rules.filter_listing(request, "agents", items) == items
+
+    def test_restricted_token_sees_only_its_rows_plus_builtin_presets(self, flask_app):
+        from flask import request
+
+        items = [{"id": AGENT_A}, {"id": AGENT_B}, {"id": "default"}]
+        with flask_app.test_request_context("/"):
+            request.decoded_token = _claims(["prompts:read"], {"prompts": [AGENT_A]})
+            assert rules.filter_listing(request, "prompts", items) == [{"id": AGENT_A}, {"id": "default"}]
+            assert rules.allowed_ids(request, "agents") is None
+            assert rules.allowed_ids(request, "prompts") == {AGENT_A}

tests/api/test_pat_tokens.py

@@ -0,0 +1,280 @@
+"""Unit tests for docsgpt/api/pat/tokens.py and the PAT branch of handle_auth."""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from datetime import datetime, timedelta, timezone
+from unittest.mock import Mock, patch
+
+import pytest
+
+from docsgpt.api.pat import tokens
+
+
+def _request(authorization=None, ip="10.1.1.1"):
+    request = Mock()
+    request.headers = {"Authorization": authorization} if authorization else {}
+    request.remote_addr = ip
+    return request
+
+
+@contextmanager
+def _db(row):
+    repo = Mock()
+    repo.find_active_by_hash.return_value = row
+
+    @contextmanager
+    def _conn():
+        yield Mock()
+
+    with patch.object(tokens, "db_readonly", _conn), patch.object(
+        tokens, "db_session", _conn
+    ), patch.object(tokens, "PersonalAccessTokensRepository", return_value=repo):
+        yield repo
+
+
+_ROW = {
+    "id": "11111111-1111-1111-1111-111111111111",
+    "user_id": "alice",
+    "name": "ci",
+    "scopes": ["agents:write", "chat:run"],
+    "resource_filter": {"agents": ["22222222-2222-2222-2222-222222222222"]},
+    "last_used_at": None,
+}
+
+
+@pytest.mark.unit
+class TestTokenFormat:
+    def test_generate_returns_prefixed_secret_hash_and_display_prefix(self):
+        token, token_hash, display = tokens.generate_token()
+        assert token.startswith("dgpt_pat_")
+        assert len(token) > 40
+        assert token_hash == tokens.hash_token(token)
+        assert len(token_hash) == 64
+        assert token.startswith(display) and len(display) == len("dgpt_pat_") + 6
+
+    def test_tokens_are_unique(self):
+        assert tokens.generate_token()[0] != tokens.generate_token()[0]
+
+    def test_redact_never_returns_the_secret(self):
+        token, _, display = tokens.generate_token()
+        assert tokens.redact(token) == display + "…"
+        assert tokens.redact("some-agent-key") == "some…"
+        assert tokens.redact(None) == ""
+
+    @pytest.mark.parametrize(
+        "value,expected",
+        [("dgpt_pat_abc", True), ("eyJhbGciOi", False), ("", False), (None, False)],
+    )
+    def test_looks_like_pat(self, value, expected):
+        assert tokens.looks_like_pat(value) is expected
+
+
+@pytest.mark.unit
+class TestScopes:
+    def test_write_implies_read(self):
+        assert tokens.expand_scopes(["agents:write"]) == {"agents:write", "agents:read"}
+
+    def test_standalone_scopes_imply_nothing(self):
+        assert tokens.expand_scopes(["agents:keys", "chat:run"]) == {"agents:keys", "chat:run"}
+
+    def test_normalize_sorts_and_dedupes(self):
+        assert tokens.normalize_scopes(["sources:read", "agents:read", "sources:read"]) == [
+            "agents:read",
+            "sources:read",
+        ]
+
+    @pytest.mark.parametrize("raw", [None, [], "agents:read", ["admin:all"], [1], ["agents:read", "x"]])
+    def test_normalize_rejects_bad_input(self, raw):
+        with pytest.raises(ValueError):
+            tokens.normalize_scopes(raw)
+
+    def test_no_admin_scope_exists(self):
+        assert not any(s.startswith("admin") for s in tokens.SCOPES)
+
+
+@pytest.mark.unit
+class TestResourceFilter:
+    UUID_A = "22222222-2222-2222-2222-222222222222"
+
+    def test_empty_is_unrestricted(self):
+        assert tokens.normalize_resource_filter(None, ["agents:read"]) == {}
+        assert tokens.normalize_resource_filter({}, ["agents:read"]) == {}
+
+    def test_canonicalizes_and_dedupes_ids(self):
+        out = tokens.normalize_resource_filter(
+            {"agents": [self.UUID_A.upper(), self.UUID_A]}, ["agents:read"]
+        )
+        assert out == {"agents": [self.UUID_A]}
+
+    def test_chat_scope_allows_agent_and_source_restrictions(self):
+        out = tokens.normalize_resource_filter(
+            {"agents": [self.UUID_A], "sources": [self.UUID_A]}, ["chat:run"]
+        )
+        assert set(out) == {"agents", "sources"}
+
+    def test_tools_restriction_cannot_be_combined_with_chat(self):
+        with pytest.raises(ValueError, match="chat:run"):
+            tokens.normalize_resource_filter({"tools": [self.UUID_A]}, ["tools:read", "chat:run"])
+        assert tokens.normalize_resource_filter({"tools": [self.UUID_A]}, ["tools:read"])
+
+    @pytest.mark.parametrize(
+        "raw,scopes",
+        [
+            ("agents", ["agents:read"]),
+            ({"conversations": [UUID_A]}, ["conversations:read"]),
+            ({"sources": [UUID_A]}, ["agents:read"]),
+            ({"agents": []}, ["agents:read"]),
+            ({"agents": "all"}, ["agents:read"]),
+            ({"agents": ["not-a-uuid"]}, ["agents:read"]),
+            ({"agents": [UUID_A] * 201}, ["agents:read"]),
+        ],
+    )
+    def test_rejects_bad_filters(self, raw, scopes):
+        with pytest.raises(ValueError):
+            tokens.normalize_resource_filter(raw, scopes)
+
+
+@pytest.mark.unit
+class TestExpiryPolicy:
+    @pytest.fixture(autouse=True)
+    def _policy(self, monkeypatch):
+        monkeypatch.setattr(tokens.settings, "PAT_DEFAULT_LIFETIME_DAYS", 90)
+        monkeypatch.setattr(tokens.settings, "PAT_MAX_LIFETIME_DAYS", 365)
+        monkeypatch.setattr(tokens.settings, "PAT_ALLOW_NON_EXPIRING", False)
+
+    def _days(self, expires_at):
+        return round((expires_at - datetime.now(timezone.utc)) / timedelta(days=1))
+
+    def test_default_lifetime(self):
+        assert self._days(tokens.resolve_expiry(None)) == 90
+
+    def test_explicit_lifetime(self):
+        assert self._days(tokens.resolve_expiry(7)) == 7
+
+    def test_max_lifetime_enforced(self):
+        assert self._days(tokens.resolve_expiry(365)) == 365
+        with pytest.raises(ValueError):
+            tokens.resolve_expiry(366)
+
+    def test_non_expiring_refused_unless_operator_allows(self, monkeypatch):
+        with pytest.raises(ValueError, match="disabled"):
+            tokens.resolve_expiry(0)
+        monkeypatch.setattr(tokens.settings, "PAT_ALLOW_NON_EXPIRING", True)
+        assert tokens.resolve_expiry(0) is None
+
+    @pytest.mark.parametrize("raw", [-1, "30", 1.5, True])
+    def test_rejects_bad_values(self, raw):
+        with pytest.raises(ValueError):
+            tokens.resolve_expiry(raw)
+
+
+@pytest.mark.unit
+class TestAuthenticatePat:
+    @pytest.fixture(autouse=True)
+    def _enabled(self, monkeypatch):
+        monkeypatch.setattr(tokens.settings, "PAT_ENABLED", True)
+        monkeypatch.setattr(tokens.settings, "AUTH_TYPE", "oidc")
+
+    def test_valid_token_yields_claims_from_the_row(self):
+        with _db(dict(_ROW)) as repo:
+            claims = tokens.authenticate_pat("dgpt_pat_secret", _request())
+        repo.find_active_by_hash.assert_called_once_with(tokens.hash_token("dgpt_pat_secret"))
+        assert claims == {
+            "sub": "alice",
+            "auth_method": "pat",
+            "pat_id": _ROW["id"],
+            "pat_name": "ci",
+            "scopes": ["agents:read", "agents:write", "chat:run"],
+            "resource_filter": _ROW["resource_filter"],
+        }
+        repo.touch_last_used.assert_called_once()
+        assert repo.touch_last_used.call_args.args[:2] == (_ROW["id"], "10.1.1.1")
+
+    def test_unknown_token_is_invalid(self):
+        with _db(None):
+            assert tokens.authenticate_pat("dgpt_pat_x", _request())["error"] == "invalid_token"
+
+    def test_lookup_failure_fails_closed(self):
+        with _db(dict(_ROW)) as repo:
+            repo.find_active_by_hash.side_effect = RuntimeError("db down")
+            assert tokens.authenticate_pat("dgpt_pat_x", _request())["error"] == "invalid_token"
+
+    def test_usage_write_failure_never_fails_the_request(self):
+        with _db(dict(_ROW)) as repo:
+            repo.touch_last_used.side_effect = RuntimeError("db down")
+            assert tokens.authenticate_pat("dgpt_pat_x", _request())["sub"] == "alice"
+
+    def test_recent_usage_skips_the_write(self):
+        row = dict(_ROW, last_used_at=datetime.now(timezone.utc).isoformat())
+        with _db(row) as repo:
+            tokens.authenticate_pat("dgpt_pat_x", _request())
+        repo.touch_last_used.assert_not_called()
+
+    @pytest.mark.parametrize("auth_type", ["simple_jwt", "session_jwt"])
+    def test_rejected_where_there_is_no_stable_user_identity(self, monkeypatch, auth_type):
+        monkeypatch.setattr(tokens.settings, "AUTH_TYPE", auth_type)
+        with _db(dict(_ROW)) as repo:
+            assert tokens.authenticate_pat("dgpt_pat_x", _request())["error"] == "invalid_token"
+        repo.find_active_by_hash.assert_not_called()
+
+    def test_rejected_when_disabled(self, monkeypatch):
+        monkeypatch.setattr(tokens.settings, "PAT_ENABLED", False)
+        with _db(dict(_ROW)):
+            assert tokens.authenticate_pat("dgpt_pat_x", _request())["error"] == "invalid_token"
+
+    def test_works_with_auth_disabled(self, monkeypatch):
+        monkeypatch.setattr(tokens.settings, "AUTH_TYPE", None)
+        with _db(dict(_ROW)):
+            assert tokens.authenticate_pat("dgpt_pat_x", _request())["sub"] == "alice"
+
+    def test_starlette_request_ip(self):
+        request = Mock(spec=["headers", "client"])
+        request.client.host = "10.9.9.9"
+        with _db(dict(_ROW)) as repo:
+            tokens.authenticate_pat("dgpt_pat_x", request)
+        assert repo.touch_last_used.call_args.args[1] == "10.9.9.9"
+
+
+@pytest.mark.unit
+class TestHandleAuthPatBranch:
+    def test_pat_bearer_goes_to_the_pat_verifier_in_any_auth_mode(self):
+        from docsgpt import auth
+
+        for auth_type in (None, "oidc", "simple_jwt"):
+            with patch.object(auth.settings, "AUTH_TYPE", auth_type), patch(
+                "docsgpt.api.pat.tokens.authenticate_pat", return_value={"sub": "alice"}
+            ) as verifier:
+                request = _request("Bearer dgpt_pat_secret")
+                assert auth.handle_auth(request) == {"sub": "alice"}
+            verifier.assert_called_once_with("dgpt_pat_secret", request)
+
+    def test_bearer_scheme_is_case_insensitive(self):
+        from docsgpt import auth
+
+        with patch("docsgpt.api.pat.tokens.authenticate_pat", return_value={"sub": "a"}) as verifier:
+            auth.handle_auth(_request("bearer dgpt_pat_secret"))
+        verifier.assert_called_once()
+
+    def test_jwt_cannot_smuggle_pat_claims(self):
+        from jose import jwt
+
+        from docsgpt import auth
+
+        forged = jwt.encode(
+            {
+                "sub": "mallory",
+                "auth_method": "pat",
+                "scopes": ["agents:write"],
+                "resource_filter": {},
+                "pat_id": "x",
+                "pat_name": "x",
+            },
+            "secret",
+            algorithm="HS256",
+        )
+        with patch.object(auth.settings, "AUTH_TYPE", "simple_jwt"), patch.object(
+            auth.settings, "JWT_SECRET_KEY", "secret"
+        ):
+            decoded = auth.handle_auth(_request(f"Bearer {forged}"))
+        assert decoded == {"sub": "mallory"}