feat(pat): authenticate personal access tokens

handle_auth resolves a dgpt_pat_ bearer against the database instead of
decoding it as a JWT, for both the Flask and the ASGI routes. Scopes and the
resource filter always come from the token row, and the claims that mark a
PAT are stripped from decoded JWTs so a session token cannot pose as one.
ASGI routes reject tokens unless they name the scope that admits them, and
/api/user/me reports what the calling token may do.
This commit is contained in:
arc53-machine committed 2026-09-19 23:39:58 +01:00
1 parent 22ecc0aee3
commit 174130607a
8 files changed
+604 -3

No files matched your search

+42
View File
@@ -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