288 lines
11 KiB
Python
288 lines
11 KiB
Python
"""Auth scope tests — DEV-1736 regression coverage.
|
|
|
|
Prior to this fix `auth()` walked the route's declared scope first and fell
|
|
through to a workspace check, so a `{w, p}` token authorized any peer in `w`.
|
|
The contract now is: authorize by the token's narrowest claim, never widen.
|
|
"""
|
|
|
|
from contextlib import asynccontextmanager
|
|
|
|
import jwt as pyjwt
|
|
import pytest
|
|
from fastapi.security import HTTPAuthorizationCredentials
|
|
|
|
from src.config import settings
|
|
from src.exceptions import AuthenticationException, ValidationException
|
|
from src.security import JWTParams, auth, create_jwt, verify_jwt
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _enable_auth(monkeypatch: pytest.MonkeyPatch): # pyright: ignore[reportUnusedFunction]
|
|
monkeypatch.setattr(settings.AUTH, "USE_AUTH", True)
|
|
monkeypatch.setattr(settings.AUTH, "JWT_SECRET", "test-secret")
|
|
|
|
|
|
def _bearer(token: str) -> HTTPAuthorizationCredentials:
|
|
return HTTPAuthorizationCredentials(scheme="Bearer", credentials=token)
|
|
|
|
|
|
class TestVerifyJWTShape:
|
|
def test_peer_token_without_workspace_rejected(self):
|
|
token = pyjwt.encode({"p": "alice"}, b"test-secret", algorithm="HS256")
|
|
with pytest.raises(AuthenticationException):
|
|
verify_jwt(token)
|
|
|
|
def test_session_token_without_workspace_rejected(self):
|
|
token = pyjwt.encode({"s": "sess-1"}, b"test-secret", algorithm="HS256")
|
|
with pytest.raises(AuthenticationException):
|
|
verify_jwt(token)
|
|
|
|
def test_workspace_only_token_ok(self):
|
|
token = create_jwt(JWTParams(w="ws-a"))
|
|
params = verify_jwt(token)
|
|
assert params.w == "ws-a"
|
|
|
|
def test_workspace_peer_token_ok(self):
|
|
token = create_jwt(JWTParams(w="ws-a", p="alice"))
|
|
params = verify_jwt(token)
|
|
assert params.w == "ws-a"
|
|
assert params.p == "alice"
|
|
|
|
|
|
class TestAuthPeerScope:
|
|
"""`{w: ws-a, p: alice}` may only act on alice in ws-a."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_matches_own_peer(self):
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a", p="alice")))
|
|
params = await auth(credentials=creds, workspace_name="ws-a", peer_name="alice")
|
|
assert params.p == "alice"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_denies_sibling_peer_same_workspace(self):
|
|
"""The original bug: peer-scoped token fell through to workspace auth."""
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a", p="alice")))
|
|
with pytest.raises(AuthenticationException):
|
|
await auth(credentials=creds, workspace_name="ws-a", peer_name="bob")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_denies_workspace_route_with_no_peer(self):
|
|
"""Peer-scoped token cannot use workspace-listing routes."""
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a", p="alice")))
|
|
with pytest.raises(AuthenticationException):
|
|
await auth(credentials=creds, workspace_name="ws-a")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_self_authorizing_route_receives_claims(self):
|
|
"""Body-scoped routes use require_auth() and compare claims in-handler."""
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a", p="alice")))
|
|
params = await auth(credentials=creds)
|
|
assert params.w == "ws-a"
|
|
assert params.p == "alice"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_denies_cross_workspace(self):
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a", p="alice")))
|
|
with pytest.raises(AuthenticationException):
|
|
await auth(credentials=creds, workspace_name="ws-b", peer_name="alice")
|
|
|
|
|
|
class TestAuthSessionScope:
|
|
@pytest.mark.asyncio
|
|
async def test_matches_own_session(self):
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a", s="sess-1")))
|
|
params = await auth(
|
|
credentials=creds, workspace_name="ws-a", session_name="sess-1"
|
|
)
|
|
assert params.s == "sess-1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_denies_sibling_session_same_workspace(self):
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a", s="sess-1")))
|
|
with pytest.raises(AuthenticationException):
|
|
await auth(credentials=creds, workspace_name="ws-a", session_name="sess-2")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_denies_workspace_route_with_no_session(self):
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a", s="sess-1")))
|
|
with pytest.raises(AuthenticationException):
|
|
await auth(credentials=creds, workspace_name="ws-a")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_self_authorizing_route_receives_claims(self):
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a", s="sess-1")))
|
|
params = await auth(credentials=creds)
|
|
assert params.w == "ws-a"
|
|
assert params.s == "sess-1"
|
|
|
|
|
|
class TestAuthWorkspaceScope:
|
|
@pytest.mark.asyncio
|
|
async def test_matches_workspace(self):
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a")))
|
|
params = await auth(credentials=creds, workspace_name="ws-a")
|
|
assert params.w == "ws-a"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_workspace_token_reaches_peer_route(self):
|
|
"""Workspace tokens still authorize narrower routes inside the workspace."""
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a")))
|
|
params = await auth(credentials=creds, workspace_name="ws-a", peer_name="alice")
|
|
assert params.w == "ws-a"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_denies_cross_workspace(self):
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a")))
|
|
with pytest.raises(AuthenticationException):
|
|
await auth(credentials=creds, workspace_name="ws-b")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_passes_self_authorizing_route(self):
|
|
"""Routes with no declared scope (e.g. POST /v3/workspaces) self-authorize
|
|
on the token's `w`. The auth dependency must let workspace tokens through."""
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a")))
|
|
params = await auth(credentials=creds)
|
|
assert params.w == "ws-a"
|
|
|
|
|
|
@asynccontextmanager
|
|
async def _fake_tracked_db(*_args: object, **_kwargs: object):
|
|
"""Stand-in for tracked_db; the membership query itself is monkeypatched."""
|
|
yield None
|
|
|
|
|
|
def _patch_membership(monkeypatch: pytest.MonkeyPatch, *, is_member: bool):
|
|
async def _is_peer_in_session(*_args: object, **_kwargs: object) -> bool:
|
|
return is_member
|
|
|
|
# Names are resolved via lazy imports inside auth(), so patch the source
|
|
# modules rather than the security namespace.
|
|
monkeypatch.setattr("src.dependencies.tracked_db", _fake_tracked_db)
|
|
monkeypatch.setattr("src.crud.session.is_peer_in_session", _is_peer_in_session)
|
|
|
|
|
|
class TestAuthMemberRead:
|
|
"""Peer-scoped key gets read-only access to sessions it is a member of."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_member_peer_allowed_on_read_route(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
_patch_membership(monkeypatch, is_member=True)
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a", p="alice")))
|
|
params = await auth(
|
|
credentials=creds,
|
|
workspace_name="ws-a",
|
|
session_name="sess-1",
|
|
allow_member_read=True,
|
|
)
|
|
assert params.p == "alice"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_member_peer_denied_on_read_route(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
_patch_membership(monkeypatch, is_member=False)
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a", p="alice")))
|
|
with pytest.raises(AuthenticationException):
|
|
await auth(
|
|
credentials=creds,
|
|
workspace_name="ws-a",
|
|
session_name="sess-1",
|
|
allow_member_read=True,
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_member_peer_denied_on_write_route(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
"""Write routes never set allow_member_read, so membership is irrelevant."""
|
|
_patch_membership(monkeypatch, is_member=True)
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a", p="alice")))
|
|
with pytest.raises(AuthenticationException):
|
|
await auth(
|
|
credentials=creds,
|
|
workspace_name="ws-a",
|
|
session_name="sess-1",
|
|
allow_member_read=False,
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_member_peer_denied_cross_workspace(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
_patch_membership(monkeypatch, is_member=True)
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a", p="alice")))
|
|
with pytest.raises(AuthenticationException):
|
|
await auth(
|
|
credentials=creds,
|
|
workspace_name="ws-b",
|
|
session_name="sess-1",
|
|
allow_member_read=True,
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_token_has_no_cross_scope_to_peer_routes(self):
|
|
"""A session key never reaches peer routes, even with allow_member_read."""
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a", s="sess-1")))
|
|
with pytest.raises(AuthenticationException):
|
|
await auth(
|
|
credentials=creds,
|
|
workspace_name="ws-a",
|
|
peer_name="alice",
|
|
allow_member_read=True,
|
|
)
|
|
|
|
|
|
class TestCreateKeyValidation:
|
|
@pytest.mark.asyncio
|
|
async def test_peer_key_without_workspace_rejected(self):
|
|
from src.routers.keys import create_key
|
|
|
|
with pytest.raises(ValidationException):
|
|
await create_key(workspace_id=None, peer_id="alice", session_id=None)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_key_without_workspace_rejected(self):
|
|
from src.routers.keys import create_key
|
|
|
|
with pytest.raises(ValidationException):
|
|
await create_key(workspace_id=None, peer_id=None, session_id="sess-1")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_peer_key_with_workspace_ok(self):
|
|
from src.routers.keys import create_key
|
|
|
|
result = await create_key(workspace_id="ws-a", peer_id="alice", session_id=None)
|
|
assert "key" in result
|
|
|
|
|
|
class TestAuthAdminAndUnscoped:
|
|
@pytest.mark.asyncio
|
|
async def test_admin_passes_any_route(self):
|
|
creds = _bearer(create_jwt(JWTParams(ad=True)))
|
|
params = await auth(credentials=creds, workspace_name="ws-a", peer_name="alice")
|
|
assert params.ad is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_admin_token_denied_on_admin_route(self):
|
|
creds = _bearer(create_jwt(JWTParams(w="ws-a")))
|
|
with pytest.raises(AuthenticationException):
|
|
await auth(credentials=creds, admin=True)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unscoped_token_on_self_authorizing_route(self):
|
|
"""A token with no scope claims and a route with no declared scope is the
|
|
escape hatch for routes that introspect jwt_params themselves."""
|
|
creds = _bearer(create_jwt(JWTParams()))
|
|
params = await auth(credentials=creds)
|
|
assert params.w is None
|
|
assert params.p is None
|
|
assert params.s is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unscoped_token_denied_on_scoped_route(self):
|
|
creds = _bearer(create_jwt(JWTParams()))
|
|
with pytest.raises(AuthenticationException):
|
|
await auth(credentials=creds, workspace_name="ws-a")
|