Phase 7: Bot/extension system

Bot accounts (User rows with is_bot=True), scoped API tokens (read:messages,
write:messages, manage:rooms) authenticated via Authorization: Bearer on both
REST and the WS handshake, live bot WebSocket access on the same /ws/chat
endpoint humans use, message editing (WS "edit" envelope -> message_update
broadcast, fans out cross-instance for free via the existing broadcaster),
incoming webhooks (room-scoped, no auth beyond the URL token), and outgoing
webhooks/event subscriptions (HMAC-SHA256 signed, backgrounded delivery,
creation-time SSRF validation against private/loopback/link-local targets).

Token auth is additive, not a parallel system: a bearer-token-authenticated
bot goes through the exact same room-membership/role checks a session-
authenticated human does everywhere; only read:messages/write:messages are
separately scope-gated (the two message endpoints). manage:rooms scope
enforcement, full per-delivery SSRF re-validation, and bot API rate limiting
were explicitly scoped out (confirmed with the repo owner) as disproportionate
to this phase -- documented as known gaps in backend/README.md rather than
silently skipped.

Admin portal gains a Bots tab (create bots, issue/revoke scoped tokens,
cross-room webhook visibility); RoomInfoPanel gains room-scoped webhook/
subscription management, mirroring how invites already work there. The chat
UI also gets a minimal "edit your own message" affordance -- not asked for
by the issue, but the only practical way to exercise the edit pipeline by
hand instead of only via a scripted bot client.

Along the way: fixed a real bug caught while writing the incoming-webhook
test -- offline-push notification relied on the sender being "connected" to
exclude themselves, true for WS-originated messages but not for the new
webhook path, which has no WS connection for the attributed sender at all.
Now explicitly excluded. Also discovered the REST-only test fixture never
triggered ASGI lifespan, so app.state.broadcaster/presence didn't exist for
it; moved their construction out of the lifespan into create_app() itself
(Redis client construction is synchronous/lazy) so both the WS and
REST-only paths always have them.

New tests/test_bots.py, test_message_edit.py, test_webhooks.py (full suite
now 78/78, stable across repeated runs) plus a scripted end-to-end smoke
test (bot WS join/post/edit, incoming webhook, SSRF rejection, outgoing
delivery) and a full browser walkthrough of the new admin/room UI.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

backend/app/services/admin_service.py

@@ -2,10 +2,10 @@ import uuid
 
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.orm import selectinload
 
-from app.models import AdminAuditLog, Room, RoomMembership, RoomRole, User
+from app.models import Room, RoomMembership, RoomRole, User
 from app.security import hash_password
+from app.services.audit import record_audit_log
 
 
 class UserNotFoundError(Exception):
