comms: use bundled synapse admin ensure image

dockerfiles/Dockerfile.synapse-admin-ensure

@@ -0,0 +1,3 @@
+FROM python:3.11-slim
+
+RUN pip install --no-cache-dir psycopg2-binary bcrypt

services/comms/oneoffs/synapse-admin-ensure-job.yaml

@@ -1,12 +1,12 @@
 # services/comms/oneoffs/synapse-admin-ensure-job.yaml
-# One-off job for comms/synapse-admin-ensure-5.
+# One-off job for comms/synapse-admin-ensure-6.
-# Purpose: synapse admin ensure 5 (see container args/env in this file).
+# Purpose: synapse admin ensure 6 (see container args/env in this file).
 # Run by setting spec.suspend to false, reconcile, then set it back to true.
 # Safe to delete the finished Job/pod; it should not run continuously.
 apiVersion: batch/v1
 kind: Job
 metadata:
-  name: synapse-admin-ensure-5
+  name: synapse-admin-ensure-6
   namespace: comms
 spec:
   suspend: true
@@ -32,7 +32,7 @@ spec:
                     values: ["arm64"]
       containers:
         - name: ensure
-          image: python:3.11-slim
+          image: registry.bstein.dev/infra/synapse-admin-ensure:0.1.0
           env:
             - name: VAULT_ADDR
               value: http://vault.vault.svc.cluster.local:8200
@@ -40,62 +40,49 @@ spec:
               value: comms-secrets
             - name: SYNAPSE_ADMIN_URL
               value: http://othrys-synapse-matrix-synapse.comms.svc.cluster.local:8008
-            - name: MAS_AUTH_URL
-              value: http://matrix-authentication-service.comms.svc.cluster.local:8080
           command:
             - /bin/sh
             - -c
             - |
               set -euo pipefail
               python - <<'PY'
-              import hashlib
-              import hmac
               import json
               import os
               import secrets
               import string
+              import time
               import urllib.error
               import urllib.request
 
+              import bcrypt
+              import psycopg2
+
               VAULT_ADDR = os.environ.get("VAULT_ADDR", "http://vault.vault.svc.cluster.local:8200").rstrip("/")
               VAULT_ROLE = os.environ.get("VAULT_ROLE", "comms-secrets")
               SA_TOKEN_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/token"
               SYNAPSE_ADMIN_URL = os.environ.get("SYNAPSE_ADMIN_URL", "").rstrip("/")
-              MAS_AUTH_URL = os.environ.get("MAS_AUTH_URL", SYNAPSE_ADMIN_URL).rstrip("/")
+              PGHOST = "postgres-service.postgres.svc.cluster.local"
+              PGPORT = 5432
+              PGDATABASE = "synapse"
+              PGUSER = "synapse"
 
               def log(msg: str) -> None:
                   print(msg, flush=True)
 
-              def request_json(url: str, payload: dict | None = None, method: str | None = None) -> tuple[int, dict]:
+              def request_json(url: str, payload: dict | None = None) -> dict:
                   data = None
                   headers = {"Content-Type": "application/json"}
                   if payload is not None:
                       data = json.dumps(payload).encode("utf-8")
-                  req = urllib.request.Request(
+                  req = urllib.request.Request(url, data=data, headers=headers, method="POST" if data else "GET")
-                      url,
+                  with urllib.request.urlopen(req, timeout=30) as resp:
-                      data=data,
+                      return json.loads(resp.read().decode("utf-8"))
-                      headers=headers,
-                      method=method or ("POST" if data else "GET"),
-                  )
-                  try:
-                      with urllib.request.urlopen(req, timeout=30) as resp:
-                          body = resp.read().decode("utf-8")
-                          return resp.getcode(), json.loads(body) if body else {}
-                  except urllib.error.HTTPError as exc:
-                      body = exc.read().decode("utf-8")
-                      try:
-                          payload = json.loads(body) if body else {}
-                      except json.JSONDecodeError:
-                          payload = {}
-                      return exc.code, payload
 
               def vault_login() -> str:
                   with open(SA_TOKEN_PATH, "r", encoding="utf-8") as f:
                       jwt = f.read().strip()
                   payload = {"jwt": jwt, "role": VAULT_ROLE}
-                  status, resp = request_json(f"{VAULT_ADDR}/v1/auth/kubernetes/login", payload)
+                  resp = request_json(f"{VAULT_ADDR}/v1/auth/kubernetes/login", payload)
-                  if status != 200:
-                      raise RuntimeError(f"vault login failed: {status} {resp}")
                   token = resp.get("auth", {}).get("client_token")
                   if not token:
                       raise RuntimeError("vault login failed")
@@ -141,53 +128,61 @@ spec:
                   vault_put(token, "comms/synapse-admin", data)
                   return data
 
