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
@@ -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 (`!REDACTEDROOM:m.example.org`) on
|
||||
`m.example.org`.
|
||||
|
||||
- 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
@@ -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.example.org"
|
||||
MATRIX_ROOM = "!REDACTEDROOM:m.example.org" # 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:
|
||||
|
||||
@@ -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.example.org/_matrix/client/r0/rooms/"
|
||||
"%21REDACTEDROOM%3Am.example.org/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.example.org/_matrix/client/r0/rooms/"
|
||||
"%21REDACTEDROOM%3Am.example.org/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.example.org/_matrix/client/r0/rooms/"
|
||||
"%21REDACTEDROOM%3Am.example.org/send/m.room.message",
|
||||
[
|
||||
{"status_code": 401, "text": "Unauthorized"},
|
||||
{"status_code": 200, "json": {"event_id": "$event1"}}
|
||||
]
|
||||
)
|
||||
|
||||
# Login endpoint
|
||||
m.post(
|
||||
"https://m.example.org/_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.example.org/_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.example.org/_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"
|
||||
|
||||
Reference in New Issue
Block a user