|
# retoor <retoor@molodetz.nl>
|
|
|
|
import json
|
|
|
|
from tests.conftest import run_async
|
|
from devplacepy.services.devii.actions.catalog import PLATFORM_CATALOG
|
|
from devplacepy.services.devii.actions.dispatcher import Dispatcher
|
|
|
|
|
|
class _Recorder:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def record_system(self, event_key, **kwargs):
|
|
self.calls.append((event_key, kwargs))
|
|
return "rec"
|
|
|
|
|
|
class _FakeClient:
|
|
def __init__(self, authenticated=True):
|
|
self.authenticated = authenticated
|
|
self.username = "voidwalker"
|
|
|
|
|
|
def _bare_dispatcher(owner_kind, owner_id, *, is_admin=False, is_primary_admin=False):
|
|
dispatcher = Dispatcher.__new__(Dispatcher)
|
|
dispatcher._actions = PLATFORM_CATALOG.by_name()
|
|
dispatcher._virtual_tools = None
|
|
dispatcher._client = _FakeClient(authenticated=True)
|
|
dispatcher._is_admin = is_admin
|
|
dispatcher._is_primary_admin = is_primary_admin
|
|
dispatcher._owner_kind = owner_kind
|
|
dispatcher._owner_id = owner_id
|
|
return dispatcher
|
|
|
|
|
|
def _patch_recorder(monkeypatch):
|
|
from devplacepy.services.audit import record as audit_record
|
|
|
|
recorder = _Recorder()
|
|
monkeypatch.setattr(audit_record, "record_system", recorder.record_system)
|
|
return recorder
|
|
|
|
|
|
def test_admin_tool_denied_for_member_is_audited(monkeypatch):
|
|
recorder = _patch_recorder(monkeypatch)
|
|
dispatcher = _bare_dispatcher("user", "member-uid-123", is_admin=False)
|
|
|
|
result = run_async(
|
|
dispatcher.dispatch(
|
|
"admin_set_user_role", {"username": "voidwalker", "role": "Admin"}
|
|
)
|
|
)
|
|
|
|
assert json.loads(result)["error"] == "auth_required"
|
|
assert len(recorder.calls) == 1
|
|
event_key, kwargs = recorder.calls[0]
|
|
assert event_key == "security.authz.denied"
|
|
assert kwargs["origin"] == "devii"
|
|
assert kwargs["via_agent"] == 1
|
|
assert kwargs["result"] == "denied"
|
|
assert kwargs["actor_kind"] == "user"
|
|
assert kwargs["actor_uid"] == "member-uid-123"
|
|
assert kwargs["actor_role"] == "member"
|
|
assert kwargs["metadata"]["tool"] == "admin_set_user_role"
|
|
|
|
|
|
def test_admin_tool_denied_for_guest_actor_shape(monkeypatch):
|
|
recorder = _patch_recorder(monkeypatch)
|
|
dispatcher = _bare_dispatcher("guest", "cookie-abc", is_admin=False)
|
|
|
|
run_async(dispatcher.dispatch("admin_list_users", {}))
|
|
|
|
assert len(recorder.calls) == 1
|
|
_, kwargs = recorder.calls[0]
|
|
assert kwargs["actor_kind"] == "guest"
|
|
assert kwargs["actor_uid"] is None
|
|
assert kwargs["actor_role"] == "guest"
|
|
|
|
|
|
def test_admin_tool_allowed_for_admin_is_not_denial_audited(monkeypatch):
|
|
recorder = _patch_recorder(monkeypatch)
|
|
dispatcher = _bare_dispatcher("user", "admin-uid-9", is_admin=True)
|
|
|
|
run_async(dispatcher.dispatch("admin_list_users", {}))
|
|
|
|
assert all(call[0] != "security.authz.denied" for call in recorder.calls)
|
|
|
|
|
|
def test_primary_admin_tool_denied_for_regular_admin_is_audited(monkeypatch):
|
|
recorder = _patch_recorder(monkeypatch)
|
|
dispatcher = _bare_dispatcher(
|
|
"user", "admin-uid-9", is_admin=True, is_primary_admin=False
|
|
)
|
|
|
|
result = run_async(dispatcher.dispatch("db_list_tables", {}))
|
|
|
|
assert json.loads(result)["error"] == "auth_required"
|
|
assert len(recorder.calls) == 1
|
|
event_key, kwargs = recorder.calls[0]
|
|
assert event_key == "security.authz.denied"
|
|
assert kwargs["metadata"]["tool"] == "db_list_tables"
|
|
assert kwargs["actor_role"] == "admin"
|