diff --git a/.gitignore b/.gitignore index c8d90b3..be5dfb5 100644 --- a/.gitignore +++ b/.gitignore @@ -11,4 +11,5 @@ build/ logs/*.log config/*.local.yaml .idea -*.db \ No newline at end of file +*.db +settings.json \ No newline at end of file diff --git a/bot/extensions/moderation/__init__.py b/bot/extensions/moderation/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/bot/extensions/moderation/kick_extension.py b/bot/extensions/moderation/kick_extension.py new file mode 100644 index 0000000..204524e --- /dev/null +++ b/bot/extensions/moderation/kick_extension.py @@ -0,0 +1,49 @@ +from matrix import Context, Extension +from matrix.errors import CheckError + +from bot.permissions import is_moderator +from .kick_service import kick_from_context +from .models import KickResult + +extension = Extension("kick") + +KICK_RESULT_TEMPLATE = ( + "{space}" + "Target: `{target}`\n" + "Kicked from: `{kicked}` rooms\n" + "Failed in: `{failed}` rooms\n" + "Reason: {reason}" +) + + +def format_kick_result(result: KickResult) -> str: + return KICK_RESULT_TEMPLATE.format( + space=f"Space: `{result.space_id}`\n" if result.space_id else "", + target=result.target_user_id, + kicked=len(result.kicked_room_ids), + failed=len(result.failed_room_ids), + reason=result.reason, + ) + + +@extension.command( + "kick", + description="Kick a user from all rooms in the current space", +) +async def kick( + ctx: Context, + user_id: str, + reason: str = "No reason provided", +) -> None: + result = await kick_from_context(ctx, user_id, reason) + await ctx.reply(format_kick_result(result)) + + +@kick.error(CheckError) +async def kick_check_error(ctx: Context, error: CheckError) -> None: + await ctx.reply(f"Could not complete kick operation: {error}") + + +@kick.check +async def can_kick(ctx: Context) -> bool: + return await is_moderator(ctx) diff --git a/bot/extensions/moderation/kick_service.py b/bot/extensions/moderation/kick_service.py new file mode 100644 index 0000000..cee7f79 --- /dev/null +++ b/bot/extensions/moderation/kick_service.py @@ -0,0 +1,70 @@ +from matrix import Context, Space, Room +from matrix.errors import MatrixError + +from .models import KickResult +from .space_service import get_parent_space + + +async def kick_from_rooms( + user_id: str, + rooms: list[Room], + *, + reason: str | None = None, + space: Space | None = None, +) -> KickResult: + kicked: list[str] = [] + failed: list[str] = [] + seen: set[str] = set() + + for room in rooms: + if room.room_id in seen: + continue + seen.add(room.room_id) + + try: + members = await room.get_members() + if user_id not in members: + # this will skip rooms where the user is not a member + continue + + await room.kick_user(user_id, reason=reason) + + members_after = await room.get_members() + if user_id in members_after: + # if the user is still in the room after the kick, + # we consider it a failure (should not happen) + failed.append(room.room_id) + else: + kicked.append(room.room_id) + + except MatrixError: + failed.append(room.room_id) + + return KickResult( + target_user_id=user_id, + reason=reason, + space_id=space.room_id if space else None, + kicked_room_ids=kicked, + failed_room_ids=failed, + ) + + +async def kick_from_context( + ctx: Context, + user_id: str, + reason: str | None = None, +) -> KickResult: + space: Space | None = get_parent_space(ctx) + + if space: + children = space.get_children(depth=3) + children.append(space) + + return await kick_from_rooms( + user_id, + children, + reason=reason, + space=space, + ) + + return await kick_from_rooms(user_id, [ctx.room], reason=reason) diff --git a/bot/extensions/moderation/models.py b/bot/extensions/moderation/models.py new file mode 100644 index 0000000..9ab20b1 --- /dev/null +++ b/bot/extensions/moderation/models.py @@ -0,0 +1,10 @@ +from dataclasses import dataclass + + +@dataclass(frozen=True) +class KickResult: + target_user_id: str + reason: str | None + space_id: str | None + kicked_room_ids: list[str] + failed_room_ids: list[str] diff --git a/bot/extensions/moderation/space_service.py b/bot/extensions/moderation/space_service.py new file mode 100644 index 0000000..0295dd0 --- /dev/null +++ b/bot/extensions/moderation/space_service.py @@ -0,0 +1,11 @@ +from matrix import Context, Space + + +def get_parent_space(ctx: Context) -> Space | None: + parent_ids: set[str] = ctx.room.matrix_room.parents + parent_id = next(iter(parent_ids), None) + + if not parent_id: + return None + + return ctx.bot.get_space(parent_id) diff --git a/bot/extensions/moderation/tests/__init__.py b/bot/extensions/moderation/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/bot/extensions/moderation/tests/test_kick_extension.py b/bot/extensions/moderation/tests/test_kick_extension.py new file mode 100644 index 0000000..d24bf83 --- /dev/null +++ b/bot/extensions/moderation/tests/test_kick_extension.py @@ -0,0 +1,163 @@ +import pytest +from matrix.errors import MatrixError +from types import SimpleNamespace +from typing import Any, cast +from unittest.mock import AsyncMock, MagicMock, patch + +from matrix import Context, Room, Space +from bot.extensions.moderation.kick_service import kick_from_context, kick_from_rooms + + +def as_async_mock(value: Any) -> AsyncMock: + return cast(AsyncMock, value) + + +def as_mock(value: Any) -> MagicMock: + return cast(MagicMock, value) + + +def mock_room( + room_id: str, + *, + kick_error: Exception | None = None, +) -> Room: + room = MagicMock(spec=Room) + room.room_id = room_id + room.name = room_id + + if kick_error: + cast(Any, room).get_members = AsyncMock(return_value=["@target:example.com"]) + else: + cast(Any, room).get_members = AsyncMock( + side_effect=[ + ["@target:example.com"], + [], + ] + ) + + cast(Any, room).kick_user = AsyncMock(side_effect=kick_error) + + return cast(Room, room) + + +def mock_space(room_id: str, children: list[Room | Space]) -> Space: + space = MagicMock(spec=Space) + space.room_id = room_id + space.name = room_id + cast(Any, space).get_children = MagicMock(return_value=children) + cast(Any, space).kick_user = AsyncMock() + return cast(Space, space) + + +@pytest.mark.asyncio +async def test_kick_from_rooms__skips_when_user_is_not_member() -> None: + room = mock_room("!room:example.com") + cast(Any, room).get_members = AsyncMock(return_value=[]) + + result = await kick_from_rooms( + "@target:example.com", + [room], + reason="spam", + ) + + as_async_mock(room.kick_user).assert_not_awaited() + assert result.kicked_room_ids == [] + assert result.failed_room_ids == [] + + +@pytest.mark.asyncio +async def test_kick_from_rooms__records_successes_failures_and_dedupes() -> None: + successful_room = mock_room("!success:example.com") + failed_room = mock_room( + "!failed:example.com", + kick_error=MatrixError("denied"), + ) + + result = await kick_from_rooms( + "@target:example.com", + [successful_room, failed_room, successful_room], + reason="spam", + ) + + as_async_mock(successful_room.kick_user).assert_awaited_once_with( + "@target:example.com", reason="spam" + ) + as_async_mock(failed_room.kick_user).assert_awaited_once_with( + "@target:example.com", reason="spam" + ) + + assert result.target_user_id == "@target:example.com" + assert result.reason == "spam" + assert result.space_id is None + assert result.kicked_room_ids == ["!success:example.com"] + assert result.failed_room_ids == ["!failed:example.com"] + + +@pytest.mark.asyncio +async def test_kick_from_rooms__includes_space_id_when_space_provided() -> None: + room = mock_room("!room:example.com") + space = mock_space("!space:example.com", []) + + result = await kick_from_rooms( + "@target:example.com", + [room], + reason="spam", + space=space, + ) + + assert result.space_id == "!space:example.com" + assert result.kicked_room_ids == ["!room:example.com"] + assert result.failed_room_ids == [] + + +@pytest.mark.asyncio +async def test_kick_from_context__without_parent_space__kicks_current_room() -> None: + current_room = mock_room("!current:example.com") + ctx = SimpleNamespace(room=current_room) + + with patch( + "bot.extensions.moderation.kick_service.get_parent_space", + return_value=None, + ): + result = await kick_from_context( + cast(Context, ctx), + "@target:example.com", + "spam", + ) + + as_async_mock(current_room.kick_user).assert_awaited_once_with( + "@target:example.com", reason="spam" + ) + assert result.kicked_room_ids == ["!current:example.com"] + assert result.failed_room_ids == [] + + +@pytest.mark.asyncio +async def test_kick_from_context__with_parent_space__kicks_room_children_only() -> None: + child_room = mock_room("!child-room:example.com") + child_space = mock_space("!child-space:example.com", []) + parent_space = mock_space( + "!parent-space:example.com", + [child_room, child_space], + ) + ctx = SimpleNamespace(room=mock_room("!current:example.com")) + + with patch( + "bot.extensions.moderation.kick_service.get_parent_space", + return_value=parent_space, + ): + result = await kick_from_context( + cast(Context, ctx), + "@target:example.com", + "spam", + ) + + as_mock(parent_space.get_children).assert_called_once_with(depth=3) + as_async_mock(child_room.kick_user).assert_awaited_once_with( + "@target:example.com", reason="spam" + ) + as_async_mock(child_space.kick_user).assert_not_awaited() + + assert result.space_id == "!parent-space:example.com" + assert result.kicked_room_ids == ["!child-room:example.com"] + assert result.failed_room_ids == [] diff --git a/bot/permissions.py b/bot/permissions.py new file mode 100644 index 0000000..dc63f77 --- /dev/null +++ b/bot/permissions.py @@ -0,0 +1,8 @@ +from matrix import Context + + +async def is_moderator(ctx: Context) -> bool: + power_levels = ctx.room.matrix_room.power_levels + sender_level = power_levels.get_user_level(ctx.sender) + + return sender_level >= power_levels.defaults.kick diff --git a/tests/test_permissions.py b/tests/test_permissions.py new file mode 100644 index 0000000..0887188 --- /dev/null +++ b/tests/test_permissions.py @@ -0,0 +1,40 @@ +import pytest +from types import SimpleNamespace + +from bot.permissions import is_moderator + + +def make_ctx(sender: str, sender_level: int, required_level: int = 50): + power_levels = SimpleNamespace( + defaults=SimpleNamespace(kick=required_level), + get_user_level=lambda user_id: sender_level, + ) + + room = SimpleNamespace( + matrix_room=SimpleNamespace( + power_levels=power_levels, + ) + ) + + return SimpleNamespace(sender=sender, room=room) + + +@pytest.mark.asyncio +async def test_is_moderator_accepts_user_at_required_power_level() -> None: + ctx = make_ctx("@mod:example.com", sender_level=50) + + assert await is_moderator(ctx) is True # type: ignore[arg-type] + + +@pytest.mark.asyncio +async def test_is_moderator_accepts_user_above_required_power_level() -> None: + ctx = make_ctx("@admin:example.com", sender_level=100) + + assert await is_moderator(ctx) is True # type: ignore[arg-type] + + +@pytest.mark.asyncio +async def test_is_moderator_rejects_user_below_required_power_level() -> None: + ctx = make_ctx("@user:example.com", sender_level=0) + + assert await is_moderator(ctx) is False # type: ignore[arg-type]