@@ -52,25 +52,6 @@ async def _get_membership(
     return membership
 
 
-def _log(
-    db: AsyncSession,
-    actor: User,
-    action: str,
-    target_type: str,
-    target_id: uuid.UUID,
-    metadata: dict | None = None,
-) -> None:
-    db.add(
-        AdminAuditLog(
-            actor_id=actor.id,
-            action=action,
-            target_type=target_type,
-            target_id=target_id,
-            metadata_=metadata,
-        )
-    )
-
-
 async def list_users(db: AsyncSession) -> list[User]:
     result = await db.execute(select(User).order_by(User.created_at))
     return list(result.scalars().all())
@@ -83,7 +64,7 @@ async def set_user_active(
         raise CannotActOnSelfError()
     user = await _get_user(db, target_user_id)
     user.is_active = active
-    _log(db, actor, "user.activate" if active else "user.deactivate", "user", user.id)
+    record_audit_log(db, actor, "user.activate" if active else "user.deactivate", "user", user.id)
     await db.commit()
     await db.refresh(user)
     return user
@@ -94,7 +75,7 @@ async def reset_user_password(
 ) -> None:
     user = await _get_user(db, target_user_id)
     user.password_hash = hash_password(new_password)
-    _log(db, actor, "user.reset_password", "user", user.id)
+    record_audit_log(db, actor, "user.reset_password", "user", user.id)
     await db.commit()
 
 
@@ -105,7 +86,7 @@ async def set_user_site_admin(
         raise CannotActOnSelfError()
     user = await _get_user(db, target_user_id)
     user.is_site_admin = is_admin
-    _log(db, actor, "user.promote" if is_admin else "user.demote", "user", user.id)
+    record_audit_log(db, actor, "user.promote" if is_admin else "user.demote", "user", user.id)
     await db.commit()
     await db.refresh(user)
     return user
@@ -126,7 +107,7 @@ async def set_room_archived(
 ) -> Room:
     room = await _get_room(db, room_id)
     room.is_archived = archived
-    _log(db, actor, "room.archive" if archived else "room.unarchive", "room", room.id)
+    record_audit_log(db, actor, "room.archive" if archived else "room.unarchive", "room", room.id)
     await db.commit()
     await db.refresh(room)
     return room
@@ -145,7 +126,7 @@ async def transfer_ownership_admin(
     new_owner_membership.role = RoomRole.owner
     current_owner_membership.role = RoomRole.admin
     room.owner_id = new_owner_id
-    _log(
+    record_audit_log(
         db,
         actor,
         "room.transfer_ownership",
@@ -156,16 +137,3 @@ async def transfer_ownership_admin(
     await db.commit()
     await db.refresh(room)
     return room
-
-
-async def list_audit_log(
-    db: AsyncSession, limit: int = 50, offset: int = 0
-) -> list[AdminAuditLog]:
-    result = await db.execute(
-        select(AdminAuditLog)
-        .options(selectinload(AdminAuditLog.actor))
-        .order_by(AdminAuditLog.created_at.desc())
-        .limit(limit)
-        .offset(offset)
-    )
-    return list(result.scalars().all())

backend/app/services/audit.py

@@ -0,0 +1,39 @@
+import uuid
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.models import AdminAuditLog, User
+
+
+def record_audit_log(
+    db: AsyncSession,
+    actor: User,
+    action: str,
+    target_type: str,
+    target_id: uuid.UUID,
+    metadata: dict | None = None,
+) -> None:
+    db.add(
+        AdminAuditLog(
+            actor_id=actor.id,
+            action=action,
+            target_type=target_type,
+            target_id=target_id,
+            metadata_=metadata,
+        )
+    )
+
+
+async def list_audit_log(
+    db: AsyncSession, limit: int = 50, offset: int = 0
+) -> list[AdminAuditLog]:
+    result = await db.execute(
+        select(AdminAuditLog)
+        .options(selectinload(AdminAuditLog.actor))
+        .order_by(AdminAuditLog.created_at.desc())
+        .limit(limit)
+        .offset(offset)
+    )
+    return list(result.scalars().all())

backend/app/services/bot_service.py

@@ -0,0 +1,119 @@
+import uuid
+from datetime import datetime, timezone
+
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.models import ApiToken, User
+from app.security import generate_token, hash_password, hash_token
+from app.services.audit import record_audit_log
+
+VALID_SCOPES = {"read:messages", "write:messages", "manage:rooms"}
+
+
+class DuplicateBotError(Exception):
+    pass
+
+
+class BotNotFoundError(Exception):
+    pass
+
+
+class InvalidScopeError(Exception):
+    pass
+
+
+class TokenNotFoundError(Exception):
+    pass
+
+
+async def create_bot(db: AsyncSession, actor: User, username: str) -> User:
+    # Bots never authenticate with a password -- generate one and discard
+    # it. email is NOT NULL/unique today; a placeholder avoids widening the
+    # schema just for accounts that will never receive real mail.
+    # bots.example.com (not e.g. bots.local/.invalid) deliberately, since
+    # pydantic's EmailStr rejects addresses whose *top-level* domain is one
+    # of the IANA special-use TLDs (.local, .invalid, .test, ...) -- a
+    # subdomain of the real (if reserved-for-docs) .com TLD isn't affected.
+    bot = User(
+        username=username,
+        email=f"{username}@bots.example.com",
+        password_hash=hash_password(generate_token()),
+        is_bot=True,
+    )
+    db.add(bot)
+    try:
+        await db.flush()
+    except IntegrityError as exc:
+        await db.rollback()
+        raise DuplicateBotError() from exc
+
+    record_audit_log(db, actor, "bot.create", "user", bot.id)
+    await db.commit()
+    await db.refresh(bot)
+    return bot
+
+
+async def list_bots(db: AsyncSession) -> list[User]:
+    result = await db.execute(
+        select(User).where(User.is_bot.is_(True)).order_by(User.created_at)
+    )
+    return list(result.scalars().all())
+
+
+async def _get_bot(db: AsyncSession, bot_id: uuid.UUID) -> User:
+    bot = await db.get(User, bot_id)
+    if bot is None or not bot.is_bot:
+        raise BotNotFoundError()
+    return bot
+
+
+async def create_api_token(
+    db: AsyncSession, actor: User, bot_id: uuid.UUID, scopes: list[str]
+) -> tuple[ApiToken, str]:
+    bot = await _get_bot(db, bot_id)
+    if not set(scopes) <= VALID_SCOPES:
+        raise InvalidScopeError()
+
+    plaintext = generate_token()
+    token = ApiToken(owner_id=bot.id, token_hash=hash_token(plaintext), scopes=scopes)
+    db.add(token)
+    record_audit_log(
+        db, actor, "bot.issue_token", "user", bot.id, {"scopes": scopes}
+    )
+    await db.commit()
+    await db.refresh(token)
+    return token, plaintext
+
+
+async def list_api_tokens(db: AsyncSession, bot_id: uuid.UUID) -> list[ApiToken]:
+    result = await db.execute(
+        select(ApiToken).where(ApiToken.owner_id == bot_id).order_by(ApiToken.created_at)
+    )
+    return list(result.scalars().all())
+
+
+async def revoke_api_token(db: AsyncSession, actor: User, token_id: uuid.UUID) -> None:
+    token = await db.get(ApiToken, token_id)
+    if token is None:
+        raise TokenNotFoundError()
+    await db.delete(token)
+    record_audit_log(db, actor, "bot.revoke_token", "user", token.owner_id)
+    await db.commit()
+
+
+async def resolve_token(db: AsyncSession, plaintext: str) -> tuple[User, ApiToken] | None:
+    result = await db.execute(
+        select(ApiToken)
+        .where(ApiToken.token_hash == hash_token(plaintext))
+        .options(selectinload(ApiToken.owner))
+    )
+    token = result.scalar_one_or_none()
+    if token is None or not token.owner.is_active:
+        return None
+
+    token.last_used_at = datetime.now(timezone.utc)
+    await db.commit()
+    return token.owner, token

backend/app/services/message_events.py

@@ -0,0 +1,79 @@
+import uuid
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models import Message, Room, RoomMembership, User
+from app.services.push_service import send_push_to_user
+from app.services.webhook_service import dispatch_event
+from app.ws.broadcaster import RoomBroadcaster
+from app.ws.presence import Presence
+
+
+async def _notify_offline_members(
+    db: AsyncSession, presence: Presence, room_id: uuid.UUID, sender: User, content: str
+) -> None:
+    result = await db.execute(
+        select(RoomMembership.user_id).where(RoomMembership.room_id == room_id)
+    )
+    member_ids = {row[0] for row in result.all()}
+    # Subtract the sender explicitly rather than relying on them being
+    # "connected" (true for the WS path, since they just sent this over an
+    # active connection -- not true for the incoming-webhook REST path,
+    # which has no WS connection for the attributed sender at all).
+    offline_ids = member_ids - await presence.connected_user_ids(room_id) - {sender.id}
+    if not offline_ids:
+        return
+
+    room = await db.get(Room, room_id)
+    payload = {
+        "title": f"#{room.name}" if room else "New message",
+        "body": f"{sender.username}: {content}"[:120],
+        "room_id": str(room_id),
+    }
+    for user_id in offline_ids:
+        await send_push_to_user(db, user_id, payload)
+
+
+def _message_payload(message: Message, username: str) -> dict:
+    return {
+        "type": "message",
+        "id": str(message.id),
+        "room_id": str(message.room_id),
+        "user_id": str(message.user_id),
+        "username": username,
+        "content": message.content,
+        "created_at": message.created_at.isoformat(),
+        "edited_at": message.edited_at.isoformat() if message.edited_at else None,
+    }
+
+
+async def broadcast_new_message(
+    db: AsyncSession,
+    broadcaster: RoomBroadcaster,
+    presence: Presence,
+    room_id: uuid.UUID,
+    message: Message,
+    sender: User,
+) -> None:
+    """The full side-effect sequence for a newly created message, shared by
+    the WS "message" handler and the incoming-webhook receiver so both
+    trigger identical fan-out/push/event behavior."""
+    payload = _message_payload(message, sender.username)
+    await broadcaster.publish(room_id, payload)
+    await _notify_offline_members(db, presence, room_id, sender, message.content)
+    await dispatch_event(db, "message.created", room_id, payload)
+
+
+async def broadcast_message_update(
+    db: AsyncSession, broadcaster: RoomBroadcaster, room_id: uuid.UUID, message: Message
+) -> None:
+    payload = {
+        "type": "message_update",
+        "id": str(message.id),
+        "room_id": str(room_id),
+        "content": message.content,
+        "edited_at": message.edited_at.isoformat() if message.edited_at else None,
+    }
+    await broadcaster.publish(room_id, payload)
+    await dispatch_event(db, "message.updated", room_id, payload)

backend/app/services/message_service.py

@@ -1,4 +1,5 @@
 import uuid
+from datetime import datetime, timezone
 
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -7,6 +8,14 @@ from sqlalchemy.orm import selectinload
 from app.models import Message
 
 
+class MessageNotFoundError(Exception):
+    pass
+
+
+class NotMessageAuthorError(Exception):
+    pass
+
+
 async def create_message(
     db: AsyncSession, room_id: uuid.UUID, user_id: uuid.UUID, content: str
 ) -> Message:
@@ -17,6 +26,22 @@ async def create_message(
     return message
 
 
+async def edit_message(
+    db: AsyncSession, message_id: uuid.UUID, editor_id: uuid.UUID, content: str
+) -> Message:
+    message = await db.get(Message, message_id)
+    if message is None:
+        raise MessageNotFoundError()
+    if message.user_id != editor_id:
+        raise NotMessageAuthorError()
+
+    message.content = content
+    message.edited_at = datetime.now(timezone.utc)
+    await db.commit()
+    await db.refresh(message)
+    return message
+
+
 async def list_recent_messages(
     db: AsyncSession, room_id: uuid.UUID, limit: int = 50
 ) -> list[Message]:

backend/app/services/ssrf.py

@@ -0,0 +1,38 @@
+import ipaddress
+import socket
+from urllib.parse import urlparse
+
+
+class UnsafeWebhookUrlError(Exception):
+    pass
+
+
+def validate_target_url(url: str) -> None:
+    """Creation-time-only SSRF check: rejects non-http(s) schemes and any
+    target whose hostname resolves to a private/loopback/link-local/
+    reserved/multicast address. Not re-checked per delivery, so this doesn't
+    defend against DNS rebinding between creation and a later send -- a
+    documented known limitation, not an oversight.
+    """
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        raise UnsafeWebhookUrlError()
+    if not parsed.hostname:
+        raise UnsafeWebhookUrlError()
+
+    try:
+        addrinfo = socket.getaddrinfo(parsed.hostname, None)
+    except socket.gaierror as exc:
+        raise UnsafeWebhookUrlError() from exc
+
+    for *_rest, sockaddr in addrinfo:
+        ip = ipaddress.ip_address(sockaddr[0])
+        if (
+            ip.is_private
+            or ip.is_loopback
+            or ip.is_link_local
+            or ip.is_reserved
+            or ip.is_multicast
+            or ip.is_unspecified
+        ):
+            raise UnsafeWebhookUrlError()

backend/app/services/webhook_delivery.py

@@ -0,0 +1,41 @@
+import hashlib
+import hmac
+import json
+import logging
+
+import httpx
+
+from app.models import EventSubscription
+
+logger = logging.getLogger(__name__)
+
+_TIMEOUT_SECONDS = 5.0
+
+
+async def deliver_event(subscription: EventSubscription, event_type: str, payload: dict) -> None:
+    """POST a signed event to a subscription's target_url.
+
+    Fire-once, best-effort: no retry/backoff, any failure is logged and
+    swallowed rather than raised -- a slow or dead third-party endpoint must
+    never affect message delivery to real room members. Safe to run in a
+    background asyncio.create_task (unlike the Phase 4 push lesson) because
+    there's no DB session involved here, just the already-serialized
+    payload and secret -- nothing that can outlive an event loop.
+    """
+    body = json.dumps({"event": event_type, "data": payload}).encode()
+    signature = hmac.new(subscription.signing_secret.encode(), body, hashlib.sha256).hexdigest()
+
+    try:
+        async with httpx.AsyncClient(timeout=_TIMEOUT_SECONDS) as client:
+            await client.post(
+                subscription.target_url,
+                content=body,
+                headers={
+                    "Content-Type": "application/json",
+                    "X-KeepItTalking-Signature": f"sha256={signature}",
+                },
+            )
+    except httpx.HTTPError:
+        logger.warning(
+            "Failed to deliver %s event to subscription %s", event_type, subscription.id
+        )

backend/app/services/webhook_service.py

@@ -0,0 +1,153 @@
+import asyncio
+import uuid
+
+from sqlalchemy import or_, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.models import EventSubscription, Message, Room, User, WebhookIncoming
+from app.security import generate_token
+from app.services.message_service import create_message
+from app.services.ssrf import validate_target_url
+from app.services.webhook_delivery import deliver_event
+
+VALID_EVENT_TYPES = {"message.created", "message.updated"}
+
+
+class WebhookNotFoundError(Exception):
+    pass
+
+
+class InvalidEventTypeError(Exception):
+    pass
+
+
+class SubscriptionNotFoundError(Exception):
+    pass
+
+
+async def create_incoming_webhook(
+    db: AsyncSession, actor: User, room_id: uuid.UUID, description: str | None
+) -> WebhookIncoming:
+    webhook = WebhookIncoming(
+        room_id=room_id, token=generate_token(), created_by=actor.id, description=description
+    )
+    db.add(webhook)
+    try:
+        await db.commit()
+    except IntegrityError:
+        # A token collision is astronomically unlikely (256 bits of
+        # randomness) -- surface it rather than silently masking it.
+        await db.rollback()
+        raise
+    await db.refresh(webhook)
+    return webhook
+
+
+async def list_incoming_webhooks(db: AsyncSession, room_id: uuid.UUID) -> list[WebhookIncoming]:
+    result = await db.execute(
+        select(WebhookIncoming)
+        .where(WebhookIncoming.room_id == room_id)
+        .order_by(WebhookIncoming.created_at)
+    )
+    return list(result.scalars().all())
+
+
+async def revoke_incoming_webhook(
+    db: AsyncSession, room_id: uuid.UUID, webhook_id: uuid.UUID
+) -> None:
+    webhook = await db.get(WebhookIncoming, webhook_id)
+    if webhook is None or webhook.room_id != room_id:
+        raise WebhookNotFoundError()
+    await db.delete(webhook)
+    await db.commit()
+
+
+async def list_all_incoming_webhooks_admin(db: AsyncSession) -> list[WebhookIncoming]:
+    result = await db.execute(
+        select(WebhookIncoming)
+        .options(selectinload(WebhookIncoming.room), selectinload(WebhookIncoming.creator))
+        .order_by(WebhookIncoming.created_at.desc())
+    )
+    return list(result.scalars().all())
+
+
+async def post_via_webhook(db: AsyncSession, token: str, content: str) -> tuple[Message, Room, User]:
+    result = await db.execute(
+        select(WebhookIncoming)
+        .where(WebhookIncoming.token == token)
+        .options(selectinload(WebhookIncoming.room), selectinload(WebhookIncoming.creator))
+    )
+    webhook = result.scalar_one_or_none()
+    if webhook is None:
+        raise WebhookNotFoundError()
+
+    message = await create_message(db, webhook.room_id, webhook.created_by, content)
+    return message, webhook.room, webhook.creator
+
+
+async def create_event_subscription(
+    db: AsyncSession,
+    actor: User,
+    room_id: uuid.UUID | None,
+    event_types: list[str],
+    target_url: str,
+) -> tuple[EventSubscription, str]:
+    if not set(event_types) <= VALID_EVENT_TYPES:
+        raise InvalidEventTypeError()
+    validate_target_url(target_url)
+
+    secret = generate_token()
+    subscription = EventSubscription(
+        room_id=room_id,
+        event_types=event_types,
+        target_url=target_url,
+        signing_secret=secret,
+        created_by=actor.id,
+    )
+    db.add(subscription)
+    await db.commit()
+    await db.refresh(subscription)
+    return subscription, secret
+
+
+async def list_event_subscriptions(db: AsyncSession, room_id: uuid.UUID) -> list[EventSubscription]:
+    result = await db.execute(
+        select(EventSubscription)
+        .where(EventSubscription.room_id == room_id)
+        .order_by(EventSubscription.created_at)
+    )
+    return list(result.scalars().all())
+
+
+async def revoke_event_subscription(
+    db: AsyncSession, room_id: uuid.UUID, subscription_id: uuid.UUID
+) -> None:
+    subscription = await db.get(EventSubscription, subscription_id)
+    if subscription is None or subscription.room_id != room_id:
+        raise SubscriptionNotFoundError()
+    await db.delete(subscription)
+    await db.commit()
+
+
+async def list_all_event_subscriptions_admin(db: AsyncSession) -> list[EventSubscription]:
+    result = await db.execute(
+        select(EventSubscription)
+        .options(selectinload(EventSubscription.room), selectinload(EventSubscription.creator))
+        .order_by(EventSubscription.created_at.desc())
+    )
+    return list(result.scalars().all())
+
+
+async def dispatch_event(
+    db: AsyncSession, event_type: str, room_id: uuid.UUID, payload: dict
+) -> None:
+    result = await db.execute(
+        select(EventSubscription).where(
+            or_(EventSubscription.room_id == room_id, EventSubscription.room_id.is_(None))
+        )
+    )
+    for subscription in result.scalars().all():
+        if event_type in subscription.event_types:
+            asyncio.create_task(deliver_event(subscription, event_type, payload))