titan-iac/services/communication/bstein-force-leave-job.yaml

# services/communication/bstein-force-leave-job.yaml
apiVersion: batch/v1
kind: Job
metadata:
  name: bstein-force-leave-1
  namespace: comms
spec:
  backoffLimit: 0
  template:
    spec:
      restartPolicy: Never
      containers:
        - name: force-leave
          image: python:3.11-slim
          env:
            - name: POSTGRES_HOST
              value: postgres-service.postgres.svc.cluster.local
            - name: POSTGRES_PORT
              value: "5432"
            - name: POSTGRES_DB
              value: synapse
            - name: POSTGRES_USER
              valueFrom:
                secretKeyRef:
                  name: synapse-db
                  key: POSTGRES_USER
            - name: POSTGRES_PASSWORD
              valueFrom:
                secretKeyRef:
                  name: synapse-db
                  key: POSTGRES_PASSWORD
            - name: SYNAPSE_BASE
              value: http://othrys-synapse-matrix-synapse:8008
            - name: AUTH_BASE
              value: http://matrix-authentication-service:8080
            - 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
              value: "!OkltaJguODUnZrbcUp:live.bstein.dev,!pMKAVvSRheIOCPIjDM:live.bstein.dev"
          command:
            - /bin/sh
            - -c
            - |
              set -euo pipefail
              pip install --no-cache-dir requests psycopg2-binary >/dev/null
              python - <<'PY'
              import json, os, sys, urllib.parse
              import requests
              import psycopg2

              DB = dict(
                  host=os.environ["POSTGRES_HOST"],
                  port=int(os.environ["POSTGRES_PORT"]),
                  dbname=os.environ["POSTGRES_DB"],
                  user=os.environ["POSTGRES_USER"],
                  password=os.environ["POSTGRES_PASSWORD"],
              )

              SYNAPSE_BASE = os.environ["SYNAPSE_BASE"]
              AUTH_BASE = os.environ["AUTH_BASE"]
              SERVER_NAME = os.environ.get("SERVER_NAME", "live.bstein.dev")
              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()]

              def db_connect():
                  return psycopg2.connect(**DB)

              def db_get_admin(conn, user_id):
                  with conn.cursor() as cur:
                      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}")
                      return bool(row[0])

              def db_set_admin(conn, user_id, is_admin):
                  with conn.cursor() as cur:
                      cur.execute("UPDATE users SET admin = %s WHERE name = %s", (bool(is_admin), user_id))

              def login(user, password):
                  r = requests.post(
                      f"{AUTH_BASE}/_matrix/client/v3/login",
                      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 admin_get_room_members(token, room_id):
                  url = f"{SYNAPSE_BASE}/_synapse/admin/v1/rooms/{urllib.parse.quote(room_id)}/members"
                  r = requests.get(url, headers={"Authorization": f"Bearer {token}"}, timeout=20)
                  if r.status_code == 404:
                      return None
                  r.raise_for_status()
                  data = r.json()
                  return data.get("members") or data.get("memberships") or data

              def admin_kick(token, room_id, user_id):
                  url = f"{SYNAPSE_BASE}/_synapse/admin/v1/rooms/{urllib.parse.quote(room_id)}/kick"
                  r = requests.post(
                      url,
                      headers={"Authorization": f"Bearer {token}"},
                      json={"user_id": user_id, "reason": "room cleanup"},
                      timeout=20,
                  )
                  if r.status_code == 404:
                      raise RuntimeError(f"kick endpoint not found on synapse ({url})")
                  if r.status_code not in (200, 202):
                      raise RuntimeError(f"kick failed for {room_id}: {r.status_code} {r.text}")

              seeder_user_id = f"@{SEEDER_USER}:{SERVER_NAME}"

              results = {"seeder_user_id": seeder_user_id, "target_user_id": TARGET_USER_ID, "rooms": {}}

              conn = db_connect()
              conn.autocommit = False
              try:
                  was_admin = db_get_admin(conn, seeder_user_id)
                  results["seeder_was_admin"] = was_admin
                  if not was_admin:
                      db_set_admin(conn, seeder_user_id, True)
                      conn.commit()

                  token = login(SEEDER_USER, SEEDER_PASS)

                  # Verify admin access now works.
                  # (If this still 403s, we fail and restore admin flag.)
                  for room_id in TARGET_ROOMS:
                      room_res = {}
                      results["rooms"][room_id] = room_res
                      try:
                          before = admin_get_room_members(token, room_id)
                          room_res["members_before"] = "unavailable" if before is None else "ok"
                      except Exception as e:
                          room_res["members_before_error"] = str(e)

                      try:
                          admin_kick(token, room_id, TARGET_USER_ID)
                          room_res["kicked"] = True
                      except Exception as e:
                          room_res["kicked"] = False
                          room_res["kick_error"] = str(e)

                      try:
                          after = admin_get_room_members(token, room_id)
                          room_res["members_after"] = "unavailable" if after is None else "ok"
                      except Exception as e:
                          room_res["members_after_error"] = str(e)

                  print(json.dumps(results, indent=2, sort_keys=True))
              finally:
                  # Restore previous admin flag
                  try:
                      if "results" in locals():
                          desired = results.get("seeder_was_admin", False)
                          db_set_admin(conn, seeder_user_id, desired)
                          conn.commit()
                  except Exception as e:
                      print(f"WARNING: failed to restore seeder admin flag: {e}", file=sys.stderr)
                  conn.close()
              PY