Cobra
45a8d6163e
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.
407 lines
14 KiB
Python
407 lines
14 KiB
Python
"""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())
|