a703fb823c
Some checks failed
Forge CI / smoke-and-build (pull_request) Failing after 36s
Phase 2 of Matrix integration — wires Hermes to any Matrix homeserver. - docs/matrix-setup.md: step-by-step guide covering matrix.org (testing) and self-hosted (sovereignty) options, auth methods, E2EE setup, room config, and troubleshooting - scripts/setup_matrix.py: interactive wizard that prompts for homeserver, supports token/password auth, generates MATRIX_DEVICE_ID, writes ~/.hermes/.env and config.yaml, and optionally creates a test room + sends a test message No config.py changes needed — all Matrix env vars (MATRIX_HOMESERVER, MATRIX_ACCESS_TOKEN, MATRIX_USER_ID, MATRIX_PASSWORD, MATRIX_ENCRYPTION, MATRIX_DEVICE_ID, MATRIX_ALLOWED_USERS, MATRIX_HOME_ROOM, etc.) are already registered in OPTIONAL_ENV_VARS and _EXTRA_ENV_KEYS. Closes #271
431 lines
14 KiB
Python
Executable File
431 lines
14 KiB
Python
Executable File
#!/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()
|