-              def registration_mac(nonce: str, username: str, password: str, admin: bool, shared_secret: str) -> str:
+              def ensure_user(cur, cols, user_id, password, admin):
-                  admin_value = "admin" if admin else "notadmin"
+                  now_ms = int(time.time() * 1000)
-                  msg = "\x00".join([nonce, username, password, admin_value]).encode("utf-8")
+                  values = {
-                  return hmac.new(shared_secret.encode("utf-8"), msg=msg, digestmod=hashlib.sha1).hexdigest()
+                      "name": user_id,
-
+                      "password_hash": bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode(),
-              def synapse_register(shared_secret: str, username: str, password: str, admin: bool) -> None:
+                      "creation_ts": now_ms,
-                  status, nonce_payload = request_json(
-                      f"{SYNAPSE_ADMIN_URL}/_synapse/admin/v1/register",
-                      method="GET",
-                  )
-                  if status != 200:
-                      raise RuntimeError(f"register nonce failed: {status}")
-                  nonce = (nonce_payload or {}).get("nonce")
-                  if not nonce:
-                      raise RuntimeError("register nonce missing")
-                  mac = registration_mac(nonce, username, password, admin, shared_secret)
-                  payload = {
-                      "nonce": nonce,
-                      "username": username,
-                      "password": password,
-                      "admin": admin,
-                      "mac": mac,
                   }
-                  status, resp = request_json(
-                      f"{SYNAPSE_ADMIN_URL}/_synapse/admin/v1/register",
-                      payload,
-                      method="POST",
-                  )
-                  if status in (200, 201):
-                      return
-                  if resp.get("errcode") == "M_USER_IN_USE":
-                      return
-                  raise RuntimeError(f"register failed: {status} {resp}")
 
-              def synapse_login(username: str, password: str) -> str:
+                  def add_flag(name, flag):
-                  payload = {
+                      if name not in cols:
-                      "type": "m.login.password",
+                          return
-                      "identifier": {"type": "m.id.user", "user": username},
+                      if cols[name]["type"] in ("smallint", "integer"):
-                      "password": password,
+                          values[name] = int(flag)
-                  }
+                      else:
-                  status, resp = request_json(f"{MAS_AUTH_URL}/_matrix/client/v3/login", payload, method="POST")
+                          values[name] = bool(flag)
-                  if status not in (200, 201):
+
-                      raise RuntimeError(f"login failed: {status} {resp}")
+                  add_flag("admin", admin)
-                  token = resp.get("access_token")
+                  add_flag("deactivated", False)
-                  if not token:
+                  add_flag("shadow_banned", False)
-                      raise RuntimeError("login returned no access token")
+                  add_flag("is_guest", False)
-                  return token
+
+                  columns = list(values.keys())
+                  placeholders = ", ".join(["%s"] * len(columns))
+                  updates = ", ".join([f"{col}=EXCLUDED.{col}" for col in columns if col != "name"])
+                  query = f"INSERT INTO users ({', '.join(columns)}) VALUES ({placeholders}) ON CONFLICT (name) DO UPDATE SET {updates};"
+                  cur.execute(query, [values[c] for c in columns])
+
+              def get_cols(cur):
+                  cur.execute(
+                      """
+                      SELECT column_name, is_nullable, column_default, data_type
+                      FROM information_schema.columns
+                      WHERE table_schema = 'public' AND table_name = 'users'
+                      """
+                  )
+                  cols = {}
+                  for name, is_nullable, default, data_type in cur.fetchall():
+                      cols[name] = {
+                          "nullable": is_nullable == "YES",
+                          "default": default,
+                          "type": data_type,
+                      }
+                  return cols
+
+              def ensure_access_token(cur, user_id, token_value):
+                  cur.execute("SELECT COALESCE(MAX(id), 0) + 1 FROM access_tokens")
+                  token_id = cur.fetchone()[0]
+                  cur.execute(
+                      """
+                      INSERT INTO access_tokens (id, user_id, token, device_id, valid_until_ms)
+                      VALUES (%s, %s, %s, %s, NULL)
+                      ON CONFLICT (token) DO NOTHING
+                      """,
+                      (token_id, user_id, token_value, "ariadne-admin"),
+                  )
 
               vault_token = vault_login()
               admin_data = ensure_admin_creds(vault_token)
@@ -195,13 +190,29 @@ spec:
                   log("synapse admin token already present")
                   raise SystemExit(0)
 
-              reg_secret = vault_get(vault_token, "comms/synapse-registration")
+              synapse_db = vault_get(vault_token, "comms/synapse-db")
-              shared_secret = (reg_secret.get("registration_shared_secret") or "").strip()
+              pg_password = synapse_db.get("POSTGRES_PASSWORD")
-              if not shared_secret:
+              if not pg_password:
-                  raise RuntimeError("registration shared secret missing")
+                  raise RuntimeError("synapse db password missing")
+
+              user_id = f"@{admin_data['username']}:live.bstein.dev"
+              conn = psycopg2.connect(
+                  host=PGHOST,
+                  port=PGPORT,
+                  dbname=PGDATABASE,
+                  user=PGUSER,
+                  password=pg_password,
+              )
+              token_value = secrets.token_urlsafe(32)
+              try:
+                  with conn:
+                      with conn.cursor() as cur:
+                          cols = get_cols(cur)
+                          ensure_user(cur, cols, user_id, admin_data["password"], True)
+                          ensure_access_token(cur, user_id, token_value)
+              finally:
+                  conn.close()
 
-              synapse_register(shared_secret, admin_data["username"], admin_data["password"], True)
-              token_value = synapse_login(admin_data["username"], admin_data["password"])
               admin_data["access_token"] = token_value
               vault_put(vault_token, "comms/synapse-admin", admin_data)
               log("synapse admin token stored")