"""Pheby HTTP + WebSocket server (aiohttp). Listens on localhost/plain HTTP behind Caddy. Routes: * ``GET /health`` — unauthenticated liveness (minimal info). * ``GET /ws`` — WebSocket; first frame must be ``hello``. * ``GET /attachments/{id}`` — authenticated attachment download (streamed). All message routing lives in :meth:`PhebyServer.handle_client_message`; Hermes integration (runs, approvals, clarifications, models) lives in :mod:`.hermes_bridge` to keep this module focused on protocol + transport. """ from __future__ import annotations import asyncio import ipaddress import logging import time from urllib.parse import quote from pathlib import Path from typing import Any, Dict, List, Optional from aiohttp import web from . import protocol as proto from .attachments import AttachmentStore, constant_time_equals from .config import PhebyConfig from .conversations import ConversationRouter from . import hermes_bridge from .ws_client import ClientConnection logger = logging.getLogger(__name__) class PhebyServer: """Owns the aiohttp app, connected clients, and shared subsystems.""" def __init__(self, config: PhebyConfig, adapter: Any = None): self.config = config self.adapter = adapter # PhebyAdapter (may be None in tests) self.router = ConversationRouter() from hermes_constants import get_hermes_home root = config.attachments_root or str( Path(get_hermes_home()) / "pheby-attachments") self.store = AttachmentStore( root=Path(root), retention_days=config.retention_days, index_path=Path(config.index_path) if config.index_path else None, ) self.bridge = hermes_bridge # bridge function module self._clients: Dict[str, ClientConnection] = {} self._auth_failures: Dict[str, List[float]] = {} self._cleanup_task: Optional[asyncio.Task] = None self._app: Optional[web.Application] = None self._runner: Optional[web.AppRunner] = None self._site: Optional[web.TCPSite] = None self._conn_counter = 0 # ── lifecycle ──────────────────────────────────────────────────────── async def start(self) -> bool: from aiohttp import web as _web # local import keeps import light self.store.hydrate_legacy_meta() app = _web.Application(client_max_size=proto.MAX_WS_MESSAGE_BYTES) app.router.add_get("/health", self._handle_health) app.router.add_get("/ws", self._handle_ws) app.router.add_get("/attachments/{attachment_id}", self._handle_attachment_download) self._app = app self._runner = web.AppRunner(app, access_log=None) await self._runner.setup() self._site = web.TCPSite(self._runner, self.config.bind_host, self.config.port) try: await self._site.start() except OSError as exc: logger.error("[pheby] failed to bind %s:%s — %s", self.config.bind_host, self.config.port, exc) await self.stop() return False self._cleanup_task = asyncio.create_task(self._cleanup_loop()) logger.info("[pheby] serving on http://%s:%d (attachments: %s)", self.config.bind_host, self.config.port, self.store.root) return True async def stop(self) -> None: if self._cleanup_task: self._cleanup_task.cancel() try: await self._cleanup_task except asyncio.CancelledError: pass self._cleanup_task = None for client in list(self._clients.values()): client.closed = True try: await client.ws.close() except Exception: pass self._clients.clear() if self._runner: await self._runner.cleanup() self._runner = None self._site = None self._app = None logger.info("[pheby] server stopped") # ── background cleanup ─────────────────────────────────────────────── async def _cleanup_loop(self) -> None: """Hourly expired-attachment sweep; first sweep after 5 minutes.""" try: await asyncio.sleep(300) while True: try: await self.store.cleanup_expired() except Exception: logger.error("[pheby] attachment cleanup failed", exc_info=True) await asyncio.sleep(3600) except asyncio.CancelledError: pass # ── HTTP handlers ──────────────────────────────────────────────────── async def _handle_health(self, request: web.Request) -> web.Response: """Minimal unauthenticated liveness probe.""" return web.json_response({"status": "ok"}) def _check_http_secret(self, request: web.Request) -> bool: header = request.headers.get("Authorization", "") if header.startswith("Bearer "): token = header[7:].strip() elif header.startswith("ApiKey "): token = header[7:].strip() else: token = request.headers.get("X-Pheby-Secret", "").strip() if not token: return False return constant_time_equals(token, self.config.secret) async def _handle_attachment_download( self, request: web.Request) -> web.StreamResponse: attachment_id = request.match_info.get("attachment_id", "") if not self._check_http_secret(request): return web.json_response( {"error": {"code": proto.ERR_UNAUTHORIZED, "message": "Authentication required"}}, status=401) blob = self.store.resolve_blob(attachment_id) if blob is None: # Expired, unknown, or malformed — same minimal response so the # endpoint leaks nothing about implementation details. return web.json_response( {"error": {"code": proto.ERR_NOT_FOUND, "message": "Attachment unavailable"}}, status=404) desc = self.store.describe(attachment_id) or {} safe_name = str(desc.get("filename", "file.bin")) ascii_name = safe_name.encode("ascii", "replace").decode("ascii") \ .replace('"', "_").replace("\\", "_") disposition = (f'attachment; filename="{ascii_name}"; ' f"filename*=UTF-8''{quote(safe_name)}") logger.info("[pheby] attachment download: id=%s bytes=%s", attachment_id, desc.get("size")) return web.FileResponse( blob, headers={ "Content-Disposition": disposition, "Content-Type": desc.get("mime_type", "application/octet-stream"), }, ) # ── WebSocket handler ──────────────────────────────────────────────── async def _handle_ws(self, request: web.Request) -> web.WebSocketResponse: # Bind server/adapter identity to THIS task's context. ContextVars set # in adapter.connect() do not propagate into aiohttp handler tasks, so # every per-connection task must bind them itself — hermes_bridge's # _current_server()/_current_adapter() resolve from here in all # message handlers and agent-thread callbacks. hermes_bridge.set_server(self) if self.adapter is not None: hermes_bridge.set_adapter(self.adapter) ws = web.WebSocketResponse(max_msg_size=proto.MAX_WS_MESSAGE_BYTES, heartbeat=30.0, autoping=True) await ws.prepare(request) self._conn_counter += 1 conn_id = f"c{self._conn_counter}" peer = self._auth_peer(request) if self._is_locked_out(peer): logger.warning("[pheby] auth lockout active for %s — refusing", peer) await ws.close(code=4401, message=b"locked out") return ws client = ClientConnection(ws, conn_id) self._clients[conn_id] = client logger.info("[pheby] client %s connected from %s", conn_id, peer) try: # Auth phase: hello must arrive within the window. try: authed = await asyncio.wait_for( self._authenticate(client, peer), timeout=proto.AUTH_TIMEOUT_SECONDS) except asyncio.TimeoutError: await client.send_json(proto.error_event( proto.ERR_AUTH_TIMEOUT, "hello not received in time")) await ws.close() return ws if not authed: await ws.close(code=4401, message=b"unauthorized") return ws await client.send_json({ "type": proto.S_READY, "protocol_version": proto.PROTOCOL_VERSION, "server": "pheby", "ts": proto.now_iso(), }) await client.read_loop(self) finally: self._clients.pop(conn_id, None) logger.info("[pheby] client %s disconnected (authed=%s, %.0fs)", conn_id, client.authenticated, time.time() - client.connected_at) return ws def _is_locked_out(self, peer: str) -> bool: fails = self._auth_failures.get(peer) if not fails: return False cutoff = time.time() - proto.AUTH_FAILURE_LOCKOUT_SECONDS recent = [t for t in fails if t > cutoff] self._auth_failures[peer] = recent return len(recent) >= proto.AUTH_FAILURE_THRESHOLD def _record_auth_failure(self, peer: str) -> None: self._auth_failures.setdefault(peer, []).append(time.time()) @staticmethod def _auth_peer(request: web.Request) -> str: """Use Caddy's client IP only when the direct peer is loopback.""" direct = request.remote or "unknown" try: if not ipaddress.ip_address(direct).is_loopback: return direct except ValueError: return direct forwarded = request.headers.get("X-Forwarded-For", "").split(",", 1)[0].strip() try: return str(ipaddress.ip_address(forwarded)) if forwarded else direct except ValueError: return direct async def _authenticate(self, client: ClientConnection, peer: str) -> bool: """Wait for the hello frame and validate the shared secret.""" msg = await client.ws.receive(timeout=proto.AUTH_TIMEOUT_SECONDS + 5) if msg.type != "text" and not hasattr(msg, "data"): return False message, err = proto.decode_message(msg.data) if err or message is None: await client.send_json( proto.error_event(err or proto.ERR_BAD_REQUEST, "Expected hello message")) return False if message.get("type") != proto.C_HELLO: await client.send_json(proto.error_event( proto.ERR_UNAUTHORIZED, "First message must be hello")) self._record_auth_failure(peer) return False supplied = str(message.get("secret", "")) if not supplied or not constant_time_equals(supplied, self.config.secret): logger.warning("[pheby] auth failure from %s", peer) self._record_auth_failure(peer) # Small delay to slow brute force; constant-time compare already # used for the secret itself. await asyncio.sleep(0.5) await client.send_json(proto.error_event( proto.ERR_UNAUTHORIZED, "Invalid secret")) return False requested = message.get("protocol_version") try: requested_version = (proto.PROTOCOL_VERSION if requested is None else int(requested)) except (TypeError, ValueError): await client.send_json(proto.error_event( proto.ERR_VERSION_MISMATCH, "protocol_version must be an integer")) return False if requested_version != proto.PROTOCOL_VERSION: await client.send_json(proto.error_event( proto.ERR_VERSION_MISMATCH, f"Protocol version mismatch: server={proto.PROTOCOL_VERSION}, " f"client={requested}")) return False client.authenticated = True client.protocol_version = proto.PROTOCOL_VERSION logger.info("[pheby] client %s authenticated", client.conn_id) return True # ── broadcast ──────────────────────────────────────────────────────── async def broadcast(self, payload: Dict[str, Any]) -> None: """Send an event to every authenticated client.""" for client in list(self._clients.values()): if client.authenticated and not client.closed: await client.send_json(payload) def has_clients(self) -> bool: return any(c.authenticated and not c.closed for c in self._clients.values()) # ── inbound dispatch ───────────────────────────────────────────────── async def handle_client_message(self, client: ClientConnection, raw: str) -> None: message, err = proto.decode_message(raw) if err or message is None: await client.send_json(proto.error_event( err or proto.ERR_BAD_REQUEST, "Malformed message")) return mtype = message.get("type", "") request_id = message.get("request_id") if self.config.debug: # Verbose protocol logging — never logs secrets; chat content # only when explicitly configured (privacy default). safe = {k: v for k, v in message.items() if k not in ("secret",)} if not self.config.log_chat_content and mtype == proto.C_CHAT_SEND: safe = dict(safe) safe["text"] = f"<{len(str(message.get('text', '')))} chars>" logger.info("[pheby] << %s", proto.safe_str(safe, 400)) try: handler = self._HANDLERS.get(mtype) if handler is None: await client.send_json(proto.error_event( proto.ERR_UNKNOWN_TYPE, f"Unknown message type: {mtype}", request_id)) return await handler(self, client, message, request_id) except Exception: logger.error("[pheby] handler failed for %s", mtype, exc_info=True) await client.send_json(proto.error_event( proto.ERR_INTERNAL, "Internal server error", request_id)) # ── simple handlers ────────────────────────────────────────────────── async def _handle_ping(self, client: ClientConnection, message: Dict, request_id: Optional[str]) -> None: await client.send_json({"type": proto.S_PONG, "ts": proto.now_iso(), **({"request_id": request_id} if request_id else {})}) async def _handle_conversation_list(self, client, message, request_id): conversations = await self.bridge.list_conversations() await client.send_json({ "type": proto.S_CONVERSATION_SNAPSHOT, "conversations": conversations, **({"request_id": request_id} if request_id else {}), }) async def _handle_conversation_open(self, client, message, request_id): conversation_id = str(message.get("conversation_id", "")) if not ConversationRouter.is_valid_conversation_id(conversation_id): await client.send_json(proto.error_event( proto.ERR_BAD_REQUEST, "Invalid conversation_id", request_id)) return limit = message.get("limit", proto.MAX_HISTORY_MESSAGES) try: limit = max(1, min(int(limit), proto.MAX_HISTORY_MESSAGES)) except (TypeError, ValueError): limit = proto.MAX_HISTORY_MESSAGES history, found = await self.bridge.conversation_history( conversation_id, limit) if not found: await client.send_json(proto.error_event( proto.ERR_CONVERSATION_NOT_FOUND, "Conversation not found", request_id)) return runtime = self.bridge.runtime_snapshot(conversation_id) await client.send_json({ "type": proto.S_CONVERSATION_HISTORY, "conversation_id": conversation_id, "messages": history, "attachments": self.store.list_for_conversation(conversation_id), **runtime, **({"request_id": request_id} if request_id else {}), }) async def _handle_conversation_create(self, client, message, request_id): name = message.get("name") cid = await self.bridge.create_conversation( self, str(name) if name else None) await client.send_json({ "type": proto.S_CONVERSATION_CREATED, "conversation_id": cid, "name": await self.router.get_name(cid), **({"request_id": request_id} if request_id else {}), }) await self.broadcast({ "type": proto.S_CONVERSATION_UPDATED, "conversation_id": cid, "name": await self.router.get_name(cid), }) async def _handle_conversation_rename(self, client, message, request_id): conversation_id = str(message.get("conversation_id", "")) name = str(message.get("name", "")).strip() if not ConversationRouter.is_valid_conversation_id(conversation_id) \ or not name or len(name) > 200: await client.send_json(proto.error_event( proto.ERR_BAD_REQUEST, "conversation_id and name (≤200 chars) required", request_id)) return ok = await self.bridge.rename_conversation(conversation_id, name) if not ok: await client.send_json(proto.error_event( proto.ERR_CONVERSATION_NOT_FOUND, "Conversation not found", request_id)) return event = { "type": proto.S_CONVERSATION_RENAMED, "conversation_id": conversation_id, "name": name, **({"request_id": request_id} if request_id else {}), } await client.send_json(event) await self.broadcast({k: v for k, v in event.items() if k != "request_id"}) async def _handle_conversation_delete(self, client, message, request_id): conversation_id = str(message.get("conversation_id", "")) if not ConversationRouter.is_valid_conversation_id(conversation_id): await client.send_json(proto.error_event( proto.ERR_BAD_REQUEST, "Invalid conversation_id", request_id)) return ok = await self.bridge.delete_conversation(conversation_id) if not ok: await client.send_json(proto.error_event( proto.ERR_CONVERSATION_NOT_FOUND, "Conversation not found", request_id)) return event = { "type": proto.S_CONVERSATION_DELETED, "conversation_id": conversation_id, **({"request_id": request_id} if request_id else {}), } await client.send_json(event) await self.broadcast({k: v for k, v in event.items() if k != "request_id"}) async def _handle_chat_send(self, client, message, request_id): conversation_id = str(message.get("conversation_id", "")) text = message.get("text") if not ConversationRouter.is_valid_conversation_id(conversation_id): await client.send_json(proto.error_event( proto.ERR_BAD_REQUEST, "Invalid conversation_id", request_id)) return if not isinstance(text, str) or not text.strip(): await client.send_json(proto.error_event( proto.ERR_BAD_REQUEST, "text is required", request_id)) return if len(text) > proto.MAX_TEXT_CHARS: await client.send_json(proto.error_event( proto.ERR_TOO_LARGE, f"text exceeds {proto.MAX_TEXT_CHARS} chars", request_id)) return await self.bridge.send_chat( self, conversation_id, text, client, request_id) async def _handle_run_cancel(self, client, message, request_id): conversation_id = str(message.get("conversation_id", "")) if not ConversationRouter.is_valid_conversation_id(conversation_id): await client.send_json(proto.error_event( proto.ERR_BAD_REQUEST, "Invalid conversation_id", request_id)) return requested_run_id = message.get("run_id") active = self.bridge.active_run(conversation_id) run_id = active.get("run_id") if active else requested_run_id ok = await self.bridge.cancel_run(conversation_id, requested_run_id) if not ok: await client.send_json(proto.error_event( proto.ERR_NOT_FOUND, "No matching active run", request_id)) return if self.adapter is not None: self.adapter._drafts.pop(conversation_id, None) await self.broadcast({ "type": proto.S_RUN_FINISHED, "conversation_id": conversation_id, "run_id": run_id, "status": "cancelled", **({"request_id": request_id} if request_id else {}), }) async def _handle_approval_respond(self, client, message, request_id): approval_id = str(message.get("approval_id", "")) choice = str(message.get("choice", "")) reason = message.get("reason") resolved = await self.bridge.resolve_approval( approval_id, choice, str(reason) if reason else None) if not resolved: await client.send_json(proto.error_event( proto.ERR_APPROVAL_NOT_FOUND, "Unknown or already-resolved approval", request_id)) return await self.broadcast({ "type": proto.S_APPROVAL_RESOLVED, "approval_id": approval_id, "choice": choice, "accepted": True, **({"request_id": request_id} if request_id else {}), }) async def _handle_clarify_respond(self, client, message, request_id): clarify_id = str(message.get("clarify_id", "")) response = message.get("response") resolved = await self.bridge.resolve_clarify( clarify_id, str(response) if response is not None else "") if not resolved: await client.send_json(proto.error_event( proto.ERR_CLARIFY_NOT_FOUND, "Unknown or already-resolved clarification", request_id)) return await self.broadcast({ "type": proto.S_CLARIFY_RESOLVED, "clarify_id": clarify_id, "accepted": True, **({"request_id": request_id} if request_id else {}), }) async def _handle_yolo_current(self, client, message, request_id): conversation_id = str(message.get("conversation_id", "")) if not ConversationRouter.is_valid_conversation_id(conversation_id): await client.send_json(proto.error_event( proto.ERR_BAD_REQUEST, "Invalid conversation_id", request_id)) return result = await self.bridge.yolo_snapshot(conversation_id) if not result.get("ok"): await client.send_json(proto.error_event( result.get("code", proto.ERR_INTERNAL), result.get("message", "YOLO state unavailable"), request_id)) return await client.send_json({ "type": proto.S_YOLO_SNAPSHOT, "enabled": bool(result.get("enabled")), "scope": result.get("scope", "conversation"), "conversation_id": conversation_id, **({"request_id": request_id} if request_id else {}), }) async def _handle_yolo_set(self, client, message, request_id): conversation_id = str(message.get("conversation_id", "")) enabled = message.get("enabled") if not ConversationRouter.is_valid_conversation_id(conversation_id) \ or not isinstance(enabled, bool): await client.send_json(proto.error_event( proto.ERR_BAD_REQUEST, "conversation_id and boolean enabled are required", request_id)) return result = await self.bridge.set_yolo(conversation_id, enabled) if not result.get("ok"): await client.send_json(proto.error_event( result.get("code", proto.ERR_INTERNAL), result.get("message", "YOLO change failed"), request_id)) return event = { "type": proto.S_YOLO_CHANGED, "enabled": bool(result.get("enabled")), "scope": result.get("scope", "conversation"), "conversation_id": conversation_id, **({"request_id": request_id} if request_id else {}), } await client.send_json(event) await self.broadcast({k: v for k, v in event.items() if k != "request_id"}) async def _handle_models_list(self, client, message, request_id): conversation_id = message.get("conversation_id") snapshot = await self.bridge.models_snapshot( str(conversation_id) if conversation_id else None) snapshot["type"] = proto.S_MODELS_SNAPSHOT if request_id: snapshot["request_id"] = request_id await client.send_json(snapshot) async def _handle_model_current(self, client, message, request_id): conversation_id = message.get("conversation_id") snapshot = await self.bridge.current_model_snapshot( str(conversation_id) if conversation_id else None) snapshot["type"] = proto.S_MODEL_CURRENT_SNAPSHOT if request_id: snapshot["request_id"] = request_id await client.send_json(snapshot) async def _handle_model_set(self, client, message, request_id): model = str(message.get("model", "")).strip() provider = message.get("provider") conversation_id = message.get("conversation_id") result = await self.bridge.set_model( model, str(provider) if provider else None, str(conversation_id) if conversation_id else None) if not result.get("ok"): await client.send_json(proto.error_event( result.get("code", proto.ERR_BAD_REQUEST), result.get("message", "Model change failed"), request_id)) return event = { "type": proto.S_MODEL_CHANGED, "model": result.get("model"), "provider": result.get("provider"), "scope": result.get("scope", "global"), **({"request_id": request_id} if request_id else {}), } await client.send_json(event) await self.broadcast({k: v for k, v in event.items() if k != "request_id"}) async def _handle_reasoning_current(self, client, message, request_id): conversation_id = message.get("conversation_id") snapshot = await self.bridge.reasoning_snapshot( str(conversation_id) if conversation_id else None) snapshot["type"] = proto.S_REASONING_SNAPSHOT if request_id: snapshot["request_id"] = request_id await client.send_json(snapshot) async def _handle_reasoning_set(self, client, message, request_id): effort = str(message.get("effort", "")).strip().lower() conversation_id = message.get("conversation_id") result = await self.bridge.set_reasoning( effort, str(conversation_id) if conversation_id else None) if not result.get("ok"): await client.send_json(proto.error_event( result.get("code", proto.ERR_BAD_REQUEST), result.get("message", "Reasoning change failed"), request_id)) return event = { "type": proto.S_REASONING_CHANGED, "effort": result.get("effort"), "scope": result.get("scope", "global"), **({"request_id": request_id} if request_id else {}), } await client.send_json(event) await self.broadcast({k: v for k, v in event.items() if k != "request_id"}) _HANDLERS = { proto.C_PING: _handle_ping, proto.C_CONVERSATION_LIST: _handle_conversation_list, proto.C_CONVERSATION_OPEN: _handle_conversation_open, proto.C_CONVERSATION_CREATE: _handle_conversation_create, proto.C_CONVERSATION_RENAME: _handle_conversation_rename, proto.C_CONVERSATION_DELETE: _handle_conversation_delete, proto.C_CHAT_SEND: _handle_chat_send, proto.C_RUN_CANCEL: _handle_run_cancel, proto.C_APPROVAL_RESPOND: _handle_approval_respond, proto.C_CLARIFY_RESPOND: _handle_clarify_respond, proto.C_MODELS_LIST: _handle_models_list, proto.C_MODEL_SET: _handle_model_set, proto.C_MODEL_CURRENT: _handle_model_current, proto.C_REASONING_SET: _handle_reasoning_set, proto.C_REASONING_CURRENT: _handle_reasoning_current, proto.C_YOLO_SET: _handle_yolo_set, proto.C_YOLO_CURRENT: _handle_yolo_current, } __all__ = ["PhebyServer"]