124 lines
3.9 KiB
Python
124 lines
3.9 KiB
Python
|
"""`AuthorizationService.require` 契约测试（替身 session，不连数据库）。
|
|||
|
|
|
|||
|
|
它是所有受保护操作的公共闸门，覆盖四点：
|
|||
|
|
1. 有权限时**直接返回**，不写审计、更不碰数据库——用"创建会话即报错"的替身证明；
|
|||
|
|
2. 拒绝时**开新事务**写 `permission.denied` 审计，并抛 `ForbiddenAgentError`；
|
|||
|
|
3. `admin=True` 时除了权限码还必须具备 admin/super_admin 角色；
|
|||
|
|
4. 审计里必须留下 permission 与 trace_id，供事后追责。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from typing import Any
|
|||
|
|
|
|||
|
|
import pytest
|
|||
|
|
|
|||
|
|
from app.core.contracts import RequestContext
|
|||
|
|
from app.core.errors import ForbiddenAgentError
|
|||
|
|
from app.service.authorization_service import AuthorizationService
|
|||
|
|
|
|||
|
|
BASE = RequestContext(user_id="9001", trace_id="trace-9")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def context_with(*permissions: str, roles: tuple[str, ...] = ()) -> RequestContext:
|
|||
|
|
return RequestContext(
|
|||
|
|
user_id="9001", trace_id="trace-9", permissions=tuple(permissions), roles=roles
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class _AsyncContext:
|
|||
|
|
async def __aenter__(self) -> None:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
async def __aexit__(self, *exc: object) -> bool:
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
|
|||
|
|
class FakeSession:
|
|||
|
|
def __init__(self, sink: list[Any]) -> None:
|
|||
|
|
self.sink = sink
|
|||
|
|
|
|||
|
|
def add(self, value: Any) -> None:
|
|||
|
|
self.sink.append(value)
|
|||
|
|
|
|||
|
|
def begin(self) -> Any:
|
|||
|
|
return _AsyncContext()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def patch_session(monkeypatch: pytest.MonkeyPatch, sink: list[Any]) -> None:
|
|||
|
|
class FakeFactory:
|
|||
|
|
async def __aenter__(self) -> FakeSession:
|
|||
|
|
return FakeSession(sink)
|
|||
|
|
|
|||
|
|
async def __aexit__(self, *exc: object) -> bool:
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
monkeypatch.setattr("app.service.authorization_service.SessionFactory", FakeFactory)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def explode_on_session() -> None:
|
|||
|
|
raise AssertionError("有权限时不应创建数据库会话")
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def test_allowed_permission_returns_without_database_access(
|
|||
|
|
monkeypatch: pytest.MonkeyPatch,
|
|||
|
|
) -> None:
|
|||
|
|
monkeypatch.setattr("app.service.authorization_service.SessionFactory", explode_on_session)
|
|||
|
|
|
|||
|
|
await AuthorizationService.require(
|
|||
|
|
context_with("conversation:feedback"), "conversation:feedback"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def test_denied_permission_writes_audit_and_raises(
|
|||
|
|
monkeypatch: pytest.MonkeyPatch,
|
|||
|
|
) -> None:
|
|||
|
|
sink: list[Any] = []
|
|||
|
|
patch_session(monkeypatch, sink)
|
|||
|
|
|
|||
|
|
with pytest.raises(ForbiddenAgentError):
|
|||
|
|
await AuthorizationService.require(context_with("agent:run"), "config:write")
|
|||
|
|
|
|||
|
|
assert len(sink) == 1
|
|||
|
|
audit = sink[0]
|
|||
|
|
assert audit.action_type == "permission.denied"
|
|||
|
|
assert audit.actor_id == 9001
|
|||
|
|
assert audit.actor_type == "user"
|
|||
|
|
assert audit.detail == {"permission": "config:write", "trace_id": "trace-9"}
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def test_admin_flag_requires_admin_role_even_with_permission(
|
|||
|
|
monkeypatch: pytest.MonkeyPatch,
|
|||
|
|
) -> None:
|
|||
|
|
"""权限码齐备但角色不是 admin：必须拒绝，否则普通角色能进管理面。"""
|
|||
|
|
sink: list[Any] = []
|
|||
|
|
patch_session(monkeypatch, sink)
|
|||
|
|
|
|||
|
|
with pytest.raises(ForbiddenAgentError):
|
|||
|
|
await AuthorizationService.require(
|
|||
|
|
context_with("config:read", roles=("customer",)), "config:read", admin=True
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert len(sink) == 1
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def test_admin_flag_passes_for_admin_role(monkeypatch: pytest.MonkeyPatch) -> None:
|
|||
|
|
monkeypatch.setattr("app.service.authorization_service.SessionFactory", explode_on_session)
|
|||
|
|
|
|||
|
|
await AuthorizationService.require(
|
|||
|
|
context_with("config:read", roles=("admin",)), "config:read", admin=True
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def test_admin_flag_still_requires_the_permission_code(
|
|||
|
|
monkeypatch: pytest.MonkeyPatch,
|
|||
|
|
) -> None:
|
|||
|
|
"""反之亦然：是 admin 角色但没有该权限码，同样拒绝。"""
|
|||
|
|
sink: list[Any] = []
|
|||
|
|
patch_session(monkeypatch, sink)
|
|||
|
|
|
|||
|
|
with pytest.raises(ForbiddenAgentError):
|
|||
|
|
await AuthorizationService.require(
|
|||
|
|
context_with(roles=("admin",)), "config:read", admin=True
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert len(sink) == 1
|