commit 45a8d6163e6f02e96ecba7c8776cf2a5b973bc4d Author: Cobra Date: Thu Jun 11 09:34:55 2026 -0400 coe-mirror: mirror Nightmare_Eclipse Forgejo repos to pCloud Standalone Python tool. Hits git.churchofmalware.org public API every 12h (00:00/12:00), downloads any repo HEAD whose SHA hasn't already been snapshotted to pCloud, with local-staging fallback when the FUSE mount is unavailable. Matrix notification stub for when the home server is back. Devil's-advocate findings folded into the design: - shutil.copy2 + size verify + unlink, not os.rename, for FUSE safety - no pCloud daemon auto-start (no user unit; DISPLAY=:0 unreliable); relies on Persistent timer + pending flush for recovery - two-call API path retained because the repo list response does not include commit.sha Audit-driven security fixes: - repo names whitelist-validated (^[a-zA-Z0-9._-]+$) - urllib.parse.quote on all interpolated URL segments - same-file no-op guard in place_into_pcloud - existence check moved to main() so cached files don't generate spurious notify entries 13 pytest tests passing. diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..55368cd --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +.venv/ +__pycache__/ +*.pyc +*.pyo +.pytest_cache/ +*.egg-info/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..0440354 --- /dev/null +++ b/README.md @@ -0,0 +1,62 @@ +# coe-mirror + +Mirror Nightmare_Eclipse repos from Church of Malware to pCloud. + +## What it does + +Monitors `Nightmare_Eclipse` on `git.churchofmalware.org` for new or updated repos. Downloads per-commit zip snapshots to `~/pCloudDrive/Hacking/Nightmare_Eclipse/`. Falls back to local staging when pCloud is down. Runs at 00:00 and 12:00 daily with boot catch-up. + +## Filename convention + +`{repo}_{short_sha}.zip` where short_sha is the first 8 chars of the HEAD commit SHA on the default branch. New commits produce new files; old snapshots stay as history. + +## Install + +``` +cd ~/tools/coe-mirror +bash install.sh +``` + +## Manual run + +``` +systemctl --user start coe-mirror.service +# or directly: +python3 ~/tools/coe-mirror/coe_mirror.py +``` + +## Logs + +- File: `~/.local/share/coe-mirror/coe-mirror.log` (rotated at 5 MB, 3 backups) +- Journal: `journalctl --user -u coe-mirror.service -n 100` + +## Storage estimate + +8 repos × 2 runs/day × ~1 commit/repo/day ≈ 16 zips/day ≈ ~5 MB/month at current zip sizes (~265 KB each). Old snapshots accumulate — no automatic pruning. Manually prune `~/pCloudDrive/Hacking/Nightmare_Eclipse/` if needed. + +## Behavior when pCloud is down + +Downloads land in `~/.local/share/coe-mirror/pending/`. Next run (or any subsequent invocation) flushes them to pCloud once the mount is healthy. No auto-start of the pCloud daemon — by design, because the daemon is a GUI AppImage with no systemd user unit. If pCloud is offline, ensure the user session is active and pCloud is running before the next scheduled run, or trigger manually. + +## Catch-up after machine was off + +The timer is `Persistent=true` and `OnBootSec=2min`. Missed windows run automatically on next boot. The tool also diffs against the pCloud directory, so even if a run is skipped entirely, the next run catches up everything that's missing. + +## 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`). + +## Testing + +``` +cd ~/tools/coe-mirror +python3 -m pytest tests/ -v +``` + +## Uninstall + +``` +systemctl --user disable --now coe-mirror.timer +rm ~/.config/systemd/user/coe-mirror.{service,timer} +systemctl --user daemon-reload +``` diff --git a/coe_mirror.py b/coe_mirror.py new file mode 100644 index 0000000..b2d9a08 --- /dev/null +++ b/coe_mirror.py @@ -0,0 +1,406 @@ +"""Mirror repos from git.churchofmalware.org Forgejo instance to pCloud. + +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 logging +import logging.handlers +import os +import re +import shutil +import sys +import time +import zipfile +from pathlib import Path +from fcntl import flock, LOCK_EX, LOCK_NB +from typing import Optional +from urllib.parse import quote + +import requests + +# Module constants (overridable via env vars for testability) +API_BASE = "https://git.churchofmalware.org" +OWNER = "Nightmare_Eclipse" +PCLOUD_TARGET = Path.home() / "pCloudDrive" / "Hacking" / "Nightmare_Eclipse" +PENDING_DIR = Path.home() / ".local" / "share" / "coe-mirror" / "pending" +LOG_DIR = Path.home() / ".local" / "share" / "coe-mirror" +LOG_FILE = LOG_DIR / "coe-mirror.log" +LOCK_NAME = "coe-mirror.lock" +SHORT_SHA_LEN = 8 +HTTP_TIMEOUT = 30 +DOWNLOAD_TIMEOUT = 60 +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._-]+$") + +# 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) + +logger: logging.Logger = None # Initialized in setup_logging() + + +def _safe_repo_name(name: str) -> bool: + """Validate repo name against whitelist to prevent path traversal.""" + return bool(REPO_NAME_RE.fullmatch(name)) + + +def setup_logging() -> None: + """Configure logging to file and stderr with rotation.""" + global logger + LOG_DIR.mkdir(parents=True, exist_ok=True) + PENDING_DIR.mkdir(parents=True, exist_ok=True) + + logger = logging.getLogger("coe-mirror") + logger.setLevel(logging.DEBUG) + + # Rotating file handler + file_handler = logging.handlers.RotatingFileHandler( + LOG_FILE, maxBytes=5 * 1024 * 1024, backupCount=3 + ) + file_handler.setLevel(logging.DEBUG) + + # Stderr handler for systemd journal + stderr_handler = logging.StreamHandler(sys.stderr) + stderr_handler.setLevel(logging.DEBUG) + + # Formatter + formatter = logging.Formatter( + "%(asctime)s %(levelname)s %(name)s: %(message)s" + ) + file_handler.setFormatter(formatter) + stderr_handler.setFormatter(formatter) + + logger.addHandler(file_handler) + logger.addHandler(stderr_handler) + + +def acquire_lock() -> Optional[int]: + """Acquire exclusive lock. Return lock fd on success, None if contended.""" + xdg_runtime = os.getenv("XDG_RUNTIME_DIR") + lock_dir = Path(xdg_runtime) if xdg_runtime else Path("/tmp") + lock_file = lock_dir / LOCK_NAME + + try: + fd = os.open(str(lock_file), os.O_CREAT | os.O_WRONLY, 0o600) + try: + flock(fd, LOCK_EX | LOCK_NB) + return fd + except OSError: + os.close(fd) + return None + except OSError as exc: + logger.error("Failed to open lock file: %s", exc) + return None + + +def release_lock(lock_fd: int) -> None: + """Release and close the lock file descriptor.""" + try: + os.close(lock_fd) + except OSError: + pass + + +def probe_pcloud() -> bool: + """Probe pCloud health. Write tiny file, read back, delete. Return True if healthy.""" + try: + probe_file = PCLOUD_TARGET / ".write_probe" + PCLOUD_TARGET.mkdir(parents=True, exist_ok=True) + + probe_file.write_text("probe") + # fsync on file descriptor, not stat + with open(probe_file, "r") as f: + os.fsync(f.fileno()) + content = probe_file.read_text() + probe_file.unlink() + + if content != "probe": + logger.warning("pCloud write verification failed (content mismatch); pcloud unavailable; queuing locally") + return False + return True + except Exception as exc: + logger.warning("pCloud probe failed: %s; pcloud unavailable; queuing locally", exc) + return False + + +def place_into_pcloud(src: Path, dst: Path) -> bool: + """Copy src → dst with size verification; unlink src only on success. + + Returns True on success, False on any failure. Never raises. + """ + # Defense in depth: if src and dst are the same file, it's a no-op success + if src.resolve() == dst.resolve(): + logger.debug("src and dst are the same file, no-op: %s", src) + return True + + try: + shutil.copy2(src, dst) + src_size = src.stat().st_size + dst_size = dst.stat().st_size + if src_size != dst_size: + logger.warning( + "size mismatch placing %s → %s (src=%d dst=%d)", + src, dst, src_size, dst_size + ) + try: + dst.unlink() + except OSError: + pass + return False + src.unlink() + return True + except Exception as exc: + logger.warning("placement failed %s → %s: %s", src, dst, exc) + return False + + +def fetch_repos(session: requests.Session) -> Optional[list]: + """Fetch list of repos owned by OWNER. Return list of repo dicts or None on fatal error.""" + url = f"{API_BASE}/api/v1/users/{OWNER}/repos" + + for attempt, backoff in enumerate(RETRY_BACKOFF): + if attempt > 0: + logger.info("Retrying repo list fetch (attempt %d), backoff %ds", attempt + 1, backoff) + time.sleep(backoff) + + try: + resp = session.get(url, timeout=HTTP_TIMEOUT) + if resp.status_code == 429: + logger.debug("Got 429 on repo list, will retry") + continue + if resp.status_code >= 500: + logger.debug("Got %d on repo list, will retry", resp.status_code) + continue + resp.raise_for_status() + return resp.json() + except requests.RequestException as exc: + logger.debug("Attempt %d: %s", attempt + 1, exc) + + logger.error("Failed to fetch repos after %d attempts", len(RETRY_BACKOFF)) + return None + + +def fetch_branch_sha(session: requests.Session, repo_name: str, default_branch: str) -> Optional[str]: + """Fetch commit SHA for repo's default branch. Return full hex SHA or None on error.""" + url = f"{API_BASE}/api/v1/repos/{OWNER}/{quote(repo_name, safe='')}/branches/{quote(default_branch, safe='')}" + + for attempt, backoff in enumerate(RETRY_BACKOFF): + if attempt > 0: + logger.debug("Retrying branch fetch for %s (attempt %d), backoff %ds", + repo_name, attempt + 1, backoff) + time.sleep(backoff) + + try: + resp = session.get(url, timeout=HTTP_TIMEOUT) + if resp.status_code == 429: + logger.debug("Got 429 on branch %s, will retry", repo_name) + continue + if resp.status_code >= 500: + logger.debug("Got %d on branch %s, will retry", resp.status_code, repo_name) + continue + resp.raise_for_status() + data = resp.json() + return data.get("commit", {}).get("id") + except requests.RequestException as exc: + logger.debug("Attempt %d for %s: %s", attempt + 1, repo_name, exc) + + logger.error("Failed to fetch branch SHA for %s after %d attempts", repo_name, len(RETRY_BACKOFF)) + return None + + +def download_repo_zip(session: requests.Session, repo_name: str, full_sha: str) -> Optional[Path]: + """Download repo zip for given SHA. Return Path to downloaded file (in PENDING_DIR) or None.""" + # Defense in depth: validate repo_name even though main() should have done so + if not _safe_repo_name(repo_name): + logger.error("refusing unsafe repo name in download_repo_zip: %r", repo_name) + return None + + filename = f"{repo_name}_{full_sha[:SHORT_SHA_LEN]}.zip" + partial_path = PENDING_DIR / f"{filename}.partial" + final_path = PENDING_DIR / filename + + # URL-encode repo_name and full_sha for safety (defense in depth after main-level validation) + url = f"{API_BASE}/{OWNER}/{quote(repo_name, safe='')}/archive/{quote(full_sha, safe='')}.zip" + + for attempt, backoff in enumerate(RETRY_BACKOFF): + if attempt > 0: + logger.debug("Retrying download for %s (attempt %d), backoff %ds", + filename, attempt + 1, backoff) + time.sleep(backoff) + + try: + resp = session.get(url, timeout=DOWNLOAD_TIMEOUT, stream=True) + if resp.status_code == 429: + logger.debug("Got 429 on download %s, will retry", filename) + continue + if resp.status_code >= 500: + logger.debug("Got %d on download %s, will retry", resp.status_code, filename) + continue + resp.raise_for_status() + + # Stream to temp file + partial_path.parent.mkdir(parents=True, exist_ok=True) + with open(partial_path, "wb") as f: + for chunk in resp.iter_content(chunk_size=65536): + if chunk: + f.write(chunk) + + # Validate zip + if not zipfile.is_zipfile(partial_path): + logger.error("Downloaded file is not a valid zip: %s", filename) + partial_path.unlink() + return None + + # Rename to final + partial_path.rename(final_path) + logger.info("Downloaded %s", filename) + return final_path + + except requests.RequestException as exc: + logger.debug("Attempt %d for %s: %s", attempt + 1, filename, exc) + except Exception as exc: + logger.error("Error downloading %s: %s", filename, exc) + if partial_path.exists(): + partial_path.unlink() + return None + + logger.error("Failed to download %s after %d attempts", filename, len(RETRY_BACKOFF)) + if partial_path.exists(): + partial_path.unlink() + return None + + +def flush_pending(pcloud_healthy: bool) -> list: + """Move all *.zip files from PENDING_DIR to PCLOUD_TARGET if healthy. + + Return list of (repo, short_sha, final_location) tuples for files moved. + """ + moved = [] + if not pcloud_healthy or not PENDING_DIR.exists(): + return moved + + for pending_file in PENDING_DIR.glob("*.zip"): + dst = PCLOUD_TARGET / pending_file.name + if place_into_pcloud(pending_file, dst): + # Extract repo name and short sha from filename + name_parts = pending_file.stem.rsplit("_", 1) + if len(name_parts) == 2: + repo, short_sha = name_parts + moved.append((repo, short_sha, str(dst))) + logger.info("Flushed %s to pCloud", pending_file.name) + + return moved + + +def notify(summary: list) -> None: + """Stub notification. Logs intention to notify Matrix when service is back.""" + 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" + ) + + +def main() -> int: + """Main entry point.""" + setup_logging() + + lock_fd = acquire_lock() + if lock_fd is None: + logger.debug("Lock already held; another instance running, exiting gracefully") + return 0 + + try: + logger.info("coe-mirror starting") + + # Probe pCloud health + pcloud_healthy = probe_pcloud() + logger.info("pCloud health check: %s", "healthy" if pcloud_healthy else "unavailable") + + # Flush pending at start + initial_flush = flush_pending(pcloud_healthy) + + # Create session and fetch repos + session = requests.Session() + session.headers.update({"User-Agent": USER_AGENT}) + + repos = fetch_repos(session) + if repos is None: + logger.error("Fatal: could not fetch repos") + return 1 + + logger.info("Fetched %d repos", len(repos)) + + new_files = [] + + # Process each repo + for repo in repos: + repo_name = repo.get("name") + default_branch = repo.get("default_branch", "main") + + if not repo_name: + logger.warning("Repo object missing 'name'") + continue + + # Fix 1: Reject unsafe repo names to prevent path traversal + if not _safe_repo_name(repo_name): + logger.warning("refusing unsafe repo name: %r", repo_name) + continue + + sha = fetch_branch_sha(session, repo_name, default_branch) + if sha is None: + logger.error("Skipping %s (could not fetch branch SHA)", repo_name) + continue + + # Fix 3: Check existence before download to avoid SameFileError + filename = f"{repo_name}_{sha[:SHORT_SHA_LEN]}.zip" + pcloud_path = PCLOUD_TARGET / filename + pending_path = PENDING_DIR / filename + already_have = (pcloud_path.exists() and pcloud_path.stat().st_size > 0) \ + or (pending_path.exists() and pending_path.stat().st_size > 0) + if already_have: + logger.info("Skipping %s (already present)", filename) + continue + + downloaded = download_repo_zip(session, repo_name, sha) + if downloaded: + short_sha = sha[:SHORT_SHA_LEN] + # Try to place it + if pcloud_healthy: + dst = PCLOUD_TARGET / downloaded.name + if place_into_pcloud(downloaded, dst): + new_files.append((repo_name, short_sha, str(dst))) + else: + logger.info("File %s left in pending (placement failed)", downloaded.name) + new_files.append((repo_name, short_sha, str(downloaded))) + else: + logger.info("pCloud unhealthy; %s queued in pending", downloaded.name) + new_files.append((repo_name, short_sha, str(downloaded))) + + # Flush pending at end + final_flush = flush_pending(pcloud_healthy) + new_files.extend(final_flush) + + # Notify + notify(new_files) + + logger.info("coe-mirror complete") + return 0 + + finally: + release_lock(lock_fd) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/install.sh b/install.sh new file mode 100644 index 0000000..956f8a7 --- /dev/null +++ b/install.sh @@ -0,0 +1,55 @@ +#!/bin/bash +set -euo pipefail + +# Resolve paths +TOOL_DIR="$HOME/tools/coe-mirror" +PCLOUD_TARGET="$HOME/pCloudDrive/Hacking/Nightmare_Eclipse" +PENDING_DIR="$HOME/.local/share/coe-mirror/pending" +LOG_DIR="$HOME/.local/share/coe-mirror" +USER_UNIT_DIR="$HOME/.config/systemd/user" + +# Check if pCloud mount exists (warning only, not fatal) +if ! mountpoint -q "$HOME/pCloudDrive" 2>/dev/null; then + echo "WARNING: pCloud mount not detected at ~/pCloudDrive — install will proceed but coe-mirror will queue everything locally until the mount appears." +fi + +# Create directories +mkdir -p "$PENDING_DIR" "$LOG_DIR" "$USER_UNIT_DIR" +if mountpoint -q "$HOME/pCloudDrive" 2>/dev/null; then + mkdir -p "$PCLOUD_TARGET" +fi + +# Install dependencies +echo "Installing dependencies..." +if ! pip install --user requests requests-mock pytest 2>&1 | tail -5; then + echo "ERROR: pip install --user failed (PEP 668 externally-managed environment?)." + echo "To install in a virtualenv instead, run:" + echo " python3 -m venv ~/tools/coe-mirror/.venv" + echo " source ~/tools/coe-mirror/.venv/bin/activate" + echo " pip install requests requests-mock pytest" + exit 1 +fi + +# Copy unit files +cp "$TOOL_DIR/systemd/coe-mirror.service" "$USER_UNIT_DIR/" +cp "$TOOL_DIR/systemd/coe-mirror.timer" "$USER_UNIT_DIR/" + +# Reload and enable +systemctl --user daemon-reload +systemctl --user enable --now coe-mirror.timer + +# Show next scheduled run +echo "" +echo "Next scheduled run:" +systemctl --user list-timers coe-mirror.timer --no-pager | head -5 + +# Summary +echo "" +echo "=== Installation Summary ===" +echo "Installed. Timer: coe-mirror.timer (enabled & active)." +echo "pCloud target: $PCLOUD_TARGET (exists? $([ -d "$PCLOUD_TARGET" ] && echo "yes" || echo "no"))" +echo "Pending dir: $PENDING_DIR" +echo "Log: ~/.local/share/coe-mirror/coe-mirror.log" +echo "" +echo "To trigger immediately: systemctl --user start coe-mirror.service" +echo "To uninstall: systemctl --user disable --now coe-mirror.timer && rm $USER_UNIT_DIR/coe-mirror.{service,timer}" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..c29f9a3 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +requests>=2.31.0 +requests-mock>=1.11.0 +pytest>=7.4.0 diff --git a/systemd/coe-mirror.service b/systemd/coe-mirror.service new file mode 100644 index 0000000..ecf7c84 --- /dev/null +++ b/systemd/coe-mirror.service @@ -0,0 +1,12 @@ +[Unit] +Description=coe-mirror: mirror Nightmare_Eclipse repos to pCloud +After=network-online.target + +[Service] +Type=oneshot +ExecStart=/usr/bin/python3 %h/tools/coe-mirror/coe_mirror.py +WorkingDirectory=%h/tools/coe-mirror +Nice=10 +IOSchedulingClass=idle +StandardOutput=journal +StandardError=journal diff --git a/systemd/coe-mirror.timer b/systemd/coe-mirror.timer new file mode 100644 index 0000000..22891b6 --- /dev/null +++ b/systemd/coe-mirror.timer @@ -0,0 +1,12 @@ +[Unit] +Description=coe-mirror schedule: 00:00 and 12:00 daily, with boot catch-up + +[Timer] +OnCalendar=*-*-* 00,12:00:00 +Persistent=true +OnBootSec=2min +RandomizedDelaySec=30 +Unit=coe-mirror.service + +[Install] +WantedBy=timers.target diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_coe_mirror.py b/tests/test_coe_mirror.py new file mode 100644 index 0000000..31ae8cb --- /dev/null +++ b/tests/test_coe_mirror.py @@ -0,0 +1,591 @@ +"""Tests for coe_mirror.py + +Lock test (test_lock_contention_exits_zero) may have ordering dependencies +and should run in isolation if flaky. Other tests are independent. +""" + +import io +import logging +import os +import shutil +import sys +import threading +import time +import zipfile +from fcntl import flock, LOCK_EX, LOCK_NB +from pathlib import Path +from unittest.mock import patch, MagicMock + +import pytest +import requests +import requests_mock + + +# Import the module under test +import coe_mirror + + +@pytest.fixture +def mock_paths(tmp_path, monkeypatch): + """Redirect all paths to tmp_path for test isolation.""" + pcloud_target = tmp_path / "pcloud" + pending_dir = tmp_path / "pending" + log_dir = tmp_path / "logs" + + pcloud_target.mkdir() + pending_dir.mkdir() + log_dir.mkdir() + + monkeypatch.setattr(coe_mirror, "PCLOUD_TARGET", pcloud_target) + monkeypatch.setattr(coe_mirror, "PENDING_DIR", pending_dir) + monkeypatch.setattr(coe_mirror, "LOG_DIR", log_dir) + monkeypatch.setattr(coe_mirror, "LOG_FILE", log_dir / "test.log") + + return { + "pcloud_target": pcloud_target, + "pending_dir": pending_dir, + "log_dir": log_dir, + } + + +@pytest.fixture +def setup_logging(mock_paths, monkeypatch): + """Set up logging for tests.""" + coe_mirror.setup_logging() + yield + # Clean up logger + logger = logging.getLogger("coe-mirror") + for handler in logger.handlers[:]: + handler.close() + logger.removeHandler(handler) + + +def create_valid_zip(path: Path, content: str = "test") -> None: + """Helper to create a valid zip file.""" + with zipfile.ZipFile(path, "w") as zf: + zf.writestr("README.md", content) + + +@pytest.fixture +def mock_lock(monkeypatch): + """Mock locking to avoid actual file locks during tests.""" + # For most tests, just return a valid fd + original_open = os.open + + def mock_open(path, flags, mode=0o777): + # Return a dummy fd that won't conflict + return original_open(path, flags, mode) + + monkeypatch.setattr(os, "open", mock_open) + + +class TestNewRepoDownloadsAndPlacesInPcloud: + """Test new repo downloads and places in pCloud.""" + + def test_new_repo_downloads_and_places_in_pcloud(self, mock_paths, setup_logging, monkeypatch): + """Mock list returning one repo, assert file lands in pCloud with expected name.""" + monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999) + monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None) + monkeypatch.setattr(time, "sleep", lambda x: None) # Skip actual sleep + + # Mock pCloud probe to return healthy + monkeypatch.setattr(coe_mirror, "probe_pcloud", lambda: True) + + with requests_mock.Mocker() as m: + # Mock repo list + m.get( + f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos", + json=[{"name": "test-repo", "default_branch": "main"}], + ) + # Mock branch info + m.get( + f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/test-repo/branches/main", + json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}}, + ) + + # Create valid zip + zip_buffer = io.BytesIO() + with zipfile.ZipFile(zip_buffer, "w") as zf: + zf.writestr("README.md", "test") + zip_buffer.seek(0) + + m.get( + f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip", + content=zip_buffer.getvalue(), + ) + + result = coe_mirror.main() + assert result == 0 + + # Verify file landed in pCloud with expected name + expected_file = mock_paths["pcloud_target"] / "test-repo_aaaabbbb.zip" + assert expected_file.exists() + assert expected_file.stat().st_size > 0 + + +class TestNewCommitCreatesSnapshot: + """Test that new commit on existing repo creates new snapshot.""" + + def test_new_commit_on_existing_repo_creates_new_snapshot(self, mock_paths, setup_logging, monkeypatch): + """Pre-create old snapshot, fetch new SHA, assert new file lands and old untouched.""" + monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999) + monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None) + monkeypatch.setattr(time, "sleep", lambda x: None) + + # Mock pCloud probe to return healthy + monkeypatch.setattr(coe_mirror, "probe_pcloud", lambda: True) + + # Pre-create old file + old_file = mock_paths["pcloud_target"] / "test-repo_aaaabbbb.zip" + create_valid_zip(old_file, "old") + + with requests_mock.Mocker() as m: + m.get( + f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos", + json=[{"name": "test-repo", "default_branch": "main"}], + ) + # Return new SHA + m.get( + f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/test-repo/branches/main", + json={"commit": {"id": "bbbbccccddddeeeeffffgggghhhhiiiijjjjkkkk"}}, + ) + + # Create valid zip + zip_buffer = io.BytesIO() + with zipfile.ZipFile(zip_buffer, "w") as zf: + zf.writestr("README.md", "new") + zip_buffer.seek(0) + + m.get( + f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/bbbbccccddddeeeeffffgggghhhhiiiijjjjkkkk.zip", + content=zip_buffer.getvalue(), + ) + + result = coe_mirror.main() + assert result == 0 + + # Verify old file still exists + assert old_file.exists() + # Verify new file landed + new_file = mock_paths["pcloud_target"] / "test-repo_bbbbcccc.zip" + assert new_file.exists() + + +class TestPcloudFailureKeepsPending: + """Test pCloud write failure keeps file in pending with WARN log.""" + + def test_pcloud_write_fails_keeps_in_pending(self, mock_paths, setup_logging, monkeypatch): + """Make pCloud target non-writable, assert download stays in pending with WARN log.""" + monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999) + monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None) + monkeypatch.setattr(time, "sleep", lambda x: None) + + # Make pCloud target a file (not a dir) to cause write failure + pcloud_as_file = mock_paths["pcloud_target"] + pcloud_as_file.rmdir() + pcloud_as_file.write_text("block") + + # Mock place_into_pcloud to fail + def mock_place(src, dst): + return False + + monkeypatch.setattr(coe_mirror, "place_into_pcloud", mock_place) + + with requests_mock.Mocker() as m: + m.get( + f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos", + json=[{"name": "test-repo", "default_branch": "main"}], + ) + m.get( + f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/test-repo/branches/main", + json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}}, + ) + + zip_buffer = io.BytesIO() + with zipfile.ZipFile(zip_buffer, "w") as zf: + zf.writestr("README.md", "test") + zip_buffer.seek(0) + + m.get( + f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip", + content=zip_buffer.getvalue(), + ) + + result = coe_mirror.main() + assert result == 0 + + # Verify file is in pending + pending_file = mock_paths["pending_dir"] / "test-repo_aaaabbbb.zip" + assert pending_file.exists() + + +class TestPendingFlushesWhenPcloudRecoveres: + """Test pending files flush when pCloud recovers.""" + + def test_pending_flushes_when_pcloud_recovers(self, mock_paths, setup_logging, monkeypatch): + """Drop valid zip in pending, make pCloud healthy, assert file lands in pCloud.""" + monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999) + monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None) + monkeypatch.setattr(time, "sleep", lambda x: None) + + # Mock pCloud probe to return healthy + monkeypatch.setattr(coe_mirror, "probe_pcloud", lambda: True) + + # Create a pending file + pending_file = mock_paths["pending_dir"] / "test-repo_aaaabbbb.zip" + create_valid_zip(pending_file, "pending") + + with requests_mock.Mocker() as m: + # Empty repo list (no new downloads) + m.get( + f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos", + json=[], + ) + + result = coe_mirror.main() + assert result == 0 + + # Verify file moved to pCloud + pcloud_file = mock_paths["pcloud_target"] / "test-repo_aaaabbbb.zip" + assert pcloud_file.exists() + assert not pending_file.exists() + + +class TestMalformedZipRejected: + """Test that malformed zip is rejected and not placed.""" + + def test_malformed_zip_rejected_and_not_placed(self, mock_paths, setup_logging, monkeypatch): + """Mock zip download returning garbage, assert nothing lands in pCloud.""" + monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999) + monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None) + monkeypatch.setattr(time, "sleep", lambda x: None) + + with requests_mock.Mocker() as m: + m.get( + f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos", + json=[{"name": "test-repo", "default_branch": "main"}], + ) + m.get( + f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/test-repo/branches/main", + json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}}, + ) + # Return binary garbage + m.get( + f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip", + content=b"GARBAGE_DATA_NOT_A_ZIP", + ) + + result = coe_mirror.main() + assert result == 0 + + # Verify nothing in pCloud or pending + pcloud_file = mock_paths["pcloud_target"] / "test-repo_aaaabbbb.zip" + assert not pcloud_file.exists() + pending_file = mock_paths["pending_dir"] / "test-repo_aaaabbbb.zip" + assert not pending_file.exists() + + +class TestNotifyStub: + """Test notify stub is no-op.""" + + def test_notify_stub_no_op(self, setup_logging, monkeypatch): + """Assert notify function does not call network code.""" + # Spy on notify + original_notify = coe_mirror.notify + call_log = [] + + def spy_notify(summary): + call_log.append(summary) + return original_notify(summary) + + monkeypatch.setattr(coe_mirror, "notify", spy_notify) + + # Test with no files + coe_mirror.notify([]) + assert len(call_log) == 1 + assert call_log[0] == [] + + # Test with files + call_log.clear() + coe_mirror.notify([("repo", "aaaabbbb", "/path")]) + assert len(call_log) == 1 + + +class TestApiRetriesOn429: + """Test API retries on 429 with backoff.""" + + def test_api_retries_on_429(self, mock_paths, setup_logging, monkeypatch): + """Mock 429 twice then 200, assert success after retries.""" + monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999) + monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None) + monkeypatch.setattr(time, "sleep", lambda x: None) # Skip actual sleep + + attempt_log = [] + + def track_attempts(status): + def callback(req, ctx): + attempt_log.append(status) + ctx.status_code = status + if status == 200: + return [{"name": "test-repo", "default_branch": "main"}] + return None + + return callback + + with requests_mock.Mocker() as m: + m.get( + f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos", + [ + {"status_code": 429, "json": None}, + {"status_code": 429, "json": None}, + {"json": [{"name": "test-repo", "default_branch": "main"}]}, + ], + ) + m.get( + f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/test-repo/branches/main", + json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}}, + ) + + zip_buffer = io.BytesIO() + with zipfile.ZipFile(zip_buffer, "w") as zf: + zf.writestr("README.md", "test") + zip_buffer.seek(0) + + m.get( + f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip", + content=zip_buffer.getvalue(), + ) + + result = coe_mirror.main() + assert result == 0 + + +class TestSizeMismatchKeepsPending: + """Test size mismatch during placement keeps file in pending.""" + + def test_size_mismatch_keeps_pending_unlinks_bad_dst(self, mock_paths, setup_logging, monkeypatch): + """Mock copy to write short file, assert bad dst removed and src in pending.""" + monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999) + monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None) + monkeypatch.setattr(time, "sleep", lambda x: None) + + # Mock shutil.copy2 to write a short file + def mock_copy2(src, dst): + # Write dst but make it smaller than src + with open(dst, "wb") as f: + f.write(b"SHORT") + + monkeypatch.setattr(shutil, "copy2", mock_copy2) + + with requests_mock.Mocker() as m: + m.get( + f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos", + json=[{"name": "test-repo", "default_branch": "main"}], + ) + m.get( + f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/test-repo/branches/main", + json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}}, + ) + + zip_buffer = io.BytesIO() + with zipfile.ZipFile(zip_buffer, "w") as zf: + zf.writestr("README.md", "test content here is longer") + zip_buffer.seek(0) + + m.get( + f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip", + content=zip_buffer.getvalue(), + ) + + result = coe_mirror.main() + assert result == 0 + + # Verify bad dst was removed + bad_dst = mock_paths["pcloud_target"] / "test-repo_aaaabbbb.zip" + assert not bad_dst.exists() + + # Verify src still in pending + pending_src = mock_paths["pending_dir"] / "test-repo_aaaabbbb.zip" + assert pending_src.exists() + + +class TestExistenceCheckSkipsZeroByte: + """Test that zero-byte file in pCloud target causes re-download.""" + + def test_existence_check_skips_when_pcloud_has_zero_byte_file(self, mock_paths, setup_logging, monkeypatch): + """Pre-create zero-byte file, assert download still happens because st_size check rejects it.""" + monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999) + monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None) + monkeypatch.setattr(time, "sleep", lambda x: None) + + # Pre-create zero-byte file in pCloud target + zero_file = mock_paths["pcloud_target"] / "test-repo_aaaabbbb.zip" + zero_file.write_bytes(b"") + assert zero_file.stat().st_size == 0 + + # Mock place_into_pcloud to succeed (make pCloud healthy) + def mock_place(src, dst): + shutil.copy2(src, dst) + src.unlink() + return True + + monkeypatch.setattr(coe_mirror, "place_into_pcloud", mock_place) + + with requests_mock.Mocker() as m: + m.get( + f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos", + json=[{"name": "test-repo", "default_branch": "main"}], + ) + m.get( + f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/test-repo/branches/main", + json={"commit": {"id": "aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj"}}, + ) + + zip_buffer = io.BytesIO() + with zipfile.ZipFile(zip_buffer, "w") as zf: + zf.writestr("README.md", "real content") + zip_buffer.seek(0) + + m.get( + f"{coe_mirror.API_BASE}/{coe_mirror.OWNER}/test-repo/archive/aaaabbbbccccddddeeeeffffgggghhhhiiiijjjj.zip", + content=zip_buffer.getvalue(), + ) + + result = coe_mirror.main() + assert result == 0 + + # Verify zero-byte file was overwritten with real content + assert zero_file.exists() + assert zero_file.stat().st_size > 0 + + +class TestLockContention: + """Test lock contention exits gracefully.""" + + def test_lock_contention_exits_zero(self, mock_paths, setup_logging, monkeypatch): + """Acquire lock externally, assert main() returns 0 without work.""" + # Create lock file and acquire it + xdg_runtime = os.getenv("XDG_RUNTIME_DIR") + lock_dir = Path(xdg_runtime) if xdg_runtime else Path("/tmp") + lock_file = lock_dir / coe_mirror.LOCK_NAME + + with open(lock_file, "w") as f: + try: + flock(f, LOCK_EX | LOCK_NB) + # Lock acquired; now test that main() respects it + result = coe_mirror.main() + assert result == 0 + finally: + try: + f.close() + except: + pass + # Clean up + try: + lock_file.unlink() + except: + pass + + +class TestUnsafeRepoNameRejected: + """Test that unsafe repo names are rejected (Fix 1 - path traversal).""" + + def test_unsafe_repo_name_rejected(self, mock_paths, setup_logging, monkeypatch): + """Mock API returning repo with unsafe name ../etc/passwd, assert rejected with WARN/ERROR log.""" + monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999) + monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None) + monkeypatch.setattr(time, "sleep", lambda x: None) + + # Mock pCloud probe to return healthy + monkeypatch.setattr(coe_mirror, "probe_pcloud", lambda: True) + + with requests_mock.Mocker() as m: + # Mock repo list returning unsafe repo name + m.get( + f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos", + json=[{"name": "../../../etc/passwd", "default_branch": "main"}], + ) + + result = coe_mirror.main() + assert result == 0 + + # Verify no file was written anywhere with traversal path + assert not (mock_paths["pcloud_target"] / "etc").exists() + assert not (mock_paths["pending_dir"] / "etc").exists() + # Also check that no file matching the unsafe name exists + unsafe_paths = list(mock_paths["pcloud_target"].glob("*..*.zip")) + list(mock_paths["pending_dir"].glob("*..*.zip")) + assert len(unsafe_paths) == 0 + + +class TestSkipWhenFileAlreadyInPcloud: + """Test that file already in pCloud is skipped (Fix 3 - SameFileError).""" + + def test_skip_when_file_already_in_pcloud(self, mock_paths, setup_logging, monkeypatch): + """Pre-create non-empty zip in pCloud, assert download_repo_zip not called and no duplicate placed.""" + monkeypatch.setattr(coe_mirror, "acquire_lock", lambda: 999) + monkeypatch.setattr(coe_mirror, "release_lock", lambda fd: None) + monkeypatch.setattr(time, "sleep", lambda x: None) + + # Mock pCloud probe to return healthy + monkeypatch.setattr(coe_mirror, "probe_pcloud", lambda: True) + + # Pre-create a non-empty file in pCloud + existing_file = mock_paths["pcloud_target"] / "RepoX_abcdef12.zip" + create_valid_zip(existing_file, "existing") + + # Spy on download_repo_zip to ensure it's not called + original_download = coe_mirror.download_repo_zip + download_calls = [] + + def spy_download(session, repo_name, full_sha): + download_calls.append((repo_name, full_sha)) + return original_download(session, repo_name, full_sha) + + monkeypatch.setattr(coe_mirror, "download_repo_zip", spy_download) + + with requests_mock.Mocker() as m: + # Mock repo list + m.get( + f"{coe_mirror.API_BASE}/api/v1/users/{coe_mirror.OWNER}/repos", + json=[{"name": "RepoX", "default_branch": "main"}], + ) + # Return matching SHA (first 8 chars = abcdef12) + m.get( + f"{coe_mirror.API_BASE}/api/v1/repos/{coe_mirror.OWNER}/RepoX/branches/main", + json={"commit": {"id": "abcdef1234567890"}}, + ) + + result = coe_mirror.main() + assert result == 0 + + # Assert download_repo_zip was not called (file already exists, skipped at main level) + assert len(download_calls) == 0 + + # Verify existing file is still there and unchanged + assert existing_file.exists() + assert existing_file.stat().st_size > 0 + + # Verify new_files does not contain this repo (no notification generated) + # This is verified indirectly: notify should be called with empty list or list not containing RepoX + # Hard to verify without capturing notify, but the skipped download is the key test + + +class TestPlaceIntoPloudSameFileIsNoop: + """Test that place_into_pcloud with same src/dst is a no-op (Fix 3 - belt and suspenders).""" + + def test_place_into_pcloud_same_file_is_noop(self, mock_paths, setup_logging): + """Call place_into_pcloud(p, p) for existing path, assert returns True and file unchanged.""" + # Create a test file + test_file = mock_paths["pcloud_target"] / "test-file.zip" + create_valid_zip(test_file, "content") + original_mtime = test_file.stat().st_mtime + + # Call with same src and dst + result = coe_mirror.place_into_pcloud(test_file, test_file) + + # Assert returns True (success) + assert result is True + + # Assert file still exists and has same content + assert test_file.exists() + assert test_file.stat().st_size > 0