mirror_daemon.py

#!/usr/bin/env python3
"""
Bidirectional GitHub-GitLab mirror daemon.

Runs via cron to ensure repository integrity between GitHub and GitLab.
Detects divergence and halts mirroring when conflicts are found.
Auto-creates missing repos on either platform.

Install path: /var/tmp/
Required envvars: GITHUB_TOKEN, GITLAB_TOKEN
Cron example: */5 * * * * cd /var/tmp && python3 mirror_daemon.py
"""

import json
import logging
import os
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

import yaml

CACHE_DIR = Path("/var/tmp/mirror-cache")
CONFIG_PATH = Path("/var/tmp/repos.yaml")

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
log = logging.getLogger("mirror-daemon")


def run(cmd, cwd=None):
    """Run a command (list form) and return stdout. Raises on failure."""
    result = subprocess.run(
        cmd, cwd=cwd, capture_output=True, text=True, check=False,
    )
    if result.returncode != 0:
        raise RuntimeError(
            f"Command failed: {cmd!r}\nstderr: {result.stderr.strip()}"
        )
    return result.stdout.strip()


def github_url(repo, token):
    return f"https://{token}@github.com/{repo}.git"


def gitlab_url(repo, token):
    return f"https://oauth2:{token}@gitlab.com/{repo}.git"


# ---------------------------------------------------------------------------
# Repo creation via REST APIs (stdlib only, no extra dependencies)
# ---------------------------------------------------------------------------

def _api_request(url, token, data=None, method="GET", token_header="Authorization",
                 token_prefix="token "):
    """Make an API request. Returns (status_code, response_body_dict)."""
    headers = {
        token_header: f"{token_prefix}{token}",
        "Content-Type": "application/json",
    }
    body = json.dumps(data).encode() if data else None
    req = urllib.request.Request(url, data=body, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req) as resp:
            return resp.status, json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        return e.code, {}


def github_repo_exists(repo_path, token):
    """Check if a GitHub repo exists."""
    url = f"https://api.github.com/repos/{repo_path}"
    status, _ = _api_request(url, token)
    return status == 200


def create_github_repo(repo_path, token):
    """Create a private GitHub repo. Tries org endpoint first, then user."""
    owner, name = repo_path.split("/", 1)
    data = {"name": name, "private": True}

    # Try as org repo
    url = f"https://api.github.com/orgs/{owner}/repos"
    status, _ = _api_request(url, token, data=data, method="POST")
    if status == 201:
        log.info("Created GitHub repo %s (org)", repo_path)
        return

    # Fall back to user repo
    url = "https://api.github.com/user/repos"
    status, body = _api_request(url, token, data=data, method="POST")
    if status == 201:
        log.info("Created GitHub repo %s (user)", repo_path)
        return

    raise RuntimeError(f"Failed to create GitHub repo {repo_path}: HTTP {status}")


def gitlab_repo_exists(repo_path, token):
    """Check if a GitLab repo exists."""
    encoded = urllib.parse.quote(repo_path, safe="")
    url = f"https://gitlab.com/api/v4/projects/{encoded}"
    status, _ = _api_request(url, token, token_header="PRIVATE-TOKEN", token_prefix="")
    return status == 200


def create_gitlab_repo(repo_path, token):
    """Create a private GitLab repo under the appropriate namespace."""
    owner, name = repo_path.split("/", 1)

    # Look up namespace ID
    url = f"https://gitlab.com/api/v4/namespaces?search={urllib.parse.quote(owner)}"
    status, namespaces = _api_request(
        url, token, token_header="PRIVATE-TOKEN", token_prefix="",
    )
    namespace_id = None
    if status == 200 and isinstance(namespaces, list):
        for ns in namespaces:
            if ns.get("path") == owner or ns.get("full_path") == owner:
                namespace_id = ns["id"]
                break

    data = {"name": name, "visibility": "private"}
    if namespace_id:
        data["namespace_id"] = namespace_id

    url = "https://gitlab.com/api/v4/projects"
    status, body = _api_request(
        url, token, data=data, method="POST",
        token_header="PRIVATE-TOKEN", token_prefix="",
    )
    if status == 201:
        log.info("Created GitLab repo %s", repo_path)
        return

    raise RuntimeError(
        f"Failed to create GitLab repo {repo_path}: HTTP {status} {body}"
    )


