commit a3b686a3720fa5ede0dd42d536f5e02d2cd35673
parent 632a6d251fc784e194e1abc93837da1c10a5d5e8
Author: MTRNord <MTRNord@users.noreply.github.com>
Date: Sat, 4 Apr 2026 00:56:11 +0200
improve matrix backup tool
Signed-off-by: MTRNord <MTRNord@users.noreply.github.com>
Diffstat:
9 files changed, 1339 insertions(+), 959 deletions(-)
diff --git a/apps/dns-a/gatus/config.yaml b/apps/dns-a/gatus/config.yaml
@@ -117,7 +117,7 @@ endpoints:
interval: 1m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- "[RESPONSE_TIME] < 5000"
- name: Grafana
@@ -126,7 +126,7 @@ endpoints:
interval: 2m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: ntfy
group: Infrastructure
@@ -135,7 +135,7 @@ endpoints:
conditions:
- "[STATUS] == 200"
- "[BODY].healthy == true"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: Flux Webhook
group: Infrastructure
@@ -143,7 +143,7 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == any(200, 404)" # endpoint exists, auth required
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
##############################################################################
# SOCIAL & COMMUNICATION
@@ -155,7 +155,7 @@ endpoints:
interval: 1m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- "[RESPONSE_TIME] < 3000"
- name: Matrix (continuwuity)
@@ -164,7 +164,7 @@ endpoints:
interval: 1m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- "has([BODY].versions) == true"
- name: Matrix Well-Known
@@ -173,7 +173,7 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: Matrix Client Well-Known
group: Social
@@ -181,7 +181,7 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: Draupnir
group: Social
@@ -189,7 +189,7 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == any(200, 404)"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
##############################################################################
# MAIL
@@ -229,7 +229,7 @@ endpoints:
interval: 15m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: WKD (nordgedanken.dev)
group: Mail
@@ -237,7 +237,7 @@ endpoints:
interval: 15m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
##############################################################################
# PERSONAL SERVICES
@@ -249,7 +249,7 @@ endpoints:
interval: 1m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: Immich
group: Personal
@@ -258,7 +258,7 @@ endpoints:
conditions:
- "[STATUS] == 200"
- "[BODY].res == pong"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: FreshRSS
group: Personal
@@ -266,7 +266,7 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: Bookwyrm
group: Personal
@@ -274,7 +274,7 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: ActualBudget
group: Personal
@@ -282,7 +282,7 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: Peertube
group: Personal
@@ -290,7 +290,7 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: Collabora
group: Personal
@@ -298,7 +298,7 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
##############################################################################
# PRODUCTIVITY
@@ -310,7 +310,7 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: Plane
group: Productivity
@@ -318,7 +318,7 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: Weblate
group: Productivity
@@ -326,7 +326,7 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
##############################################################################
# PUBLISHING & DEV
@@ -338,7 +338,7 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: cgit
group: Publishing
@@ -346,7 +346,7 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: Connectivity Tester
group: Publishing
@@ -354,7 +354,7 @@ endpoints:
interval: 2m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
- name: Connectivity Tester API
group: Publishing
@@ -362,4 +362,4 @@ endpoints:
interval: 5m
conditions:
- "[STATUS] == 200"
- - "[CERTIFICATE_EXPIRATION] > 48h"
+ - "[CERTIFICATE_EXPIRATION] > 24h"
diff --git a/apps/talos_cluster/image-builder/build-matrix-backup-cronjob.yaml b/apps/talos_cluster/image-builder/build-matrix-backup-cronjob.yaml
@@ -0,0 +1,136 @@
+apiVersion: batch/v1
+kind: CronJob
+metadata:
+ name: build-matrix-backup
+ namespace: image-builder
+spec:
+ schedule: "0 4 * * *"
+ concurrencyPolicy: Forbid
+ successfulJobsHistoryLimit: 3
+ failedJobsHistoryLimit: 3
+ jobTemplate:
+ spec:
+ ttlSecondsAfterFinished: 86400
+ template:
+ spec:
+ serviceAccountName: image-builder-cronjob
+ restartPolicy: Never
+ initContainers:
+ - name: setup-buildctl
+ image: moby/buildkit:v0.29.0
+ command: [cp, /usr/bin/buildctl, /tools/buildctl]
+ volumeMounts:
+ - name: tools
+ mountPath: /tools
+
+ - name: setup-cosign
+ image: alpine:3
+ command:
+ - /bin/sh
+ - -c
+ - |
+ wget -qO /tools/cosign https://github.com/sigstore/cosign/releases/download/v3.0.5/cosign-linux-arm64 \
+ && chmod +x /tools/cosign
+ volumeMounts:
+ - name: tools
+ mountPath: /tools
+
+ - name: setup-crane
+ image: alpine:3
+ command:
+ - /bin/sh
+ - -c
+ - |
+ wget -qO /tmp/crane.tar.gz https://github.com/google/go-containerregistry/releases/download/v0.20.2/go-containerregistry_Linux_arm64.tar.gz \
+ && tar -xzf /tmp/crane.tar.gz -C /tools crane \
+ && chmod +x /tools/crane
+ volumeMounts:
+ - name: tools
+ mountPath: /tools
+
+ - name: clone-repo
+ image: alpine/git:latest
+ command:
+ - git
+ - clone
+ - --depth=1
+ - https://github.com/MTRNord/cluster.git
+ - /workspace
+ volumeMounts:
+ - name: workspace
+ mountPath: /workspace
+
+ containers:
+ - name: build
+ image: alpine:3
+ command:
+ - /bin/sh
+ - -c
+ - |
+ set -eu
+ apk add --no-cache git >/dev/null
+ export PATH="/tools:$PATH"
+
+ mkdir -p /root/.docker
+ cp /registry-secret/dockerconfig.json /root/.docker/config.json
+
+ TAG_TS="$(date -u +%Y%m%d-%H%M%S)"
+ TAG_SHA="sha-$(git -C /workspace rev-parse --short HEAD)"
+ IMAGE="registry.midnightthoughts.space/mtrnord/cluster/matrix-backup"
+
+ echo "==> Building ${IMAGE}:${TAG_TS}"
+
+ buildctl \
+ --addr tcp://buildkitd.image-builder.svc.cluster.local:1234 \
+ build \
+ --frontend dockerfile.v0 \
+ --opt platform=linux/amd64,linux/arm64 \
+ --local context=/workspace/apps/talos_cluster/matrix-backup/backup-tool \
+ --local dockerfile=/workspace/apps/talos_cluster/matrix-backup/backup-tool \
+ --output "type=image,name=${IMAGE}:${TAG_TS},push=true,oci-mediatypes=true" \
+ --metadata-file /tmp/build-meta.json
+
+ DIGEST="$(crane digest ${IMAGE}:${TAG_TS})"
+ echo "==> Digest: ${DIGEST}"
+
+ crane cp "${IMAGE}:${TAG_TS}" "${IMAGE}:main"
+ crane cp "${IMAGE}:${TAG_TS}" "${IMAGE}:${TAG_SHA}"
+
+ COSIGN_PASSWORD="$(cat /cosign-secret/cosign.password)" \
+ cosign sign --yes \
+ --key /cosign-secret/cosign.key \
+ --new-bundle-format=false \
+ --use-signing-config=false \
+ --registry-referrers-mode=oci-1-1 \
+ "${IMAGE}@${DIGEST}"
+
+ echo "==> Done: ${IMAGE}:${TAG_TS}"
+ env:
+ - name: COSIGN_OCI_EXPERIMENTAL
+ value: "1"
+ - name: COSIGN_EXPERIMENTAL
+ value: "1"
+ volumeMounts:
+ - name: tools
+ mountPath: /tools
+ - name: workspace
+ mountPath: /workspace
+ readOnly: true
+ - name: registry-secret
+ mountPath: /registry-secret
+ readOnly: true
+ - name: cosign-secret
+ mountPath: /cosign-secret
+ readOnly: true
+
+ volumes:
+ - name: tools
+ emptyDir: {}
+ - name: workspace
+ emptyDir: {}
+ - name: registry-secret
+ secret:
+ secretName: image-builder-registry-secret
+ - name: cosign-secret
+ secret:
+ secretName: image-builder-cosign-secret
diff --git a/apps/talos_cluster/image-builder/kustomization.yaml b/apps/talos_cluster/image-builder/kustomization.yaml
@@ -11,3 +11,4 @@ resources:
- build-blog-cronjob.yaml
- build-continuwuity-cronjob.yaml
- build-bookwyrm-cronjob.yaml
+ - build-matrix-backup-cronjob.yaml
diff --git a/apps/talos_cluster/matrix-backup/backup-cronjob.yaml b/apps/talos_cluster/matrix-backup/backup-cronjob.yaml
@@ -18,40 +18,14 @@ spec:
restartPolicy: OnFailure
securityContext:
runAsNonRoot: true
- runAsUser: 1000
- runAsGroup: 1000
- fsGroup: 1000
+ runAsUser: 65532
+ runAsGroup: 65532
+ fsGroup: 65532
seccompProfile:
type: RuntimeDefault
- initContainers:
- - name: install-deps
- image: python:3.12-slim
- command:
- - /bin/sh
- - -c
- - pip install --quiet --no-cache-dir --target /deps 'matrix-nio[e2e]' boto3 cryptography
- securityContext:
- allowPrivilegeEscalation: false
- readOnlyRootFilesystem: false
- capabilities:
- drop:
- - ALL
- resources:
- requests:
- cpu: 200m
- memory: 256Mi
- limits:
- memory: 512Mi
- volumeMounts:
- - name: deps
- mountPath: /deps
containers:
- name: backup
- image: python:3.12-slim
- command:
- - /bin/sh
- - -c
- - PYTHONPATH=/deps python /scripts/backup.py
+ image: registry.midnightthoughts.space/mtrnord/cluster/matrix-backup:main
securityContext:
allowPrivilegeEscalation: false
readOnlyRootFilesystem: true
@@ -61,7 +35,7 @@ spec:
resources:
requests:
cpu: 100m
- memory: 256Mi
+ memory: 128Mi
limits:
memory: 512Mi
env:
@@ -107,23 +81,12 @@ spec:
name: matrix-backup-secret
key: key_export_passphrase
volumeMounts:
- - name: scripts
- mountPath: /scripts
- readOnly: true
- - name: crypto-store
- mountPath: /data/crypto
- - name: deps
- mountPath: /deps
- readOnly: true
+ - name: data
+ mountPath: /data
- name: tmp
mountPath: /tmp
volumes:
- - name: scripts
- configMap:
- name: matrix-backup-script
- - name: crypto-store
- emptyDir: {}
- - name: deps
+ - name: data
emptyDir: {}
- name: tmp
emptyDir: {}
diff --git a/apps/talos_cluster/matrix-backup/backup-script-configmap.yaml b/apps/talos_cluster/matrix-backup/backup-script-configmap.yaml
@@ -1,888 +0,0 @@
-apiVersion: v1
-kind: ConfigMap
-metadata:
- name: matrix-backup-script
- namespace: matrix-backup
-data:
- backup.py: |
- #!/usr/bin/env python3
- """
- Matrix backup script.
-
- Auth lifecycle:
- - First run: login with password → creates a dedicated "matrix-backup" device.
- session.json (access_token + device_id) is saved into the crypto store.
- - Subsequent runs: restore_login from session.json.
- - The password in the secret is ONLY used on first run.
-
- SSSS bootstrap (first run only):
- - Decrypts the megolm key backup private key from Secure Secret Storage using
- the provided recovery key.
- - Fetches all backed-up room keys from the server.
- - Decrypts them with the backup private key (olm PK) and imports them into
- the local crypto store so the bot can decrypt E2EE messages.
-
- Per-run backup:
- - Crypto store round-tripped via S3 tarball (download at start, upload at end).
- - Room list with IDs, aliases, server hints, names, types.
- - E2EE exported key file (encrypted with KEY_EXPORT_PASSPHRASE).
- - Incremental message history as JSONL.
- - Images sent by *:mtrnord.blog users (downloaded and stored to S3).
- - Display-name and avatar changes for *:mtrnord.blog users.
- """
- import asyncio
- import base64
- import hashlib
- import hmac as hmac_lib
- import io
- import json
- import mimetypes
- import os
- import re
- import struct
- import sys
- import tarfile
- import tempfile
- from collections import Counter
- from datetime import datetime, timezone
-
- import boto3
- from botocore.exceptions import ClientError
- from cryptography.hazmat.backends import default_backend
- from cryptography.hazmat.primitives import hashes
- from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
- from cryptography.hazmat.primitives.kdf.hkdf import HKDF
- from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
- from nio import (
- AsyncClient,
- AsyncClientConfig,
- DownloadError,
- LoginError,
- MessageDirection,
- RoomMessagesError,
- )
-
- # Environment
- HOMESERVER = os.environ["HOMESERVER"]
- S3_ENDPOINT = os.environ["S3_ENDPOINT"]
- S3_BUCKET = os.environ["S3_BUCKET"]
- S3_ACCESS_KEY = os.environ["S3_ACCESS_KEY"]
- S3_SECRET_KEY = os.environ["S3_SECRET_KEY"]
- KEY_EXPORT_PASSPHRASE = os.environ["KEY_EXPORT_PASSPHRASE"]
- MTRNORD_PASSWORD = os.environ["MTRNORD_PASSWORD"]
- MTRNORD_SSSS_KEY = os.environ["MTRNORD_SSSS_KEY"]
- LEXI_PASSWORD = os.environ["LEXI_PASSWORD"]
- LEXI_SSSS_KEY = os.environ["LEXI_SSSS_KEY"]
-
- OUR_HOMESERVER_SUFFIX = ":mtrnord.blog"
- DATE_STR = datetime.now(timezone.utc).strftime("%Y%m%d")
-
- ACCOUNTS = [
- {
- "user_id": "@mtrnord:mtrnord.blog",
- "password": MTRNORD_PASSWORD,
- "ssss_key": MTRNORD_SSSS_KEY,
- "store": "/data/crypto/mtrnord",
- "prefix": "mtrnord",
- },
- {
- "user_id": "@lexi:mtrnord.blog",
- "password": LEXI_PASSWORD,
- "ssss_key": LEXI_SSSS_KEY,
- "store": "/data/crypto/lexi",
- "prefix": "lexi",
- },
- ]
-
- # S3 helpers
- s3 = boto3.client(
- "s3",
- endpoint_url=f"https://{S3_ENDPOINT}",
- aws_access_key_id=S3_ACCESS_KEY,
- aws_secret_access_key=S3_SECRET_KEY,
- region_name="hel1",
- )
-
-
- def s3_get_json(key):
- try:
- obj = s3.get_object(Bucket=S3_BUCKET, Key=key)
- return json.loads(obj["Body"].read())
- except ClientError as e:
- if e.response["Error"]["Code"] in ("NoSuchKey", "404"):
- return None
- raise
-
-
- def s3_get_bytes(key):
- try:
- obj = s3.get_object(Bucket=S3_BUCKET, Key=key)
- return obj["Body"].read()
- except ClientError as e:
- if e.response["Error"]["Code"] in ("NoSuchKey", "404"):
- return None
- raise
-
-
- def s3_put(key, data: bytes, content_type: str = "application/octet-stream"):
- s3.put_object(Bucket=S3_BUCKET, Key=key, Body=data, ContentType=content_type)
-
-
- def s3_exists(key):
- try:
- s3.head_object(Bucket=S3_BUCKET, Key=key)
- return True
- except ClientError as e:
- if e.response["Error"]["Code"] in ("404", "NoSuchKey"):
- return False
- raise
-
-
- # Crypto store S3 round-trip
- def download_crypto_store(store_path: str, s3_key: str):
- data = s3_get_bytes(s3_key)
- if not data:
- print(f" No existing crypto store at {s3_key} — starting fresh.")
- return
- os.makedirs(store_path, exist_ok=True)
- with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as tar:
- tar.extractall(store_path, filter="data")
- print(f" Crypto store restored from S3 ({len(data)} bytes).")
-
-
- def upload_crypto_store(store_path: str, s3_key: str):
- buf = io.BytesIO()
- with tarfile.open(fileobj=buf, mode="w:gz") as tar:
- tar.add(store_path, arcname=".")
- data = buf.getvalue()
- s3_put(s3_key, data)
- print(f" Crypto store saved to S3 ({len(data)} bytes).")
-
-
- # Session management (password login, once only)
- def load_session(store_path: str):
- """Return stored session dict or None if not yet created."""
- session_file = os.path.join(store_path, "session.json")
- if not os.path.exists(session_file):
- return None
- try:
- with open(session_file) as f:
- return json.load(f)
- except Exception:
- return None
-
-
- def save_session(store_path: str, session: dict):
- os.makedirs(store_path, exist_ok=True)
- with open(os.path.join(store_path, "session.json"), "w") as f:
- json.dump(session, f)
-
-
- async def ensure_session(account: dict) -> dict:
- """
- Return a session dict with access_token, device_id, ssss_bootstrapped.
- Performs a password login on first run.
- """
- session = load_session(account["store"])
- if session:
- return session
-
- print(f" First run: logging in with password for {account['user_id']} ...")
- tmp_client = AsyncClient(HOMESERVER, account["user_id"])
- try:
- resp = await tmp_client.login(
- password=account["password"],
- device_name="matrix-backup",
- )
- if isinstance(resp, LoginError):
- raise RuntimeError(f"Login failed: {resp}")
- finally:
- await tmp_client.close()
-
- session = {
- "access_token": resp.access_token,
- "device_id": resp.device_id,
- "ssss_bootstrapped": False,
- }
- save_session(account["store"], session)
- print(f" Logged in. Device ID: {resp.device_id}")
- return session
-
-
- # SSSS recovery key decoding
- _BASE58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
-
-
- def decode_recovery_key(key_str: str) -> bytes:
- """
- Decode a Matrix recovery/security key to its 32 raw key bytes.
- Accepts keys with or without spaces/dashes.
- """
- cleaned = key_str.replace(" ", "").replace("-", "")
- n = 0
- for ch in cleaned:
- n = n * 58 + _BASE58.index(ch)
- raw = n.to_bytes(35, "big")
- if raw[0] != 0x8B or raw[1] != 0x01:
- raise ValueError("Invalid recovery key prefix")
- parity = 0
- for b in raw[:34]:
- parity ^= b
- if parity != raw[34]:
- raise ValueError("Invalid recovery key parity byte")
- return raw[2:34] # 32-byte key
-
-
- # SSSS AES-HMAC-SHA2 decryption
- def _ssss_derive(raw_key: bytes, secret_name: str):
- """Derive (aes_key, mac_key) for a given secret name via HKDF-SHA256."""
- derived = HKDF(
- algorithm=hashes.SHA256(),
- length=64,
- salt=b"\x00" * 32,
- info=secret_name.encode(),
- backend=default_backend(),
- ).derive(raw_key)
- return derived[:32], derived[32:]
-
-
- def ssss_decrypt(encrypted: dict, raw_key: bytes, secret_name: str) -> bytes:
- """Decrypt one entry from the 'encrypted' map of an SSSS account-data event."""
- aes_key, mac_key = _ssss_derive(raw_key, secret_name)
- ciphertext = base64.b64decode(encrypted["ciphertext"])
- iv = base64.b64decode(encrypted["iv"])
- stored_mac = base64.b64decode(encrypted["mac"])
-
- expected_mac = hmac_lib.new(mac_key, ciphertext, hashlib.sha256).digest()
- if not hmac_lib.compare_digest(expected_mac, stored_mac):
- raise ValueError("SSSS MAC mismatch — wrong recovery key?")
-
- cipher = Cipher(algorithms.AES(aes_key), modes.CTR(iv), backend=default_backend())
- dec = cipher.decryptor()
- return dec.update(ciphertext) + dec.finalize()
-
-
- # Megolm export file creation (for client.import_keys)
- def build_megolm_export(sessions: list, passphrase: str = "_backup_import_") -> bytes:
- """
- Encode a list of session dicts into the standard Megolm key export format
- so they can be loaded with AsyncClient.import_keys().
-
- sessions: [{"algorithm","room_id","sender_key","session_id","session_key",
- "sender_claimed_keys","forwarding_curve25519_key_chain"}] # gitleaks:allow
- """
- salt = os.urandom(16)
- iv_bytes = bytearray(os.urandom(16))
- iv_bytes[0] &= 0x7F # ensure highest bit is 0 as spec requires
- iv = bytes(iv_bytes)
- iterations = 100_000
-
- kdf = PBKDF2HMAC(
- algorithm=hashes.SHA512(),
- length=64,
- salt=salt,
- iterations=iterations,
- backend=default_backend(),
- )
- key_material = kdf.derive(passphrase.encode())
- aes_key = key_material[:32]
- hmac_key = key_material[32:]
-
- plaintext = json.dumps(sessions).encode()
- cipher = Cipher(algorithms.AES(aes_key), modes.CTR(iv), backend=default_backend())
- encryptor = cipher.encryptor()
- ciphertext = encryptor.update(plaintext) + encryptor.finalize()
-
- payload = b"\x01" + salt + iv + struct.pack(">I", iterations) + ciphertext
- mac = hmac_lib.new(hmac_key, payload, hashlib.sha256).digest()
-
- encoded = base64.b64encode(payload + mac).decode()
- lines = "\n".join(encoded[i : i + 76] for i in range(0, len(encoded), 76))
- return (
- "-----BEGIN MEGOLM SESSION DATA-----\n"
- + lines
- + "\n-----END MEGOLM SESSION DATA-----\n"
- ).encode()
-
-
- # SSSS bootstrap: fetch and import key backup
- def _parse_account_data(sync_resp) -> dict:
- """Extract account data events from a SyncResponse into a {type: event} dict."""
- result = {}
- for event in getattr(sync_resp, "account_data", []):
- key = getattr(event, "type", None) or getattr(event, "event_type", None)
- if key:
- result[key] = event
- return result
-
-
- def _fetch_account_data_event(access_token: str, user_id: str, event_type: str) -> dict | None:
- """
- Fetch a single account data event directly from the server (read-only GET).
- Returns the content dict, or None if the event doesn't exist.
- Incremental syncs don't re-deliver unchanged account data, so we fetch directly.
- """
- import urllib.request, urllib.error
- url = f"{HOMESERVER}/_matrix/client/v3/user/{urllib.request.quote(user_id, safe='')}/account_data/{urllib.request.quote(event_type, safe='')}"
- req = urllib.request.Request(url, headers={"Authorization": f"Bearer {access_token}"})
- try:
- with urllib.request.urlopen(req, timeout=15) as r:
- return json.loads(r.read())
- except urllib.error.HTTPError as e:
- if e.code == 404:
- return None
- raise
-
-
- async def _decrypt_ssss_backup_key(client: AsyncClient, user_id: str, recovery_key_str: str) -> str | None:
- """
- Decrypt and return the base64-encoded megolm backup private key from SSSS.
- Fetches the required account data events directly from the server (read-only GETs)
- so this works on both first run and incremental syncs.
- Returns None if SSSS data is missing or decryption fails.
- """
- raw_key = decode_recovery_key(recovery_key_str)
-
- default_key_content = _fetch_account_data_event(client.access_token, user_id, "m.secret_storage.default_key")
- if not default_key_content:
- print(" Warning: no m.secret_storage.default_key in account data")
- return None
- key_id = default_key_content.get("key")
- if not key_id:
- print(" Warning: m.secret_storage.default_key has no 'key' field")
- return None
-
- backup_secret_content = _fetch_account_data_event(client.access_token, user_id, "m.megolm_backup.v1")
- if not backup_secret_content:
- print(" Warning: no m.megolm_backup.v1 in account data")
- return None
- encrypted_map = backup_secret_content.get("encrypted", {})
- if key_id not in encrypted_map:
- print(f" Warning: key_id {key_id} not found in backup secret encrypted map")
- return None
-
- backup_key_bytes = ssss_decrypt(encrypted_map[key_id], raw_key, "m.megolm_backup.v1")
- return backup_key_bytes.decode().strip()
-
-
- def _olm_pk_decrypt(private_key_bytes: bytes, ciphertext_b64: str, mac_b64: str, ephemeral_b64: str) -> str:
- """
- Decrypt an olm PK-encrypted payload using the cryptography library directly,
- avoiding the need for PkDecryption.from_private_key (requires libolm >= 3.2.3).
-
- Olm PK decryption spec:
- 1. ECDH(our_private, ephemeral_public) → shared_secret (X25519)
- 2. HKDF-SHA256(salt=\\x00*32, ikm=shared_secret, info=b"OLM_KEYS") → 80 bytes
- → aes_key[0:32], mac_key[32:64], iv[64:80]
- 3. Verify HMAC-SHA256(mac_key, ciphertext) == mac
- 4. AES-256-CBC decrypt(aes_key, iv, ciphertext), unpad PKCS7
- """
- from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey, X25519PublicKey
- from cryptography.hazmat.primitives import padding as sym_padding
-
- def _b64decode(s: str) -> bytes:
- return base64.b64decode(s + "=" * (-len(s) % 4))
-
- ciphertext = _b64decode(ciphertext_b64)
- mac = _b64decode(mac_b64)
- ephemeral_pub = _b64decode(ephemeral_b64)
-
- # X25519 ECDH
- priv = X25519PrivateKey.from_private_bytes(private_key_bytes)
- shared_secret = priv.exchange(X25519PublicKey.from_public_bytes(ephemeral_pub))
-
- # HKDF key derivation
- derived = HKDF(
- algorithm=hashes.SHA256(),
- length=80,
- salt=b"\x00" * 32,
- info=b"OLM_KEYS",
- backend=default_backend(),
- ).derive(shared_secret)
- aes_key = derived[:32]
- mac_key = derived[32:64]
- iv = derived[64:80]
-
- # MAC verification
- expected_mac = hmac_lib.new(mac_key, ciphertext, hashlib.sha256).digest()
- if not hmac_lib.compare_digest(expected_mac, mac):
- raise ValueError("BAD_MESSAGE_MAC")
-
- # AES-256-CBC decryption + PKCS7 unpad
- cipher = Cipher(algorithms.AES(aes_key), modes.CBC(iv), backend=default_backend())
- dec = cipher.decryptor()
- padded = dec.update(ciphertext) + dec.finalize()
- unpadder = sym_padding.PKCS7(128).unpadder()
- plaintext = unpadder.update(padded) + unpadder.finalize()
- return plaintext.decode()
-
-
- async def import_key_backup(client: AsyncClient, backup_private_key_b64: str):
- """
- Fetch all sessions from the server-side key backup, decrypt them with the
- backup private key, and import them into the local crypto store.
- Called on every run so newly backed-up sessions are always available.
- """
- import urllib.request
-
- # Resolve backup version and expected public key
- url = f"{HOMESERVER}/_matrix/client/v3/room_keys/version"
- req = urllib.request.Request(
- url, headers={"Authorization": f"Bearer {client.access_token}"}
- )
- with urllib.request.urlopen(req, timeout=15) as r:
- backup_info = json.loads(r.read())
- backup_version = backup_info["version"]
- print(f" Key backup version: {backup_version}")
-
- # Fetch all backed-up sessions
- url = f"{HOMESERVER}/_matrix/client/v3/room_keys/keys?version={backup_version}"
- req = urllib.request.Request(
- url, headers={"Authorization": f"Bearer {client.access_token}"}
- )
- with urllib.request.urlopen(req, timeout=60) as r:
- rooms = json.loads(r.read()).get("rooms", {})
- print(f" Fetched key backup: {len(rooms)} rooms")
-
- # Decode the private key (Matrix stores as unpadded base64)
- private_key_bytes = base64.b64decode(backup_private_key_b64 + "=" * (-len(backup_private_key_b64) % 4))
-
- sessions_to_import = []
- for room_id, room_data in rooms.items():
- for session_id, session_info in room_data.get("sessions", {}).items():
- sd = session_info.get("session_data", {})
- try:
- plaintext = _olm_pk_decrypt(
- private_key_bytes,
- sd["ciphertext"],
- sd["mac"],
- sd["ephemeral"],
- )
- session_obj = json.loads(plaintext)
- sessions_to_import.append({
- "algorithm": "m.megolm.v1.aes-sha2",
- "room_id": room_id,
- "session_id": session_id,
- "session_key": session_obj["session_key"],
- "sender_key": session_obj.get("sender_key", ""),
- "sender_claimed_keys": session_obj.get("sender_claimed_keys", {}),
- "forwarding_curve25519_key_chain":
- session_obj.get("forwarding_curve25519_key_chain", []),
- })
- except Exception as e:
- print(f" Warning: failed to decrypt session {session_id} in {room_id}: {e}")
-
- print(f" Decrypted {len(sessions_to_import)} sessions; importing...")
- if sessions_to_import:
- tmp = tempfile.mktemp(suffix=".megolm")
- try:
- with open(tmp, "wb") as f:
- f.write(build_megolm_export(sessions_to_import))
- await client.import_keys(tmp, "_backup_import_")
- finally:
- if os.path.exists(tmp):
- os.unlink(tmp)
- print(f" Imported {len(sessions_to_import)} sessions from key backup.")
-
-
- async def bootstrap_crypto_from_ssss(client: AsyncClient, user_id: str, recovery_key_str: str):
- """
- First-run only: fetch SSSS account data directly from the server (read-only GETs),
- decrypt the backup key, and return it for storage in session.json.
- """
- print(" Bootstrapping crypto from SSSS...")
- backup_private_key_b64 = await _decrypt_ssss_backup_key(client, user_id, recovery_key_str)
- if not backup_private_key_b64:
- print(" Warning: could not decrypt SSSS backup key — skipping")
- return None
- print(" Decrypted megolm backup private key from SSSS.")
- return backup_private_key_b64
-
-
- # Room helpers
- def safe_room_key(room_id: str) -> str:
- return room_id.replace("/", "_").replace(":", "_")
-
-
- def derive_server_hints(room) -> list:
- counter = Counter()
- # matrix-nio exposes room members as room.users (dict of user_id -> RoomMember)
- for user_id in room.users:
- if ":" in user_id:
- counter[user_id.split(":", 1)[1]] += 1
- return [s for s, _ in counter.most_common(5)]
-
-
- def get_aliases(room) -> list:
- aliases = []
- if room.canonical_alias:
- aliases.append(room.canonical_alias)
- if hasattr(room, "alt_aliases"):
- for a in room.alt_aliases:
- if a not in aliases:
- aliases.append(a)
- if hasattr(room, "aliases") and isinstance(room.aliases, dict):
- for server_aliases in room.aliases.values():
- for a in server_aliases:
- if a not in aliases:
- aliases.append(a)
- return aliases
-
-
- async def get_room_type(client, room_id: str, room, dm_rooms: set) -> str:
- if room_id in dm_rooms:
- return "dm"
- try:
- resp = await client.room_get_state_event(room_id, "m.room.create", "")
- if hasattr(resp, "content") and resp.content.get("type") == "m.space":
- return "space"
- except Exception:
- pass
- return "normal"
-
-
- # Media helpers
- _MXC_RE = re.compile(r"^mxc://([^/]+)/(.+)$")
-
-
- def parse_mxc(mxc_url: str):
- m = _MXC_RE.match(mxc_url or "")
- return (m.group(1), m.group(2)) if m else (None, None)
-
-
- def ext_from_content_type(ct: str) -> str:
- ext = mimetypes.guess_extension(ct.split(";")[0].strip()) or ""
- return {".jpe": ".jpg", ".jpeg": ".jpg"}.get(ext, ext)
-
-
- async def store_media(client, mxc_url: str, prefix: str, label: str = "media"):
- server, media_id = parse_mxc(mxc_url)
- if not server or not media_id:
- return
- base_key = f"{prefix}/{label}/{server}/{media_id}"
- if s3_exists(base_key):
- return
- resp = await client.download(server_name=server, media_id=media_id)
- if isinstance(resp, DownloadError):
- print(f" Warning: failed to download {mxc_url}")
- return
- ext = ext_from_content_type(resp.content_type or "")
- key = base_key + ext if ext else base_key
- s3_put(key, resp.body, resp.content_type or "application/octet-stream")
- print(f" Stored {key} ({len(resp.body)} bytes)")
-
-
- # Profile change tracking
- async def record_profile_update(client, src: dict, prefix: str):
- state_key = src.get("state_key", "")
- if not state_key.endswith(OUR_HOMESERVER_SUFFIX):
- return
- content = src.get("content", {})
- prev_content = src.get("unsigned", {}).get("prev_content", {})
- new_name = content.get("displayname")
- prev_name = prev_content.get("displayname")
- new_avatar = content.get("avatar_url")
- prev_avatar = prev_content.get("avatar_url")
- if new_name == prev_name and new_avatar == prev_avatar:
- return
- record = {
- "ts": src.get("origin_server_ts"),
- "event_id": src.get("event_id"),
- "user_id": state_key,
- "room_id": src.get("room_id"),
- }
- if new_name != prev_name:
- record["displayname_old"] = prev_name
- record["displayname_new"] = new_name
- if new_avatar != prev_avatar:
- record["avatar_old"] = prev_avatar
- record["avatar_new"] = new_avatar
- if new_avatar:
- await store_media(client, new_avatar, prefix, label="avatars")
- log_key = f"{prefix}/profile-updates.jsonl"
- existing = s3_get_bytes(log_key) or b""
- s3_put(log_key, existing + json.dumps(record).encode() + b"\n", "application/x-ndjson")
-
-
- # Media message types to download
- MEDIA_MSGTYPES = {"m.image", "m.file", "m.video", "m.audio"}
-
-
- async def _process_event(client, src: dict, room_id: str, prefix: str, is_dm: bool):
- """Handle media download and profile-update side-effects for one event."""
- content = src.get("content", {})
- event_type = src.get("type", "")
- sender = str(src.get("sender", ""))
-
- # Media: all types in DMs; only our-user files in other rooms
- if event_type == "m.room.message" and content.get("msgtype") in MEDIA_MSGTYPES:
- if is_dm or sender.endswith(OUR_HOMESERVER_SUFFIX):
- mxc = content.get("url", "")
- if mxc:
- await store_media(client, mxc, prefix, label="media")
-
- # Stickers (m.sticker) always have a url directly in content
- if event_type == "m.sticker":
- if is_dm or sender.endswith(OUR_HOMESERVER_SUFFIX):
- mxc = content.get("url", "")
- if mxc:
- await store_media(client, mxc, prefix, label="media")
-
- # Profile changes (display name / avatar in m.room.member state events)
- if event_type == "m.room.member":
- src["room_id"] = room_id
- await record_profile_update(client, src, prefix)
-
-
- def _event_to_record(src: dict) -> dict:
- return {
- "event_id": src.get("event_id"),
- "sender": src.get("sender"),
- "type": src.get("type"),
- "origin_server_ts": src.get("origin_server_ts"),
- "content": src.get("content", {}),
- }
-
-
- # History pagination
- async def paginate_history(client, room_id: str, prefix: str, is_dm: bool = False):
- room_key = safe_room_key(room_id)
- cursor_s3 = f"{prefix}/history-cursor/{room_key}.json"
-
- cursor_data = s3_get_json(cursor_s3)
-
- if not cursor_data:
- # First time we've seen this room.
- room = client.rooms.get(room_id)
- token = getattr(room, "prev_batch", None)
- if not token:
- return
-
- if is_dm:
- # Paginate backward through the full DM history before setting the cursor.
- print(f" DM first-run: fetching full history for {room_id} ...")
- messages = []
- back_token = token
- pages = 0
- while True:
- resp = await client.room_messages(
- room_id,
- start=back_token,
- limit=100,
- direction=MessageDirection.back,
- )
- if isinstance(resp, RoomMessagesError):
- print(f" Warning: room_messages error for {room_id}: {resp}")
- break
- if not resp.chunk:
- break
- for event in resp.chunk:
- src = event.source if hasattr(event, "source") else {}
- messages.append(_event_to_record(src))
- await _process_event(client, src, room_id, prefix, is_dm=True)
- new_end = getattr(resp, "end", None)
- if not new_end or new_end == back_token:
- break
- back_token = new_end
- pages += 1
-
- if messages:
- # Events arrived in reverse-chron order; reverse so oldest is first.
- messages.reverse()
- history_s3 = f"{prefix}/history/{room_key}/{DATE_STR}.jsonl"
- chunk = b"\n".join(json.dumps(m).encode() for m in messages) + b"\n"
- s3_put(history_s3, chunk, "application/x-ndjson")
- print(f" DM history: {len(messages)} events ({pages} pages) for {room_id}")
-
- # Record the forward cursor so subsequent runs pick up new messages.
- s3_put(cursor_s3, json.dumps({"token": token}).encode(), "application/json")
- print(f" History cursor initialised for {room_id}")
- return
-
- # Subsequent runs: paginate forward from the stored cursor.
- start_token = cursor_data["token"]
- messages = []
- next_token = start_token
- history_s3 = f"{prefix}/history/{room_key}/{DATE_STR}.jsonl"
-
- for _ in range(100):
- resp = await client.room_messages(
- room_id,
- start=next_token,
- limit=100,
- direction=MessageDirection.forward,
- )
- if isinstance(resp, RoomMessagesError):
- print(f" Warning: room_messages error for {room_id}: {resp}")
- break
- if not resp.chunk:
- break
- for event in resp.chunk:
- src = event.source if hasattr(event, "source") else {}
- messages.append(_event_to_record(src))
- await _process_event(client, src, room_id, prefix, is_dm)
- new_end = getattr(resp, "end", None)
- if not new_end or new_end == next_token:
- next_token = None
- break
- next_token = new_end
-
- if messages:
- existing = s3_get_bytes(history_s3) or b""
- chunk = b"\n".join(json.dumps(m).encode() for m in messages) + b"\n"
- s3_put(history_s3, existing + chunk, "application/x-ndjson")
- print(f" History: +{len(messages)} events for {room_id}")
-
- if next_token and next_token != start_token:
- s3_put(cursor_s3, json.dumps({"token": next_token}).encode(), "application/json")
-
-
- # Per-account backup
- async def backup_account(account: dict):
- prefix = account["prefix"]
- print(f"\n=== Backing up {account['user_id']} ===")
-
- # 1. Restore crypto store from S3 (must happen before opening the client)
- store_s3_key = f"{prefix}/crypto-store.tar.gz"
- download_crypto_store(account["store"], store_s3_key)
-
- # 2. Ensure we have a session (password-login on first run, persisted inside the store)
- session = await ensure_session(account)
-
- # 3. Open the client (only after store is ready and session is known)
- config = AsyncClientConfig(store_sync_tokens=True)
- client = AsyncClient(
- HOMESERVER,
- account["user_id"],
- store_path=account["store"],
- config=config,
- )
- client.restore_login(
- user_id=account["user_id"],
- device_id=session["device_id"],
- access_token=session["access_token"],
- )
-
- try:
- # Load sync token for incremental syncs
- token_file = os.path.join(account["store"], "sync_token.json")
- since = None
- if os.path.exists(token_file):
- try:
- with open(token_file) as f:
- since = json.load(f).get("next_batch")
- except Exception:
- pass
-
- print(f" Syncing (full_state=True, since={since!r})...")
- sync_resp = await client.sync(full_state=True, since=since, timeout=60000)
- with open(token_file, "w") as f:
- json.dump({"next_batch": sync_resp.next_batch}, f)
- print(f" Sync done. Rooms: {len(client.rooms)}")
-
- # Account data from sync response (nio doesn't accumulate it on the client object)
- account_data = _parse_account_data(sync_resp)
-
- # 4. SSSS bootstrap on first login (after sync so account_data is available).
- # Only reads from account_data and decrypts locally — no server writes.
- if not session.get("ssss_bootstrapped") or not session.get("backup_private_key"):
- try:
- backup_private_key_b64 = await bootstrap_crypto_from_ssss(
- client, account["user_id"], account["ssss_key"]
- )
- if backup_private_key_b64:
- session["backup_private_key"] = backup_private_key_b64
- session["ssss_bootstrapped"] = True
- save_session(account["store"], session)
- except Exception as e:
- print(f" Warning: SSSS bootstrap failed: {e}", file=sys.stderr)
-
- # 5. Import key backup on every run so newly backed-up sessions are available.
- if session.get("backup_private_key"):
- try:
- await import_key_backup(client, session["backup_private_key"])
- except Exception as e:
- print(f" Warning: key backup import failed: {e}", file=sys.stderr)
- else:
- print(" Skipping key backup import (no backup_private_key in session yet)")
-
- # 6. DM room set — prefer list_direct_rooms() (nio PR #451) with fallback
- dm_rooms: set = set()
- try:
- dm_resp = await client.list_direct_rooms()
- if hasattr(dm_resp, "rooms") and isinstance(dm_resp.rooms, dict):
- for room_ids in dm_resp.rooms.values():
- if isinstance(room_ids, list):
- dm_rooms.update(room_ids)
- else:
- raise AttributeError("unexpected list_direct_rooms response")
- except Exception:
- # Fallback: parse m.direct from sync account data manually
- ad_event = account_data.get("m.direct")
- dm_map = getattr(ad_event, "content", {}) or {} if ad_event else {}
- for room_ids in dm_map.values():
- if isinstance(room_ids, list):
- dm_rooms.update(room_ids)
-
- # 7. Room list
- print(" Building room list...")
- rooms_data = []
- for room_id, room in client.rooms.items():
- room_type = await get_room_type(client, room_id, room, dm_rooms)
- aliases = get_aliases(room)
- via = derive_server_hints(room)
- rooms_data.append({
- "room_id": room_id,
- "name": room.display_name or room.name or room_id,
- "type": room_type,
- "aliases": aliases,
- "canonical_alias": room.canonical_alias,
- "via": via,
- "member_count": room.member_count,
- "encrypted": room.encrypted,
- })
- print(f" [{room_type:6s}] {room.display_name or room_id}")
- rooms_json = json.dumps(rooms_data, indent=2).encode()
- s3_put(f"{prefix}/rooms-{DATE_STR}.json", rooms_json, "application/json")
- s3_put(f"{prefix}/rooms-latest.json", rooms_json, "application/json")
- print(f" Uploaded rooms ({len(rooms_data)} rooms)")
-
- # 8. E2EE key export
- print(" Exporting E2EE keys...")
- try:
- key_file = tempfile.mktemp(suffix=".bin")
- await client.export_keys(key_file, KEY_EXPORT_PASSPHRASE)
- with open(key_file, "rb") as f:
- key_data = f.read()
- os.unlink(key_file)
- s3_put(f"{prefix}/crypto-keys-{DATE_STR}.bin", key_data)
- s3_put(f"{prefix}/crypto-keys-latest.bin", key_data)
- print(f" Uploaded crypto-keys-latest.bin ({len(key_data)} bytes)")
- except Exception as e:
- print(f" Warning: key export failed: {e}", file=sys.stderr)
-
- # 9. Incremental history + media + profile updates
- print(" Backing up history / media / profile updates...")
- for room_id in list(client.rooms.keys()):
- try:
- await paginate_history(client, room_id, prefix, is_dm=(room_id in dm_rooms))
- except Exception as e:
- print(f" Warning: history error for {room_id}: {e}", file=sys.stderr)
-
- finally:
- # Close the client BEFORE uploading the store so nothing is mid-write
- await client.close()
- upload_crypto_store(account["store"], store_s3_key)
-
- print(f" Done: {account['user_id']}")
-
-
- # Entry point
- async def main():
- for account in ACCOUNTS:
- await backup_account(account)
- print("\nAll accounts backed up successfully.")
-
-
- if __name__ == "__main__":
- asyncio.run(main())
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/Dockerfile b/apps/talos_cluster/matrix-backup/backup-tool/Dockerfile
@@ -0,0 +1,28 @@
+# Multi-stage build — produces a fully static binary (no CGO, no libolm).
+# The goolm build tag enables mautrix-go's pure-Go Olm/Megolm implementation.
+
+# Build stage runs on the native host arch; cross-compiles via GOARCH.
+FROM --platform=$BUILDPLATFORM golang:1.23-bookworm AS builder
+ARG TARGETARCH
+
+WORKDIR /app
+
+# Copy module files first for layer caching. go.sum is generated by go mod tidy
+# at build time since we cannot run it locally in the gitops pipeline.
+COPY go.mod ./
+COPY . .
+
+RUN go mod tidy
+RUN CGO_ENABLED=0 GOARCH=${TARGETARCH} \
+ go build \
+ -tags goolm \
+ -ldflags="-s -w" \
+ -o /matrix-backup \
+ .
+
+# Runtime stage — distroless/static has no shell, no libc, minimal attack surface.
+FROM gcr.io/distroless/static:nonroot
+
+COPY --from=builder /matrix-backup /matrix-backup
+
+ENTRYPOINT ["/matrix-backup"]
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/go.mod b/apps/talos_cluster/matrix-backup/backup-tool/go.mod
@@ -0,0 +1,12 @@
+module github.com/mtrnord/matrix-backup
+
+go 1.23
+
+require (
+ github.com/aws/aws-sdk-go-v2 v1.30.3
+ github.com/aws/aws-sdk-go-v2/config v1.27.26
+ github.com/aws/aws-sdk-go-v2/credentials v1.17.26
+ github.com/aws/aws-sdk-go-v2/service/s3 v1.61.0
+ golang.org/x/crypto v0.27.0
+ maunium.net/go/mautrix v0.26.4
+)
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/main.go b/apps/talos_cluster/matrix-backup/backup-tool/main.go
@@ -0,0 +1,1129 @@
+// Matrix backup tool — rewrites the Python matrix-nio backup in Go using mautrix-go.
+//
+// Key improvements over the Python version:
+// - SSSS decryption uses mautrix-go's proven implementation (fixes BAD_MESSAGE_MAC)
+// - Olm PK decryption uses mautrix-go with the goolm pure-Go backend (no libolm needed)
+// - No runtime pip install; compiled static binary via Dockerfile multi-stage build
+// - No SQLite/OlmMachine needed; session and sync token stored in S3
+// - Megolm export format implemented in pure Go (PBKDF2-SHA512 + AES-CTR + HMAC-SHA256)
+//
+// Build: CGO_ENABLED=0 go build -tags goolm -o matrix-backup .
+package main
+
+import (
+ "archive/tar"
+ "bytes"
+ "compress/gzip"
+ "context"
+ "crypto/aes"
+ "crypto/cipher"
+ "crypto/hmac"
+ "crypto/rand"
+ "crypto/sha256"
+ "crypto/sha512"
+ "encoding/base64"
+ "encoding/binary"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "net/http"
+ "net/url"
+ "os"
+ "strings"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ awsconfig "github.com/aws/aws-sdk-go-v2/config"
+ "github.com/aws/aws-sdk-go-v2/credentials"
+ "github.com/aws/aws-sdk-go-v2/service/s3"
+ s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
+ "golang.org/x/crypto/pbkdf2"
+ "maunium.net/go/mautrix"
+ "maunium.net/go/mautrix/crypto/backup"
+ "maunium.net/go/mautrix/crypto/ssss"
+ "maunium.net/go/mautrix/event"
+ "maunium.net/go/mautrix/id"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Config
+// ─────────────────────────────────────────────────────────────────────────────
+
+const ourServerName = "mtrnord.blog"
+
+type accountCfg struct {
+ UserID id.UserID
+ Password string
+ SSSSKey string
+ Prefix string
+ StoreDir string
+}
+
+var (
+ homeserver = mustEnv("HOMESERVER")
+ s3Endpoint = mustEnv("S3_ENDPOINT")
+ s3BucketName = mustEnv("S3_BUCKET")
+ s3AccessKey = mustEnv("S3_ACCESS_KEY")
+ s3SecretKey = mustEnv("S3_SECRET_KEY")
+ keyExportPass = mustEnv("KEY_EXPORT_PASSPHRASE")
+ dateStr = time.Now().UTC().Format("20060102")
+
+ accounts = []accountCfg{
+ {
+ UserID: "@mtrnord:mtrnord.blog",
+ Password: os.Getenv("MTRNORD_PASSWORD"),
+ SSSSKey: os.Getenv("MTRNORD_SSSS_KEY"),
+ Prefix: "mtrnord",
+ StoreDir: "/data/crypto/mtrnord",
+ },
+ {
+ UserID: "@lexi:mtrnord.blog",
+ Password: os.Getenv("LEXI_PASSWORD"),
+ SSSSKey: os.Getenv("LEXI_SSSS_KEY"),
+ Prefix: "lexi",
+ StoreDir: "/data/crypto/lexi",
+ },
+ }
+)
+
+func mustEnv(key string) string {
+ v := os.Getenv(key)
+ if v == "" {
+ slog.Error("Required environment variable not set", "key", key)
+ os.Exit(1)
+ }
+ return v
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// S3 helpers
+// ─────────────────────────────────────────────────────────────────────────────
+
+var s3c *s3.Client
+
+func initS3(ctx context.Context) error {
+ cfg, err := awsconfig.LoadDefaultConfig(ctx,
+ awsconfig.WithRegion("hel1"),
+ awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(
+ s3AccessKey, s3SecretKey, "",
+ )),
+ awsconfig.WithBaseEndpoint("https://"+s3Endpoint),
+ )
+ if err != nil {
+ return err
+ }
+ s3c = s3.NewFromConfig(cfg, func(o *s3.Options) {
+ o.UsePathStyle = true
+ })
+ return nil
+}
+
+func s3Get(ctx context.Context, key string) ([]byte, error) {
+ out, err := s3c.GetObject(ctx, &s3.GetObjectInput{
+ Bucket: aws.String(s3BucketName),
+ Key: aws.String(key),
+ })
+ if err != nil {
+ var nsk *s3types.NoSuchKey
+ if errors.As(err, &nsk) {
+ return nil, nil
+ }
+ return nil, err
+ }
+ defer out.Body.Close()
+ return io.ReadAll(out.Body)
+}
+
+func s3GetJSON(ctx context.Context, key string, out interface{}) error {
+ data, err := s3Get(ctx, key)
+ if err != nil || data == nil {
+ return err
+ }
+ return json.Unmarshal(data, out)
+}
+
+func s3Put(ctx context.Context, key string, data []byte, contentType string) error {
+ _, err := s3c.PutObject(ctx, &s3.PutObjectInput{
+ Bucket: aws.String(s3BucketName),
+ Key: aws.String(key),
+ Body: bytes.NewReader(data),
+ ContentType: aws.String(contentType),
+ })
+ return err
+}
+
+func s3Exists(ctx context.Context, key string) bool {
+ _, err := s3c.HeadObject(ctx, &s3.HeadObjectInput{
+ Bucket: aws.String(s3BucketName),
+ Key: aws.String(key),
+ })
+ return err == nil
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Crypto-store tarball (persists sessions and sync tokens across CronJob runs)
+// ─────────────────────────────────────────────────────────────────────────────
+
+func downloadStore(ctx context.Context, storeDir, s3Key string) error {
+ data, err := s3Get(ctx, s3Key)
+ if err != nil {
+ return err
+ }
+ if data == nil {
+ slog.Info("No existing store in S3, starting fresh", "key", s3Key)
+ return nil
+ }
+ if err := os.MkdirAll(storeDir, 0700); err != nil {
+ return err
+ }
+ gr, err := gzip.NewReader(bytes.NewReader(data))
+ if err != nil {
+ return err
+ }
+ defer gr.Close()
+ tr := tar.NewReader(gr)
+ for {
+ hdr, err := tr.Next()
+ if errors.Is(err, io.EOF) {
+ break
+ }
+ if err != nil {
+ return err
+ }
+ target := storeDir + "/" + hdr.Name
+ switch hdr.Typeflag {
+ case tar.TypeDir:
+ os.MkdirAll(target, 0700) //nolint:errcheck
+ case tar.TypeReg:
+ f, err := os.OpenFile(target, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0600)
+ if err != nil {
+ return err
+ }
+ if _, err := io.Copy(f, tr); err != nil {
+ f.Close()
+ return err
+ }
+ f.Close()
+ }
+ }
+ slog.Info("Store restored from S3", "key", s3Key, "bytes", len(data))
+ return nil
+}
+
+func uploadStore(ctx context.Context, storeDir, s3Key string) error {
+ var buf bytes.Buffer
+ gw := gzip.NewWriter(&buf)
+ tw := tar.NewWriter(gw)
+ if err := addDirToTar(tw, storeDir, "."); err != nil {
+ return err
+ }
+ tw.Close() //nolint:errcheck
+ gw.Close() //nolint:errcheck
+ data := buf.Bytes()
+ if err := s3Put(ctx, s3Key, data, "application/gzip"); err != nil {
+ return err
+ }
+ slog.Info("Store saved to S3", "key", s3Key, "bytes", len(data))
+ return nil
+}
+
+func addDirToTar(tw *tar.Writer, baseDir, arcBase string) error {
+ entries, err := os.ReadDir(baseDir)
+ if err != nil {
+ if os.IsNotExist(err) {
+ return nil
+ }
+ return err
+ }
+ for _, e := range entries {
+ srcPath := baseDir + "/" + e.Name()
+ arcPath := arcBase + "/" + e.Name()
+ if e.IsDir() {
+ _ = tw.WriteHeader(&tar.Header{
+ Typeflag: tar.TypeDir,
+ Name: arcPath + "/",
+ Mode: 0700,
+ })
+ if err := addDirToTar(tw, srcPath, arcPath); err != nil {
+ return err
+ }
+ continue
+ }
+ info, err := e.Info()
+ if err != nil {
+ continue
+ }
+ f, err := os.Open(srcPath)
+ if err != nil {
+ continue
+ }
+ if err := tw.WriteHeader(&tar.Header{
+ Typeflag: tar.TypeReg,
+ Name: arcPath,
+ Size: info.Size(),
+ Mode: 0600,
+ }); err != nil {
+ f.Close()
+ return err
+ }
+ _, copyErr := io.Copy(tw, f)
+ f.Close()
+ if copyErr != nil {
+ return copyErr
+ }
+ }
+ return nil
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Session management
+// ─────────────────────────────────────────────────────────────────────────────
+
+type sessionData struct {
+ AccessToken string `json:"access_token"`
+ DeviceID id.DeviceID `json:"device_id"`
+}
+
+func ensureSession(ctx context.Context, client *mautrix.Client, acc accountCfg, s3Key string) (*sessionData, error) {
+ var sess sessionData
+ if err := s3GetJSON(ctx, s3Key, &sess); err != nil {
+ return nil, fmt.Errorf("load session: %w", err)
+ }
+ if sess.AccessToken != "" {
+ slog.Info("Restored existing session", "device_id", sess.DeviceID)
+ client.UserID = acc.UserID
+ client.DeviceID = sess.DeviceID
+ client.AccessToken = sess.AccessToken
+ return &sess, nil
+ }
+
+ slog.Info("First run — logging in with password", "user_id", acc.UserID)
+ resp, err := client.Login(ctx, &mautrix.ReqLogin{
+ Type: mautrix.AuthTypePassword,
+ Identifier: mautrix.UserIdentifier{
+ Type: mautrix.IdentifierTypeUser,
+ User: string(acc.UserID),
+ },
+ Password: acc.Password,
+ DeviceName: "matrix-backup",
+ })
+ if err != nil {
+ return nil, fmt.Errorf("login: %w", err)
+ }
+ sess = sessionData{
+ AccessToken: resp.AccessToken,
+ DeviceID: resp.DeviceID,
+ }
+ client.UserID = resp.UserID
+ client.DeviceID = resp.DeviceID
+ client.AccessToken = resp.AccessToken
+ slog.Info("Logged in", "device_id", resp.DeviceID)
+
+ data, err := json.Marshal(sess)
+ if err != nil {
+ return nil, err
+ }
+ if err := s3Put(ctx, s3Key, data, "application/json"); err != nil {
+ return nil, fmt.Errorf("save session: %w", err)
+ }
+ return &sess, nil
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Matrix API helpers
+// ─────────────────────────────────────────────────────────────────────────────
+
+func matrixGetJSON(ctx context.Context, client *mautrix.Client, path string, out interface{}) error {
+ base := strings.TrimRight(client.HomeserverURL.String(), "/")
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+path, nil)
+ if err != nil {
+ return err
+ }
+ req.Header.Set("Authorization", "Bearer "+client.AccessToken)
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+ body, err := io.ReadAll(resp.Body)
+ if err != nil {
+ return err
+ }
+ if resp.StatusCode != http.StatusOK {
+ return fmt.Errorf("HTTP %d: %s", resp.StatusCode, body)
+ }
+ return json.Unmarshal(body, out)
+}
+
+func getAccountData(ctx context.Context, client *mautrix.Client, eventType string, out interface{}) error {
+ path := "/_matrix/client/v3/user/" +
+ url.PathEscape(client.UserID.String()) +
+ "/account_data/" +
+ url.PathEscape(eventType)
+ return matrixGetJSON(ctx, client, path, out)
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// SSSS + Megolm key backup
+// ─────────────────────────────────────────────────────────────────────────────
+
+type defaultKeyEventContent struct {
+ Key string `json:"key"`
+}
+
+// encryptedSecretContent matches the `{"encrypted": {"<keyID>": {...}}}` format.
+type encryptedSecretContent struct {
+ Encrypted map[string]ssss.EncryptedKeyData `json:"encrypted"`
+}
+
+type keyBackupVersionResp struct {
+ Version string `json:"version"`
+ Algorithm string `json:"algorithm"`
+}
+
+type keyBackupRoomSession struct {
+ FirstMessageIndex int `json:"first_message_index"`
+ ForwardedCount int `json:"forwarded_count"`
+ IsVerified bool `json:"is_verified"`
+ SessionData json.RawMessage `json:"session_data"`
+}
+
+type keyBackupRoom struct {
+ Sessions map[string]keyBackupRoomSession `json:"sessions"`
+}
+
+type keyBackupAllRooms struct {
+ Rooms map[id.RoomID]keyBackupRoom `json:"rooms"`
+}
+
+// exportedSessionEntry matches the standard Megolm key export session format
+// (https://spec.matrix.org/v1.9/client-server-api/#key-exports).
+type exportedSessionEntry struct {
+ Algorithm string `json:"algorithm"`
+ ForwardingCurve25519KeyChain []string `json:"forwarding_curve25519_key_chain"`
+ RoomID string `json:"room_id"`
+ SenderClaimedKeys map[string]string `json:"sender_claimed_keys"`
+ SenderKey string `json:"sender_key"`
+ SessionID string `json:"session_id"`
+ SessionKey string `json:"session_key"`
+}
+
+func fetchAndExportKeyBackup(ctx context.Context, client *mautrix.Client, recoveryKeyStr, prefix string) error {
+ // 1. Resolve the default SSSS key ID from account data.
+ var defaultKey defaultKeyEventContent
+ if err := getAccountData(ctx, client, "m.secret_storage.default_key", &defaultKey); err != nil {
+ return fmt.Errorf("get default key: %w", err)
+ }
+ if defaultKey.Key == "" {
+ return fmt.Errorf("m.secret_storage.default_key has no 'key' field")
+ }
+ keyID := defaultKey.Key
+ slog.Info("SSSS default key ID", "key_id", keyID)
+
+ // 2. Fetch the key metadata for the default key.
+ // The event type is m.secret_storage.key.<keyID>.
+ var keyMetadata ssss.KeyMetadata
+ if err := getAccountData(ctx, client, "m.secret_storage.key."+keyID, &keyMetadata); err != nil {
+ return fmt.Errorf("get key metadata: %w", err)
+ }
+
+ // 3. Verify the recovery key against the stored metadata and derive the raw key.
+ // VerifyRecoveryKey may return ErrUnverifiableKey alongside a valid key when
+ // no MAC/IV is stored in the metadata — treat that as success.
+ sssKey, err := keyMetadata.VerifyRecoveryKey(keyID, recoveryKeyStr)
+ if err != nil && !errors.Is(err, ssss.ErrUnverifiableKey) {
+ return fmt.Errorf("verify recovery key: %w", err)
+ }
+ slog.Info("Recovery key verified (or unverifiable but accepted)")
+
+ // 4. Fetch the encrypted backup secret and decrypt with the SSSS key.
+ var backupSecret encryptedSecretContent
+ if err := getAccountData(ctx, client, "m.megolm_backup.v1", &backupSecret); err != nil {
+ return fmt.Errorf("get backup secret: %w", err)
+ }
+ encData, ok := backupSecret.Encrypted[keyID]
+ if !ok {
+ return fmt.Errorf("backup secret not encrypted with key %q", keyID)
+ }
+ backupKeyRaw, err := sssKey.Decrypt("m.megolm_backup.v1", encData)
+ if err != nil {
+ return fmt.Errorf("decrypt backup key from SSSS: %w", err)
+ }
+ // backupKeyRaw is the base64-encoded X25519 private key bytes.
+ backupKeyB64 := strings.TrimSpace(string(backupKeyRaw))
+
+ // 5. Decode the private key bytes and create the MegolmBackupKey.
+ privateKeyBytes, err := base64DecodeUnpadded(backupKeyB64)
+ if err != nil {
+ return fmt.Errorf("decode backup private key: %w", err)
+ }
+ megolmKey, err := backup.MegolmBackupKeyFromBytes(privateKeyBytes)
+ if err != nil {
+ return fmt.Errorf("create megolm backup key: %w", err)
+ }
+ slog.Info("Decoded megolm backup private key from SSSS")
+
+ // 6. Get the current backup version.
+ var backupVersion keyBackupVersionResp
+ if err := matrixGetJSON(ctx, client, "/_matrix/client/v3/room_keys/version", &backupVersion); err != nil {
+ return fmt.Errorf("get backup version: %w", err)
+ }
+ slog.Info("Key backup version", "version", backupVersion.Version)
+
+ // 7. Fetch all backed-up sessions.
+ var allRooms keyBackupAllRooms
+ if err := matrixGetJSON(ctx, client,
+ "/_matrix/client/v3/room_keys/keys?version="+url.QueryEscape(backupVersion.Version),
+ &allRooms,
+ ); err != nil {
+ return fmt.Errorf("get backup keys: %w", err)
+ }
+ slog.Info("Fetched key backup", "rooms", len(allRooms.Rooms))
+
+ // 8. Decrypt each session using mautrix-go's Olm PK (via goolm).
+ // session_data JSON → EncryptedSessionData[MegolmSessionData] → Decrypt → MegolmSessionData
+ var sessions []exportedSessionEntry
+ decOK, decFail := 0, 0
+ for roomID, room := range allRooms.Rooms {
+ for sessionID, sessionInfo := range room.Sessions {
+ var encSD backup.EncryptedSessionData[backup.MegolmSessionData]
+ if err := json.Unmarshal(sessionInfo.SessionData, &encSD); err != nil {
+ slog.Warn("Failed to parse session_data",
+ "room_id", roomID, "session_id", sessionID, "error", err)
+ decFail++
+ continue
+ }
+ sessionData, err := encSD.Decrypt(megolmKey)
+ if err != nil {
+ slog.Warn("Failed to decrypt session",
+ "room_id", roomID, "session_id", sessionID, "error", err)
+ decFail++
+ continue
+ }
+ sessions = append(sessions, exportedSessionEntry{
+ Algorithm: string(sessionData.Algorithm),
+ ForwardingCurve25519KeyChain: sessionData.ForwardingKeyChain,
+ RoomID: string(roomID),
+ SenderClaimedKeys: map[string]string{"ed25519": string(sessionData.SenderClaimedKeys.Ed25519)},
+ SenderKey: string(sessionData.SenderKey),
+ SessionID: sessionID,
+ SessionKey: sessionData.SessionKey,
+ })
+ decOK++
+ }
+ }
+ slog.Info("Session decryption complete", "ok", decOK, "failed", decFail)
+
+ if len(sessions) == 0 {
+ slog.Warn("No sessions decrypted — skipping key export")
+ return nil
+ }
+
+ // 9. Archive raw session JSON for debugging / future re-import.
+ if rawJSON, err := json.MarshalIndent(sessions, "", " "); err == nil {
+ _ = s3Put(ctx, prefix+"/backup-sessions-latest.json", rawJSON, "application/json")
+ _ = s3Put(ctx, prefix+"/backup-sessions-"+dateStr+".json", rawJSON, "application/json")
+ }
+
+ // 10. Build the standard Megolm key export file and upload.
+ exportData, err := buildMegolmExport(sessions, keyExportPass)
+ if err != nil {
+ return fmt.Errorf("build megolm export: %w", err)
+ }
+ if err := s3Put(ctx, prefix+"/crypto-keys-latest.bin", exportData, "application/octet-stream"); err != nil {
+ return err
+ }
+ if err := s3Put(ctx, prefix+"/crypto-keys-"+dateStr+".bin", exportData, "application/octet-stream"); err != nil {
+ return err
+ }
+ slog.Info("Exported E2EE keys", "sessions", len(sessions), "bytes", len(exportData))
+ return nil
+}
+
+func base64DecodeUnpadded(s string) ([]byte, error) {
+ switch len(s) % 4 {
+ case 2:
+ s += "=="
+ case 3:
+ s += "="
+ }
+ return base64.StdEncoding.DecodeString(s)
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Standard Megolm key export format (pure Go)
+// Spec: https://spec.matrix.org/v1.9/client-server-api/#key-exports
+// ─────────────────────────────────────────────────────────────────────────────
+
+func buildMegolmExport(sessions []exportedSessionEntry, passphrase string) ([]byte, error) {
+ salt := make([]byte, 16)
+ if _, err := rand.Read(salt); err != nil {
+ return nil, err
+ }
+ ivBytes := make([]byte, 16)
+ if _, err := rand.Read(ivBytes); err != nil {
+ return nil, err
+ }
+ ivBytes[0] &= 0x7F // spec: highest bit of IV must be 0
+
+ const iterations = 100_000
+ keyMaterial := pbkdf2.Key([]byte(passphrase), salt, iterations, 64, sha512.New)
+ aesKey := keyMaterial[:32]
+ hmacKey := keyMaterial[32:]
+
+ plaintext, err := json.Marshal(sessions)
+ if err != nil {
+ return nil, err
+ }
+
+ block, err := aes.NewCipher(aesKey)
+ if err != nil {
+ return nil, err
+ }
+ ciphertext := make([]byte, len(plaintext))
+ cipher.NewCTR(block, ivBytes).XORKeyStream(ciphertext, plaintext)
+
+ iterBuf := make([]byte, 4)
+ binary.BigEndian.PutUint32(iterBuf, uint32(iterations))
+
+ payload := []byte{0x01}
+ payload = append(payload, salt...)
+ payload = append(payload, ivBytes...)
+ payload = append(payload, iterBuf...)
+ payload = append(payload, ciphertext...)
+
+ mac := hmac.New(sha256.New, hmacKey)
+ mac.Write(payload)
+ payload = append(payload, mac.Sum(nil)...)
+
+ encoded := base64.StdEncoding.EncodeToString(payload)
+ var sb strings.Builder
+ sb.WriteString("-----BEGIN MEGOLM SESSION DATA-----\n")
+ for i := 0; i < len(encoded); i += 76 {
+ end := i + 76
+ if end > len(encoded) {
+ end = len(encoded)
+ }
+ sb.WriteString(encoded[i:end])
+ sb.WriteByte('\n')
+ }
+ sb.WriteString("-----END MEGOLM SESSION DATA-----\n")
+ return []byte(sb.String()), nil
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Room list
+// ─────────────────────────────────────────────────────────────────────────────
+
+type roomEntry struct {
+ RoomID string `json:"room_id"`
+ Name string `json:"name"`
+ Type string `json:"type"`
+ Aliases []string `json:"aliases"`
+ CanonicalAlias string `json:"canonical_alias,omitempty"`
+ MemberCount int `json:"member_count"`
+ Encrypted bool `json:"encrypted"`
+}
+
+func saveRoomList(ctx context.Context, client *mautrix.Client, syncResp *mautrix.RespSync, prefix string) error {
+ dmRooms := getDMRooms(syncResp)
+
+ var rooms []roomEntry
+ for roomID, joinedRoom := range syncResp.Rooms.Join {
+ var name, canonicalAlias string
+ var aliases []string
+ encrypted := false
+ rtype := "normal"
+ if dmRooms[roomID] {
+ rtype = "dm"
+ }
+ for _, ev := range joinedRoom.State.Events {
+ switch ev.Type {
+ case event.StateRoomName:
+ if c := ev.Content.AsName(); c != nil {
+ name = c.Name
+ }
+ case event.StateCanonicalAlias:
+ if c := ev.Content.AsCanonicalAlias(); c != nil {
+ canonicalAlias = string(c.Alias)
+ for _, a := range c.AltAliases {
+ aliases = append(aliases, string(a))
+ }
+ }
+ case event.StateEncryption:
+ encrypted = true
+ case event.StateCreate:
+ if ev.Content.VeryRaw != nil {
+ var createContent struct {
+ Type string `json:"type"`
+ }
+ if json.Unmarshal(ev.Content.VeryRaw, &createContent) == nil && createContent.Type == "m.space" {
+ rtype = "space"
+ }
+ }
+ }
+ }
+ if name == "" {
+ name = string(roomID)
+ }
+ rooms = append(rooms, roomEntry{
+ RoomID: string(roomID),
+ Name: name,
+ Type: rtype,
+ Aliases: aliases,
+ CanonicalAlias: canonicalAlias,
+ MemberCount: joinedRoom.Summary.JoinedMemberCount,
+ Encrypted: encrypted,
+ })
+ slog.Info("Room", "type", rtype, "name", name)
+ }
+
+ data, err := json.MarshalIndent(rooms, "", " ")
+ if err != nil {
+ return err
+ }
+ if err := s3Put(ctx, prefix+"/rooms-"+dateStr+".json", data, "application/json"); err != nil {
+ return err
+ }
+ if err := s3Put(ctx, prefix+"/rooms-latest.json", data, "application/json"); err != nil {
+ return err
+ }
+ slog.Info("Uploaded room list", "rooms", len(rooms))
+ return nil
+}
+
+func getDMRooms(syncResp *mautrix.RespSync) map[id.RoomID]bool {
+ dmRooms := make(map[id.RoomID]bool)
+ for _, ev := range syncResp.AccountData.Events {
+ if ev.Type == event.AccountDataDirectChats {
+ // DirectChatsEventContent is map[id.UserID][]id.RoomID
+ var direct event.DirectChatsEventContent
+ if err := json.Unmarshal(ev.Content.VeryRaw, &direct); err != nil {
+ continue
+ }
+ for _, roomIDs := range direct {
+ for _, rid := range roomIDs {
+ dmRooms[rid] = true
+ }
+ }
+ break
+ }
+ }
+ return dmRooms
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Media download
+// ─────────────────────────────────────────────────────────────────────────────
+
+func downloadAndStoreMedia(ctx context.Context, client *mautrix.Client, mxcURL, prefix, label string) error {
+ if !strings.HasPrefix(mxcURL, "mxc://") {
+ return nil
+ }
+ rest := mxcURL[len("mxc://"):]
+ slashIdx := strings.Index(rest, "/")
+ if slashIdx < 0 {
+ return nil
+ }
+ server := rest[:slashIdx]
+ mediaID := rest[slashIdx+1:]
+ if server == "" || mediaID == "" {
+ return nil
+ }
+
+ s3Key := prefix + "/" + label + "/" + server + "/" + mediaID
+ if s3Exists(ctx, s3Key) {
+ return nil
+ }
+
+ base := strings.TrimRight(client.HomeserverURL.String(), "/")
+ downloadURL := base + "/_matrix/media/v3/download/" +
+ url.PathEscape(server) + "/" + url.PathEscape(mediaID)
+
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
+ if err != nil {
+ return err
+ }
+ req.Header.Set("Authorization", "Bearer "+client.AccessToken)
+
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+
+ if resp.StatusCode != http.StatusOK {
+ return fmt.Errorf("media download HTTP %d for %s", resp.StatusCode, mxcURL)
+ }
+
+ data, err := io.ReadAll(resp.Body)
+ if err != nil {
+ return err
+ }
+
+ ct := resp.Header.Get("Content-Type")
+ if ct == "" {
+ ct = "application/octet-stream"
+ }
+
+ finalKey := s3Key
+ if ext := extFromContentType(ct); ext != "" {
+ finalKey = s3Key + ext
+ }
+ if err := s3Put(ctx, finalKey, data, ct); err != nil {
+ return err
+ }
+ slog.Info("Stored media", "key", finalKey, "bytes", len(data))
+ return nil
+}
+
+func extFromContentType(ct string) string {
+ if i := strings.Index(ct, ";"); i >= 0 {
+ ct = ct[:i]
+ }
+ switch strings.TrimSpace(ct) {
+ case "image/jpeg":
+ return ".jpg"
+ case "image/png":
+ return ".png"
+ case "image/gif":
+ return ".gif"
+ case "image/webp":
+ return ".webp"
+ case "video/mp4":
+ return ".mp4"
+ case "video/webm":
+ return ".webm"
+ case "audio/mpeg":
+ return ".mp3"
+ case "audio/ogg":
+ return ".ogg"
+ case "audio/opus":
+ return ".opus"
+ case "application/pdf":
+ return ".pdf"
+ }
+ return ""
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Event processing
+// ─────────────────────────────────────────────────────────────────────────────
+
+var mediaMsgTypes = map[string]bool{
+ "m.image": true,
+ "m.file": true,
+ "m.video": true,
+ "m.audio": true,
+}
+
+// processEvent handles media downloads and profile-update tracking for one event.
+// Encrypted events are stored as-is but media/profiles are skipped (no OlmMachine).
+func processEvent(ctx context.Context, client *mautrix.Client, ev *event.Event, roomID id.RoomID, prefix string, isDM bool) {
+ if ev.Type == event.EventEncrypted {
+ return
+ }
+
+ if ev.Type == event.EventMessage {
+ var content struct {
+ MsgType string `json:"msgtype"`
+ URL string `json:"url"`
+ }
+ if err := json.Unmarshal(ev.Content.VeryRaw, &content); err == nil {
+ if mediaMsgTypes[content.MsgType] && content.URL != "" {
+ if isDM || strings.HasSuffix(string(ev.Sender), ":"+ourServerName) {
+ if err := downloadAndStoreMedia(ctx, client, content.URL, prefix, "media"); err != nil {
+ slog.Warn("Media download failed", "url", content.URL, "error", err)
+ }
+ }
+ }
+ }
+ }
+
+ if ev.Type == event.EventSticker {
+ var content struct{ URL string `json:"url"` }
+ if err := json.Unmarshal(ev.Content.VeryRaw, &content); err == nil && content.URL != "" {
+ if isDM || strings.HasSuffix(string(ev.Sender), ":"+ourServerName) {
+ if err := downloadAndStoreMedia(ctx, client, content.URL, prefix, "media"); err != nil {
+ slog.Warn("Sticker download failed", "url", content.URL, "error", err)
+ }
+ }
+ }
+ }
+
+ if ev.Type == event.StateMember {
+ recordProfileUpdate(ctx, ev, roomID, prefix)
+ }
+}
+
+type profileUpdateRecord struct {
+ TS int64 `json:"ts"`
+ EventID string `json:"event_id"`
+ UserID string `json:"user_id"`
+ RoomID string `json:"room_id"`
+ DisplayNameOld string `json:"displayname_old,omitempty"`
+ DisplayNameNew string `json:"displayname_new,omitempty"`
+ AvatarOld string `json:"avatar_old,omitempty"`
+ AvatarNew string `json:"avatar_new,omitempty"`
+}
+
+func recordProfileUpdate(ctx context.Context, ev *event.Event, roomID id.RoomID, prefix string) {
+ stateKey := string(ev.StateKey)
+ if !strings.HasSuffix(stateKey, ":"+ourServerName) {
+ return
+ }
+ var content, prevContent struct {
+ Displayname string `json:"displayname"`
+ AvatarURL string `json:"avatar_url"`
+ }
+ if err := json.Unmarshal(ev.Content.VeryRaw, &content); err != nil {
+ return
+ }
+ if ev.Unsigned.PrevContent != nil {
+ _ = json.Unmarshal(ev.Unsigned.PrevContent.VeryRaw, &prevContent)
+ }
+ if content.Displayname == prevContent.Displayname && content.AvatarURL == prevContent.AvatarURL {
+ return
+ }
+ rec := profileUpdateRecord{
+ TS: ev.Timestamp,
+ EventID: string(ev.ID),
+ UserID: stateKey,
+ RoomID: string(roomID),
+ }
+ if content.Displayname != prevContent.Displayname {
+ rec.DisplayNameOld = prevContent.Displayname
+ rec.DisplayNameNew = content.Displayname
+ }
+ if content.AvatarURL != prevContent.AvatarURL {
+ rec.AvatarOld = prevContent.AvatarURL
+ rec.AvatarNew = content.AvatarURL
+ if content.AvatarURL != "" {
+ if err := downloadAndStoreMedia(ctx, client, content.AvatarURL, prefix, "avatars"); err != nil {
+ slog.Warn("Avatar download failed", "url", content.AvatarURL, "error", err)
+ }
+ }
+ }
+ line, err := json.Marshal(rec)
+ if err != nil {
+ return
+ }
+ s3Key := prefix + "/profile-updates.jsonl"
+ existing, _ := s3Get(ctx, s3Key)
+ _ = s3Put(ctx, s3Key, append(existing, append(line, '\n')...), "application/x-ndjson")
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// History pagination
+// ─────────────────────────────────────────────────────────────────────────────
+
+type historyEvent struct {
+ EventID string `json:"event_id"`
+ Sender string `json:"sender"`
+ Type string `json:"type"`
+ Timestamp int64 `json:"origin_server_ts"`
+ Content json.RawMessage `json:"content"`
+}
+
+func eventToRecord(ev *event.Event) historyEvent {
+ return historyEvent{
+ EventID: string(ev.ID),
+ Sender: string(ev.Sender),
+ Type: ev.Type.Type,
+ Timestamp: ev.Timestamp,
+ Content: ev.Content.VeryRaw,
+ }
+}
+
+func paginateRoom(ctx context.Context, client *mautrix.Client, roomID id.RoomID, prefix string, isDM bool, prevBatch string) error {
+ safeKey := strings.NewReplacer("/", "_", ":", "_").Replace(string(roomID))
+ cursorKey := prefix + "/history-cursor/" + safeKey + ".json"
+
+ var cursor struct {
+ Token string `json:"token"`
+ }
+ if err := s3GetJSON(ctx, cursorKey, &cursor); err != nil {
+ return err
+ }
+
+ if cursor.Token == "" {
+ if prevBatch == "" {
+ return nil
+ }
+ if isDM {
+ slog.Info("DM first-run: fetching full history", "room_id", roomID)
+ var messages []historyEvent
+ token := prevBatch
+ for {
+ resp, err := client.Messages(ctx, roomID, token, "", mautrix.DirectionBackward, nil, 100)
+ if err != nil {
+ slog.Warn("room_messages error (backward)", "room_id", roomID, "error", err)
+ break
+ }
+ if len(resp.Chunk) == 0 {
+ break
+ }
+ for _, ev := range resp.Chunk {
+ messages = append(messages, eventToRecord(ev))
+ processEvent(ctx, client, ev, roomID, prefix, true)
+ }
+ if resp.End == "" || resp.End == token {
+ break
+ }
+ token = resp.End
+ }
+ if len(messages) > 0 {
+ for i, j := 0, len(messages)-1; i < j; i, j = i+1, j-1 {
+ messages[i], messages[j] = messages[j], messages[i]
+ }
+ histKey := prefix + "/history/" + safeKey + "/" + dateStr + ".jsonl"
+ var buf bytes.Buffer
+ for _, m := range messages {
+ line, _ := json.Marshal(m)
+ buf.Write(line)
+ buf.WriteByte('\n')
+ }
+ _ = s3Put(ctx, histKey, buf.Bytes(), "application/x-ndjson")
+ slog.Info("DM history stored", "room_id", roomID, "events", len(messages))
+ }
+ }
+ cursorJSON, _ := json.Marshal(map[string]string{"token": prevBatch})
+ _ = s3Put(ctx, cursorKey, cursorJSON, "application/json")
+ return nil
+ }
+
+ var messages []historyEvent
+ nextToken := cursor.Token
+ histKey := prefix + "/history/" + safeKey + "/" + dateStr + ".jsonl"
+
+ for range 100 {
+ resp, err := client.Messages(ctx, roomID, nextToken, "", mautrix.DirectionForward, nil, 100)
+ if err != nil {
+ slog.Warn("room_messages error (forward)", "room_id", roomID, "error", err)
+ break
+ }
+ if len(resp.Chunk) == 0 {
+ break
+ }
+ for _, ev := range resp.Chunk {
+ messages = append(messages, eventToRecord(ev))
+ processEvent(ctx, client, ev, roomID, prefix, isDM)
+ }
+ if resp.End == "" || resp.End == nextToken {
+ nextToken = ""
+ break
+ }
+ nextToken = resp.End
+ }
+
+ if len(messages) > 0 {
+ existing, _ := s3Get(ctx, histKey)
+ var buf bytes.Buffer
+ buf.Write(existing)
+ for _, m := range messages {
+ line, _ := json.Marshal(m)
+ buf.Write(line)
+ buf.WriteByte('\n')
+ }
+ _ = s3Put(ctx, histKey, buf.Bytes(), "application/x-ndjson")
+ slog.Info("History updated", "room_id", roomID, "new_events", len(messages))
+ }
+
+ if nextToken != "" && nextToken != cursor.Token {
+ cursorJSON, _ := json.Marshal(map[string]string{"token": nextToken})
+ _ = s3Put(ctx, cursorKey, cursorJSON, "application/json")
+ }
+ return nil
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Per-account backup
+// ─────────────────────────────────────────────────────────────────────────────
+
+func backupAccount(ctx context.Context, acc accountCfg) error {
+ slog.Info("=== Backing up account ===", "user_id", acc.UserID)
+
+ storeS3Key := acc.Prefix + "/crypto-store.tar.gz"
+ sessionS3Key := acc.Prefix + "/session.json"
+ syncTokenS3Key := acc.Prefix + "/sync-token.json"
+
+ if err := downloadStore(ctx, acc.StoreDir, storeS3Key); err != nil {
+ slog.Warn("Could not restore store", "error", err)
+ }
+
+ client, err := mautrix.NewClient(homeserver, "", "")
+ if err != nil {
+ return fmt.Errorf("create client: %w", err)
+ }
+
+ if _, err := ensureSession(ctx, client, acc, sessionS3Key); err != nil {
+ return fmt.Errorf("session: %w", err)
+ }
+
+ var syncToken struct {
+ NextBatch string `json:"next_batch"`
+ }
+ _ = s3GetJSON(ctx, syncTokenS3Key, &syncToken)
+
+ slog.Info("Syncing", "since", syncToken.NextBatch)
+ syncResp, err := client.SyncRequest(ctx, 60000, syncToken.NextBatch, "", true, event.PresenceUnavailable)
+ if err != nil {
+ return fmt.Errorf("sync: %w", err)
+ }
+ if data, err := json.Marshal(map[string]string{"next_batch": syncResp.NextBatch}); err == nil {
+ _ = s3Put(ctx, syncTokenS3Key, data, "application/json")
+ }
+ slog.Info("Sync done", "rooms", len(syncResp.Rooms.Join))
+
+ if acc.SSSSKey != "" {
+ if err := fetchAndExportKeyBackup(ctx, client, acc.SSSSKey, acc.Prefix); err != nil {
+ slog.Warn("Key backup export failed", "error", err)
+ }
+ }
+
+ if err := saveRoomList(ctx, client, syncResp, acc.Prefix); err != nil {
+ slog.Warn("Room list failed", "error", err)
+ }
+
+ dmRooms := getDMRooms(syncResp)
+ for roomID, joinedRoom := range syncResp.Rooms.Join {
+ isDM := dmRooms[roomID]
+ if err := paginateRoom(ctx, client, roomID, acc.Prefix, isDM, joinedRoom.Timeline.PrevBatch); err != nil {
+ slog.Warn("History error", "room_id", roomID, "error", err)
+ }
+ }
+
+ if err := uploadStore(ctx, acc.StoreDir, storeS3Key); err != nil {
+ slog.Warn("Could not upload store", "error", err)
+ }
+
+ slog.Info("Done", "user_id", acc.UserID)
+ return nil
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// main
+// ─────────────────────────────────────────────────────────────────────────────
+
+func main() {
+ slog.SetDefault(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
+ Level: slog.LevelInfo,
+ })))
+
+ ctx := context.Background()
+
+ if err := initS3(ctx); err != nil {
+ slog.Error("Failed to init S3", "error", err)
+ os.Exit(1)
+ }
+
+ for _, acc := range accounts {
+ if err := backupAccount(ctx, acc); err != nil {
+ slog.Error("Account backup failed", "user_id", acc.UserID, "error", err)
+ }
+ }
+
+ slog.Info("All accounts backed up successfully")
+}
diff --git a/apps/talos_cluster/matrix-backup/kustomization.yaml b/apps/talos_cluster/matrix-backup/kustomization.yaml
@@ -4,5 +4,4 @@ namespace: matrix-backup
resources:
- namespace.yaml
- secret.yaml
- - backup-script-configmap.yaml
- backup-cronjob.yaml
