MAID Network Protocol Enhancements Design Specification¶
Document Version: 1.0
Date: January 30, 2026
Status: Draft
Authors: MAID Development Team
Executive Summary¶
This document specifies enhancements to MAID's network layer to achieve feature parity with Evennia's mature network infrastructure. The three major enhancement areas are:
- SSL/TLS Support - Encrypted connections for Telnet and WebSocket
- External Service Bridges - IRC, Discord, and RSS integration
- Enhanced REST API - Full administrative and game API endpoints
These enhancements address the gaps identified in the MAID vs Evennia comparison where Evennia leads in secure connections, external service integration, and web API capabilities.
Table of Contents¶
- Feature 1: SSL/TLS Support
- Feature 2: External Service Bridges
- Feature 3: Enhanced REST API
- Appendix A: Security Considerations
- Appendix B: Testing Requirements
Feature 1: SSL/TLS Support¶
1.1 Feature Overview¶
What it does:
Provides encrypted connections for both Telnet (TLS) and WebSocket (WSS/HTTPS) protocols, protecting player credentials and game data in transit.
Why it's needed: - Current state: All connections are plaintext (Telnet port 4000, WebSocket port 8080) - Player passwords transmitted in clear text - Session tokens can be intercepted - Evennia provides native SSL/TLS and SSH support
Current Limitation:
asyncio.start_server() in server.py and uvicorn config in web/server.py have no SSL context configured.
1.2 User Stories¶
US-1.1: Secure Telnet Connections
As a server administrator, I want to enable TLS encryption for Telnet connections so that player credentials are protected.
US-1.2: Secure WebSocket Connections
As a server administrator, I want to enable WSS (WebSocket Secure) so that browser clients connect securely.
US-1.3: Certificate Management
As a server administrator, I want to configure SSL certificates via environment variables so that I can use Let's Encrypt or other CAs.
US-1.4: Mixed Mode Support
As a server administrator, I want to run both encrypted and unencrypted ports simultaneously so that I can support legacy clients during migration.
US-1.5: Automatic Certificate Reload
As a server administrator, I want certificates to reload without server restart so that Let's Encrypt renewals don't cause downtime.
1.3 Technical Requirements¶
1.3.1 Core Requirements¶
| ID | Requirement | Priority |
|---|---|---|
| SSL-001 | System SHALL support TLS 1.2+ for Telnet connections | P0 |
| SSL-002 | System SHALL support WSS (HTTPS) for WebSocket connections | P0 |
| SSL-003 | System SHALL support configurable certificate and key paths | P0 |
| SSL-004 | System SHALL support optional client certificate verification | P2 |
| SSL-005 | System SHALL support running secure and insecure ports simultaneously | P1 |
| SSL-006 | System SHALL support certificate chain files | P1 |
| SSL-007 | System SHALL support hot-reload of certificates | P2 |
| SSL-008 | System SHALL log SSL connection details (cipher, protocol version) | P1 |
| SSL-009 | System SHALL support configurable minimum TLS version | P1 |
| SSL-010 | System SHALL support configurable cipher suites | P2 |
1.3.2 Configuration Requirements¶
| ID | Requirement | Priority |
|---|---|---|
| CFG-001 | SSL settings SHALL be configurable via environment variables | P0 |
| CFG-002 | SSL settings SHALL be optional (default to disabled) | P0 |
| CFG-003 | System SHALL validate certificate files on startup | P0 |
| CFG-004 | System SHALL provide clear error messages for invalid certificates | P0 |
1.4 API/Interface Design¶
1.4.1 Configuration Settings¶
# packages/maid-engine/src/maid_engine/config/settings.py
from pydantic import BaseModel, Field, field_validator
from pathlib import Path
from typing import Literal
import ssl
class SSLSettings(BaseModel):
"""SSL/TLS configuration settings."""
enabled: bool = Field(
default=False,
description="Enable SSL/TLS encryption",
)
cert_path: Path | None = Field(
default=None,
description="Path to SSL certificate file (PEM format)",
)
key_path: Path | None = Field(
default=None,
description="Path to SSL private key file (PEM format)",
)
key_password: str | None = Field(
default=None,
description="Password for encrypted private key",
)
ca_path: Path | None = Field(
default=None,
description="Path to CA bundle for certificate chain",
)
min_version: Literal["TLS1_2", "TLS1_3"] = Field(
default="TLS1_2",
description="Minimum TLS version to accept",
)
verify_client: bool = Field(
default=False,
description="Require client certificates",
)
client_ca_path: Path | None = Field(
default=None,
description="Path to CA for client certificate verification",
)
ciphers: str | None = Field(
default=None,
description="OpenSSL cipher string (e.g., 'HIGH:!aNULL:!MD5')",
)
@field_validator("cert_path", "key_path", "ca_path", "client_ca_path", mode="after")
@classmethod
def validate_path_exists(cls, v: Path | None) -> Path | None:
if v is not None and not v.exists():
raise ValueError(f"File not found: {v}")
return v
class TelnetSettings(BaseModel):
"""Telnet server settings - extended with SSL."""
host: str = Field(default="0.0.0.0")
port: int = Field(default=4000, ge=1, le=65535)
max_connections: int = Field(default=1000)
idle_timeout: int = Field(default=300)
# SSL for Telnet
ssl: SSLSettings = Field(default_factory=SSLSettings)
ssl_port: int | None = Field(
default=None,
description="Separate port for TLS connections (if different from main port)",
)
class WebSettings(BaseModel):
"""Web server settings - extended with SSL."""
host: str = Field(default="0.0.0.0")
port: int = Field(default=8080, ge=1, le=65535)
cors_enabled: bool = Field(default=True)
cors_origins: list[str] = Field(default_factory=lambda: ["*"])
# SSL for WebSocket/HTTPS
ssl: SSLSettings = Field(default_factory=SSLSettings)
https_port: int | None = Field(
default=None,
description="Separate port for HTTPS (if different from main port)",
)
redirect_http_to_https: bool = Field(
default=False,
description="Redirect HTTP requests to HTTPS",
)
1.4.2 SSL Context Builder¶
# packages/maid-engine/src/maid_engine/net/ssl_context.py
from __future__ import annotations
import ssl
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from maid_engine.config.settings import SSLSettings
logger = logging.getLogger(__name__)
class SSLError(Exception):
"""Error creating or managing SSL context."""
pass
@dataclass
class SSLInfo:
"""Information about an SSL connection."""
version: str
cipher: str
cipher_bits: int
client_cert_subject: str | None = None
class SSLContextBuilder:
"""Builds SSL contexts from settings."""
@staticmethod
def create_server_context(settings: SSLSettings) -> ssl.SSLContext:
"""Create SSL context for server socket.
Args:
settings: SSL configuration settings
Returns:
Configured SSLContext for server use
Raises:
SSLError: If configuration is invalid
"""
if not settings.enabled:
raise SSLError("SSL is not enabled")
if not settings.cert_path or not settings.key_path:
raise SSLError("SSL enabled but cert_path or key_path not provided")
try:
# Create context
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
# Set minimum TLS version
if settings.min_version == "TLS1_3":
context.minimum_version = ssl.TLSVersion.TLSv1_3
else:
context.minimum_version = ssl.TLSVersion.TLSv1_2
# Load certificate chain
context.load_cert_chain(
certfile=str(settings.cert_path),
keyfile=str(settings.key_path),
password=settings.key_password,
)
# Load CA bundle if provided
if settings.ca_path:
context.load_verify_locations(cafile=str(settings.ca_path))
# Configure client verification
if settings.verify_client:
context.verify_mode = ssl.CERT_REQUIRED
if settings.client_ca_path:
context.load_verify_locations(cafile=str(settings.client_ca_path))
else:
context.verify_mode = ssl.CERT_NONE
# Set ciphers if specified
if settings.ciphers:
context.set_ciphers(settings.ciphers)
else:
# Use secure defaults
context.set_ciphers("HIGH:!aNULL:!MD5:!RC4")
logger.info(
f"SSL context created: TLS {settings.min_version}+, "
f"cert={settings.cert_path}"
)
return context
except ssl.SSLError as e:
raise SSLError(f"Failed to create SSL context: {e}")
except FileNotFoundError as e:
raise SSLError(f"SSL file not found: {e}")
@staticmethod
def get_connection_info(ssl_object: ssl.SSLObject | ssl.SSLSocket) -> SSLInfo:
"""Extract SSL connection information.
Args:
ssl_object: Active SSL connection
Returns:
SSLInfo with connection details
"""
cipher = ssl_object.cipher()
version = ssl_object.version() or "Unknown"
cipher_name = cipher[0] if cipher else "Unknown"
cipher_bits = cipher[2] if cipher and len(cipher) > 2 else 0
# Get client cert info if available
client_cert = ssl_object.getpeercert()
client_subject = None
if client_cert:
subject = client_cert.get("subject", ())
for rdn in subject:
for name, value in rdn:
if name == "commonName":
client_subject = value
break
return SSLInfo(
version=version,
cipher=cipher_name,
cipher_bits=cipher_bits,
client_cert_subject=client_subject,
)
class CertificateReloader:
"""Handles hot-reloading of SSL certificates."""
def __init__(self, settings: SSLSettings):
self.settings = settings
self._context: ssl.SSLContext | None = None
self._cert_mtime: float = 0
self._key_mtime: float = 0
@property
def context(self) -> ssl.SSLContext:
"""Get current SSL context, reloading if certificates changed."""
self._check_reload()
if self._context is None:
self._load_context()
return self._context
def _check_reload(self) -> None:
"""Check if certificates have changed and reload if so."""
if not self.settings.cert_path or not self.settings.key_path:
return
cert_mtime = self.settings.cert_path.stat().st_mtime
key_mtime = self.settings.key_path.stat().st_mtime
if cert_mtime != self._cert_mtime or key_mtime != self._key_mtime:
logger.info("Certificate files changed, reloading SSL context")
self._load_context()
self._cert_mtime = cert_mtime
self._key_mtime = key_mtime
def _load_context(self) -> None:
"""Load a fresh SSL context."""
self._context = SSLContextBuilder.create_server_context(self.settings)
1.4.3 Integration with Telnet Server¶
# Modifications to packages/maid-engine/src/maid_engine/net/server.py
import ssl
from maid_engine.net.ssl_context import SSLContextBuilder, CertificateReloader
class MAIDServer:
"""Extended with SSL support."""
async def _start_telnet_servers(self) -> None:
"""Start Telnet servers (optionally with TLS)."""
settings = self._settings.telnet
# Start plain Telnet (if SSL not enabled, or if we want both)
if not settings.ssl.enabled or settings.ssl_port:
self._telnet_server = await asyncio.start_server(
self._handle_telnet_connection,
host=settings.host,
port=settings.port,
)
logger.info(f"Telnet server listening on {settings.host}:{settings.port}")
# Start TLS Telnet
if settings.ssl.enabled:
self._cert_reloader = CertificateReloader(settings.ssl)
ssl_port = settings.ssl_port or settings.port
self._telnet_ssl_server = await asyncio.start_server(
self._handle_telnet_connection,
host=settings.host,
port=ssl_port,
ssl=self._cert_reloader.context,
)
logger.info(f"Telnet+TLS server listening on {settings.host}:{ssl_port}")
async def _handle_telnet_connection(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
) -> None:
"""Handle Telnet connection (plain or TLS)."""
# Check if TLS connection
transport = writer.transport
ssl_object = transport.get_extra_info("ssl_object")
if ssl_object:
ssl_info = SSLContextBuilder.get_connection_info(ssl_object)
logger.info(
f"TLS connection: {ssl_info.version}, "
f"cipher={ssl_info.cipher} ({ssl_info.cipher_bits} bits)"
)
# Continue with normal connection handling...
session = TelnetSession(reader, writer)
session.ssl_info = ssl_info if ssl_object else None
# ... rest of connection handling
1.4.4 Integration with Web Server¶
# Modifications to packages/maid-engine/src/maid_engine/net/web/server.py
import uvicorn
from maid_engine.net.ssl_context import SSLContextBuilder
class WebServer:
"""Extended with SSL support."""
def _create_uvicorn_config(self) -> uvicorn.Config:
"""Create uvicorn configuration with optional SSL."""
settings = self._settings.web
config_kwargs = {
"app": self._app,
"host": settings.host,
"port": settings.port,
"log_level": "info",
}
if settings.ssl.enabled:
config_kwargs.update({
"ssl_certfile": str(settings.ssl.cert_path),
"ssl_keyfile": str(settings.ssl.key_path),
"ssl_keyfile_password": settings.ssl.key_password,
})
if settings.ssl.ca_path:
config_kwargs["ssl_ca_certs"] = str(settings.ssl.ca_path)
logger.info(
f"WebSocket server will use HTTPS on {settings.host}:{settings.port}"
)
return uvicorn.Config(**config_kwargs)
def _add_https_redirect_middleware(self) -> None:
"""Add middleware to redirect HTTP to HTTPS."""
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware
if self._settings.web.ssl.enabled and self._settings.web.redirect_http_to_https:
self._app.add_middleware(HTTPSRedirectMiddleware)
1.5 Environment Variable Configuration¶
# Telnet TLS
MAID_TELNET__SSL__ENABLED=true
MAID_TELNET__SSL__CERT_PATH=/etc/maid/cert.pem
MAID_TELNET__SSL__KEY_PATH=/etc/maid/key.pem
MAID_TELNET__SSL__KEY_PASSWORD=secret
MAID_TELNET__SSL__MIN_VERSION=TLS1_2
MAID_TELNET__SSL_PORT=4443
# WebSocket HTTPS
MAID_WEB__SSL__ENABLED=true
MAID_WEB__SSL__CERT_PATH=/etc/maid/cert.pem
MAID_WEB__SSL__KEY_PATH=/etc/maid/key.pem
MAID_WEB__HTTPS_PORT=8443
MAID_WEB__REDIRECT_HTTP_TO_HTTPS=true
1.6 Acceptance Criteria¶
| ID | Criterion | Verification |
|---|---|---|
| AC-1.1 | Telnet connections work with TLS 1.2+ | Manual test with openssl s_client |
| AC-1.2 | WebSocket connections work with WSS | Manual test with browser |
| AC-1.3 | Invalid certificates produce clear startup error | Unit test |
| AC-1.4 | Missing key produces clear startup error | Unit test |
| AC-1.5 | SSL disabled by default (backward compatible) | Unit test |
| AC-1.6 | Both secure and insecure ports can run simultaneously | Integration test |
| AC-1.7 | SSL connection details logged | Log verification |
| AC-1.8 | Certificate hot-reload works without restart | Integration test |
Feature 2: External Service Bridges¶
2.1 Feature Overview¶
What it does:
Provides bidirectional bridges between MAID game channels and external communication services (Discord, IRC, RSS). Players can chat with users on external platforms, and vice versa.
Why it's needed: - Current state: No external service integration - Modern gaming communities use Discord as primary communication - IRC remains popular for some MUD communities - RSS provides game news/updates to external readers - Evennia provides IRC, Discord, and RSS bridge contribs
Current Opportunity:
MAID's GMCP already has External.Discord.Hello package defined, indicating design consideration for Discord integration.
2.2 User Stories¶
US-2.1: Discord Channel Bridge
As a server administrator, I want to link a Discord text channel to an in-game chat channel so that Discord users and players can communicate.
US-2.2: Discord Webhook Notifications
As a server administrator, I want game events (player logins, achievements, world events) sent to Discord so that the community stays informed.
US-2.3: IRC Channel Bridge
As a server administrator, I want to link an IRC channel to an in-game chat channel so that IRC users and players can communicate.
US-2.4: RSS Game Feed
As a server administrator, I want to expose game news and announcements as an RSS feed so that players can subscribe in their feed readers.
US-2.5: Multi-Service Bridge
As a server administrator, I want to bridge one in-game channel to multiple external services so that Discord, IRC, and in-game all share messages.
2.3 Technical Requirements¶
2.3.1 Discord Bridge Requirements¶
| ID | Requirement | Priority |
|---|---|---|
| DISC-001 | System SHALL support Discord bot integration via discord.py | P0 |
| DISC-002 | System SHALL support bidirectional message relay | P0 |
| DISC-003 | System SHALL format Discord messages for MUD display (strip markdown, handle embeds) | P0 |
| DISC-004 | System SHALL format MUD messages for Discord display (escape markdown) | P0 |
| DISC-005 | System SHALL support webhook-only mode for one-way notifications | P1 |
| DISC-006 | System SHALL support multiple channel mappings | P1 |
| DISC-007 | System SHALL handle Discord rate limits gracefully | P0 |
| DISC-008 | System SHALL reconnect on disconnect | P0 |
| DISC-009 | System SHALL support Discord slash commands for game queries | P2 |
| DISC-010 | System SHALL support rich embeds for game events | P1 |
2.3.2 IRC Bridge Requirements¶
| ID | Requirement | Priority |
|---|---|---|
| IRC-001 | System SHALL support IRC client connection via irc3 or similar | P0 |
| IRC-002 | System SHALL support bidirectional message relay | P0 |
| IRC-003 | System SHALL support IRC user authentication (NickServ) | P1 |
| IRC-004 | System SHALL support SSL/TLS IRC connections | P1 |
| IRC-005 | System SHALL support multiple channel mappings | P1 |
| IRC-006 | System SHALL handle IRC disconnects with reconnection | P0 |
| IRC-007 | System SHALL support IRC colors to/from MUD colors | P2 |
2.3.3 RSS Feed Requirements¶
| ID | Requirement | Priority |
|---|---|---|
| RSS-001 | System SHALL expose RSS 2.0 feed endpoint | P0 |
| RSS-002 | System SHALL support Atom feed format | P1 |
| RSS-003 | System SHALL include game announcements in feed | P0 |
| RSS-004 | System SHALL include major world events in feed | P1 |
| RSS-005 | System SHALL support configurable feed size (last N items) | P0 |
| RSS-006 | System SHALL support ETag/If-Modified-Since caching | P1 |
2.4 API/Interface Design¶
2.4.1 Bridge Protocol¶
# packages/maid-engine/src/maid_engine/bridges/protocol.py
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, Callable
from uuid import UUID
if TYPE_CHECKING:
from maid_engine.core.engine import GameEngine
class BridgeState(Enum):
"""State of an external bridge connection."""
DISCONNECTED = auto()
CONNECTING = auto()
CONNECTED = auto()
RECONNECTING = auto()
ERROR = auto()
@dataclass
class ExternalUser:
"""Represents a user from an external service."""
service: str # "discord", "irc"
user_id: str
display_name: str
avatar_url: str | None = None
roles: list[str] = field(default_factory=list)
is_admin: bool = False
@dataclass
class BridgeMessage:
"""A message crossing the bridge."""
content: str
source_service: str # "game", "discord", "irc"
source_channel: str
author: ExternalUser | str # ExternalUser or player name
timestamp: datetime = field(default_factory=datetime.utcnow)
attachments: list[str] = field(default_factory=list)
reply_to: str | None = None
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass
class ChannelMapping:
"""Maps a game channel to external channel(s)."""
game_channel: str
discord_channel_id: str | None = None
discord_webhook_url: str | None = None
irc_channel: str | None = None
bidirectional: bool = True
format_template: str = "[{service}] {author}: {content}"
class ExternalBridge(ABC):
"""Base class for external service bridges."""
def __init__(self, engine: GameEngine, config: dict[str, Any]):
self.engine = engine
self.config = config
self.state = BridgeState.DISCONNECTED
self._message_handlers: list[Callable[[BridgeMessage], None]] = []
@property
@abstractmethod
def service_name(self) -> str:
"""Return the service name (e.g., 'discord', 'irc')."""
pass
@abstractmethod
async def connect(self) -> None:
"""Establish connection to external service."""
pass
@abstractmethod
async def disconnect(self) -> None:
"""Disconnect from external service."""
pass
@abstractmethod
async def send_message(self, channel_id: str, message: str, **kwargs) -> None:
"""Send a message to an external channel."""
pass
@abstractmethod
async def send_embed(self, channel_id: str, embed: dict[str, Any]) -> None:
"""Send a rich embed to an external channel."""
pass
def on_message(self, handler: Callable[[BridgeMessage], None]) -> None:
"""Register a message handler for incoming messages."""
self._message_handlers.append(handler)
async def _dispatch_message(self, message: BridgeMessage) -> None:
"""Dispatch incoming message to handlers."""
for handler in self._message_handlers:
try:
await handler(message)
except Exception as e:
import logging
logging.getLogger(__name__).exception(f"Bridge message handler error: {e}")
class BridgeManager:
"""Manages all external bridges and channel mappings."""
def __init__(self, engine: GameEngine):
self.engine = engine
self._bridges: dict[str, ExternalBridge] = {}
self._mappings: list[ChannelMapping] = []
def register_bridge(self, bridge: ExternalBridge) -> None:
"""Register an external bridge."""
self._bridges[bridge.service_name] = bridge
bridge.on_message(self._handle_external_message)
def add_channel_mapping(self, mapping: ChannelMapping) -> None:
"""Add a channel mapping."""
self._mappings.append(mapping)
async def start_all(self) -> None:
"""Start all registered bridges."""
for bridge in self._bridges.values():
await bridge.connect()
async def stop_all(self) -> None:
"""Stop all registered bridges."""
for bridge in self._bridges.values():
await bridge.disconnect()
async def relay_to_external(
self,
game_channel: str,
player_name: str,
message: str,
) -> None:
"""Relay a game message to external services."""
for mapping in self._mappings:
if mapping.game_channel != game_channel:
continue
if mapping.discord_channel_id:
discord = self._bridges.get("discord")
if discord:
formatted = mapping.format_template.format(
service="MUD",
author=player_name,
content=message,
)
await discord.send_message(mapping.discord_channel_id, formatted)
if mapping.irc_channel:
irc = self._bridges.get("irc")
if irc:
formatted = mapping.format_template.format(
service="MUD",
author=player_name,
content=message,
)
await irc.send_message(mapping.irc_channel, formatted)
async def _handle_external_message(self, message: BridgeMessage) -> None:
"""Handle incoming message from external service."""
for mapping in self._mappings:
if not mapping.bidirectional:
continue
# Match message to mapping
matched = False
if message.source_service == "discord" and mapping.discord_channel_id == message.source_channel:
matched = True
elif message.source_service == "irc" and mapping.irc_channel == message.source_channel:
matched = True
if matched:
# Relay to game channel
await self._relay_to_game(mapping.game_channel, message)
async def _relay_to_game(self, game_channel: str, message: BridgeMessage) -> None:
"""Relay external message to game channel."""
author_name = message.author if isinstance(message.author, str) else message.author.display_name
formatted = f"[{message.source_service.upper()}] {author_name}: {message.content}"
# Broadcast to all players subscribed to this channel
# This would use the game's channel system
await self.engine.broadcast_channel(game_channel, formatted)
2.4.2 Discord Bridge Implementation¶
# packages/maid-engine/src/maid_engine/bridges/discord_bridge.py
from __future__ import annotations
import asyncio
import logging
from typing import TYPE_CHECKING, Any
try:
import discord
from discord.ext import commands
DISCORD_AVAILABLE = True
except ImportError:
DISCORD_AVAILABLE = False
from .protocol import (
BridgeMessage,
BridgeState,
ExternalBridge,
ExternalUser,
)
if TYPE_CHECKING:
from maid_engine.core.engine import GameEngine
logger = logging.getLogger(__name__)
class DiscordBridgeConfig:
"""Discord bridge configuration."""
def __init__(self, config: dict[str, Any]):
self.bot_token: str = config["bot_token"]
self.guild_id: int | None = config.get("guild_id")
self.intents: list[str] = config.get("intents", ["guilds", "guild_messages", "message_content"])
self.status_message: str = config.get("status_message", "Playing MAID")
self.reconnect_delay: int = config.get("reconnect_delay", 30)
self.webhook_urls: dict[str, str] = config.get("webhook_urls", {})
class DiscordBridge(ExternalBridge):
"""Discord integration bridge."""
def __init__(self, engine: GameEngine, config: dict[str, Any]):
super().__init__(engine, config)
if not DISCORD_AVAILABLE:
raise ImportError("discord.py not installed. Install with: pip install discord.py")
self.bridge_config = DiscordBridgeConfig(config)
# Set up intents
intents = discord.Intents.default()
if "message_content" in self.bridge_config.intents:
intents.message_content = True
if "guild_messages" in self.bridge_config.intents:
intents.guild_messages = True
# Create bot
self._bot = commands.Bot(
command_prefix="!",
intents=intents,
)
# Register event handlers
self._bot.event(self._on_ready)
self._bot.event(self._on_message)
self._bot.event(self._on_disconnect)
self._task: asyncio.Task | None = None
@property
def service_name(self) -> str:
return "discord"
async def connect(self) -> None:
"""Connect to Discord."""
if self.state == BridgeState.CONNECTED:
return
self.state = BridgeState.CONNECTING
logger.info("Connecting to Discord...")
self._task = asyncio.create_task(
self._bot.start(self.bridge_config.bot_token)
)
async def disconnect(self) -> None:
"""Disconnect from Discord."""
logger.info("Disconnecting from Discord...")
await self._bot.close()
if self._task:
self._task.cancel()
self.state = BridgeState.DISCONNECTED
async def send_message(self, channel_id: str, message: str, **kwargs) -> None:
"""Send message to Discord channel."""
channel = self._bot.get_channel(int(channel_id))
if channel:
# Escape Discord markdown in game messages
escaped = discord.utils.escape_markdown(message)
await channel.send(escaped)
else:
logger.warning(f"Discord channel {channel_id} not found")
async def send_embed(self, channel_id: str, embed_data: dict[str, Any]) -> None:
"""Send rich embed to Discord channel."""
channel = self._bot.get_channel(int(channel_id))
if channel:
embed = discord.Embed(
title=embed_data.get("title"),
description=embed_data.get("description"),
color=embed_data.get("color", 0x00ff00),
)
if "thumbnail" in embed_data:
embed.set_thumbnail(url=embed_data["thumbnail"])
for field in embed_data.get("fields", []):
embed.add_field(
name=field["name"],
value=field["value"],
inline=field.get("inline", False),
)
await channel.send(embed=embed)
async def send_webhook(
self,
webhook_url: str,
content: str,
username: str | None = None,
avatar_url: str | None = None,
embed: dict[str, Any] | None = None,
) -> None:
"""Send message via webhook (one-way, no bot needed)."""
import aiohttp
payload = {"content": content}
if username:
payload["username"] = username
if avatar_url:
payload["avatar_url"] = avatar_url
if embed:
payload["embeds"] = [embed]
async with aiohttp.ClientSession() as session:
async with session.post(webhook_url, json=payload) as resp:
if resp.status >= 400:
logger.error(f"Webhook failed: {resp.status}")
async def _on_ready(self) -> None:
"""Handle Discord ready event."""
self.state = BridgeState.CONNECTED
logger.info(f"Discord bot connected as {self._bot.user}")
# Set status
await self._bot.change_presence(
activity=discord.Game(name=self.bridge_config.status_message)
)
async def _on_message(self, message: discord.Message) -> None:
"""Handle incoming Discord message."""
# Ignore bot messages
if message.author.bot:
return
# Create bridge message
user = ExternalUser(
service="discord",
user_id=str(message.author.id),
display_name=message.author.display_name,
avatar_url=str(message.author.avatar.url) if message.author.avatar else None,
roles=[role.name for role in message.author.roles] if hasattr(message.author, "roles") else [],
is_admin=message.author.guild_permissions.administrator if hasattr(message.author, "guild_permissions") else False,
)
# Strip Discord formatting for MUD display
content = self._strip_discord_formatting(message.content)
bridge_msg = BridgeMessage(
content=content,
source_service="discord",
source_channel=str(message.channel.id),
author=user,
attachments=[a.url for a in message.attachments],
)
await self._dispatch_message(bridge_msg)
async def _on_disconnect(self) -> None:
"""Handle Discord disconnect."""
if self.state != BridgeState.DISCONNECTED:
self.state = BridgeState.RECONNECTING
logger.warning("Discord disconnected, will reconnect automatically")
def _strip_discord_formatting(self, content: str) -> str:
"""Strip Discord markdown for MUD display."""
import re
# Remove custom emoji
content = re.sub(r"<a?:\w+:\d+>", "", content)
# Remove user/role/channel mentions (keep display name)
content = re.sub(r"<@!?\d+>", "[user]", content)
content = re.sub(r"<@&\d+>", "[role]", content)
content = re.sub(r"<#\d+>", "[channel]", content)
# Remove markdown formatting
content = re.sub(r"\*\*(.+?)\*\*", r"\1", content) # Bold
content = re.sub(r"\*(.+?)\*", r"\1", content) # Italic
content = re.sub(r"__(.+?)__", r"\1", content) # Underline
content = re.sub(r"~~(.+?)~~", r"\1", content) # Strikethrough
content = re.sub(r"```.*?```", "[code]", content, flags=re.DOTALL) # Code blocks
content = re.sub(r"`(.+?)`", r"\1", content) # Inline code
return content.strip()
2.4.3 IRC Bridge Implementation¶
# packages/maid-engine/src/maid_engine/bridges/irc_bridge.py
from __future__ import annotations
import asyncio
import logging
from typing import TYPE_CHECKING, Any
try:
import irc3
IRC_AVAILABLE = True
except ImportError:
IRC_AVAILABLE = False
from .protocol import (
BridgeMessage,
BridgeState,
ExternalBridge,
ExternalUser,
)
if TYPE_CHECKING:
from maid_engine.core.engine import GameEngine
logger = logging.getLogger(__name__)
class IRCBridgeConfig:
"""IRC bridge configuration."""
def __init__(self, config: dict[str, Any]):
self.server: str = config["server"]
self.port: int = config.get("port", 6667)
self.ssl: bool = config.get("ssl", False)
self.nickname: str = config["nickname"]
self.username: str = config.get("username", self.nickname)
self.realname: str = config.get("realname", "MAID MUD Bridge")
self.password: str | None = config.get("password")
self.nickserv_password: str | None = config.get("nickserv_password")
self.channels: list[str] = config.get("channels", [])
self.reconnect_delay: int = config.get("reconnect_delay", 30)
class IRCBridge(ExternalBridge):
"""IRC integration bridge."""
def __init__(self, engine: GameEngine, config: dict[str, Any]):
super().__init__(engine, config)
if not IRC_AVAILABLE:
raise ImportError("irc3 not installed. Install with: pip install irc3")
self.bridge_config = IRCBridgeConfig(config)
self._bot: Any = None
self._connected_channels: set[str] = set()
@property
def service_name(self) -> str:
return "irc"
async def connect(self) -> None:
"""Connect to IRC server."""
if self.state == BridgeState.CONNECTED:
return
self.state = BridgeState.CONNECTING
logger.info(f"Connecting to IRC: {self.bridge_config.server}:{self.bridge_config.port}")
# irc3 configuration
config = {
"nick": self.bridge_config.nickname,
"username": self.bridge_config.username,
"realname": self.bridge_config.realname,
"host": self.bridge_config.server,
"port": self.bridge_config.port,
"ssl": self.bridge_config.ssl,
"autojoins": self.bridge_config.channels,
}
if self.bridge_config.password:
config["password"] = self.bridge_config.password
# Create and start bot in background task
asyncio.create_task(self._run_irc_bot(config))
async def _run_irc_bot(self, config: dict) -> None:
"""Run IRC bot (irc3 is synchronous, so we run in executor)."""
import asyncio
loop = asyncio.get_event_loop()
# Create bot instance with plugins
self._bot = irc3.IrcBot.from_config(config)
# Register message handler
@irc3.event(irc3.rfc.PRIVMSG)
def on_message(bot, mask, target, data):
if target.startswith("#"): # Channel message
asyncio.run_coroutine_threadsafe(
self._handle_irc_message(mask, target, data),
loop
)
self._bot.include(on_message)
# Run in thread pool
await loop.run_in_executor(None, self._bot.run, False)
async def disconnect(self) -> None:
"""Disconnect from IRC."""
logger.info("Disconnecting from IRC...")
if self._bot:
self._bot.quit("MAID server shutting down")
self.state = BridgeState.DISCONNECTED
async def send_message(self, channel: str, message: str, **kwargs) -> None:
"""Send message to IRC channel."""
if self._bot:
# IRC messages are limited to ~400 chars, split if needed
for line in self._split_message(message, 400):
self._bot.privmsg(channel, line)
async def send_embed(self, channel: str, embed_data: dict[str, Any]) -> None:
"""IRC doesn't support embeds, send as formatted text."""
title = embed_data.get("title", "")
desc = embed_data.get("description", "")
if title:
await self.send_message(channel, f"** {title} **")
if desc:
await self.send_message(channel, desc)
for field in embed_data.get("fields", []):
await self.send_message(channel, f" {field['name']}: {field['value']}")
async def _handle_irc_message(self, mask: str, target: str, data: str) -> None:
"""Handle incoming IRC message."""
# Parse nick from mask (nick!user@host)
nick = mask.split("!")[0] if "!" in mask else mask
user = ExternalUser(
service="irc",
user_id=mask,
display_name=nick,
)
bridge_msg = BridgeMessage(
content=data,
source_service="irc",
source_channel=target,
author=user,
)
await self._dispatch_message(bridge_msg)
def _split_message(self, message: str, max_length: int) -> list[str]:
"""Split message into IRC-safe chunks."""
lines = []
for line in message.split("\n"):
while len(line) > max_length:
# Find last space before limit
split_at = line.rfind(" ", 0, max_length)
if split_at == -1:
split_at = max_length
lines.append(line[:split_at])
line = line[split_at:].strip()
if line:
lines.append(line)
return lines
2.4.4 RSS Feed Implementation¶
# packages/maid-engine/src/maid_engine/bridges/rss_feed.py
from __future__ import annotations
import hashlib
from dataclasses import dataclass, field
from datetime import datetime
from typing import TYPE_CHECKING, Any
from xml.etree import ElementTree as ET
from fastapi import APIRouter, Request, Response
from fastapi.responses import Response as FastAPIResponse
if TYPE_CHECKING:
from maid_engine.core.engine import GameEngine
@dataclass
class FeedItem:
"""A single RSS feed item."""
title: str
description: str
link: str
pub_date: datetime
guid: str | None = None
author: str | None = None
category: str | None = None
def __post_init__(self):
if not self.guid:
# Generate GUID from content hash
content = f"{self.title}{self.description}{self.pub_date.isoformat()}"
self.guid = hashlib.md5(content.encode()).hexdigest()
@dataclass
class FeedConfig:
"""RSS feed configuration."""
title: str = "MAID Game News"
description: str = "News and updates from the MAID MUD server"
link: str = "http://localhost:8080"
language: str = "en-us"
max_items: int = 50
cache_ttl_seconds: int = 300
class RSSFeedManager:
"""Manages RSS/Atom feed generation."""
def __init__(self, engine: GameEngine, config: dict[str, Any] | None = None):
self.engine = engine
self.config = FeedConfig(**(config or {}))
self._items: list[FeedItem] = []
self._last_modified: datetime = datetime.utcnow()
self._etag: str = ""
def add_item(self, item: FeedItem) -> None:
"""Add item to feed."""
self._items.insert(0, item)
# Trim to max items
if len(self._items) > self.config.max_items:
self._items = self._items[:self.config.max_items]
self._last_modified = datetime.utcnow()
self._update_etag()
def add_announcement(self, title: str, content: str, author: str = "System") -> None:
"""Convenience method to add an announcement."""
item = FeedItem(
title=title,
description=content,
link=self.config.link,
pub_date=datetime.utcnow(),
author=author,
category="announcement",
)
self.add_item(item)
def add_world_event(self, event_name: str, description: str) -> None:
"""Add a world event to the feed."""
item = FeedItem(
title=f"World Event: {event_name}",
description=description,
link=self.config.link,
pub_date=datetime.utcnow(),
category="world_event",
)
self.add_item(item)
def generate_rss(self) -> str:
"""Generate RSS 2.0 XML."""
rss = ET.Element("rss", version="2.0")
channel = ET.SubElement(rss, "channel")
# Channel metadata
ET.SubElement(channel, "title").text = self.config.title
ET.SubElement(channel, "link").text = self.config.link
ET.SubElement(channel, "description").text = self.config.description
ET.SubElement(channel, "language").text = self.config.language
ET.SubElement(channel, "lastBuildDate").text = self._format_rfc822(self._last_modified)
# Items
for item in self._items:
item_elem = ET.SubElement(channel, "item")
ET.SubElement(item_elem, "title").text = item.title
ET.SubElement(item_elem, "description").text = item.description
ET.SubElement(item_elem, "link").text = item.link
ET.SubElement(item_elem, "pubDate").text = self._format_rfc822(item.pub_date)
ET.SubElement(item_elem, "guid").text = item.guid
if item.author:
ET.SubElement(item_elem, "author").text = item.author
if item.category:
ET.SubElement(item_elem, "category").text = item.category
return ET.tostring(rss, encoding="unicode", xml_declaration=True)
def generate_atom(self) -> str:
"""Generate Atom 1.0 XML."""
nsmap = {"": "http://www.w3.org/2005/Atom"}
feed = ET.Element("feed", xmlns="http://www.w3.org/2005/Atom")
# Feed metadata
ET.SubElement(feed, "title").text = self.config.title
ET.SubElement(feed, "link", href=self.config.link)
ET.SubElement(feed, "updated").text = self._last_modified.isoformat() + "Z"
ET.SubElement(feed, "id").text = self.config.link
# Entries
for item in self._items:
entry = ET.SubElement(feed, "entry")
ET.SubElement(entry, "title").text = item.title
ET.SubElement(entry, "link", href=item.link)
ET.SubElement(entry, "id").text = item.guid
ET.SubElement(entry, "updated").text = item.pub_date.isoformat() + "Z"
ET.SubElement(entry, "summary").text = item.description
if item.author:
author_elem = ET.SubElement(entry, "author")
ET.SubElement(author_elem, "name").text = item.author
return ET.tostring(feed, encoding="unicode", xml_declaration=True)
def _format_rfc822(self, dt: datetime) -> str:
"""Format datetime as RFC 822 for RSS."""
return dt.strftime("%a, %d %b %Y %H:%M:%S +0000")
def _update_etag(self) -> None:
"""Update ETag for caching."""
content = f"{len(self._items)}{self._last_modified.isoformat()}"
self._etag = hashlib.md5(content.encode()).hexdigest()
def check_cache(self, if_none_match: str | None, if_modified_since: str | None) -> bool:
"""Check if client cache is still valid."""
if if_none_match == self._etag:
return True
# Could also check If-Modified-Since header
return False
def create_rss_router(feed_manager: RSSFeedManager) -> APIRouter:
"""Create FastAPI router for RSS endpoints."""
router = APIRouter(prefix="/feeds", tags=["RSS"])
@router.get("/rss.xml")
async def get_rss_feed(request: Request) -> Response:
"""Get RSS 2.0 feed."""
if_none_match = request.headers.get("If-None-Match")
if feed_manager.check_cache(if_none_match, None):
return Response(status_code=304)
content = feed_manager.generate_rss()
return Response(
content=content,
media_type="application/rss+xml",
headers={
"ETag": feed_manager._etag,
"Cache-Control": f"max-age={feed_manager.config.cache_ttl_seconds}",
},
)
@router.get("/atom.xml")
async def get_atom_feed(request: Request) -> Response:
"""Get Atom 1.0 feed."""
if_none_match = request.headers.get("If-None-Match")
if feed_manager.check_cache(if_none_match, None):
return Response(status_code=304)
content = feed_manager.generate_atom()
return Response(
content=content,
media_type="application/atom+xml",
headers={
"ETag": feed_manager._etag,
"Cache-Control": f"max-age={feed_manager.config.cache_ttl_seconds}",
},
)
return router
2.5 Configuration¶
# packages/maid-engine/src/maid_engine/config/settings.py (additions)
class DiscordBridgeSettings(BaseModel):
"""Discord bridge configuration."""
enabled: bool = False
bot_token: str | None = None
guild_id: int | None = None
status_message: str = "Playing MAID"
intents: list[str] = ["guilds", "guild_messages", "message_content"]
class IRCBridgeSettings(BaseModel):
"""IRC bridge configuration."""
enabled: bool = False
server: str | None = None
port: int = 6667
ssl: bool = False
nickname: str = "MAIDBot"
username: str = "maid"
realname: str = "MAID MUD Bridge"
password: str | None = None
nickserv_password: str | None = None
channels: list[str] = []
class RSSFeedSettings(BaseModel):
"""RSS feed configuration."""
enabled: bool = True
title: str = "MAID Game News"
description: str = "News and updates from the MAID MUD server"
max_items: int = 50
cache_ttl_seconds: int = 300
class BridgeSettings(BaseModel):
"""External service bridge settings."""
discord: DiscordBridgeSettings = Field(default_factory=DiscordBridgeSettings)
irc: IRCBridgeSettings = Field(default_factory=IRCBridgeSettings)
rss: RSSFeedSettings = Field(default_factory=RSSFeedSettings)
channel_mappings: list[dict[str, Any]] = Field(
default_factory=list,
description="Channel mappings between game and external services",
)
# Environment variables
MAID_BRIDGES__DISCORD__ENABLED=true
MAID_BRIDGES__DISCORD__BOT_TOKEN=your_discord_bot_token
MAID_BRIDGES__DISCORD__GUILD_ID=123456789
MAID_BRIDGES__IRC__ENABLED=true
MAID_BRIDGES__IRC__SERVER=irc.libera.chat
MAID_BRIDGES__IRC__PORT=6697
MAID_BRIDGES__IRC__SSL=true
MAID_BRIDGES__IRC__NICKNAME=MAIDBot
MAID_BRIDGES__IRC__CHANNELS=["#maid-game"]
MAID_BRIDGES__RSS__ENABLED=true
MAID_BRIDGES__RSS__TITLE="My MUD News"
2.6 Acceptance Criteria¶
| ID | Criterion | Verification |
|---|---|---|
| AC-2.1 | Discord bot connects and appears online | Manual test |
| AC-2.2 | Messages from game appear in Discord channel | Integration test |
| AC-2.3 | Messages from Discord appear in game channel | Integration test |
| AC-2.4 | Discord embeds display for game events | Manual test |
| AC-2.5 | IRC bot connects and joins channels | Manual test |
| AC-2.6 | Messages relay between IRC and game | Integration test |
| AC-2.7 | RSS feed validates as RSS 2.0 | Validation tool |
| AC-2.8 | Atom feed validates as Atom 1.0 | Validation tool |
| AC-2.9 | Feed caching with ETag works correctly | Unit test |
| AC-2.10 | Bridges reconnect after disconnect | Integration test |
Feature 3: Enhanced REST API¶
3.1 Feature Overview¶
What it does:
Provides a comprehensive REST API for game administration, player management, world queries, and third-party integrations. Enables external tools and dashboards to interact with the running game.
Why it's needed:
- Current state: Basic /api/health, /api/status, /api/admin/broadcast endpoints only
- No CRUD operations for game entities
- No player management endpoints
- Evennia provides full Django REST framework API
3.2 User Stories¶
US-3.1: Player Management API
As an external tool developer, I want REST endpoints to list, search, and manage player accounts so that I can build admin dashboards.
US-3.2: World Query API
As an external tool developer, I want REST endpoints to query rooms, items, and NPCs so that I can build world viewers and maps.
US-3.3: Real-Time Events API
As an external tool developer, I want a WebSocket endpoint for real-time game events so that I can build live dashboards.
US-3.4: API Authentication
As a server administrator, I want API authentication with API keys and OAuth so that I can control access to the API.
US-3.5: Rate Limiting
As a server administrator, I want API rate limiting so that the API cannot be abused.
3.3 Technical Requirements¶
| ID | Requirement | Priority |
|---|---|---|
| API-001 | System SHALL provide OpenAPI/Swagger documentation | P0 |
| API-002 | System SHALL support API key authentication | P0 |
| API-003 | System SHALL support JWT token authentication | P1 |
| API-004 | System SHALL provide rate limiting per client | P0 |
| API-005 | System SHALL provide player management CRUD endpoints | P0 |
| API-006 | System SHALL provide world query endpoints (rooms, items, NPCs) | P1 |
| API-007 | System SHALL provide game statistics endpoints | P1 |
| API-008 | System SHALL provide real-time event WebSocket | P1 |
| API-009 | System SHALL support pagination for list endpoints | P0 |
| API-010 | System SHALL support filtering and sorting for list endpoints | P1 |
| API-011 | System SHALL log all API access for audit | P0 |
3.4 API/Interface Design¶
3.4.1 API Authentication¶
# packages/maid-engine/src/maid_engine/api/auth.py
from __future__ import annotations
import hashlib
import secrets
import time
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum, auto
from typing import Any
from fastapi import Depends, HTTPException, Security, status
from fastapi.security import APIKeyHeader, HTTPBearer, HTTPAuthorizationCredentials
import jwt
from pydantic import BaseModel
class APIPermission(Enum):
"""API permission levels."""
READ_PUBLIC = auto() # Public game info
READ_PLAYERS = auto() # Player list/info
WRITE_PLAYERS = auto() # Player management
READ_WORLD = auto() # World data
WRITE_WORLD = auto() # World modification
ADMIN = auto() # Full admin access
@dataclass
class APIKey:
"""An API key with permissions."""
key_id: str
key_hash: str
name: str
permissions: set[APIPermission]
created_at: datetime
last_used: datetime | None = None
expires_at: datetime | None = None
rate_limit: int = 100 # Requests per minute
owner: str | None = None
@dataclass
class APIUser:
"""Authenticated API user."""
key_id: str
name: str
permissions: set[APIPermission]
class APIKeyStore:
"""In-memory API key store (use database in production)."""
def __init__(self):
self._keys: dict[str, APIKey] = {}
def generate_key(
self,
name: str,
permissions: set[APIPermission],
expires_in_days: int | None = None,
rate_limit: int = 100,
owner: str | None = None,
) -> tuple[str, APIKey]:
"""Generate a new API key.
Returns:
(raw_key, APIKey) - raw_key must be given to user, it cannot be recovered
"""
raw_key = secrets.token_urlsafe(32)
key_id = secrets.token_urlsafe(8)
key_hash = hashlib.sha256(raw_key.encode()).hexdigest()
expires_at = None
if expires_in_days:
expires_at = datetime.utcnow() + timedelta(days=expires_in_days)
api_key = APIKey(
key_id=key_id,
key_hash=key_hash,
name=name,
permissions=permissions,
created_at=datetime.utcnow(),
expires_at=expires_at,
rate_limit=rate_limit,
owner=owner,
)
self._keys[key_id] = api_key
return f"{key_id}.{raw_key}", api_key
def validate_key(self, raw_key: str) -> APIKey | None:
"""Validate an API key and return the APIKey if valid."""
if "." not in raw_key:
return None
key_id, secret = raw_key.split(".", 1)
api_key = self._keys.get(key_id)
if not api_key:
return None
# Verify hash
key_hash = hashlib.sha256(secret.encode()).hexdigest()
if key_hash != api_key.key_hash:
return None
# Check expiration
if api_key.expires_at and api_key.expires_at < datetime.utcnow():
return None
# Update last used
api_key.last_used = datetime.utcnow()
return api_key
def revoke_key(self, key_id: str) -> bool:
"""Revoke an API key."""
if key_id in self._keys:
del self._keys[key_id]
return True
return False
class RateLimiter:
"""Simple sliding window rate limiter."""
def __init__(self):
self._requests: dict[str, list[float]] = {}
def check(self, key_id: str, limit: int, window_seconds: int = 60) -> bool:
"""Check if request is allowed.
Returns True if allowed, False if rate limited.
"""
now = time.time()
window_start = now - window_seconds
# Get requests in window
requests = self._requests.get(key_id, [])
requests = [r for r in requests if r > window_start]
if len(requests) >= limit:
return False
requests.append(now)
self._requests[key_id] = requests
return True
# FastAPI dependencies
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
bearer_scheme = HTTPBearer(auto_error=False)
# Global instances (inject via dependency)
api_key_store = APIKeyStore()
rate_limiter = RateLimiter()
async def get_api_user(
api_key: str | None = Security(api_key_header),
) -> APIUser:
"""Dependency to get authenticated API user."""
if not api_key:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="API key required",
)
key = api_key_store.validate_key(api_key)
if not key:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired API key",
)
# Check rate limit
if not rate_limiter.check(key.key_id, key.rate_limit):
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail=f"Rate limit exceeded ({key.rate_limit}/min)",
)
return APIUser(
key_id=key.key_id,
name=key.name,
permissions=key.permissions,
)
def require_permission(*permissions: APIPermission):
"""Dependency to require specific API permissions."""
async def check_permissions(user: APIUser = Depends(get_api_user)) -> APIUser:
for perm in permissions:
if perm not in user.permissions and APIPermission.ADMIN not in user.permissions:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Missing permission: {perm.name}",
)
return user
return check_permissions
3.4.2 API Endpoints¶
# packages/maid-engine/src/maid_engine/api/v1/__init__.py
from fastapi import APIRouter
from .players import router as players_router
from .world import router as world_router
from .stats import router as stats_router
from .admin import router as admin_router
router = APIRouter(prefix="/api/v1")
router.include_router(players_router)
router.include_router(world_router)
router.include_router(stats_router)
router.include_router(admin_router)
# packages/maid-engine/src/maid_engine/api/v1/players.py
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from maid_engine.api.auth import APIPermission, APIUser, require_permission
if TYPE_CHECKING:
from maid_engine.core.engine import GameEngine
router = APIRouter(prefix="/players", tags=["Players"])
class PlayerSummary(BaseModel):
"""Summary player information."""
id: UUID
name: str
level: int
race: str
character_class: str
is_online: bool
last_seen: datetime | None
class PlayerDetail(BaseModel):
"""Detailed player information."""
id: UUID
name: str
level: int
race: str
character_class: str
is_online: bool
last_seen: datetime | None
current_room: str | None
guild: str | None
stats: dict[str, int]
created_at: datetime
play_time_hours: float
class PlayerListResponse(BaseModel):
"""Paginated player list response."""
items: list[PlayerSummary]
total: int
page: int
per_page: int
pages: int
class PlayerUpdateRequest(BaseModel):
"""Request to update player."""
level: int | None = None
race: str | None = None
character_class: str | None = None
@router.get("", response_model=PlayerListResponse)
async def list_players(
page: int = Query(1, ge=1),
per_page: int = Query(20, ge=1, le=100),
online_only: bool = Query(False),
search: str | None = Query(None, min_length=1, max_length=50),
sort_by: str = Query("name", regex="^(name|level|last_seen)$"),
sort_order: str = Query("asc", regex="^(asc|desc)$"),
user: APIUser = Depends(require_permission(APIPermission.READ_PLAYERS)),
) -> PlayerListResponse:
"""List all players with pagination and filtering."""
# Implementation would query player data
# This is a skeleton showing the interface
...
@router.get("/{player_id}", response_model=PlayerDetail)
async def get_player(
player_id: UUID,
user: APIUser = Depends(require_permission(APIPermission.READ_PLAYERS)),
) -> PlayerDetail:
"""Get detailed player information."""
...
@router.patch("/{player_id}", response_model=PlayerDetail)
async def update_player(
player_id: UUID,
update: PlayerUpdateRequest,
user: APIUser = Depends(require_permission(APIPermission.WRITE_PLAYERS)),
) -> PlayerDetail:
"""Update player information."""
...
@router.post("/{player_id}/kick")
async def kick_player(
player_id: UUID,
reason: str = Query(..., min_length=1, max_length=200),
user: APIUser = Depends(require_permission(APIPermission.ADMIN)),
) -> dict:
"""Kick a player from the game."""
...
@router.post("/{player_id}/ban")
async def ban_player(
player_id: UUID,
reason: str = Query(..., min_length=1, max_length=200),
duration_hours: int | None = Query(None, ge=1),
user: APIUser = Depends(require_permission(APIPermission.ADMIN)),
) -> dict:
"""Ban a player."""
...
@router.delete("/{player_id}/ban")
async def unban_player(
player_id: UUID,
user: APIUser = Depends(require_permission(APIPermission.ADMIN)),
) -> dict:
"""Unban a player."""
...
# packages/maid-engine/src/maid_engine/api/v1/world.py
from __future__ import annotations
from typing import TYPE_CHECKING
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from maid_engine.api.auth import APIPermission, APIUser, require_permission
router = APIRouter(prefix="/world", tags=["World"])
class RoomSummary(BaseModel):
"""Summary room information."""
id: UUID
name: str
area: str
sector_type: str
player_count: int
class RoomDetail(BaseModel):
"""Detailed room information."""
id: UUID
name: str
description: str
area: str
sector_type: str
exits: dict[str, UUID]
coordinates: tuple[int, int, int] | None
flags: list[str]
player_count: int
npc_count: int
item_count: int
class NPCSummary(BaseModel):
"""Summary NPC information."""
id: UUID
name: str
level: int
room_id: UUID | None
is_alive: bool
class ItemSummary(BaseModel):
"""Summary item information."""
id: UUID
name: str
item_type: str
rarity: str
location: str # "room:uuid", "inventory:uuid", "ground"
@router.get("/rooms", response_model=list[RoomSummary])
async def list_rooms(
area: str | None = Query(None),
page: int = Query(1, ge=1),
per_page: int = Query(50, ge=1, le=100),
user: APIUser = Depends(require_permission(APIPermission.READ_WORLD)),
) -> list[RoomSummary]:
"""List rooms with optional area filter."""
...
@router.get("/rooms/{room_id}", response_model=RoomDetail)
async def get_room(
room_id: UUID,
user: APIUser = Depends(require_permission(APIPermission.READ_WORLD)),
) -> RoomDetail:
"""Get detailed room information."""
...
@router.get("/rooms/{room_id}/contents")
async def get_room_contents(
room_id: UUID,
user: APIUser = Depends(require_permission(APIPermission.READ_WORLD)),
) -> dict:
"""Get players, NPCs, and items in a room."""
...
@router.get("/npcs", response_model=list[NPCSummary])
async def list_npcs(
room_id: UUID | None = Query(None),
area: str | None = Query(None),
alive_only: bool = Query(False),
page: int = Query(1, ge=1),
per_page: int = Query(50, ge=1, le=100),
user: APIUser = Depends(require_permission(APIPermission.READ_WORLD)),
) -> list[NPCSummary]:
"""List NPCs with optional filters."""
...
@router.get("/items", response_model=list[ItemSummary])
async def list_items(
item_type: str | None = Query(None),
rarity: str | None = Query(None),
page: int = Query(1, ge=1),
per_page: int = Query(50, ge=1, le=100),
user: APIUser = Depends(require_permission(APIPermission.READ_WORLD)),
) -> list[ItemSummary]:
"""List items with optional filters."""
...
@router.get("/areas")
async def list_areas(
user: APIUser = Depends(require_permission(APIPermission.READ_WORLD)),
) -> list[dict]:
"""List all areas with room counts."""
...
# packages/maid-engine/src/maid_engine/api/v1/stats.py
from __future__ import annotations
from datetime import datetime
from typing import Any
from fastapi import APIRouter, Depends
from pydantic import BaseModel
from maid_engine.api.auth import APIPermission, APIUser, require_permission
router = APIRouter(prefix="/stats", tags=["Statistics"])
class ServerStats(BaseModel):
"""Server statistics."""
uptime_seconds: float
tick_count: int
ticks_per_second: float
players_online: int
players_peak_today: int
total_entities: int
total_rooms: int
memory_usage_mb: float
cpu_percent: float
class PlayerStats(BaseModel):
"""Player-related statistics."""
total_accounts: int
active_last_24h: int
active_last_7d: int
active_last_30d: int
new_today: int
new_this_week: int
average_session_minutes: float
level_distribution: dict[str, int]
class EconomyStats(BaseModel):
"""Economy statistics."""
total_gold_circulation: int
gold_per_player_average: float
auction_house_listings: int
auction_house_value: int
npc_shop_sales_today: int
player_trades_today: int
class CombatStats(BaseModel):
"""Combat statistics."""
total_kills_today: int
total_deaths_today: int
pvp_kills_today: int
boss_kills_today: int
most_killed_mob: str
top_killer: str
@router.get("/server", response_model=ServerStats)
async def get_server_stats(
user: APIUser = Depends(require_permission(APIPermission.READ_PUBLIC)),
) -> ServerStats:
"""Get server statistics."""
...
@router.get("/players", response_model=PlayerStats)
async def get_player_stats(
user: APIUser = Depends(require_permission(APIPermission.READ_PLAYERS)),
) -> PlayerStats:
"""Get player statistics."""
...
@router.get("/economy", response_model=EconomyStats)
async def get_economy_stats(
user: APIUser = Depends(require_permission(APIPermission.READ_WORLD)),
) -> EconomyStats:
"""Get economy statistics."""
...
@router.get("/combat", response_model=CombatStats)
async def get_combat_stats(
user: APIUser = Depends(require_permission(APIPermission.READ_PUBLIC)),
) -> CombatStats:
"""Get combat statistics."""
...
@router.get("/commands")
async def get_command_stats(
user: APIUser = Depends(require_permission(APIPermission.ADMIN)),
) -> dict[str, Any]:
"""Get command execution statistics."""
...
3.4.3 Real-Time Events WebSocket¶
# packages/maid-engine/src/maid_engine/api/v1/events_ws.py
from __future__ import annotations
import asyncio
import json
from datetime import datetime
from enum import Enum
from typing import TYPE_CHECKING, Any, Callable
from uuid import UUID
from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Query
from pydantic import BaseModel
if TYPE_CHECKING:
from maid_engine.core.engine import GameEngine
router = APIRouter(tags=["Events"])
class EventType(str, Enum):
"""Types of events that can be subscribed to."""
PLAYER_LOGIN = "player.login"
PLAYER_LOGOUT = "player.logout"
PLAYER_LEVEL_UP = "player.level_up"
PLAYER_DEATH = "player.death"
CHAT_MESSAGE = "chat.message"
WORLD_EVENT = "world.event"
SERVER_STATUS = "server.status"
COMBAT_KILL = "combat.kill"
class GameEvent(BaseModel):
"""A game event for WebSocket streaming."""
event_type: EventType
timestamp: datetime
data: dict[str, Any]
class EventSubscription:
"""A WebSocket subscription to game events."""
def __init__(
self,
websocket: WebSocket,
event_types: set[EventType],
api_key: str,
):
self.websocket = websocket
self.event_types = event_types
self.api_key = api_key
self.connected_at = datetime.utcnow()
async def send_event(self, event: GameEvent) -> bool:
"""Send event to subscriber. Returns False if connection lost."""
try:
await self.websocket.send_json(event.dict())
return True
except Exception:
return False
class EventBroadcaster:
"""Manages WebSocket subscriptions and broadcasts events."""
def __init__(self):
self._subscriptions: list[EventSubscription] = []
self._lock = asyncio.Lock()
async def subscribe(
self,
websocket: WebSocket,
event_types: set[EventType],
api_key: str,
) -> EventSubscription:
"""Add a new subscription."""
sub = EventSubscription(websocket, event_types, api_key)
async with self._lock:
self._subscriptions.append(sub)
return sub
async def unsubscribe(self, subscription: EventSubscription) -> None:
"""Remove a subscription."""
async with self._lock:
if subscription in self._subscriptions:
self._subscriptions.remove(subscription)
async def broadcast(self, event: GameEvent) -> int:
"""Broadcast event to all matching subscribers.
Returns number of subscribers reached.
"""
dead_subs = []
sent_count = 0
async with self._lock:
for sub in self._subscriptions:
if event.event_type in sub.event_types:
success = await sub.send_event(event)
if success:
sent_count += 1
else:
dead_subs.append(sub)
# Remove dead subscriptions
for sub in dead_subs:
self._subscriptions.remove(sub)
return sent_count
# Global broadcaster
broadcaster = EventBroadcaster()
@router.websocket("/events")
async def events_websocket(
websocket: WebSocket,
api_key: str = Query(...),
events: str = Query("*"), # Comma-separated event types or * for all
):
"""WebSocket endpoint for real-time game events.
Connect with API key and optional event filter.
Example: ws://host/api/v1/events?api_key=xxx&events=player.login,player.logout
"""
# Validate API key (simplified)
from maid_engine.api.auth import api_key_store
key = api_key_store.validate_key(api_key)
if not key:
await websocket.close(code=4001, reason="Invalid API key")
return
# Parse event types
if events == "*":
event_types = set(EventType)
else:
event_types = set()
for event_name in events.split(","):
try:
event_types.add(EventType(event_name.strip()))
except ValueError:
pass
if not event_types:
await websocket.close(code=4002, reason="No valid event types specified")
return
# Accept connection
await websocket.accept()
# Subscribe
subscription = await broadcaster.subscribe(websocket, event_types, api_key)
try:
# Keep connection alive
while True:
# Wait for ping/pong or client message
message = await websocket.receive_text()
if message == "ping":
await websocket.send_text("pong")
except WebSocketDisconnect:
pass
finally:
await broadcaster.unsubscribe(subscription)
# Integration with game engine
def create_event_hooks(engine: "GameEngine") -> None:
"""Create event hooks to broadcast game events."""
async def on_player_login(event):
game_event = GameEvent(
event_type=EventType.PLAYER_LOGIN,
timestamp=datetime.utcnow(),
data={
"player_id": str(event.player_id),
"player_name": event.player_name,
},
)
await broadcaster.broadcast(game_event)
async def on_player_logout(event):
game_event = GameEvent(
event_type=EventType.PLAYER_LOGOUT,
timestamp=datetime.utcnow(),
data={
"player_id": str(event.player_id),
"player_name": event.player_name,
"session_duration": event.session_duration,
},
)
await broadcaster.broadcast(game_event)
# Subscribe to game events
engine.events.subscribe("PlayerLoginEvent", on_player_login)
engine.events.subscribe("PlayerLogoutEvent", on_player_logout)
# ... more event subscriptions
3.5 OpenAPI Documentation¶
The FastAPI framework automatically generates OpenAPI documentation. Access at:
- /docs - Swagger UI
- /redoc - ReDoc UI
- /openapi.json - Raw OpenAPI spec
3.6 Acceptance Criteria¶
| ID | Criterion | Verification |
|---|---|---|
| AC-3.1 | API key authentication works | Unit test |
| AC-3.2 | Rate limiting rejects excess requests | Unit test |
| AC-3.3 | Permission checks block unauthorized access | Unit test |
| AC-3.4 | Player list pagination works correctly | Integration test |
| AC-3.5 | Player search returns matching results | Integration test |
| AC-3.6 | Room query returns correct data | Integration test |
| AC-3.7 | Statistics endpoints return valid data | Integration test |
| AC-3.8 | WebSocket events stream in real-time | Manual test |
| AC-3.9 | OpenAPI documentation is complete | Manual review |
| AC-3.10 | API responses use consistent format | Unit test |
Appendix A: Security Considerations¶
SSL/TLS Security¶
- Minimum TLS Version: Enforce TLS 1.2 or higher; TLS 1.3 preferred
- Cipher Selection: Use HIGH ciphers, exclude NULL, MD5, RC4
- Certificate Validation: Validate certificate paths on startup
- Key Protection: Support encrypted private keys
- Certificate Renewal: Support hot-reload for Let's Encrypt
API Security¶
- Key Storage: Hash API keys with SHA-256; never store raw keys
- Rate Limiting: Implement per-key rate limits to prevent abuse
- Audit Logging: Log all API access with key ID and endpoint
- CORS: Restrict origins in production
- Input Validation: Validate all input with Pydantic
Bridge Security¶
- Discord Tokens: Store bot tokens in environment variables, never commit
- IRC Password: Support NickServ authentication
- Message Sanitization: Strip potentially harmful content from external messages
- User Verification: Don't trust external user claims without verification
Appendix B: Testing Requirements¶
Unit Test Coverage¶
| Component | Minimum Coverage | Key Test Cases |
|---|---|---|
| SSLContextBuilder | 90% | Valid certs, invalid certs, missing files |
| CertificateReloader | 85% | File change detection, reload |
| DiscordBridge | 80% | Connection, message relay, formatting |
| IRCBridge | 80% | Connection, message relay, splitting |
| RSSFeedManager | 90% | Item add, feed generation, caching |
| APIKeyStore | 95% | Generation, validation, expiration |
| RateLimiter | 90% | Allow/deny, window sliding |
| API Endpoints | 85% | CRUD operations, pagination, filtering |
Integration Test Scenarios¶
- SSL Telnet: Connect with openssl s_client, verify handshake
- SSL WebSocket: Connect with browser, verify HTTPS
- Discord Full Flow: Send message in game, verify in Discord, reply, verify in game
- IRC Full Flow: Same as Discord
- API Auth: Generate key, use key, exceed rate limit, verify rejection
- WebSocket Events: Subscribe, trigger game event, verify receipt
Performance Benchmarks¶
| Operation | Target | Measurement |
|---|---|---|
| SSL handshake | < 100ms | pytest-benchmark |
| API authentication | < 5ms | pytest-benchmark |
| Rate limit check | < 0.1ms | pytest-benchmark |
| RSS generation (50 items) | < 50ms | pytest-benchmark |
| Event broadcast (100 subs) | < 100ms | pytest-benchmark |
Document History¶
| Version | Date | Author | Changes |
|---|---|---|---|
| 1.0 | 2026-01-30 | MAID Team | Initial specification |