def ensure_repo_exists(platform, repo_path, token):
    """Ensure a repo exists on the given platform, creating it if needed."""
    if platform == "github":
        if not github_repo_exists(repo_path, token):
            log.info("GitHub repo %s not found, creating...", repo_path)
            create_github_repo(repo_path, token)
    elif platform == "gitlab":
        if not gitlab_repo_exists(repo_path, token):
            log.info("GitLab repo %s not found, creating...", repo_path)
            create_gitlab_repo(repo_path, token)


# ---------------------------------------------------------------------------
# Git operations
# ---------------------------------------------------------------------------

def ensure_clone(repo_id, gh_url, gl_url, cache_dir):
    """Ensure a bare clone exists with both remotes using namespaced fetch refspecs.

    Both remotes store all their refs under refs/remotes/<name>/ so we can
    cleanly compare branches and tags from each side without collisions.
    """
    path = cache_dir / repo_id
    if not path.exists():
        log.info("Creating bare repo for %s", repo_id)
        run(["git", "init", "--bare", str(path)])
        run(["git", "remote", "add", "origin", gh_url], cwd=str(path))
        run(["git", "remote", "add", "gitlab", gl_url], cwd=str(path))
        # Configure fetch refspecs so all refs are namespaced per remote
        for remote in ("origin", "gitlab"):
            run(
                ["git", "config", "--replace-all",
                 f"remote.{remote}.fetch",
                 f"+refs/heads/*:refs/remotes/{remote}/heads/*"],
                cwd=str(path),
            )
            run(
                ["git", "config", "--add",
                 f"remote.{remote}.fetch",
                 f"+refs/tags/*:refs/remotes/{remote}/tags/*"],
                cwd=str(path),
            )
    else:
        # Update remote URLs in case tokens changed
        run(["git", "remote", "set-url", "origin", gh_url], cwd=str(path))
        run(["git", "remote", "set-url", "gitlab", gl_url], cwd=str(path))
    return path


def fetch_remote(path, remote):
    """Fetch a single remote. Returns True on success, False if the remote is empty."""
    result = subprocess.run(
        ["git", "fetch", remote],
        cwd=str(path), capture_output=True, text=True, check=False,
    )
    if result.returncode != 0:
        # An empty repo returns error but that's OK — no refs to fetch
        if "no matching remote head" in result.stderr.lower():
            log.info("Remote %s exists but is empty", remote)
            return True
        raise RuntimeError(
            f"Failed to fetch {remote}: {result.stderr.strip()}"
        )
    return True


def get_refs(path, remote):
    """Get all branch and tag refs for a remote.

    Returns dict mapping canonical ref name (e.g. 'heads/main', 'tags/v1.0')
    to commit hash.
    """
    prefix = f"refs/remotes/{remote}/"
    output = run(
        ["git", "for-each-ref", "--format=%(refname) %(objectname)", prefix],
        cwd=str(path),
    )
    refs = {}
    for line in output.splitlines():
        if not line.strip():
            continue
        refname, sha = line.split()
        # Strip remote prefix: refs/remotes/origin/heads/main -> heads/main
        short = refname[len(prefix):]
        refs[short] = sha
    return refs


def is_ancestor(path, ancestor, descendant):
    """Check if ancestor is an ancestor of descendant."""
    result = subprocess.run(
        ["git", "merge-base", "--is-ancestor", ancestor, descendant],
        cwd=str(path), capture_output=True, text=True, check=False,
    )
    return result.returncode == 0


