n0mad1k/coe-mirror
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Mirror all four Church of Malware owners, per-owner pcloud subdirs

Browse Source 
- OWNERS list replaces single OWNER: Nightmare_Eclipse, shai_hulud, bikini_mirror, ek0mssavi0r
- PCLOUD_TARGET moves to ChurchOfMalware/ parent, per-owner subdirs at runtime
- Owner threaded through fetch_repos, fetch_branch_sha, download_repo_zip
- Pending filenames get owner__ prefix to prevent cross-owner collisions
- flush_pending strips owner prefix when routing to owner subdir
- Per-owner fetch failure logs and continues; failed_owners reported in notify
- notify groups new files by owner; also picks up flushes done at run start
- Tests: parametrize on OWNERS via monkeypatch, add multi-owner collision and
  per-owner failure isolation tests

DevTrack #1284
This commit is contained in:
n0mad1k
2026-07-06 12:35:56 -04:00
parent 8ea01c898f
commit 5912ca92d2
3 changed files with 252 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coe_mirror.py
+121 -74
View File
@@ -1,7 +1,8 @@
"""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.
Fetches all repos owned by each user in OWNERS and stores per-commit zip snapshots
under `PCLOUD_TARGET/<owner>/`. Runs unattended via systemd timer twice daily.
Local staging when pCloud unavailable.
"""
import json
@@ -24,8 +25,8 @@ 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"
OWNERS = ["Nightmare_Eclipse", "shai_hulud", "bikini_mirror", "ek0mssavi0r"]
PCLOUD_TARGET = Path.home() / "pCloudDrive" / "Hacking" / "ChurchOfMalware"
PENDING_DIR = Path.home() / ".local" / "share" / "coe-mirror" / "pending"
LOG_DIR = Path.home() / ".local" / "share" / "coe-mirror"
LOG_FILE = LOG_DIR / "coe-mirror.log"
@@ -174,9 +175,9 @@ def place_into_pcloud(src: Path, dst: Path) -> bool:
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"
def fetch_repos(session: requests.Session, owner: str) -> 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:
@@ -200,9 +201,9 @@ def fetch_repos(session: requests.Session) -> Optional[list]:
return None
def fetch_branch_sha(session: requests.Session, repo_name: str, default_branch: str) -> Optional[str]:
def fetch_branch_sha(session: requests.Session, owner: str, 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='')}"
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:
@@ -228,19 +229,24 @@ def fetch_branch_sha(session: requests.Session, repo_name: str, default_branch:
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."""
def download_repo_zip(session: requests.Session, owner: str, repo_name: str, full_sha: str) -> Optional[Path]:
"""Download repo zip for given SHA. Return Path to downloaded file (in PENDING_DIR) or None.
Pending filename is `{owner}__{repo}_{sha8}.zip` so owners with identical repo names
don't collide in the shared PENDING_DIR. flush_pending() strips the owner prefix
when routing to the per-owner pcloud subdir.
"""
# 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)
if not _safe_repo_name(repo_name) or not _safe_repo_name(owner):
logger.error("refusing unsafe owner/repo name in download_repo_zip: %r/%r", owner, repo_name)
return None
filename = f"{repo_name}_{full_sha[:SHORT_SHA_LEN]}.zip"
filename = f"{owner}__{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"
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:
@@ -291,22 +297,35 @@ def download_repo_zip(session: requests.Session, repo_name: str, full_sha: str)
def flush_pending(pcloud_healthy: bool) -> list:
"""Move all *.zip files from PENDING_DIR to PCLOUD_TARGET if healthy.
"""Move all *.zip files from PENDING_DIR to PCLOUD_TARGET/<owner>/ if healthy.
Return list of (repo, short_sha, final_location) tuples for files moved.
Pending filenames are `{owner}__{repo}_{sha8}.zip`. The owner prefix is stripped
when placing into pcloud (final name is `{repo}_{sha8}.zip` inside owner subdir).
Return list of (owner, 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
# Expect `{owner}__{repo}_{sha8}.zip`
if "__" not in pending_file.name:
logger.warning("skipping pending file without owner prefix: %s", pending_file.name)
continue
owner, rest = pending_file.name.split("__", 1)
if not _safe_repo_name(owner):
logger.warning("refusing unsafe owner prefix in pending file: %r", owner)
continue
owner_dir = PCLOUD_TARGET / owner
owner_dir.mkdir(parents=True, exist_ok=True)
dst = owner_dir / rest
if place_into_pcloud(pending_file, dst):
# Extract repo name and short sha from filename
name_parts = pending_file.stem.rsplit("_", 1)
# Extract repo name and short sha from remainder (`{repo}_{sha8}.zip`)
name_parts = Path(rest).stem.rsplit("_", 1)
if len(name_parts) == 2:
repo, short_sha = name_parts
moved.append((repo, short_sha, str(dst)))
moved.append((owner, repo, short_sha, str(dst)))
logger.info("Flushed %s to pCloud", pending_file.name)
return moved
@@ -454,15 +473,25 @@ def _send_matrix(message: str) -> bool:
return False
def notify(summary: List[Tuple[str, str, str]]) -> None:
"""Build a Matrix message and POST it. Non-fatal."""
if not summary:
def notify(summary: List[Tuple[str, str, str, str]], failed_owners: Optional[List[str]] = None) -> None:
"""Build a Matrix message and POST it. Non-fatal.
`summary` is a list of (owner, repo, short_sha, final_location).
`failed_owners` is a list of owners whose repo-list fetch failed this run.
"""
failed_owners = failed_owners or []
if not summary and not failed_owners:
return
repos = [f"{repo}@{short_sha}" for repo, short_sha, _ in summary]
repos_str = ", ".join(repos)
text = (
f"coe-mirror: {len(repos)} new from Nightmare_Eclipse: {repos_str}"
)
parts = []
if summary:
by_owner: dict = {}
for owner, repo, short_sha, _ in summary:
by_owner.setdefault(owner, []).append(f"{repo}@{short_sha}")
chunks = [f"{owner}: {', '.join(items)}" for owner, items in by_owner.items()]
parts.append(f"coe-mirror: {len(summary)} new — " + " | ".join(chunks))
if failed_owners:
parts.append(f"{len(failed_owners)} owner(s) failed: {', '.join(failed_owners)}")
text = " ; ".join(parts)
# Always log the intended message so behavior is visible even when Matrix is down
logger.info("notify: %s", text)
_send_matrix(text)
@@ -487,69 +516,87 @@ def main() -> int:
# Flush pending at start
initial_flush = flush_pending(pcloud_healthy)
# Create session and fetch repos
# Create session and iterate owners
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
new_files = list(initial_flush)
failed_owners: List[str] = []
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'")
for owner in OWNERS:
if not _safe_repo_name(owner):
logger.warning("refusing unsafe owner name: %r", owner)
failed_owners.append(owner)
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)
repos = fetch_repos(session, owner)
if repos is None:
logger.error("Could not fetch repo list for %s; continuing to next owner", owner)
failed_owners.append(owner)
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
logger.info("Fetched %d repos for %s", len(repos), owner)
# 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
# Ensure per-owner pcloud subdir exists (best-effort; probe already covered parent)
owner_pcloud = PCLOUD_TARGET / owner
if pcloud_healthy:
try:
owner_pcloud.mkdir(parents=True, exist_ok=True)
except OSError as exc:
logger.warning("could not create owner subdir %s: %s", owner_pcloud, exc)
# Process each repo for this owner
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' (owner=%s)", owner)
continue
# Reject unsafe repo names to prevent path traversal
if not _safe_repo_name(repo_name):
logger.warning("refusing unsafe repo name: %r (owner=%s)", repo_name, owner)
continue
sha = fetch_branch_sha(session, owner, repo_name, default_branch)
if sha is None:
logger.error("Skipping %s/%s (could not fetch branch SHA)", owner, repo_name)
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)))
# pcloud final name: {repo}_{sha8}.zip inside owner subdir
pcloud_final = f"{repo_name}_{short_sha}.zip"
pcloud_path = owner_pcloud / pcloud_final
# pending name carries owner prefix to avoid cross-owner collisions
pending_name = f"{owner}__{pcloud_final}"
pending_path = PENDING_DIR / pending_name
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)", pending_name)
continue
downloaded = download_repo_zip(session, owner, repo_name, sha)
if downloaded:
# Try to place it
if pcloud_healthy:
dst = owner_pcloud / pcloud_final
if place_into_pcloud(downloaded, dst):
new_files.append((owner, repo_name, short_sha, str(dst)))
else:
logger.info("File %s left in pending (placement failed)", downloaded.name)
new_files.append((owner, repo_name, short_sha, str(downloaded)))
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)))
logger.info("pCloud unhealthy; %s queued in pending", downloaded.name)
new_files.append((owner, 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)
notify(new_files, failed_owners=failed_owners)
logger.info("coe-mirror complete")
return 0