1411fded99
- Verify #183 scaffold completeness (MATRIX_SCAFFOLD_VERIFICATION.md) - Add bootstrap-fleet-rooms.py for automated Matrix room creation (#166) - Add CUTOVER_PLAN.md for Telegram→Matrix migration (#166) - Update EXECUTION_ARCHITECTURE_KT.md with new automation references Progresses #166, verifies #183
225 lines
8.2 KiB
Python
Executable File
225 lines
8.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""bootstrap-fleet-rooms.py — Automate Matrix room creation for Timmy fleet.
|
|
|
|
Issue: #166 (timmy-config)
|
|
Usage:
|
|
export MATRIX_HOMESERVER=https://matrix.timmytime.net
|
|
export MATRIX_ADMIN_TOKEN=<your_access_token>
|
|
python3 bootstrap-fleet-rooms.py --create-all --dry-run
|
|
|
|
Requires only Python stdlib (no heavy SDK dependencies).
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.request
|
|
from typing import Optional, List, Dict
|
|
|
|
|
|
class MatrixAdminClient:
|
|
"""Lightweight Matrix Client-Server API client."""
|
|
|
|
def __init__(self, homeserver: str, access_token: str):
|
|
self.homeserver = homeserver.rstrip("/")
|
|
self.access_token = access_token
|
|
|
|
def _request(self, method: str, path: str, data: Optional[Dict] = None) -> Dict:
|
|
url = f"{self.homeserver}/_matrix/client/v3{path}"
|
|
req = urllib.request.Request(url, method=method)
|
|
req.add_header("Authorization", f"Bearer {self.access_token}")
|
|
req.add_header("Content-Type", "application/json")
|
|
body = json.dumps(data).encode() if data else None
|
|
try:
|
|
with urllib.request.urlopen(req, data=body, timeout=30) as resp:
|
|
return json.loads(resp.read().decode())
|
|
except urllib.error.HTTPError as e:
|
|
try:
|
|
err = json.loads(e.read().decode())
|
|
except Exception:
|
|
err = {"error": str(e)}
|
|
return {"error": err, "status": e.code}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
def whoami(self) -> Dict:
|
|
return self._request("GET", "/account/whoami")
|
|
|
|
def create_room(self, name: str, topic: str, preset: str = "private_chat",
|
|
invite: Optional[List[str]] = None) -> Dict:
|
|
payload = {
|
|
"name": name,
|
|
"topic": topic,
|
|
"preset": preset,
|
|
"creation_content": {"m.federate": False},
|
|
}
|
|
if invite:
|
|
payload["invite"] = invite
|
|
return self._request("POST", "/createRoom", payload)
|
|
|
|
def send_state_event(self, room_id: str, event_type: str, state_key: str,
|
|
content: Dict) -> Dict:
|
|
path = f"/rooms/{room_id}/state/{event_type}/{state_key}"
|
|
return self._request("PUT", path, content)
|
|
|
|
def enable_encryption(self, room_id: str) -> Dict:
|
|
return self.send_state_event(
|
|
room_id, "m.room.encryption", "",
|
|
{"algorithm": "m.megolm.v1.aes-sha2"}
|
|
)
|
|
|
|
def set_room_avatar(self, room_id: str, url: str) -> Dict:
|
|
return self.send_state_event(
|
|
room_id, "m.room.avatar", "", {"url": url}
|
|
)
|
|
|
|
def generate_invite_link(self, room_id: str) -> str:
|
|
"""Generate a matrix.to invite link."""
|
|
localpart = room_id.split(":")[0].lstrip("#")
|
|
server = room_id.split(":")[1]
|
|
return f"https://matrix.to/#/{room_id}?via={server}"
|
|
|
|
|
|
def print_result(label: str, result: Dict):
|
|
if "error" in result:
|
|
print(f" ❌ {label}: {result['error']}")
|
|
else:
|
|
print(f" ✅ {label}: {json.dumps(result, indent=2)[:200]}")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Bootstrap Matrix rooms for Timmy fleet")
|
|
parser.add_argument("--homeserver", default=os.environ.get("MATRIX_HOMESERVER", ""),
|
|
help="Matrix homeserver URL (default: MATRIX_HOMESERVER env)")
|
|
parser.add_argument("--token", default=os.environ.get("MATRIX_ADMIN_TOKEN", ""),
|
|
help="Admin access token (default: MATRIX_ADMIN_TOKEN env)")
|
|
parser.add_argument("--operator-user", default="@alexander:matrix.timmytime.net",
|
|
help="Operator Matrix user ID")
|
|
parser.add_argument("--domain", default="matrix.timmytime.net",
|
|
help="Server domain for room aliases")
|
|
parser.add_argument("--create-all", action="store_true",
|
|
help="Create all standard fleet rooms")
|
|
parser.add_argument("--dry-run", action="store_true",
|
|
help="Preview actions without executing API calls")
|
|
args = parser.parse_args()
|
|
|
|
if not args.homeserver or not args.token:
|
|
print("Error: --homeserver and --token are required (or set env vars).")
|
|
sys.exit(1)
|
|
|
|
if args.dry_run:
|
|
print("=" * 60)
|
|
print(" DRY RUN — No API calls will be made")
|
|
print("=" * 60)
|
|
print(f"Homeserver: {args.homeserver}")
|
|
print(f"Operator: {args.operator_user}")
|
|
print(f"Domain: {args.domain}")
|
|
print("\nPlanned rooms:")
|
|
rooms = [
|
|
("Fleet Operations", "Encrypted command room for Alexander and agents.", "#fleet-ops"),
|
|
("General Chat", "Open fleet chatter and status updates.", "#fleet-general"),
|
|
("Alerts", "Automated alerts and monitoring notifications.", "#fleet-alerts"),
|
|
]
|
|
for name, topic, alias in rooms:
|
|
print(f" - {name} ({alias}:{args.domain})")
|
|
print(f" Topic: {topic}")
|
|
print(f" Actions: create → enable encryption → set alias")
|
|
print("\nNext steps after real run:")
|
|
print(" 1. Open Element Web and join with your operator account")
|
|
print(" 2. Share room invite links with fleet agents")
|
|
print(" 3. Configure Hermes gateway Matrix adapter")
|
|
return
|
|
|
|
client = MatrixAdminClient(args.homeserver, args.token)
|
|
|
|
print("Verifying credentials...")
|
|
identity = client.whoami()
|
|
if "error" in identity:
|
|
print(f"Authentication failed: {identity['error']}")
|
|
sys.exit(1)
|
|
print(f"Authenticated as: {identity.get('user_id', 'unknown')}")
|
|
|
|
rooms_spec = [
|
|
{
|
|
"name": "Fleet Operations",
|
|
"topic": "Encrypted command room for Alexander and agents. | Issue #166",
|
|
"alias": f"#fleet-ops:{args.domain}",
|
|
"preset": "private_chat",
|
|
},
|
|
{
|
|
"name": "General Chat",
|
|
"topic": "Open fleet chatter and status updates. | Issue #166",
|
|
"alias": f"#fleet-general:{args.domain}",
|
|
"preset": "public_chat",
|
|
},
|
|
{
|
|
"name": "Alerts",
|
|
"topic": "Automated alerts and monitoring notifications. | Issue #166",
|
|
"alias": f"#fleet-alerts:{args.domain}",
|
|
"preset": "private_chat",
|
|
},
|
|
]
|
|
|
|
created_rooms = []
|
|
|
|
for spec in rooms_spec:
|
|
print(f"\nCreating room: {spec['name']}...")
|
|
result = client.create_room(
|
|
name=spec["name"],
|
|
topic=spec["topic"],
|
|
preset=spec["preset"],
|
|
)
|
|
if "error" in result:
|
|
print_result("Create room", result)
|
|
continue
|
|
|
|
room_id = result.get("room_id")
|
|
print(f" ✅ Room created: {room_id}")
|
|
|
|
# Enable encryption
|
|
enc = client.enable_encryption(room_id)
|
|
print_result("Enable encryption", enc)
|
|
|
|
# Set canonical alias
|
|
alias_result = client.send_state_event(
|
|
room_id, "m.room.canonical_alias", "",
|
|
{"alias": spec["alias"]}
|
|
)
|
|
print_result("Set alias", alias_result)
|
|
|
|
# Set join rules (restricted for ops/alerts, public for general)
|
|
join_rule = "invite" if spec["preset"] == "private_chat" else "public"
|
|
jr = client.send_state_event(
|
|
room_id, "m.room.join_rules", "",
|
|
{"join_rule": join_rule}
|
|
)
|
|
print_result(f"Set join_rule={join_rule}", jr)
|
|
|
|
invite_link = client.generate_invite_link(room_id)
|
|
created_rooms.append({
|
|
"name": spec["name"],
|
|
"room_id": room_id,
|
|
"alias": spec["alias"],
|
|
"invite_link": invite_link,
|
|
})
|
|
|
|
print("\n" + "=" * 60)
|
|
print(" BOOTSTRAP COMPLETE")
|
|
print("=" * 60)
|
|
for room in created_rooms:
|
|
print(f"\n{room['name']}")
|
|
print(f" Alias: {room['alias']}")
|
|
print(f" Room ID: {room['room_id']}")
|
|
print(f" Invite: {room['invite_link']}")
|
|
|
|
print("\nNext steps:")
|
|
print(" 1. Join rooms from Element Web as operator")
|
|
print(" 2. Pin Fleet Operations as primary room")
|
|
print(" 3. Configure Hermes Matrix gateway with room aliases")
|
|
print(" 4. Follow docs/matrix-fleet-comms/CUTOVER_PLAN.md for Telegram transition")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|