From 82ec6cba0e0acff04b7ae4a9c90cf99c9d1bec01 Mon Sep 17 00:00:00 2001 From: arc53-machine <232052973+arc53-machine@users.noreply.github.com> Date: Sat, 19 Sep 2026 23:39:58 +0100 Subject: [PATCH] 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. --- docsgpt/api/pat/rules.py | 447 +++++++++++++++++++++++++ docsgpt/api/user/agents/portability.py | 31 ++ docsgpt/api/user/agents/routes.py | 3 +- docsgpt/api/user/prompts/routes.py | 4 +- docsgpt/api/user/sources/routes.py | 3 +- docsgpt/api/user/tools/routes.py | 4 + docsgpt/app.py | 19 +- tests/api/test_agent_portability.py | 69 ++++ tests/api/test_pat_routes.py | 244 ++++++++++++++ tests/api/test_pat_rules.py | 294 ++++++++++++++++ 10 files changed, 1114 insertions(+), 4 deletions(-) create mode 100644 docsgpt/api/pat/rules.py create mode 100644 tests/api/test_pat_routes.py create mode 100644 tests/api/test_pat_rules.py diff --git a/docsgpt/api/pat/rules.py b/docsgpt/api/pat/rules.py new file mode 100644 index 00000000..f340d698 --- /dev/null +++ b/docsgpt/api/pat/rules.py @@ -0,0 +1,447 @@ +"""What a personal access token may call: the scope and resource rule table. + +Authorization for PATs is central and deny by default. ``RULES`` maps a Flask +route (its rule string and method) to the scope it needs; a PAT request to a +route that is not listed is refused, so a new endpoint is unreachable by token +until someone classifies it here. ``tests/api/test_pat_rules.py`` fails when a +registered route is in neither ``RULES`` nor ``DENIED``. + +A token may also carry a resource filter (``{"agents": [ids]}``). For a +restricted family the rule must be able to prove the request stays inside the +allowlist: it names where the id travels (``ids``), or declares that the route +filters its own listing (``listing``), or delegates to the route +(``in_route``). Anything else, creation included, is refused. ``refs`` cover +ids of *other* families a route accepts (an agent update naming a source), and +``blocked_by`` closes routes whose rows hang off a family the rule cannot see +(a schedule belongs to an agent). + +Session (JWT) callers never pass through here. +""" + +from __future__ import annotations + +import json +import uuid +from dataclasses import dataclass +from typing import Any, Callable, Iterable, Optional + +VIEW, QUERY, JSON, FORM, BODY = "view", "query", "json", "form", "body" + +Locator = tuple[str, str] + + +@dataclass(frozen=True) +class Rule: + """Requirement for one route+method. ``scopes`` is any-of; empty means any valid token.""" + + scopes: tuple[str, ...] = () + family: Optional[str] = None + ids: tuple[Locator, ...] = () + refs: tuple[tuple[str, Locator], ...] = () + listing: bool = False + open: bool = False + in_route: bool = False + blocked_by: tuple[str, ...] = () + check: Optional[Callable[[Any, dict], Optional[str]]] = None + + +def _rule(scope: Optional[str] = None, *ids: Locator, any_of: tuple[str, ...] = (), **kwargs) -> Rule: + scopes = any_of or ((scope,) if scope else ()) + family = kwargs.pop("family", None) + if family is None and scope: + family = scope.partition(":")[0] + return Rule(scopes=scopes, family=family, ids=tuple(ids), **kwargs) + + +# Ids of other families that agent create/update accept in their JSON-or-form body. +_AGENT_BODY_REFS = ( + ("sources", (BODY, "source")), + ("sources", (BODY, "sources")), + ("prompts", (BODY, "prompt_id")), + ("tools", (BODY, "tools")), + ("workflows", (BODY, "workflow")), +) + + +def _chat_check(request, resource_filter: dict) -> Optional[str]: + """Keep a restricted token's chat traffic inside its allowlists. + + An agent brings its own sources, prompt and tools, which this table cannot + see, so a token restricted on any family must name an allowed agent (or, + when only sources are restricted, chat against allowed sources directly). + An agent ``api_key`` in the body would swap in an arbitrary agent. + """ + body = _json_body(request) + if body.get("api_key"): + return "A restricted token cannot chat with an agent API key; pass agent_id" + if body.get("workflow"): + # An inline workflow graph (builder preview) can reference any resource. + return "A restricted token cannot run an inline workflow" + agent_ids = _as_ids(body.get("agent_id")) + if "agents" in resource_filter: + if not agent_ids: + return "This token is restricted to specific agents; pass agent_id" + return None # the agent id itself is verified through ``refs`` + if agent_ids: + return "This token is restricted to specific resources and cannot run arbitrary agents" + return None + + +_CHAT = dict( + family=None, + refs=( + ("agents", (JSON, "agent_id")), + ("sources", (JSON, "active_docs")), + ("prompts", (JSON, "prompt_id")), + ("workflows", (JSON, "workflow_id")), + ), + check=_chat_check, +) + +RULES: dict[tuple[str, str], Rule] = { + # Identity and public metadata: any valid token. + ("/api/user/me", "GET"): _rule(open=True), + ("/api/health", "GET"): _rule(open=True), + ("/api/config", "GET"): _rule(open=True), + # Agents + ("/api/get_agent", "GET"): _rule("agents:read", (QUERY, "id")), + ("/api/get_agents", "GET"): _rule("agents:read", listing=True), + ("/api/pinned_agents", "GET"): _rule("agents:read"), + ("/api/shared_agents", "GET"): _rule("agents:read"), + ("/api/template_agents", "GET"): _rule("agents:read", open=True), + ("/api/export_agent", "GET"): _rule("agents:read", (QUERY, "id")), + ("/api/guardrails/catalog", "GET"): _rule("agents:read", open=True), + ("/api/guardrails/events", "GET"): _rule("agents:read", (QUERY, "agent_id")), + ("/api/guardrails/summary", "GET"): _rule("agents:read", (QUERY, "agent_id")), + ("/api/agents/folders/", "GET"): _rule("agents:read", open=True), + ("/api/agents/folders/", "GET"): _rule("agents:read"), + ("/api/create_agent", "POST"): _rule("agents:write", refs=_AGENT_BODY_REFS), + ("/api/update_agent/", "PUT"): _rule( + "agents:write", (VIEW, "agent_id"), refs=_AGENT_BODY_REFS + ), + ("/api/delete_agent", "DELETE"): _rule("agents:write", (QUERY, "id")), + ("/api/adopt_agent", "POST"): _rule("agents:write"), + ("/api/pin_agent", "POST"): _rule("agents:write", (QUERY, "id")), + ("/api/remove_shared_agent", "DELETE"): _rule("agents:write", (QUERY, "id")), + ("/api/share_agent", "PUT"): _rule("agents:write", (JSON, "id")), + ("/api/import_agent/plan", "POST"): _rule("agents:write", in_route=True), + ("/api/import_agent", "POST"): _rule("agents:write", in_route=True), + ("/api/agents/folders/", "POST"): _rule("agents:write"), + ("/api/agents/folders/", "PUT"): _rule("agents:write"), + ("/api/agents/folders/", "DELETE"): _rule("agents:write"), + ("/api/agents/folders/move_agent", "POST"): _rule("agents:write", (JSON, "agent_id")), + ("/api/agents/folders/bulk_move", "POST"): _rule("agents:write", (JSON, "agent_ids")), + ("/api/regenerate_agent_key/", "POST"): _rule("agents:keys", (VIEW, "agent_id")), + ("/api/agent_webhook", "GET"): _rule("agents:keys", (QUERY, "id")), + # Schedules hang off an agent. + ("/api/agents//schedules", "GET"): _rule( + "schedules:read", refs=(("agents", (VIEW, "agent_id")),) + ), + ("/api/agents//schedules", "POST"): _rule( + "schedules:write", refs=(("agents", (VIEW, "agent_id")),), blocked_by=("tools",) + ), + ("/api/schedules/", "GET"): _rule("schedules:read", blocked_by=("agents",)), + ("/api/schedules//runs", "GET"): _rule("schedules:read", blocked_by=("agents",)), + ("/api/schedules//runs/", "GET"): _rule( + "schedules:read", blocked_by=("agents",) + ), + ("/api/schedules/", "PUT"): _rule("schedules:write", blocked_by=("agents", "tools")), + ("/api/schedules/", "PATCH"): _rule("schedules:write", blocked_by=("agents", "tools")), + ("/api/schedules/", "DELETE"): _rule("schedules:write", blocked_by=("agents",)), + ("/api/schedules//run", "POST"): _rule("schedules:write", blocked_by=("agents",)), + # Sources + ("/api/sources", "GET"): _rule("sources:read", listing=True), + # Counted and paged in SQL, so it cannot be narrowed here; restricted tokens use /api/sources. + ("/api/sources/paginated", "GET"): _rule("sources:read"), + ("/api/directory_structure", "GET"): _rule("sources:read", (QUERY, "id")), + ("/api/get_chunks", "GET"): _rule("sources:read", (QUERY, "id")), + ("/api/sources//wiki/pages", "GET"): _rule("sources:read", (VIEW, "source_id")), + ("/api/sources//wiki/page", "GET"): _rule("sources:read", (VIEW, "source_id")), + ("/api/sources//graph", "GET"): _rule("sources:read", (VIEW, "source_id")), + ("/api/sources//graph/node/", "GET"): _rule( + "sources:read", (VIEW, "source_id") + ), + # Ingestion and attachment extraction both report through this poll. + ("/api/task_status", "GET"): _rule(any_of=("sources:read", "sources:write", "chat:run"), open=True), + ("/api/upload", "POST"): _rule("sources:write"), + ("/api/remote", "POST"): _rule("sources:write"), + ("/api/sources/wiki", "POST"): _rule("sources:write"), + ("/api/delete_old", "GET"): _rule("sources:write", (QUERY, "source_id")), + ("/api/manage_sync", "POST"): _rule("sources:write", (JSON, "source_id")), + ("/api/sync_source", "POST"): _rule("sources:write", (JSON, "source_id")), + ("/api/sources/reingest", "POST"): _rule("sources:write", (JSON, "source_id")), + ("/api/manage_source_files", "POST"): _rule("sources:write", (FORM, "source_id")), + ("/api/sources//config", "PATCH"): _rule("sources:write", (VIEW, "source_id")), + ("/api/sources//wiki/page", "PUT"): _rule("sources:write", (VIEW, "source_id")), + ("/api/sources//wiki/convert", "POST"): _rule("sources:write", (VIEW, "source_id")), + ("/api/sources//graphrag/enable", "POST"): _rule( + "sources:write", (VIEW, "source_id") + ), + ("/api/add_chunk", "POST"): _rule("sources:write", (JSON, "id")), + ("/api/update_chunk", "PUT"): _rule("sources:write", (JSON, "id")), + ("/api/delete_chunk", "DELETE"): _rule("sources:write", (QUERY, "id")), + # Prompts + ("/api/get_prompts", "GET"): _rule("prompts:read", listing=True), + ("/api/get_single_prompt", "GET"): _rule("prompts:read", (QUERY, "id")), + ("/api/create_prompt", "POST"): _rule("prompts:write"), + ("/api/update_prompt", "POST"): _rule("prompts:write", (JSON, "id")), + ("/api/delete_prompt", "POST"): _rule("prompts:write", (JSON, "id")), + # Tools + ("/api/available_tools", "GET"): _rule("tools:read", open=True), + ("/api/get_tools", "GET"): _rule("tools:read", listing=True), + ("/api/create_tool", "POST"): _rule("tools:write"), + ("/api/parse_spec", "POST"): _rule("tools:write", open=True), + ("/api/update_tool", "POST"): _rule("tools:write", (JSON, "id")), + ("/api/update_tool_config", "POST"): _rule("tools:write", (JSON, "id")), + ("/api/update_tool_actions", "POST"): _rule("tools:write", (JSON, "id")), + ("/api/update_tool_status", "POST"): _rule("tools:write", (JSON, "id")), + ("/api/delete_tool", "POST"): _rule("tools:write", (JSON, "id")), + ("/api/mcp_server/test", "POST"): _rule("tools:write", open=True), + ("/api/mcp_server/save", "POST"): _rule("tools:write", (JSON, "id")), + # Models + ("/api/models", "GET"): _rule(any_of=("models:read", "chat:run"), open=True), + ("/api/user/models", "GET"): _rule("models:read"), + ("/api/user/models/", "GET"): _rule("models:read"), + ("/api/user/models", "POST"): _rule("models:write"), + ("/api/user/models/", "PATCH"): _rule("models:write"), + ("/api/user/models/", "DELETE"): _rule("models:write"), + ("/api/user/models/test", "POST"): _rule("models:write"), + ("/api/user/models//test", "POST"): _rule("models:write"), + # Workflows + ("/api/workflows", "POST"): _rule("workflows:write"), + ("/api/workflows/", "GET"): _rule("workflows:read", (VIEW, "workflow_id")), + ("/api/workflows/", "PUT"): _rule("workflows:write", (VIEW, "workflow_id")), + ("/api/workflows/", "DELETE"): _rule("workflows:write", (VIEW, "workflow_id")), + # Conversations and analytics span every agent, so an agent-restricted token is kept out. + ("/api/get_conversations", "GET"): _rule("conversations:read", blocked_by=("agents",)), + ("/api/search_conversations", "GET"): _rule("conversations:read", blocked_by=("agents",)), + ("/api/get_single_conversation", "GET"): _rule("conversations:read", blocked_by=("agents",)), + ("/api/messages//tail", "GET"): _rule( + any_of=("conversations:read", "chat:run"), family=None + ), + ("/api/delete_conversation", "POST"): _rule("conversations:write", blocked_by=("agents",)), + ("/api/delete_all_conversations", "GET"): _rule("conversations:write", blocked_by=("agents",)), + ("/api/update_conversation_name", "POST"): _rule("conversations:write", blocked_by=("agents",)), + ("/api/feedback", "POST"): _rule("conversations:write", blocked_by=("agents",)), + ("/api/get_message_analytics", "POST"): _rule("analytics:read", blocked_by=("agents",)), + ("/api/get_token_analytics", "POST"): _rule("analytics:read", blocked_by=("agents",)), + ("/api/get_feedback_analytics", "POST"): _rule("analytics:read", blocked_by=("agents",)), + ("/api/get_tool_analytics", "POST"): _rule("analytics:read", blocked_by=("agents",)), + ("/api/get_schedule_analytics", "POST"): _rule("analytics:read", blocked_by=("agents",)), + ("/api/get_user_logs", "POST"): _rule("analytics:read", blocked_by=("agents",)), + # Teams (read only) + ("/api/teams", "GET"): _rule("teams:read"), + ("/api/teams/", "GET"): _rule("teams:read"), + ("/api/teams//members", "GET"): _rule("teams:read"), + ("/api/teams//grants", "GET"): _rule("teams:read"), + ("/api/resource_shares", "GET"): _rule("teams:read"), + # Chat + ("/api/answer", "POST"): _rule("chat:run", **_CHAT), + ("/stream", "POST"): _rule("chat:run", **_CHAT), + ("/api/search", "POST"): _rule("chat:run", **_CHAT), + ("/api/store_attachment", "POST"): _rule("chat:run", family=None), + ("/api/sources//search", "POST"): _rule( + "chat:run", family=None, refs=(("sources", (VIEW, "source_id")),) + ), +} + +#: Routes a PAT may never call, by exact rule string ("*" = every method) or prefix. +#: Token management, admin, login flows and interactive OAuth handshakes need a +#: signed-in session; the rest have no scope yet. Listing them keeps the +#: classification test honest: a new route must land here or in ``RULES``. +DENIED: dict[str, tuple[str, ...]] = { + "/": ("*",), + "/api/user/tokens": ("*",), + "/api/user/tokens/": ("*",), + "/api/generate_token": ("*",), + "/api/combine": ("*",), + "/api/download": ("*",), + "/api/upload_index": ("*",), + "/api/share": ("*",), + "/api/shared_agent": ("*",), + "/api/shared_conversation/": ("*",), + "/api/webhooks/agents/": ("*",), + "/api/images//": ("*",), + "/api/mcp_server/callback": ("*",), + "/api/mcp_server/auth_status": ("*",), + "/api/artifact/": ("*",), + "/api/artifacts": ("*",), + "/api/artifacts/": ("*",), + "/api/artifacts//restore": ("*",), + "/api/artifacts//versions/": ("*",), + "/api/stt": ("*",), + "/api/stt/live/start": ("*",), + "/api/stt/live/chunk": ("*",), + "/api/stt/live/finish": ("*",), + "/api/tts": ("*",), + "/api/teams": ("POST",), + "/api/teams/": ("PUT", "DELETE"), + "/api/teams//members": ("POST",), + "/api/teams//members/": ("*",), + "/api/teams//grants": ("POST", "DELETE"), + "/api/teams//transfer_owner": ("*",), + "/swagger.json": ("*",), +} +DENIED_PREFIXES = ( + "/api/admin/", + "/api/auth/oidc/", + "/api/connectors/", + "/api/devices", + "/scim/", + "/static/", + "/swaggerui/", + "/v1/", +) + + +def is_denied(rule: str, method: str) -> bool: + if rule.startswith(DENIED_PREFIXES): + return True + methods = DENIED.get(rule) + return bool(methods) and ("*" in methods or method in methods) + + +def _json_body(request) -> dict: + body = request.get_json(silent=True) + return body if isinstance(body, dict) else {} + + +def _as_ids(value: Any) -> list[str]: + """Flatten whatever a route accepts as ids: a string, a JSON-encoded or plain list, or an ``{id}`` dict.""" + if value is None or value == "": + return [] + if isinstance(value, dict): + return _as_ids(value.get("id") or value.get("_id") or value.get("workflow_id")) + if isinstance(value, (list, tuple)): + out: list[str] = [] + for item in value: + out.extend(_as_ids(item)) + return out + text = str(value).strip() + if text[:1] in "[{": + try: + return _as_ids(json.loads(text)) + except ValueError: + return [text] + return [text] + + +def _read(request, locator: Locator) -> list[str]: + where, key = locator + if where == VIEW: + return _as_ids((request.view_args or {}).get(key)) + if where == QUERY: + return _as_ids(request.args.get(key)) + if where == JSON: + return _as_ids(_json_body(request).get(key)) + if where == FORM: + return _as_ids(request.form.get(key)) + # BODY: routes that accept JSON or a multipart form interchangeably. + if request.is_json: + return _as_ids(_json_body(request).get(key)) + return _as_ids(request.form.get(key)) + + +def _canonical(value: str) -> str: + try: + return str(uuid.UUID(value)) + except (ValueError, AttributeError, TypeError): + return value + + +def _all_allowed(ids: Iterable[str], allowed: Iterable[str]) -> bool: + allowlist = {_canonical(a) for a in allowed} + return all(_canonical(i) in allowlist for i in ids) + + +def authorize(request, decoded_token: dict) -> Optional[tuple[dict, int]]: + """Check a PAT request against the table. ``None`` allows; otherwise ``(body, status)``.""" + url_rule = getattr(request, "url_rule", None) + rule = RULES.get((url_rule.rule, request.method)) if url_rule is not None else None + if rule is None: + return ( + { + "success": False, + "error": "not_available_to_tokens", + "message": "This endpoint cannot be called with a personal access token", + }, + 403, + ) + granted = set(decoded_token.get("scopes") or []) + if rule.scopes and not granted.intersection(rule.scopes): + return ( + { + "success": False, + "error": "insufficient_scope", + "message": f"Token lacks the required scope: {' or '.join(rule.scopes)}", + "required_scope": rule.scopes[0], + }, + 403, + ) + resource_filter = decoded_token.get("resource_filter") or {} + if not resource_filter: + return None + reason = _check_resources(request, rule, resource_filter) + if reason is None: + return None + return ({"success": False, "error": "resource_not_allowed", "message": reason}, 403) + + +def _check_resources(request, rule: Rule, resource_filter: dict) -> Optional[str]: + for family in rule.blocked_by: + if family in resource_filter: + return f"This endpoint is not available to a token restricted to specific {family}" + if rule.check is not None: + reason = rule.check(request, resource_filter) + if reason: + return reason + for family, locator in rule.refs: + if family not in resource_filter: + continue + ids = _read(request, locator) + if ids and not _all_allowed(ids, resource_filter[family]): + return f"Token is not allowed to use one of the referenced {family}" + family = rule.family + if family is None or family not in resource_filter or rule.open or rule.listing or rule.in_route: + return None + ids = [i for locator in rule.ids for i in _read(request, locator)] + if not ids: + return f"This token is restricted to specific {family} and cannot use this endpoint" + if not _all_allowed(ids, resource_filter[family]): + return f"Token is not allowed to access this resource ({family})" + return None + + +def allowed_ids(request, family: str) -> Optional[set[str]]: + """The caller's allowlist for ``family``, or ``None`` when unrestricted (or not a PAT). + + Used by listing routes (``listing=True``) and by ``in_route`` handlers. + """ + decoded = getattr(request, "decoded_token", None) or {} + if decoded.get("auth_method") != "pat": + return None + ids = (decoded.get("resource_filter") or {}).get(family) + if ids is None: + return None + return {_canonical(str(i)) for i in ids} + + +def filter_listing(request, family: str, items: list, key: str = "id") -> list: + """Drop rows outside the caller's allowlist. Rows without a UUID id (built-in presets) are kept.""" + allowed = allowed_ids(request, family) + if allowed is None: + return items + kept = [] + for item in items: + value = str(item.get(key, "")) + if not _is_uuid(value) or _canonical(value) in allowed: + kept.append(item) + return kept + + +def _is_uuid(value: str) -> bool: + try: + uuid.UUID(value) + except (ValueError, AttributeError, TypeError): + return False + return True diff --git a/docsgpt/api/user/agents/portability.py b/docsgpt/api/user/agents/portability.py index 621d0c2d..a9121532 100644 --- a/docsgpt/api/user/agents/portability.py +++ b/docsgpt/api/user/agents/portability.py @@ -36,6 +36,7 @@ from docsgpt.agents.default_tools import ( synthesized_tool_name_for_id, ) from docsgpt.api import api +from docsgpt.api.pat.rules import allowed_ids from docsgpt.core.model_utils import validate_model_id from docsgpt.core.url_validation import SSRFError, validate_url from docsgpt.security.safe_url import UnsafeUserUrlError, validate_user_base_url @@ -1717,6 +1718,32 @@ def _read_import_payload(req): return raw.decode("utf-8", "replace"), {} +def _restricted_token_denial(conn, user: str, doc: dict) -> Optional[str]: + """Why a resource-restricted personal access token may not import ``doc``, if it may not. + + An import resolves sources, prompts and tools by name and may create them, + so a token restricted on any of those families cannot be held to its + allowlist here. A token restricted to specific agents may update exactly + those; it can never create one. + """ + for family in ("sources", "prompts", "tools", "workflows"): + if allowed_ids(request, family) is not None: + return f"A token restricted to specific {family} cannot import agents" + allowed_agents = allowed_ids(request, "agents") + if allowed_agents is None: + return None + target = _resolve_target(conn, user, doc.get("metadata") or {}) + if target["action"] != "update" or target["agent_id"] not in allowed_agents: + return "This token is restricted to specific agents and may only update those" + return None + + +def _token_denied_response(reason: str): + return make_response( + jsonify({"success": False, "error": "resource_not_allowed", "message": reason}), 403 + ) + + @agents_portability_ns.route("/export_agent") class ExportAgent(Resource): @api.doc(params={"id": "Agent ID"}, description="Export an agent as YAML") @@ -1763,6 +1790,8 @@ class ImportAgentPlan(Resource): return make_response(jsonify({"success": False, "message": str(exc)}), 400) try: with db_readonly() as conn: + if reason := _restricted_token_denial(conn, user, doc): + return _token_denied_response(reason) plan = plan_import(conn, user, doc) except Exception: current_app.logger.error("Agent import plan failed", exc_info=True) @@ -1786,6 +1815,8 @@ class ImportAgent(Resource): return make_response(jsonify({"success": False, "message": str(exc)}), 400) try: with db_session() as conn: + if reason := _restricted_token_denial(conn, user, doc): + return _token_denied_response(reason) result = apply_import(conn, user, doc, resolution) except AgentImportError as exc: # Apply-time rejection of the user's document (e.g. the workflow diff --git a/docsgpt/api/user/agents/routes.py b/docsgpt/api/user/agents/routes.py index 2ddcda5d..757dd933 100644 --- a/docsgpt/api/user/agents/routes.py +++ b/docsgpt/api/user/agents/routes.py @@ -9,6 +9,7 @@ from flask_restx import fields, Namespace, Resource from pydantic import ValidationError as PydanticValidationError from docsgpt.api import api +from docsgpt.api.pat.rules import filter_listing from docsgpt.guardrails.config import AgentConfig from docsgpt.api.user.base import ( copy_agent_image_for_user, @@ -498,7 +499,7 @@ class GetAgents(Resource): except Exception as err: current_app.logger.error(f"Error retrieving agents: {err}", exc_info=True) return make_response(jsonify({"success": False}), 400) - return make_response(jsonify(list_agents), 200) + return make_response(jsonify(filter_listing(request, "agents", list_agents)), 200) @agents_ns.route("/create_agent") diff --git a/docsgpt/api/user/prompts/routes.py b/docsgpt/api/user/prompts/routes.py index d798d01b..af5b9867 100644 --- a/docsgpt/api/user/prompts/routes.py +++ b/docsgpt/api/user/prompts/routes.py @@ -5,6 +5,7 @@ from flask import current_app, jsonify, make_response, request from flask_restx import fields, Namespace, Resource from docsgpt.api import api +from docsgpt.api.pat.rules import filter_listing from docsgpt.api.user.team_sharing import team_access_for, visible_with_access from docsgpt.storage.db.repositories.prompts import PromptsRepository from docsgpt.prompts.composer import compose_preset, is_composed_preset @@ -91,7 +92,8 @@ class GetPrompts(Resource): except Exception as err: current_app.logger.error(f"Error retrieving prompts: {err}", exc_info=True) return make_response(jsonify({"success": False}), 400) - return make_response(jsonify(list_prompts), 200) + # Presets (default/creative/strict) have no row id and stay visible to a restricted token. + return make_response(jsonify(filter_listing(request, "prompts", list_prompts)), 200) @prompts_ns.route("/get_single_prompt") diff --git a/docsgpt/api/user/sources/routes.py b/docsgpt/api/user/sources/routes.py index dc675063..2708b387 100644 --- a/docsgpt/api/user/sources/routes.py +++ b/docsgpt/api/user/sources/routes.py @@ -10,6 +10,7 @@ from pydantic import ValidationError from docsgpt.agents.tools.path_utils import validate_tool_path from docsgpt.api import api +from docsgpt.api.pat.rules import filter_listing from docsgpt.api.user.tasks import ( convert_source_to_wiki, extract_graph, @@ -130,7 +131,7 @@ class CombinedJson(Resource): except Exception as err: current_app.logger.error(f"Error retrieving sources: {err}", exc_info=True) return make_response(jsonify({"success": False}), 400) - return make_response(jsonify(data), 200) + return make_response(jsonify(filter_listing(request, "sources", data)), 200) @sources_ns.route("/sources/paginated") diff --git a/docsgpt/api/user/tools/routes.py b/docsgpt/api/user/tools/routes.py index ecd6f719..8b70abac 100644 --- a/docsgpt/api/user/tools/routes.py +++ b/docsgpt/api/user/tools/routes.py @@ -16,6 +16,7 @@ from docsgpt.agents.default_tools import ( from docsgpt.agents.tools.spec_parser import parse_spec from docsgpt.agents.tools.tool_manager import ToolManager from docsgpt.api import api +from docsgpt.api.pat.rules import filter_listing from docsgpt.api.user.artifacts.authz import Principal, authorize_artifact from docsgpt.api.user.team_sharing import effective_write_owner, visible_with_access from docsgpt.core.settings import settings @@ -265,6 +266,9 @@ class GetTools(Resource): shaped = _shape_tool(row, ownership="team", force_strip_secret=True) shaped["team_access"] = team_shared.get(str(row["id"])) user_tools.append(shaped) + # A resource-restricted token sees only its allowed tools; the + # default and builtin rows appended below belong to no user. + user_tools = filter_listing(request, "tools", user_tools) # ``scheduler`` is dual-registered (default chat tool + agent- # selectable builtin) and resolves to the same synthetic uuid5 id. diff --git a/docsgpt/app.py b/docsgpt/app.py index 4a11937e..88e2d5cd 100644 --- a/docsgpt/app.py +++ b/docsgpt/app.py @@ -22,8 +22,11 @@ from docsgpt.api.devices import devices_bp # noqa: E402 from docsgpt.api.internal.routes import internal # noqa: E402 from docsgpt.api.oidc import oidc_bp # noqa: E402 from docsgpt.api.oidc.denylist import is_denied as oidc_session_denied # noqa: E402 +from docsgpt.api.pat.routes import pat_ns # noqa: E402 +from docsgpt.api.pat.rules import authorize as authorize_pat # noqa: E402 +from docsgpt.api.pat.tokens import is_pat # noqa: E402 from docsgpt.api.scim import scim_bp # noqa: E402 -from docsgpt.api.user.authz import resolve_roles # noqa: E402 +from docsgpt.api.user.authz import ROLE_USER, resolve_roles # noqa: E402 from docsgpt.api.user.routes import user # noqa: E402 from docsgpt.api.connector.routes import connector # noqa: E402 from docsgpt.api.v1 import v1_bp # noqa: E402 @@ -109,6 +112,8 @@ app.register_blueprint(v1_bp) # first app and raise "add_url_rule can no longer be called". if admin_ns not in api.namespaces: api.add_namespace(admin_ns) +if pat_ns not in api.namespaces: + api.add_namespace(pat_ns) app.config.update( UPLOAD_FOLDER="inputs", CELERY_BROKER_URL=settings.CELERY_BROKER_URL, @@ -317,6 +322,18 @@ def authenticate_request(): request.decoded_token = None elif "error" in decoded_token: return jsonify(decoded_token), 401 + elif is_pat(decoded_token): + # Scopes and resource restrictions are enforced here, centrally and + # deny by default (docsgpt/api/pat/rules.py). A token never carries + # admin, whatever its owner holds, and the session denylist does not + # apply: the token lookup already excludes revoked tokens and + # deactivated users. + denied = authorize_pat(request, decoded_token) + if denied is not None: + body, status = denied + return jsonify(body), status + decoded_token["roles"] = [ROLE_USER] + request.decoded_token = decoded_token elif settings.AUTH_TYPE == "oidc" and oidc_session_denied(decoded_token): # Back-channel logout / SCIM deactivation revoked this session. return ( diff --git a/tests/api/test_agent_portability.py b/tests/api/test_agent_portability.py index 4f64cbd9..a3cc85f6 100644 --- a/tests/api/test_agent_portability.py +++ b/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() diff --git a/tests/api/test_pat_routes.py b/tests/api/test_pat_routes.py new file mode 100644 index 00000000..cd157284 --- /dev/null +++ b/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 diff --git a/tests/api/test_pat_rules.py b/tests/api/test_pat_rules.py new file mode 100644 index 00000000..3a3cd75a --- /dev/null +++ b/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/", "DELETE"), + ("/api/admin/users", "GET"), + ("/api/admin/tokens/", "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}