"""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())