267 lines
9.7 KiB
Python
267 lines
9.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Mock Synapse server for testing MAS UI without a real homeserver.
|
|
|
|
Handles both Modern (_synapse/mas/...) and Legacy (_synapse/admin/...)
|
|
API endpoints that MAS calls on Synapse.
|
|
|
|
Usage:
|
|
python3 mock_synapse.py
|
|
|
|
Then start MAS normally:
|
|
cargo run -- server -c config.yaml
|
|
"""
|
|
|
|
from http.server import HTTPServer, BaseHTTPRequestHandler
|
|
from urllib.parse import urlparse, parse_qs
|
|
import json
|
|
import sys
|
|
|
|
|
|
# In-memory store of provisioned users: localpart -> user data
|
|
provisioned_users = {}
|
|
|
|
|
|
class MockSynapseHandler(BaseHTTPRequestHandler):
|
|
|
|
def _send_json(self, data, status=200):
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(data).encode())
|
|
|
|
def _read_body(self):
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
if length:
|
|
return json.loads(self.rfile.read(length))
|
|
return {}
|
|
|
|
# ── GET endpoints ──────────────────────────────────────────────
|
|
|
|
def do_GET(self):
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path.rstrip("/")
|
|
qs = parse_qs(parsed.query)
|
|
|
|
# Modern: check if localpart is available
|
|
if path == "/_synapse/mas/is_localpart_available":
|
|
localpart = qs.get("localpart", [""])[0]
|
|
if localpart in provisioned_users:
|
|
print(f" [mock] is_localpart_available: {localpart} → in use")
|
|
self._send_json({"errcode": "M_USER_IN_USE", "error": "User ID already taken"}, 409)
|
|
else:
|
|
print(f" [mock] is_localpart_available: {localpart} → available")
|
|
self._send_json({}) # 200 = available
|
|
return
|
|
|
|
# Modern: query user info
|
|
if path == "/_synapse/mas/query_user":
|
|
localpart = qs.get("localpart", [""])[0]
|
|
if localpart in provisioned_users:
|
|
user = provisioned_users[localpart]
|
|
print(f" [mock] query_user: {localpart} → found")
|
|
self._send_json({
|
|
"user_id": f"@{localpart}:localhost:8008",
|
|
"display_name": user.get("display_name"),
|
|
"avatar_url": user.get("avatar_url"),
|
|
"is_suspended": False,
|
|
"is_deactivated": False,
|
|
})
|
|
else:
|
|
print(f" [mock] query_user: {localpart} → not found (404)")
|
|
self._send_json({"errcode": "M_NOT_FOUND", "error": "User not found"}, 404)
|
|
return
|
|
|
|
# Legacy: username available
|
|
if path == "/_synapse/admin/v1/username_available":
|
|
username = qs.get("username", [""])[0]
|
|
print(f" [mock] username_available: {username} → available=true")
|
|
self._send_json({"available": True})
|
|
return
|
|
|
|
# Legacy: get user info
|
|
if "/_synapse/admin/v2/users/" in path and "/devices" not in path:
|
|
print(f" [mock] get user → not found (404)")
|
|
self._send_json({"errcode": "M_NOT_FOUND", "error": "User not found"}, 404)
|
|
return
|
|
|
|
# Legacy: list devices
|
|
if "/_synapse/admin/v2/users/" in path and path.endswith("/devices"):
|
|
print(f" [mock] list devices → empty")
|
|
self._send_json({"devices": []})
|
|
return
|
|
|
|
# Matrix client versions
|
|
if path == "/_matrix/client/versions":
|
|
self._send_json({"versions": ["v1.1", "v1.2", "v1.3"]})
|
|
return
|
|
|
|
# Catch-all
|
|
print(f" [mock] Unhandled GET: {self.path}")
|
|
self._send_json({})
|
|
|
|
# ── POST endpoints ─────────────────────────────────────────────
|
|
|
|
def do_POST(self):
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path.rstrip("/")
|
|
body = self._read_body()
|
|
|
|
# Modern: provision user
|
|
if path == "/_synapse/mas/provision_user":
|
|
localpart = body.get("localpart", "?")
|
|
is_new = localpart not in provisioned_users
|
|
provisioned_users[localpart] = {
|
|
"display_name": body.get("set_displayname"),
|
|
"avatar_url": body.get("set_avatar_url"),
|
|
"emails": body.get("set_emails", []),
|
|
}
|
|
status = 201 if is_new else 200
|
|
print(f" [mock] provision_user: {localpart} → {status} {'created' if is_new else 'updated'}")
|
|
self._send_json({}, status)
|
|
return
|
|
|
|
# Modern: upsert device
|
|
if path == "/_synapse/mas/upsert_device":
|
|
print(f" [mock] upsert_device: {body.get('device_id', '?')} → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Modern: update device display name
|
|
if path == "/_synapse/mas/update_device_display_name":
|
|
print(f" [mock] update_device_display_name → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Modern: delete device
|
|
if path == "/_synapse/mas/delete_device":
|
|
print(f" [mock] delete_device → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Modern: sync devices
|
|
if path == "/_synapse/mas/sync_devices":
|
|
print(f" [mock] sync_devices → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Modern: delete user
|
|
if path == "/_synapse/mas/delete_user":
|
|
print(f" [mock] delete_user → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Modern: reactivate user
|
|
if path == "/_synapse/mas/reactivate_user":
|
|
print(f" [mock] reactivate_user → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Modern: set displayname
|
|
if path == "/_synapse/mas/set_displayname":
|
|
localpart = body.get("localpart", "?")
|
|
if localpart in provisioned_users:
|
|
provisioned_users[localpart]["display_name"] = body.get("displayname")
|
|
print(f" [mock] set_displayname: {localpart} → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Modern: unset displayname
|
|
if path == "/_synapse/mas/unset_displayname":
|
|
localpart = body.get("localpart", "?")
|
|
if localpart in provisioned_users:
|
|
provisioned_users[localpart]["display_name"] = None
|
|
print(f" [mock] unset_displayname: {localpart} → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Modern: allow cross-signing reset
|
|
if path == "/_synapse/mas/allow_cross_signing_reset":
|
|
print(f" [mock] allow_cross_signing_reset → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Legacy: create device
|
|
if "/_synapse/admin/v2/users/" in path and path.endswith("/devices"):
|
|
print(f" [mock] legacy create device → 201")
|
|
self._send_json({}, 201)
|
|
return
|
|
|
|
# Legacy: bulk delete devices
|
|
if "/_synapse/admin/v2/users/" in path and "delete_devices" in path:
|
|
print(f" [mock] legacy bulk delete devices → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Legacy: deactivate user
|
|
if "/_synapse/admin/v1/deactivate/" in path:
|
|
print(f" [mock] legacy deactivate → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Legacy: allow cross-signing reset
|
|
if "_allow_cross_signing_replacement_without_uia" in path:
|
|
print(f" [mock] legacy allow cross-signing reset → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Catch-all
|
|
print(f" [mock] Unhandled POST: {path} body={body}")
|
|
self._send_json({})
|
|
|
|
# ── PUT endpoints ──────────────────────────────────────────────
|
|
|
|
def do_PUT(self):
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path.rstrip("/")
|
|
body = self._read_body()
|
|
|
|
# Legacy: create/update user
|
|
if "/_synapse/admin/v2/users/" in path and "/devices/" not in path:
|
|
print(f" [mock] legacy put user → 201 created")
|
|
self._send_json({}, 201)
|
|
return
|
|
|
|
# Legacy: update device display name
|
|
if "/_synapse/admin/v2/users/" in path and "/devices/" in path:
|
|
print(f" [mock] legacy update device display name → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Client API: set displayname
|
|
if "/_matrix/client/v3/profile/" in path and "/displayname" in path:
|
|
print(f" [mock] set profile displayname → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Catch-all
|
|
print(f" [mock] Unhandled PUT: {path} body={body}")
|
|
self._send_json({})
|
|
|
|
# ── DELETE endpoints ───────────────────────────────────────────
|
|
|
|
def do_DELETE(self):
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path.rstrip("/")
|
|
|
|
# Legacy: delete device
|
|
if "/_synapse/admin/v2/users/" in path and "/devices/" in path:
|
|
print(f" [mock] legacy delete device → ok")
|
|
self._send_json({})
|
|
return
|
|
|
|
# Catch-all
|
|
print(f" [mock] Unhandled DELETE: {path}")
|
|
self._send_json({})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
port = int(sys.argv[1]) if len(sys.argv) > 1 else 8008
|
|
server = HTTPServer(("localhost", port), MockSynapseHandler)
|
|
print(f"Mock Synapse running on http://localhost:{port}")
|
|
print("Press Ctrl+C to stop\n")
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
print("\nStopped.")
|