feat(pat): enforce scopes and resource restrictions, deny by default

A central rule table maps each route and method to the scope a token needs;
a route that is not listed cannot be called with a token, and a test fails
when a registered route is left unclassified. Token management, admin, team
management, sign-in, device pairing and OAuth handshakes are never token
reachable, and a token never carries the admin role.

A token restricted to specific agents, sources, prompts, tools or workflows
is held to its allowlist: ids are checked wherever a route carries them,
listings are filtered, creation is refused, and routes whose rows cannot be
tied to the allowlist are closed. Agent import checks the resolved target.

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_pat_routes.py

@@ -0,0 +1,244 @@
+"""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
+
+    def test_duplicate_name_conflicts(self, client, db):
+        assert _create(client).status_code == 201
+        assert _create(client).status_code == 409
+
+    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 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 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
+
+    def test_unknown_and_malformed_ids(self, client, db):
+        with _session():
+            assert client.delete(f"/api/user/tokens/{AGENT_A}").status_code == 404
+            assert client.delete("/api/user/tokens/not-a-uuid").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

tests/api/test_pat_rules.py

@@ -0,0 +1,294 @@
+"""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/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_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"
+
+    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"
+
+
+@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"
+
+    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}