def sync_repo(repo, gh_token, gl_token):
    """Sync a single repo pair. Returns True if OK, False if diverged."""
    repo_id = repo["github"].replace("/", "_")
    gh_url = github_url(repo["github"], gh_token)
    gl_url = gitlab_url(repo["gitlab"], gl_token)

    log.info("Syncing %s <-> %s", repo["github"], repo["gitlab"])

    # Ensure repos exist on both platforms
    ensure_repo_exists("github", repo["github"], gh_token)
    ensure_repo_exists("gitlab", repo["gitlab"], gl_token)

    path = ensure_clone(repo_id, gh_url, gl_url, CACHE_DIR)

    # Fetch each remote individually
    fetch_remote(path, "origin")
    fetch_remote(path, "gitlab")

    # Get refs from both remotes (keys like 'heads/main', 'tags/v1.0')
    gh_refs = get_refs(path, "origin")
    gl_refs = get_refs(path, "gitlab")

    all_ref_names = sorted(set(gh_refs.keys()) | set(gl_refs.keys()))

    push_to_gitlab = []
    push_to_github = []
    diverged = False

    for ref in all_ref_names:
        gh_sha = gh_refs.get(ref)
        gl_sha = gl_refs.get(ref)

        # Build the correct destination ref: heads/main -> refs/heads/main
        dest_ref = f"refs/{ref}"

        if gh_sha and not gl_sha:
            log.info("  %s: only on GitHub, pushing to GitLab", ref)
            push_to_gitlab.append(
                f"refs/remotes/origin/{ref}:{dest_ref}"
            )
        elif gl_sha and not gh_sha:
            log.info("  %s: only on GitLab, pushing to GitHub", ref)
            push_to_github.append(
                f"refs/remotes/gitlab/{ref}:{dest_ref}"
            )
        elif gh_sha == gl_sha:
            continue
        elif is_ancestor(path, gl_sha, gh_sha):
            log.info("  %s: GitHub ahead, pushing to GitLab", ref)
            push_to_gitlab.append(
                f"refs/remotes/origin/{ref}:{dest_ref}"
            )
        elif is_ancestor(path, gh_sha, gl_sha):
            log.info("  %s: GitLab ahead, pushing to GitHub", ref)
            push_to_github.append(
                f"refs/remotes/gitlab/{ref}:{dest_ref}"
            )
        else:
            log.error(
                "  DIVERGENCE on %s: GitHub=%s GitLab=%s",
                ref, gh_sha, gl_sha,
            )
            diverged = True

    if diverged:
        log.error(
            "HALTING sync for %s <-> %s due to divergence. Manual repair required.",
            repo["github"], repo["gitlab"],
        )
        return False

    if push_to_gitlab:
        log.info("Pushing %d ref(s) to GitLab", len(push_to_gitlab))
        run(
            ["git", "push", "--atomic", "gitlab"] + push_to_gitlab,
            cwd=str(path),
        )

    if push_to_github:
        log.info("Pushing %d ref(s) to GitHub", len(push_to_github))
        run(
            ["git", "push", "--atomic", "origin"] + push_to_github,
            cwd=str(path),
        )

    if not push_to_gitlab and not push_to_github:
        log.info("Already in sync")

    return True


def main():
    gh_token = os.environ.get("GITHUB_TOKEN")
    gl_token = os.environ.get("GITLAB_TOKEN")

    if not gh_token or not gl_token:
        log.error("GITHUB_TOKEN and GITLAB_TOKEN envvars are required")
        sys.exit(1)

    CACHE_DIR.mkdir(parents=True, exist_ok=True)

    with open(CONFIG_PATH) as f:
        config = yaml.safe_load(f)

    all_ok = True
    for repo in config["repos"]:
        try:
            ok = sync_repo(repo, gh_token, gl_token)
            if not ok:
                all_ok = False
        except Exception:
            log.exception("Error syncing %s", repo)
            all_ok = False

    if not all_ok:
        sys.exit(1)


if __name__ == "__main__":
    main()