fix(pat): align replay scopes, keep 404/405, retire expired names, document the PAT_ENABLED switch

The ASGI message events route now accepts the same scopes as its Flask
sibling (conversations:read or chat:run) through a shared constant. A token
request that fails routing gets Flask's 404/405 instead of a 403. Creating a
token retires an expired token that still held the name. allowed_ids uses
is_pat instead of a bare literal. PAT_ENABLED is documented as the master
switch it is: turning it off stops every existing token from authenticating.
The docs explain that sources are matched by name (oldest wins) and point CI
flows at sources upload --replace.
This commit is contained in:
arc53-machine committed 2026-09-20 12:12:42 +01:00
1 parent 934ae1afb3
commit 7ddb5f6400
12 files changed
+105 -15

No files matched your search

+16
View File
@@ -164,3 +164,19 @@ class TestPersonalAccessTokens:
assert decoded is None
assert error.status_code == 403
assert json.loads(error.body)["error"] == "resource_not_allowed"
async def test_any_of_several_scopes_admits_the_token(self):
claims = dict(_PAT_CLAIMS, scopes=["conversations:read"])
with patch.object(asgi_auth, "handle_auth", return_value=claims):
decoded, error = await asgi_auth.authenticate(
_request(), pat_scope=("conversations:read", "chat:run")
)
assert error is None and decoded["sub"] == "alice"
async def test_message_events_accepts_the_same_scopes_as_message_tail(self):
from docsgpt.api import async_sse
from docsgpt.api.pat import rules
tail = rules.RULES[("/api/messages/<string:message_id>/tail", "GET")]
assert tail.scopes == rules.MESSAGE_REPLAY_SCOPES
assert async_sse.MESSAGE_REPLAY_SCOPES is rules.MESSAGE_REPLAY_SCOPES
+9
View File
@@ -127,6 +127,15 @@ class TestCreate:
assert _create(client).status_code == 201
assert _create(client).status_code == 409
def test_expired_token_does_not_reserve_its_name(self, client, db):
assert _create(client).status_code == 201
db.execute(text("UPDATE personal_access_tokens SET expires_at = now() - interval '1 day'"))
assert _create(client).status_code == 201
rows = db.execute(
text("SELECT status, revoke_reason FROM personal_access_tokens ORDER BY created_at")
).all()
assert [tuple(r) for r in rows] == [("revoked", "expired"), ("active", None)]
def test_per_user_cap(self, client, db, monkeypatch):
monkeypatch.setattr(pat_tokens.settings, "PAT_MAX_PER_USER", 1)
assert _create(client, name="one").status_code == 201
+5
View File
@@ -114,6 +114,11 @@ class TestScopeEnforcement:
response = _call(client, "GET", "/api/admin/users", _claims(list(SCOPES)))
assert _denied(response) == "not_available_to_tokens"
def test_unknown_path_and_wrong_method_keep_their_own_status(self, client):
claims = _claims(list(SCOPES))
assert _call(client, "GET", "/api/no_such_route", claims).status_code == 404
assert _call(client, "DELETE", "/api/get_agents", claims).status_code == 405
def test_missing_scope_is_refused_and_names_the_scope(self, client):
response = _call(client, "GET", "/api/get_agents", _claims(["sources:read"]))
assert _denied(response) == "insufficient_scope"
@@ -64,6 +64,24 @@ class TestUniqueness:
assert not repo.name_in_use("u1", "ci")
assert _create(repo, token_hash="h2")["name"] == "ci"
def test_retire_expired_name_only_touches_expired_tokens_with_that_name(self, pg_conn):
repo = PersonalAccessTokensRepository(pg_conn)
past = datetime.now(timezone.utc) - timedelta(days=1)
_create(repo, name="ci", token_hash="h1", expires_at=past)
_create(repo, name="other", token_hash="h2", expires_at=past)
_create(repo, user_id="u2", name="ci", token_hash="h3", expires_at=past)
assert repo.retire_expired_name("u1", "ci") == 1
assert not repo.name_in_use("u1", "ci")
assert repo.name_in_use("u1", "other") and repo.name_in_use("u2", "ci")
def test_retire_expired_name_leaves_live_tokens(self, pg_conn):
repo = PersonalAccessTokensRepository(pg_conn)
_create(repo, token_hash="h1")
_create(repo, name="later", token_hash="h2", expires_at=datetime.now(timezone.utc) + timedelta(days=1))
assert repo.retire_expired_name("u1", "ci") == 0
assert repo.retire_expired_name("u1", "later") == 0
assert repo.name_in_use("u1", "ci") and repo.name_in_use("u1", "later")
def test_same_name_for_other_user_is_fine(self, pg_conn):
repo = PersonalAccessTokensRepository(pg_conn)
_create(repo, user_id="u1", token_hash="h1")