#!/usr/bin/env python3 """Interactive Matrix setup wizard for Hermes Agent. Guides you through configuring Matrix integration: - Homeserver URL - Token auth or password auth - Device ID generation - Config/env file writing - Optional: test room creation and message send - E2EE verification Usage: python scripts/setup_matrix.py """ import getpass import json import os import secrets import sys import urllib.error import urllib.request from pathlib import Path # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _hermes_home() -> Path: """Resolve ~/.hermes (or HERMES_HOME override).""" return Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes")) def _prompt(msg: str, default: str = "") -> str: """Prompt with optional default. Returns stripped input or default.""" suffix = f" [{default}]" if default else "" val = input(f"{msg}{suffix}: ").strip() return val or default def _prompt_bool(msg: str, default: bool = True) -> bool: """Yes/no prompt.""" d = "Y/n" if default else "y/N" val = input(f"{msg} [{d}]: ").strip().lower() if not val: return default return val in ("y", "yes") def _http_post_json(url: str, data: dict, timeout: int = 15) -> dict: """POST JSON and return parsed response. Raises on HTTP errors.""" body = json.dumps(data).encode() req = urllib.request.Request( url, data=body, headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read()) except urllib.error.HTTPError as exc: detail = exc.read().decode(errors="replace") raise RuntimeError(f"HTTP {exc.code}: {detail}") from exc except urllib.error.URLError as exc: raise RuntimeError(f"Connection error: {exc.reason}") from exc def _http_get_json(url: str, token: str = "", timeout: int = 15) -> dict: """GET JSON, optionally with Bearer auth.""" req = urllib.request.Request(url, method="GET") if token: req.add_header("Authorization", f"Bearer {token}") try: with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read()) except urllib.error.HTTPError as exc: detail = exc.read().decode(errors="replace") raise RuntimeError(f"HTTP {exc.code}: {detail}") from exc except urllib.error.URLError as exc: raise RuntimeError(f"Connection error: {exc.reason}") from exc def _write_env_file(env_path: Path, vars: dict) -> None: """Write/update ~/.hermes/.env with given variables.""" existing: dict[str, str] = {} if env_path.exists(): for line in env_path.read_text().splitlines(): line = line.strip() if line and not line.startswith("#") and "=" in line: k, v = line.split("=", 1) existing[k.strip()] = v.strip().strip("'\"") existing.update(vars) lines = ["# Hermes Agent environment variables"] for k, v in sorted(existing.items()): # Quote values with spaces or special chars if any(c in v for c in " \t#\"'$"): lines.append(f'{k}="{v}"') else: lines.append(f"{k}={v}") env_path.parent.mkdir(parents=True, exist_ok=True) env_path.write_text("\n".join(lines) + "\n") try: os.chmod(str(env_path), 0o600) except (OSError, NotImplementedError): pass print(f" -> Wrote {len(vars)} vars to {env_path}") def _write_config_yaml(config_path: Path, matrix_section: dict) -> None: """Add/update matrix: section in config.yaml (creates file if needed).""" try: import yaml except ImportError: print(" [!] PyYAML not installed — skipping config.yaml update.") print(" Add manually under 'matrix:' key.") return config: dict = {} if config_path.exists(): try: config = yaml.safe_load(config_path.read_text()) or {} except Exception: config = {} config["matrix"] = matrix_section config_path.parent.mkdir(parents=True, exist_ok=True) config_path.write_text(yaml.dump(config, default_flow_style=False, sort_keys=False)) try: os.chmod(str(config_path), 0o600) except (OSError, NotImplementedError): pass print(f" -> Updated matrix section in {config_path}") def _generate_device_id() -> str: """Generate a stable, human-readable device ID.""" return f"HERMES_{secrets.token_hex(4).upper()}" # --------------------------------------------------------------------------- # Login flows # --------------------------------------------------------------------------- def login_with_token(homeserver: str) -> dict: """Validate an existing access token via whoami.""" token = getpass.getpass("Access token (hidden): ").strip() if not token: print(" [!] Token cannot be empty.") sys.exit(1) whoami_url = f"{homeserver}/_matrix/client/v3/account/whoami" print(" Validating token...") resp = _http_get_json(whoami_url, token=token) user_id = resp.get("user_id", "") device_id = resp.get("device_id", "") print(f" Authenticated as: {user_id}") if device_id: print(f" Server device ID: {device_id}") return { "MATRIX_ACCESS_TOKEN": token, "MATRIX_USER_ID": user_id, } def login_with_password(homeserver: str) -> dict: """Login with username + password, get access token.""" user_id = _prompt("Full user ID (e.g. @bot:matrix.org)") if not user_id: print(" [!] User ID cannot be empty.") sys.exit(1) password = getpass.getpass("Password (hidden): ").strip() if not password: print(" [!] Password cannot be empty.") sys.exit(1) login_url = f"{homeserver}/_matrix/client/v3/login" print(" Logging in...") resp = _http_post_json(login_url, { "type": "m.login.password", "identifier": { "type": "m.id.user", "user": user_id, }, "password": password, "device_name": "Hermes Agent", }) access_token = resp.get("access_token", "") device_id = resp.get("device_id", "") resolved_user = resp.get("user_id", user_id) if not access_token: print(" [!] Login succeeded but no access_token in response.") sys.exit(1) print(f" Authenticated as: {resolved_user}") if device_id: print(f" Device ID: {device_id}") return { "MATRIX_ACCESS_TOKEN": access_token, "MATRIX_USER_ID": resolved_user, "_server_device_id": device_id, } # --------------------------------------------------------------------------- # Test room + message # --------------------------------------------------------------------------- def create_test_room(homeserver: str, token: str) -> str | None: """Create a private test room and return the room ID.""" create_url = f"{homeserver}/_matrix/client/v3/createRoom" try: resp = _http_post_json(create_url, { "name": "Hermes Test Room", "topic": "Auto-created by hermes setup_matrix.py — safe to delete", "preset": "private_chat", "visibility": "private", }, timeout=30) # Set auth header manually (createRoom needs proper auth) room_id = resp.get("room_id", "") if room_id: print(f" Created test room: {room_id}") return room_id except Exception: pass # Fallback: use curl-style with auth req = urllib.request.Request( create_url, data=json.dumps({ "name": "Hermes Test Room", "topic": "Auto-created by hermes setup_matrix.py — safe to delete", "preset": "private_chat", "visibility": "private", }).encode(), headers={ "Content-Type": "application/json", "Authorization": f"Bearer {token}", }, method="POST", ) try: with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read()) room_id = data.get("room_id", "") if room_id: print(f" Created test room: {room_id}") return room_id except Exception as exc: print(f" [!] Room creation failed: {exc}") return None def send_test_message(homeserver: str, token: str, room_id: str) -> bool: """Send a test message to a room. Returns True on success.""" txn_id = secrets.token_hex(8) url = ( f"{homeserver}/_matrix/client/v3/rooms/" f"{urllib.request.quote(room_id, safe='')}/send/m.room.message/{txn_id}" ) req = urllib.request.Request( url, data=json.dumps({ "msgtype": "m.text", "body": "Hermes Agent setup verified successfully!", }).encode(), headers={ "Content-Type": "application/json", "Authorization": f"Bearer {token}", }, method="PUT", ) try: with urllib.request.urlopen(req, timeout=15) as resp: data = json.loads(resp.read()) event_id = data.get("event_id", "") if event_id: print(f" Test message sent: {event_id}") return True except Exception as exc: print(f" [!] Test message failed: {exc}") return False def check_e2ee_support() -> bool: """Check if E2EE dependencies are available.""" try: import nio from nio.crypto import ENCRYPTION_ENABLED return bool(ENCRYPTION_ENABLED) except (ImportError, AttributeError): return False # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): print("=" * 60) print(" Hermes Agent — Matrix Setup Wizard") print("=" * 60) print() # -- Homeserver -- print("Step 1: Homeserver") print(" A) matrix.org (public, for testing)") print(" B) Custom homeserver (self-hosted)") choice = _prompt("Choose [A/B]", "A").upper() if choice == "B": homeserver = _prompt("Homeserver URL (e.g. https://matrix.example.com)") if not homeserver: print(" [!] Homeserver URL is required.") sys.exit(1) else: homeserver = "https://matrix-client.matrix.org" homeserver = homeserver.rstrip("/") print(f" Using: {homeserver}") print() # -- Authentication -- print("Step 2: Authentication") print(" A) Access token (recommended)") print(" B) Username + password") auth_choice = _prompt("Choose [A/B]", "A").upper() if auth_choice == "B": auth_vars = login_with_password(homeserver) else: auth_vars = login_with_token(homeserver) print() # -- Device ID -- print("Step 3: Device ID (for E2EE persistence)") server_device = auth_vars.pop("_server_device_id", "") default_device = server_device or _generate_device_id() device_id = _prompt("Device ID", default_device) auth_vars["MATRIX_DEVICE_ID"] = device_id print() # -- E2EE -- print("Step 4: End-to-End Encryption") e2ee_available = check_e2ee_support() if e2ee_available: enable_e2ee = _prompt_bool("Enable E2EE?", default=False) if enable_e2ee: auth_vars["MATRIX_ENCRYPTION"] = "true" print(" E2EE enabled. Keys will be stored in:") print(" ~/.hermes/platforms/matrix/store/") else: print(" E2EE dependencies not found. Skipping.") print(" To enable later: pip install 'matrix-nio[e2e]'") print() # -- Optional settings -- print("Step 5: Optional Settings") allowed = _prompt("Allowed user IDs (comma-separated, or empty for all)") if allowed: auth_vars["MATRIX_ALLOWED_USERS"] = allowed home_room = _prompt("Home room ID for notifications (or empty)") if home_room: auth_vars["MATRIX_HOME_ROOM"] = home_room require_mention = _prompt_bool("Require @mention in rooms?", default=True) auto_thread = _prompt_bool("Auto-create threads?", default=True) print() # -- Write files -- print("Step 6: Writing Configuration") hermes_home = _hermes_home() env_path = hermes_home / ".env" _write_env_file(env_path, auth_vars) config_path = hermes_home / "config.yaml" matrix_cfg = { "require_mention": require_mention, "auto_thread": auto_thread, } _write_config_yaml(config_path, matrix_cfg) print() # -- Verify connection -- print("Step 7: Verification") token = auth_vars.get("MATRIX_ACCESS_TOKEN", "") do_test = _prompt_bool("Create test room and send message?", default=True) if do_test and token: room_id = create_test_room(homeserver, token) if room_id: send_test_message(homeserver, token, room_id) print() # -- Summary -- print("=" * 60) print(" Setup Complete!") print("=" * 60) print() print(" Config written to:") print(f" {env_path}") print(f" {config_path}") print() print(" To start the Matrix gateway:") print(" hermes gateway --platform matrix") print() if not e2ee_available: print(" To enable E2EE later:") print(" pip install 'matrix-nio[e2e]'") print(" Then set MATRIX_ENCRYPTION=true in .env") print() print(" Docs: docs/matrix-setup.md") print() if __name__ == "__main__": main()