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

comms: rotate invalid synapse admin token

Browse Source
This commit is contained in:
Brad Stein 2026-01-28 17:57:39 -03:00
parent 250fe22288
commit 171356a351
1 changed files with 29 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
services/comms/oneoffs/synapse-admin-ensure-job.yaml
View File

@ -1,12 +1,12 @@
# services/comms/oneoffs/synapse-admin-ensure-job.yaml
# One-off job for comms/synapse-admin-ensure-7.
# Purpose: synapse admin ensure 7 (see container args/env in this file).
# One-off job for comms/synapse-admin-ensure-8.
# Purpose: synapse admin ensure 8 (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-7
name: synapse-admin-ensure-8
namespace: comms
spec:
suspend: false
@ -53,6 +53,7 @@ spec:
import string
import time
import urllib.error
import urllib.parse
import urllib.request
import bcrypt
@ -185,18 +186,40 @@ spec:
(token_id, user_id, token_value, "ariadne-admin"),
)
def admin_token_valid(token: str, user_id: str) -> bool:
if not token or not SYNAPSE_ADMIN_URL:
return False
encoded = urllib.parse.quote(user_id, safe="")
url = f"{SYNAPSE_ADMIN_URL}/_synapse/admin/v2/users/{encoded}"
req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
try:
with urllib.request.urlopen(req, timeout=30) as resp:
resp.read()
return True
except urllib.error.HTTPError as exc:
if exc.code == 404:
return True
if exc.code in (401, 403):
return False
raise
vault_token = vault_login()
admin_data = ensure_admin_creds(vault_token)
if admin_data.get("access_token"):
log("synapse admin token already present")
user_id = f"@{admin_data['username']}:live.bstein.dev"
existing_token = admin_data.get("access_token")
if existing_token and admin_token_valid(existing_token, user_id):
log("synapse admin token already present and valid")
raise SystemExit(0)
if existing_token:
log("synapse admin token invalid; rotating")
admin_data.pop("access_token", None)
vault_put(vault_token, "comms/synapse-admin", admin_data)
synapse_db = vault_get(vault_token, "comms/synapse-db")
pg_password = synapse_db.get("POSTGRES_PASSWORD")
if not pg_password:
raise RuntimeError("synapse db password missing")
user_id = f"@{admin_data['username']}:live.bstein.dev"
conn = psycopg2.connect(
host=PGHOST,
port=PGPORT,