Implement real Matrix notification for coe-mirror

Replace stub notify() with full Synapse integration targeting General room.
- Add _send_matrix(), _get_matrix_token(), _matrix_login() helpers
- Implement token cache with 30-min refresh threshold and 401 retry
- Fetch credentials via Infisical (MATRIX_USER, MATRIX_PASSWORD from matrix/)
- Add MATRIX_* env overrides for testing (homeserver, room, token cache path)
- Non-fatal failures: Matrix outage logs warning, does not break mirror run
- Add 5 comprehensive test cases covering success, failure, 401 refresh, env override, chmod 600
- All 18 tests passing
- Update README and notes.md with implementation details
This commit is contained in:
n0mad1k
2026-06-14 14:36:14 -04:00
parent 98cab71a21
commit c322488869
4 changed files with 417 additions and 9 deletions
+11 -1
View File
@@ -44,7 +44,17 @@ The timer is `Persistent=true` and `OnBootSec=2min`. Missed windows run automati
## Matrix notification
Currently a stub. Real implementation deferred until the home Matrix server is back. Reference: `~/tools/devtrack/scripts/matrix_alerts.py` shows the auth + send pattern (Infisical creds `MATRIX_USER_ID` and `MATRIX_ACCESS_TOKEN` from the `devtrack` folder, POST to `/_matrix/client/r0/rooms/{room_id}/send/m.room.message`).
After every run that lands at least one new zip, coe-mirror posts a summary
to the General room (`!MBpPtYlQYMcngtpvJt:m.mealeyfamily.com`) on
`m.mealeyfamily.com`.
- Authentication uses `MATRIX_USER` + `MATRIX_PASSWORD` from Infisical folder
`matrix`. Tokens are cached at `~/.local/share/coe-mirror/matrix-token.json`
(mode 0600) and auto-refreshed on expiry or 401.
- Failures are non-fatal: a Matrix outage logs a warning and the rest of the
run proceeds normally.
- Override the target room for testing: `COE_MIRROR_MATRIX_ROOM=! ...:host`.
- Override the homeserver: `COE_MIRROR_MATRIX_HOMESERVER=https://...`.
## Testing
+162 -8
View File
@@ -4,14 +4,17 @@ Fetches all repos owned by Nightmare_Eclipse and stores per-commit zip snapshots
Runs unattended via systemd timer twice daily. Local staging when pCloud unavailable.
"""
import json
import logging
import logging.handlers
import os
import re
import shutil
import subprocess
import sys
import time
import zipfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from fcntl import flock, LOCK_EX, LOCK_NB
from typing import Optional
@@ -34,11 +37,22 @@ RETRY_BACKOFF = [0, 5, 30] # seconds between attempts; len = max attempts
USER_AGENT = "coe-mirror/1.0"
REPO_NAME_RE = re.compile(r"^[a-zA-Z0-9._-]+$")
# Matrix configuration
MATRIX_HOMESERVER = "https://m.mealeyfamily.com"
MATRIX_ROOM = "!MBpPtYlQYMcngtpvJt:m.mealeyfamily.com" # General
MATRIX_TOKEN_CACHE = LOG_DIR / "matrix-token.json"
MATRIX_TIMEOUT = 8
MATRIX_REFRESH_THRESHOLD_SECONDS = 1800 # refresh if <30 min left
# Allow env var overrides for testing and deployment flexibility
if env_pcloud := os.getenv("COE_MIRROR_PCLOUD_TARGET"):
PCLOUD_TARGET = Path(env_pcloud)
if env_pending := os.getenv("COE_MIRROR_PENDING_DIR"):
PENDING_DIR = Path(env_pending)
if env_hs := os.getenv("COE_MIRROR_MATRIX_HOMESERVER"):
MATRIX_HOMESERVER = env_hs
if env_room := os.getenv("COE_MIRROR_MATRIX_ROOM"):
MATRIX_ROOM = env_room
logger: logging.Logger = None # Initialized in setup_logging()
@@ -296,20 +310,160 @@ def flush_pending(pcloud_healthy: bool) -> list:
return moved
def _read_token_cache() -> Optional[dict]:
"""Return cached token dict or None on any failure. Never raises."""
try:
if not MATRIX_TOKEN_CACHE.exists():
return None
with open(MATRIX_TOKEN_CACHE) as f:
data = json.load(f)
# Shape check
if "access_token" in data and "expires_at" in data:
return data
except Exception as exc:
logger.debug("token cache read failed: %s", exc)
return None
def _write_token_cache(access_token: str, expires_at_iso: str) -> None:
"""Write token cache with 0600 permissions. Never raises."""
try:
MATRIX_TOKEN_CACHE.parent.mkdir(parents=True, exist_ok=True)
payload = {"access_token": access_token, "expires_at": expires_at_iso}
# Write then chmod for atomicity-ish behavior
tmp = MATRIX_TOKEN_CACHE.with_suffix(".tmp")
with open(tmp, "w") as f:
json.dump(payload, f)
os.chmod(tmp, 0o600)
os.replace(tmp, MATRIX_TOKEN_CACHE)
except Exception as exc:
logger.warning("token cache write failed: %s", exc)
def _fetch_creds() -> Optional[tuple]:
"""Fetch (user, password) from Infisical via creds. Return None on failure."""
try:
user = subprocess.run(
["creds", "get", "MATRIX_USER", "matrix"],
capture_output=True, text=True, timeout=5
)
pw = subprocess.run(
["creds", "get", "MATRIX_PASSWORD", "matrix"],
capture_output=True, text=True, timeout=5
)
if user.returncode == 0 and pw.returncode == 0:
u, p = user.stdout.strip(), pw.stdout.strip()
if u and p:
return u, p
logger.warning("creds returned non-zero for MATRIX_USER/MATRIX_PASSWORD")
except FileNotFoundError:
logger.warning("creds binary not found on PATH")
except subprocess.TimeoutExpired:
logger.warning("creds timed out fetching Matrix credentials")
except Exception as exc:
logger.warning("creds fetch failed: %s", exc)
return None
def _matrix_login(session: requests.Session) -> Optional[str]:
"""Login via password, cache token, return access_token. None on failure."""
creds = _fetch_creds()
if not creds:
return None
user, password = creds
url = f"{MATRIX_HOMESERVER}/_matrix/client/r0/login"
body = {"type": "m.login.password", "user": user, "password": password}
try:
resp = session.post(url, json=body, timeout=MATRIX_TIMEOUT)
if resp.status_code != 200:
logger.warning("matrix login failed: HTTP %d", resp.status_code)
return None
data = resp.json()
token = data.get("access_token")
expires_in_ms = data.get("expires_in_ms", 0)
if not token:
logger.warning("matrix login returned no access_token")
return None
expires_at = datetime.now(timezone.utc) + timedelta(milliseconds=expires_in_ms)
_write_token_cache(token, expires_at.isoformat())
return token
except requests.RequestException as exc:
logger.warning("matrix login network error: %s", exc)
except Exception as exc:
logger.warning("matrix login unexpected error: %s", exc)
return None
def _get_matrix_token(session: requests.Session, force_refresh: bool = False) -> Optional[str]:
"""Return a valid token. Read cache → check expiry → login if needed."""
if not force_refresh:
cached = _read_token_cache()
if cached:
try:
expires_at = datetime.fromisoformat(cached["expires_at"])
remaining = (expires_at - datetime.now(timezone.utc)).total_seconds()
if remaining > MATRIX_REFRESH_THRESHOLD_SECONDS:
return cached["access_token"]
except Exception as exc:
logger.debug("cache expiry parse failed: %s", exc)
return _matrix_login(session)
def _send_matrix(message: str) -> bool:
"""Send a Matrix message to MATRIX_ROOM. Return True on success.
Non-fatal: any failure logs WARN and returns False. Never raises.
Implements one 401 retry with token refresh.
"""
session = requests.Session()
session.headers.update({"User-Agent": USER_AGENT})
encoded_room = quote(MATRIX_ROOM, safe="")
url = f"{MATRIX_HOMESERVER}/_matrix/client/r0/rooms/{encoded_room}/send/m.room.message"
body = {"msgtype": "m.text", "body": message}
for attempt in (1, 2):
token = _get_matrix_token(session, force_refresh=(attempt == 2))
if not token:
logger.warning("matrix send aborted: no token (attempt %d)", attempt)
return False
try:
resp = session.post(
url,
json=body,
headers={"Authorization": f"Bearer {token}"},
timeout=MATRIX_TIMEOUT,
)
if resp.status_code == 200:
logger.info("matrix sent: %s", message[:80])
return True
if resp.status_code == 401 and attempt == 1:
logger.info("matrix 401 on send; refreshing token and retrying")
continue
logger.warning("matrix send failed: HTTP %d body=%s",
resp.status_code, resp.text[:200])
return False
except requests.RequestException as exc:
logger.warning("matrix send network error: %s", exc)
return False
except Exception as exc:
logger.warning("matrix send unexpected error: %s", exc)
return False
return False
def notify(summary: list) -> None:
"""Stub notification. Logs intention to notify Matrix when service is back."""
"""Build a Matrix message and POST it. Non-fatal."""
if not summary:
return
repos = [f"{repo}@{short_sha}" for repo, short_sha, _ in summary]
repos_str = ", ".join(repos)
logger.info(
"notify(matrix-stub): coe-mirror: %d new from Nightmare_Eclipse: %s",
len(repos), repos_str
)
logger.debug(
"TODO: Implement real Matrix notification via ~/tools/devtrack/scripts/matrix_alerts.py when service is available"
text = (
f"coe-mirror: {len(repos)} new from Nightmare_Eclipse: {repos_str}"
)
# Always log the intended message so behavior is visible even when Matrix is down
logger.info("notify: %s", text)
_send_matrix(text)
def main() -> int:
+9
View File
@@ -11,3 +11,12 @@
- `commit.sha` is NOT in the repo-list API response — two-call approach (list, then per-repo branch) is retained.
- **DevTrack**: Not created at creation time (oraculo offline, Tailscale last seen 2d ago, carbon roaming off home LAN). Add a `coe-mirror` project + item retroactively when oraculo returns.
- **Matrix notification**: Stub-only at creation time, Matrix server down. Real implementation deferred until home Matrix server is back.
## 2026-06-14 — Matrix notification swap
- Replaced `notify()` stub with real Synapse send. Targets General room (`!MBpPtYlQYMcngtpvJt:m.mealeyfamily.com`).
- Auth: password login via `MATRIX_USER` + `MATRIX_PASSWORD` from Infisical `matrix/` at refresh time. Token cache at `~/.local/share/coe-mirror/matrix-token.json` (chmod 0600), refreshed when <30 min left or on 401.
- Non-fatal failures — Matrix outage does not break the mirror.
- Env overrides: `COE_MIRROR_MATRIX_ROOM`, `COE_MIRROR_MATRIX_HOMESERVER`.
- Pre-deploy step: invite `@intel-bot:m.mealeyfamily.com` to General once Synapse is back. Without joined membership, sends will 403.
- DevTrack: #1172.
+235
View File
@@ -589,3 +589,238 @@ class TestPlaceIntoPloudSameFileIsNoop:
# Assert file still exists and has same content
assert test_file.exists()
assert test_file.stat().st_size > 0
# Tests for Matrix notification implementation.
# Note: Tests requiring monkeypatched subprocess.run must use raising=False because
# subprocess.run is not a class attribute on the coe_mirror module by default
# — patch via monkeypatch.setattr("coe_mirror.subprocess.run", fake_run).
class TestMatrixSendSuccess:
"""Test successful Matrix message send with cached token."""
def test_matrix_send_success(self, mock_paths, setup_logging, monkeypatch):
"""Pre-write valid token, mock send endpoint 200, assert returns True."""
# Pre-write a valid (un-expired) token to cache
token_cache_file = mock_paths["log_dir"] / "matrix-token.json"
import json
from datetime import datetime, timedelta, timezone
future_time = datetime.now(timezone.utc) + timedelta(days=1)
cache_data = {
"access_token": "cached-token-12345",
"expires_at": future_time.isoformat()
}
token_cache_file.parent.mkdir(parents=True, exist_ok=True)
with open(token_cache_file, "w") as f:
json.dump(cache_data, f)
os.chmod(token_cache_file, 0o600)
# Patch MATRIX_TOKEN_CACHE to point to our temp file
monkeypatch.setattr(coe_mirror, "MATRIX_TOKEN_CACHE", token_cache_file)
# Mock send endpoint to return 200
with requests_mock.Mocker() as m:
m.post(
"https://m.mealeyfamily.com/_matrix/client/r0/rooms/"
"%21MBpPtYlQYMcngtpvJt%3Am.mealeyfamily.com/send/m.room.message",
json={"event_id": "$event1"},
status_code=200
)
# Call _send_matrix
result = coe_mirror._send_matrix("hello")
# Assert success
assert result is True
# Verify the request was made
assert m.called
class TestMatrixSendFailureNonfatal:
"""Test that Matrix send failure is non-fatal."""
def test_matrix_send_failure_nonfatal(self, mock_paths, setup_logging, monkeypatch):
"""Pre-write token, mock send 500, assert returns False, no exception."""
token_cache_file = mock_paths["log_dir"] / "matrix-token.json"
import json
from datetime import datetime, timedelta, timezone
future_time = datetime.now(timezone.utc) + timedelta(days=1)
cache_data = {
"access_token": "cached-token-12345",
"expires_at": future_time.isoformat()
}
token_cache_file.parent.mkdir(parents=True, exist_ok=True)
with open(token_cache_file, "w") as f:
json.dump(cache_data, f)
monkeypatch.setattr(coe_mirror, "MATRIX_TOKEN_CACHE", token_cache_file)
with requests_mock.Mocker() as m:
m.post(
"https://m.mealeyfamily.com/_matrix/client/r0/rooms/"
"%21MBpPtYlQYMcngtpvJt%3Am.mealeyfamily.com/send/m.room.message",
status_code=500,
text="Internal Server Error"
)
result = coe_mirror._send_matrix("hello")
assert result is False
class TestMatrixTokenRefreshOn401:
"""Test that 401 on send triggers token refresh and retry."""
def test_matrix_token_refresh_on_401(self, mock_paths, setup_logging, monkeypatch):
"""Pre-write token, mock send 401 then 200, mock login, assert refreshes and succeeds."""
token_cache_file = mock_paths["log_dir"] / "matrix-token.json"
import json
from datetime import datetime, timedelta, timezone
future_time = datetime.now(timezone.utc) + timedelta(days=1)
cache_data = {
"access_token": "old-token",
"expires_at": future_time.isoformat()
}
token_cache_file.parent.mkdir(parents=True, exist_ok=True)
with open(token_cache_file, "w") as f:
json.dump(cache_data, f)
monkeypatch.setattr(coe_mirror, "MATRIX_TOKEN_CACHE", token_cache_file)
# Mock creds to return user/pass
fake_result_user = MagicMock()
fake_result_user.returncode = 0
fake_result_user.stdout = "test_user"
fake_result_pass = MagicMock()
fake_result_pass.returncode = 0
fake_result_pass.stdout = "test_pass"
def mock_subprocess_run(cmd, **kwargs):
if cmd[2] == "MATRIX_USER":
return fake_result_user
elif cmd[2] == "MATRIX_PASSWORD":
return fake_result_pass
return MagicMock(returncode=1, stdout="")
monkeypatch.setattr(coe_mirror.subprocess, "run", mock_subprocess_run)
with requests_mock.Mocker() as m:
# First send returns 401
m.post(
"https://m.mealeyfamily.com/_matrix/client/r0/rooms/"
"%21MBpPtYlQYMcngtpvJt%3Am.mealeyfamily.com/send/m.room.message",
[
{"status_code": 401, "text": "Unauthorized"},
{"status_code": 200, "json": {"event_id": "$event1"}}
]
)
# Login endpoint
m.post(
"https://m.mealeyfamily.com/_matrix/client/r0/login",
json={"access_token": "new-token", "expires_in_ms": 604800000}
)
result = coe_mirror._send_matrix("hello")
# Assert succeeded after refresh
assert result is True
# Verify new token is in cache
with open(token_cache_file) as f:
cached = json.load(f)
assert cached["access_token"] == "new-token"
class TestMatrixRoomEnvOverride:
"""Test that COE_MIRROR_MATRIX_ROOM env var overrides the default room."""
def test_matrix_room_env_override(self, mock_paths, setup_logging, monkeypatch):
"""Set env override, assert send POSTs to override room URL."""
token_cache_file = mock_paths["log_dir"] / "matrix-token.json"
import json
from datetime import datetime, timedelta, timezone
future_time = datetime.now(timezone.utc) + timedelta(days=1)
cache_data = {
"access_token": "test-token",
"expires_at": future_time.isoformat()
}
token_cache_file.parent.mkdir(parents=True, exist_ok=True)
with open(token_cache_file, "w") as f:
json.dump(cache_data, f)
monkeypatch.setattr(coe_mirror, "MATRIX_TOKEN_CACHE", token_cache_file)
# Override the room
override_room = "!override:example.org"
monkeypatch.setattr(coe_mirror, "MATRIX_ROOM", override_room)
with requests_mock.Mocker() as m:
m.post(
"https://m.mealeyfamily.com/_matrix/client/r0/rooms/"
"%21override%3Aexample.org/send/m.room.message",
json={"event_id": "$event1"},
status_code=200
)
result = coe_mirror._send_matrix("hello")
assert result is True
assert m.called
class TestMatrixLoginWritesCacheChmod600:
"""Test that _matrix_login writes cache file with 0600 permissions."""
def test_matrix_login_writes_cache_chmod_600(self, mock_paths, setup_logging, monkeypatch):
"""Start with no cache, mock creds and login, assert cache exists with 0o600."""
token_cache_file = mock_paths["log_dir"] / "matrix-token.json"
monkeypatch.setattr(coe_mirror, "MATRIX_TOKEN_CACHE", token_cache_file)
# Mock creds to return user/pass
fake_result_user = MagicMock()
fake_result_user.returncode = 0
fake_result_user.stdout = "testuser"
fake_result_pass = MagicMock()
fake_result_pass.returncode = 0
fake_result_pass.stdout = "testpass"
def mock_subprocess_run(cmd, **kwargs):
if cmd[2] == "MATRIX_USER":
return fake_result_user
elif cmd[2] == "MATRIX_PASSWORD":
return fake_result_pass
return MagicMock(returncode=1, stdout="")
monkeypatch.setattr(coe_mirror.subprocess, "run", mock_subprocess_run)
with requests_mock.Mocker() as m:
m.post(
"https://m.mealeyfamily.com/_matrix/client/r0/login",
json={"access_token": "new-token", "expires_in_ms": 604800000}
)
session = requests.Session()
token = coe_mirror._get_matrix_token(session, force_refresh=True)
assert token == "new-token"
assert token_cache_file.exists()
# Check permissions
import stat
mode = stat.S_IMODE(token_cache_file.stat().st_mode)
assert mode == 0o600
# Check content
import json
with open(token_cache_file) as f:
cached = json.load(f)
assert cached["access_token"] == "new-token"