Add kick extension and user management services

.gitignore

@@ -11,4 +11,5 @@ build/
 logs/*.log
 config/*.local.yaml
 .idea
-*.db
+*.db
+settings.json

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)

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)

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]

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)

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 == []

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

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]