diff --git a/docsgpt/api/asgi_auth.py b/docsgpt/api/asgi_auth.py index 93f08a23..aa2ac5bd 100644 --- a/docsgpt/api/asgi_auth.py +++ b/docsgpt/api/asgi_auth.py @@ -16,27 +16,43 @@ from starlette.requests import Request from starlette.responses import JSONResponse from docsgpt.api.oidc.denylist import is_denied as oidc_session_denied +from docsgpt.api.pat.tokens import is_pat from docsgpt.auth import handle_auth from docsgpt.core import log_context from docsgpt.core.settings import settings -async def authenticate(request: Request) -> Tuple[Optional[dict], Optional[JSONResponse]]: +async def authenticate( + request: Request, *, pat_scope: Optional[str] = None +) -> Tuple[Optional[dict], Optional[JSONResponse]]: """Decode the caller's JWT the way Flask's ``authenticate_request`` does. Args: request: The incoming Starlette request. + pat_scope: Scope a personal access token needs for this route. Left + unset, the route rejects PATs outright (deny by default, matching + the Flask rule table in ``docsgpt/api/pat/rules.py``). Returns: tuple: ``(claims, None)`` for an authenticated caller, ``(None, None)`` when no token was sent (the route decides whether that is allowed), or ``(None, response)`` carrying the 401 to return. """ - decoded = handle_auth(request) + # A personal access token resolves against Postgres; keep that sync read off the event loop. + decoded = await anyio.to_thread.run_sync(handle_auth, request) if not decoded: return None, None if "error" in decoded: return None, JSONResponse(decoded, status_code=401) + if is_pat(decoded): + # A PAT lookup already excludes revoked tokens and deactivated users, + # so the session denylist below does not apply to it. + if pat_scope is None or pat_scope not in (decoded.get("scopes") or []): + return None, JSONResponse( + {"success": False, "message": "Token lacks the required scope", "error": "insufficient_scope"}, + status_code=403, + ) + return decoded, None # The denylist is a sync Redis read; keep it off the event loop. if settings.AUTH_TYPE == "oidc" and await anyio.to_thread.run_sync(oidc_session_denied, decoded): return None, JSONResponse( diff --git a/docsgpt/api/async_sse.py b/docsgpt/api/async_sse.py index a5064e2c..1bb3af1d 100644 --- a/docsgpt/api/async_sse.py +++ b/docsgpt/api/async_sse.py @@ -94,7 +94,7 @@ async def stream_message_events(request: Request) -> Response: """ # Same JWT decoder and OIDC revocation check as the Flask routes. With # AUTH_TYPE unset the caller resolves to ``{"sub": "local"}``. - decoded, error = await authenticate(request) + decoded, error = await authenticate(request, pat_scope="chat:run") if error is not None: return error user_id = decoded.get("sub") if isinstance(decoded, dict) else None diff --git a/docsgpt/api/pat/__init__.py b/docsgpt/api/pat/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/docsgpt/api/pat/tokens.py b/docsgpt/api/pat/tokens.py new file mode 100644 index 00000000..fa4cb9d0 --- /dev/null +++ b/docsgpt/api/pat/tokens.py @@ -0,0 +1,234 @@ +"""Personal access tokens: format, scope catalog and the per-request verifier. + +A PAT is ``dgpt_pat_`` + 32 random bytes (urlsafe). Only its SHA-256 is stored +(``personal_access_tokens.token_hash``), the same shape as device session +tokens. Scopes and the resource filter are always read from the database row; +nothing about a token's authority is encoded in the credential itself. +""" + +from __future__ import annotations + +import hashlib +import logging +import secrets +import uuid +from datetime import datetime, timedelta, timezone +from typing import Any, Optional + +from docsgpt.core.settings import settings +from docsgpt.storage.db.repositories.personal_access_tokens import ( + PersonalAccessTokensRepository, +) +from docsgpt.storage.db.session import db_readonly, db_session + +logger = logging.getLogger(__name__) + +TOKEN_PREFIX = "dgpt_pat_" +# Characters of the secret kept in ``token_prefix`` so users can tell tokens apart. +_DISPLAY_CHARS = 6 + +AUTH_METHOD_PAT = "pat" + +#: Every grantable scope with the description shown in the UI and docs. +SCOPES: dict[str, str] = { + "agents:read": "View agents, folders, guardrail events and export agent definitions", + "agents:write": "Create, update, delete, share and import (apply) agents and folders", + "agents:keys": "Regenerate agent API keys and read incoming webhook URLs", + "sources:read": "View sources, their files, chunks and ingestion task status", + "sources:write": "Upload, ingest, sync, edit and delete sources and chunks", + "prompts:read": "View prompts", + "prompts:write": "Create, update and delete prompts", + "tools:read": "View configured tools", + "tools:write": "Create, update and delete tools and MCP servers", + "models:read": "View available and custom models", + "models:write": "Create, update, test and delete custom models", + "workflows:read": "View workflows", + "workflows:write": "Create, update and delete workflows", + "schedules:read": "View agent schedules and their runs", + "schedules:write": "Create, update, run and delete agent schedules", + "conversations:read": "View conversations and messages", + "conversations:write": "Rename, delete and give feedback on conversations", + "analytics:read": "View usage analytics and logs", + "teams:read": "View teams, members and resource shares", + "chat:run": "Ask agents and search sources (answer, stream, search); used for benchmarking", +} + +#: Resource families whose tokens can be narrowed to specific ids. +FILTERABLE_FAMILIES = ("agents", "sources", "prompts", "tools", "workflows") +_MAX_FILTER_IDS = 200 + + +def auth_type_supports_pats() -> bool: + """PATs bind to a stable user id, which simple_jwt/session_jwt don't have.""" + return bool(settings.PAT_ENABLED) and settings.AUTH_TYPE in (None, "oidc") + + +def generate_token() -> tuple[str, str, str]: + """Mint a token. Returns ``(plaintext, sha256_hex, display_prefix)``.""" + secret = secrets.token_urlsafe(32) + token = TOKEN_PREFIX + secret + return token, hash_token(token), TOKEN_PREFIX + secret[:_DISPLAY_CHARS] + + +def hash_token(token: str) -> str: + return hashlib.sha256(token.encode("utf-8")).hexdigest() + + +def looks_like_pat(value: Optional[str]) -> bool: + return bool(value) and value.startswith(TOKEN_PREFIX) + + +def redact(value: Optional[str]) -> str: + """Log-safe form of a credential: the display prefix only.""" + if not value: + return "" + if looks_like_pat(value): + return value[: len(TOKEN_PREFIX) + _DISPLAY_CHARS] + "…" + return value[:4] + "…" + + +def expand_scopes(scopes) -> set[str]: + """Granted scopes plus what they imply (``x:write`` includes ``x:read``).""" + granted = set(scopes or []) + for scope in list(granted): + family, _, action = scope.partition(":") + if action == "write" and f"{family}:read" in SCOPES: + granted.add(f"{family}:read") + return granted + + +def normalize_scopes(raw: Any) -> list[str]: + """Validate a requested scope list. Raises ``ValueError`` with a user-facing message.""" + if not isinstance(raw, list) or not raw: + raise ValueError("scopes must be a non-empty list") + unknown = sorted({s for s in raw if not isinstance(s, str) or s not in SCOPES}, key=str) + if unknown: + raise ValueError(f"Unknown scopes: {', '.join(map(str, unknown))}") + return sorted(set(raw)) + + +def normalize_resource_filter(raw: Any, scopes: list[str]) -> dict[str, list[str]]: + """Validate ``{"": ["", ...]}``. Raises ``ValueError`` with a user-facing message. + + A family may only be restricted when the token holds a scope in it; + otherwise the restriction would be dead weight that reads as protection. + """ + if raw in (None, {}): + return {} + if not isinstance(raw, dict): + raise ValueError("resource_filter must be an object") + families = {s.partition(":")[0] for s in scopes} + # chat:run acts on agents and sources, so both may be restricted alongside it. + if "chat" in families: + families.update({"agents", "sources"}) + out: dict[str, list[str]] = {} + for family, ids in raw.items(): + if family not in FILTERABLE_FAMILIES: + raise ValueError( + f"resource_filter supports only: {', '.join(FILTERABLE_FAMILIES)}" + ) + if family not in families: + raise ValueError(f"resource_filter.{family} needs a {family} scope on the token") + if not isinstance(ids, list) or not ids: + raise ValueError(f"resource_filter.{family} must be a non-empty list of ids") + if len(ids) > _MAX_FILTER_IDS: + raise ValueError(f"resource_filter.{family} allows at most {_MAX_FILTER_IDS} ids") + normalized = [] + for value in ids: + try: + normalized.append(str(uuid.UUID(str(value)))) + except (ValueError, AttributeError, TypeError): + raise ValueError(f"resource_filter.{family} contains an invalid id: {value!r}") + out[family] = sorted(set(normalized)) + return out + + +def resolve_expiry(expires_in_days: Any) -> Optional[datetime]: + """Map the requested lifetime to ``expires_at``. Raises ``ValueError`` with a user-facing message. + + ``None`` means "use the default"; ``0`` asks for a non-expiring token, + which only an operator setting can allow. + """ + if expires_in_days is None: + days = settings.PAT_DEFAULT_LIFETIME_DAYS + elif isinstance(expires_in_days, bool) or not isinstance(expires_in_days, int): + raise ValueError("expires_in_days must be an integer") + elif expires_in_days == 0: + if not settings.PAT_ALLOW_NON_EXPIRING: + raise ValueError("Non-expiring tokens are disabled on this server") + return None + elif expires_in_days < 0: + raise ValueError("expires_in_days must be positive") + else: + days = expires_in_days + if days > settings.PAT_MAX_LIFETIME_DAYS: + raise ValueError(f"expires_in_days must not exceed {settings.PAT_MAX_LIFETIME_DAYS}") + return datetime.now(timezone.utc) + timedelta(days=days) + + +def _client_ip(request) -> Optional[str]: + # Flask exposes remote_addr; Starlette exposes client.host. + ip = getattr(request, "remote_addr", None) + if ip: + return ip + client = getattr(request, "client", None) + return getattr(client, "host", None) + + +_TOUCH_INTERVAL_SECONDS = 60 + + +def _usage_is_stale(last_used_at: Any) -> bool: + """True when ``last_used_at`` is old enough to be worth a write transaction.""" + if not last_used_at: + return True + try: + seen = last_used_at if isinstance(last_used_at, datetime) else datetime.fromisoformat(str(last_used_at)) + except ValueError: + return True + if seen.tzinfo is None: + seen = seen.replace(tzinfo=timezone.utc) + return (datetime.now(timezone.utc) - seen).total_seconds() >= _TOUCH_INTERVAL_SECONDS + + +_INVALID = {"message": "Authentication error: invalid token", "error": "invalid_token"} + + +def authenticate_pat(token: str, request) -> dict: + """Resolve a PAT into the claims dict the rest of the app reads. + + Fails closed: an unknown, revoked or expired token, a disabled feature or + a database error all yield the same ``invalid_token`` error. + """ + if not auth_type_supports_pats(): + return dict(_INVALID) + try: + with db_readonly() as conn: + row = PersonalAccessTokensRepository(conn).find_active_by_hash(hash_token(token)) + except Exception: + logger.error("PAT lookup failed for %s", redact(token), exc_info=True) + return dict(_INVALID) + if not row: + logger.warning("Rejected personal access token %s", redact(token)) + return dict(_INVALID) + if _usage_is_stale(row.get("last_used_at")): + try: + with db_session() as conn: + PersonalAccessTokensRepository(conn).touch_last_used( + str(row["id"]), _client_ip(request), min_interval_seconds=_TOUCH_INTERVAL_SECONDS + ) + except Exception: + # Usage telemetry must never fail a request. + logger.debug("PAT last-used update failed", exc_info=True) + return { + "sub": row["user_id"], + "auth_method": AUTH_METHOD_PAT, + "pat_id": str(row["id"]), + "pat_name": row["name"], + "scopes": sorted(expand_scopes(row.get("scopes"))), + "resource_filter": row.get("resource_filter") or {}, + } + + +def is_pat(decoded_token: Optional[dict]) -> bool: + return bool(decoded_token) and decoded_token.get("auth_method") == AUTH_METHOD_PAT diff --git a/docsgpt/api/user/me/routes.py b/docsgpt/api/user/me/routes.py index 34e108ad..67e31dfb 100644 --- a/docsgpt/api/user/me/routes.py +++ b/docsgpt/api/user/me/routes.py @@ -12,6 +12,8 @@ from __future__ import annotations from flask import jsonify, make_response, request from flask_restx import Namespace, Resource +from docsgpt.api.pat.tokens import is_pat + me_ns = Namespace("me", description="Current user identity and roles", path="/api") @@ -31,4 +33,13 @@ class MeResource(Resource): value = decoded_token.get(field) if value: body[field] = value + if is_pat(decoded_token): + # Lets a CLI or pipeline confirm what its token is allowed to do. + body["auth_method"] = "pat" + body["token"] = { + "id": decoded_token.get("pat_id"), + "name": decoded_token.get("pat_name"), + "scopes": decoded_token.get("scopes") or [], + "resource_filter": decoded_token.get("resource_filter") or {}, + } return make_response(jsonify(body), 200) diff --git a/docsgpt/auth.py b/docsgpt/auth.py index d56140d7..a38623a5 100644 --- a/docsgpt/auth.py +++ b/docsgpt/auth.py @@ -4,7 +4,28 @@ from jose.exceptions import ExpiredSignatureError from docsgpt.core.settings import settings +# Claims only the PAT verifier may set. Dropped from decoded JWTs so a session +# token can never present itself as a (differently scoped) personal access token. +_PAT_ONLY_CLAIMS = ("auth_method", "pat_id", "pat_name", "scopes", "resource_filter") + + +def _bearer_value(request): + header = request.headers.get("Authorization") + if not header or not isinstance(header, str): + return None + scheme, _, value = header.partition(" ") + return value.strip() if scheme.lower() == "bearer" and value else header.strip() + + def handle_auth(request, data={}): + # Personal access tokens are opaque (not JWTs) and resolve against the + # database in every auth mode that supports them, including AUTH_TYPE unset. + from docsgpt.api.pat.tokens import authenticate_pat, looks_like_pat + + bearer = _bearer_value(request) + if looks_like_pat(bearer): + return authenticate_pat(bearer, request) + if settings.AUTH_TYPE in ["simple_jwt", "session_jwt", "oidc"]: jwt_token = request.headers.get("Authorization") if not jwt_token: @@ -26,6 +47,8 @@ def handle_auth(request, data={}): # requirement is scoped to oidc. options={"verify_exp": is_oidc, "require_exp": is_oidc}, ) + for claim in _PAT_ONLY_CLAIMS: + decoded_token.pop(claim, None) return decoded_token except ExpiredSignatureError: return { diff --git a/tests/api/test_asgi_auth.py b/tests/api/test_asgi_auth.py index cf72eb43..341ded0d 100644 --- a/tests/api/test_asgi_auth.py +++ b/tests/api/test_asgi_auth.py @@ -114,3 +114,45 @@ 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 diff --git a/tests/api/test_pat_tokens.py b/tests/api/test_pat_tokens.py new file mode 100644 index 00000000..1e38d0ce --- /dev/null +++ b/tests/api/test_pat_tokens.py @@ -0,0 +1,275 @@ +"""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"} + + @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"}