bstein/titan-iac
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

comms: leave stuck rooms via MAS admin

Browse Source
This commit is contained in:
Brad Stein 2026-01-01 18:26:50 -03:00
parent 144467dfe2
commit d43e40d515
1 changed files with 136 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

286
services/communication/bstein-force-leave-job.yaml
View File

@ -2,48 +2,40 @@
apiVersion: batch/v1 apiVersion: batch/v1
kind: Job kind: Job
metadata: metadata:
name: bstein-force-leave-6 name: bstein-leave-rooms-1
namespace: comms namespace: comms
spec: spec:
backoffLimit: 0 backoffLimit: 0
template: template:
spec: spec:
restartPolicy: Never restartPolicy: Never
volumes:
- name: mas-admin-client
secret:
secretName: mas-admin-client-runtime
items:
- key: client_secret
path: client_secret
containers: containers:
- name: force-leave - name: leave
image: python:3.11-slim image: python:3.11-slim
volumeMounts:
- name: mas-admin-client
mountPath: /etc/mas-admin-client
readOnly: true
env: env:
- name: POSTGRES_HOST - name: MAS_ADMIN_CLIENT_ID
value: postgres-service.postgres.svc.cluster.local value: 01KDXMVQBQ5JNY6SEJPZW6Z8BM
- name: POSTGRES_PORT - name: MAS_ADMIN_CLIENT_SECRET_FILE
value: "5432" value: /etc/mas-admin-client/client_secret
- name: POSTGRES_DB - name: MAS_TOKEN_URL
value: synapse value: http://matrix-authentication-service:8080/oauth2/token
- name: POSTGRES_USER - name: MAS_ADMIN_API_BASE
valueFrom: value: http://matrix-authentication-service:8081/api/admin/v1
secretKeyRef:
name: synapse-db
key: POSTGRES_USER
- name: POSTGRES_PASSWORD
valueFrom:
secretKeyRef:
name: synapse-db
key: POSTGRES_PASSWORD
- name: SYNAPSE_BASE - name: SYNAPSE_BASE
value: http://othrys-synapse-matrix-synapse:8008 value: http://othrys-synapse-matrix-synapse:8008
- name: AUTH_BASE - name: TARGET_USERNAME
value: http://matrix-authentication-service:8080 value: bstein
- name: SERVER_NAME
value: live.bstein.dev
- name: SEEDER_USER
value: othrys-seeder
- name: SEEDER_PASS
valueFrom:
secretKeyRef:
name: atlasbot-credentials-runtime
key: seeder-password
- name: TARGET_USER_ID
value: "@bstein:live.bstein.dev"
- name: TARGET_ROOMS - name: TARGET_ROOMS
value: "!OkltaJguODUnZrbcUp:live.bstein.dev,!pMKAVvSRheIOCPIjDM:live.bstein.dev" value: "!OkltaJguODUnZrbcUp:live.bstein.dev,!pMKAVvSRheIOCPIjDM:live.bstein.dev"
command: command:
@ -51,138 +43,132 @@ spec:
- -c - -c
- | - |
set -euo pipefail set -euo pipefail
pip install --no-cache-dir requests psycopg2-binary >/dev/null
python - <<'PY' python - <<'PY'
import json, os, sys, urllib.parse import base64
import requests import json
import psycopg2 import os
import urllib.error
import urllib.parse
import urllib.request
DB = dict( MAS_ADMIN_CLIENT_ID = os.environ["MAS_ADMIN_CLIENT_ID"]
host=os.environ["POSTGRES_HOST"], MAS_ADMIN_CLIENT_SECRET_FILE = os.environ["MAS_ADMIN_CLIENT_SECRET_FILE"]
port=int(os.environ["POSTGRES_PORT"]), MAS_TOKEN_URL = os.environ["MAS_TOKEN_URL"]
dbname=os.environ["POSTGRES_DB"], MAS_ADMIN_API_BASE = os.environ["MAS_ADMIN_API_BASE"].rstrip("/")
user=os.environ["POSTGRES_USER"], SYNAPSE_BASE = os.environ["SYNAPSE_BASE"].rstrip("/")
password=os.environ["POSTGRES_PASSWORD"], TARGET_USERNAME = os.environ["TARGET_USERNAME"]
)
SYNAPSE_BASE = os.environ["SYNAPSE_BASE"]
AUTH_BASE = os.environ["AUTH_BASE"]
SEEDER_USER = os.environ["SEEDER_USER"]
SEEDER_PASS = os.environ["SEEDER_PASS"]
TARGET_USER_ID = os.environ["TARGET_USER_ID"]
TARGET_ROOMS = [r.strip() for r in os.environ["TARGET_ROOMS"].split(",") if r.strip()] TARGET_ROOMS = [r.strip() for r in os.environ["TARGET_ROOMS"].split(",") if r.strip()]
def db_connect(): def http_json(method, url, *, headers=None, json_body=None, form=None, timeout=30):
return psycopg2.connect(**DB) req_headers = dict(headers or {})
data = None
def db_get_admin(conn, user_id): if json_body is not None and form is not None:
with conn.cursor() as cur: raise ValueError("choose json_body or form, not both")
cur.execute("SELECT admin FROM users WHERE name = %s", (user_id,))
row = cur.fetchone()
if not row:
raise RuntimeError(f"user not found in synapse db: {user_id}")
# Synapse stores admin as an int (0/1)
return int(row[0])
def db_set_admin(conn, user_id, is_admin): if json_body is not None:
with conn.cursor() as cur: data = json.dumps(json_body).encode()
cur.execute("UPDATE users SET admin = %s WHERE name = %s", (1 if is_admin else 0, user_id)) req_headers.setdefault("Content-Type", "application/json")
def login(user, password): if form is not None:
r = requests.post( data = urllib.parse.urlencode(form).encode()
f"{AUTH_BASE}/_matrix/client/v3/login", req_headers.setdefault("Content-Type", "application/x-www-form-urlencoded")
json={
"type": "m.login.password",
"identifier": {"type": "m.id.user", "user": user},
"password": password,
},
timeout=20,
)
if r.status_code != 200:
raise RuntimeError(f"login failed: {r.status_code} {r.text}")
return r.json()["access_token"]
def whoami(token): req = urllib.request.Request(url, data=data, method=method, headers=req_headers)
r = requests.get( try:
f"{SYNAPSE_BASE}/_matrix/client/v3/account/whoami", with urllib.request.urlopen(req, timeout=timeout) as resp:
headers={"Authorization": f"Bearer {token}"}, raw = resp.read()
timeout=20, payload = json.loads(raw.decode("utf-8")) if raw else None
) return resp.status, payload
r.raise_for_status() except urllib.error.HTTPError as e:
return r.json()["user_id"] raw = e.read()
try:
payload = json.loads(raw.decode("utf-8")) if raw else None
except Exception:
payload = None
return e.code, payload
def admin_delete_room(token, room_id): with open(MAS_ADMIN_CLIENT_SECRET_FILE, "r", encoding="utf-8") as f:
url = f"{SYNAPSE_BASE}/_synapse/admin/v1/rooms/{urllib.parse.quote(room_id)}" mas_admin_client_secret = f.read().strip()
r = requests.delete( if not mas_admin_client_secret:
url, raise RuntimeError("MAS admin client secret file is empty")
headers={"Authorization": f"Bearer {token}"},
json={"purge": False, "block": False},
timeout=60,
)
if r.status_code != 200:
raise RuntimeError(f"admin delete room failed: {room_id}: {r.status_code} {r.text}")
return r.json()
def admin_joined_rooms(token, user_id): basic = base64.b64encode(f"{MAS_ADMIN_CLIENT_ID}:{mas_admin_client_secret}".encode()).decode()
url = f"{SYNAPSE_BASE}/_synapse/admin/v1/users/{urllib.parse.quote(user_id)}/joined_rooms" token_status, token_payload = http_json(
r = requests.get(url, headers={"Authorization": f"Bearer {token}"}, timeout=20) "POST",
if r.status_code != 200: MAS_TOKEN_URL,
raise RuntimeError(f"admin joined_rooms failed: {r.status_code} {r.text}") headers={"Authorization": f"Basic {basic}"},
return r.json().get("joined_rooms", []) form={"grant_type": "client_credentials", "scope": "urn:mas:admin"},
timeout=30,
)
if token_status != 200 or not token_payload or "access_token" not in token_payload:
raise RuntimeError(f"MAS admin token request failed (HTTP {token_status})")
mas_admin_token = token_payload["access_token"]
results = {"target_user_id": TARGET_USER_ID, "rooms": {}} user_status, user_payload = http_json(
"GET",
f"{MAS_ADMIN_API_BASE}/users/by-username/{urllib.parse.quote(TARGET_USERNAME)}",
headers={"Authorization": f"Bearer {mas_admin_token}"},
timeout=30,
)
if user_status != 200 or not user_payload or "data" not in user_payload or "id" not in user_payload["data"]:
raise RuntimeError(f"MAS user lookup failed (HTTP {user_status})")
actor_user_id = user_payload["data"]["id"]
sess_status, sess_payload = http_json(
"POST",
f"{MAS_ADMIN_API_BASE}/personal-sessions",
headers={"Authorization": f"Bearer {mas_admin_token}"},
json_body={
"actor_user_id": actor_user_id,
"human_name": "bstein room cleanup",
"scope": "urn:matrix:client:api:*",
"expires_in": 300,
},
timeout=30,
)
if sess_status != 201 or not sess_payload or "data" not in sess_payload:
raise RuntimeError(f"MAS personal session create failed (HTTP {sess_status})")
personal_session_id = sess_payload["data"]["id"]
personal_token = (sess_payload.get("data", {}).get("attributes", {}) or {}).get("access_token")
if not personal_token:
raise RuntimeError("MAS personal session did not return an access token")
results = {"rooms": {}, "revoke": None}
failures = []
conn = db_connect()
conn.autocommit = False
try: try:
try:
token = login(SEEDER_USER, SEEDER_PASS)
results["seeder_login"] = "ok"
except Exception as e:
results["seeder_login"] = "error"
results["seeder_login_error"] = str(e)
print(json.dumps(results, indent=2, sort_keys=True))
raise
try:
seeder_user_id = whoami(token)
results["seeder_user_id"] = seeder_user_id
except Exception as e:
results["seeder_user_id_error"] = str(e)
seeder_user_id = None
if seeder_user_id:
try:
results["seeder_admin_db"] = db_get_admin(conn, seeder_user_id)
except Exception as e:
results["seeder_admin_db_error"] = str(e)
if results.get("seeder_admin_db") == 0:
try:
db_set_admin(conn, seeder_user_id, True)
conn.commit()
results["seeder_admin_db_promoted"] = True
except Exception as e:
results["seeder_admin_db_promote_error"] = str(e)
else:
results["seeder_admin_db_promoted"] = False
for room_id in TARGET_ROOMS: for room_id in TARGET_ROOMS:
room_res = {} room_q = urllib.parse.quote(room_id, safe="")
results["rooms"][room_id] = room_res leave_status, _ = http_json(
try: "POST",
room_res["delete"] = admin_delete_room(token, room_id) f"{SYNAPSE_BASE}/_matrix/client/v3/rooms/{room_q}/leave",
room_res["deleted"] = True headers={"Authorization": f"Bearer {personal_token}"},
except Exception as e: json_body={},
room_res["deleted"] = False timeout=30,
room_res["delete_error"] = str(e) )
forget_status, _ = http_json(
try: "POST",
results["target_joined_rooms_after"] = admin_joined_rooms(token, TARGET_USER_ID) f"{SYNAPSE_BASE}/_matrix/client/v3/rooms/{room_q}/forget",
except Exception as e: headers={"Authorization": f"Bearer {personal_token}"},
results["target_joined_rooms_after_error"] = str(e) json_body={},
timeout=30,
print(json.dumps(results, indent=2, sort_keys=True)) )
results["rooms"][room_id] = {"leave": leave_status, "forget": forget_status}
if leave_status != 200 or forget_status != 200:
failures.append(room_id)
finally: finally:
conn.close() revoke_status, _ = http_json(
"POST",
f"{MAS_ADMIN_API_BASE}/personal-sessions/{urllib.parse.quote(personal_session_id)}/revoke",
headers={"Authorization": f"Bearer {mas_admin_token}"},
json_body={},
timeout=30,
)
results["revoke"] = revoke_status
print(json.dumps(results, indent=2, sort_keys=True))
if failures:
raise SystemExit(f"failed to leave/forget rooms: {', '.join(failures)}")
PY PY