scripts/sync_sources.py hinzugefügt

scripts/sync_sources.py

@@ -0,0 +1,120 @@
+
+#!/usr/bin/env python3
+
+import json
+import os
+import re
+import subprocess
+import urllib.request
+from pathlib import Path
+
+GITEA_URL = "https://git.apex-project.de"
+SOURCES_URL = (
+    f"{GITEA_URL}/VULN/vuln-status/"
+    "raw/branch/main/sources.json"
+)
+
+CACHE_ROOT = Path(
+    os.getenv("CACHE_ROOT", "/vuln-cache/vuln-builder")
+)
+SOURCES_DIR = CACHE_ROOT / "sources"
+
+TOKEN = os.getenv("SOURCE_READ_TOKEN", "")
+REPO_PATTERN = re.compile(r"^[a-zA-Z0-9_.-]+$")
+
+
+def load_sources():
+    request = urllib.request.Request(SOURCES_URL)
+
+    if TOKEN:
+        request.add_header("Authorization", f"token {TOKEN}")
+
+    with urllib.request.urlopen(request, timeout=60) as response:
+        config = json.load(response)
+
+    if not isinstance(config.get("sources"), list):
+        raise ValueError("Ungültige sources.json")
+
+    if not REPO_PATTERN.fullmatch(config.get("owner", "")):
+        raise ValueError("Ungültiger Repository-Owner")
+
+    if config.get("branch") != "main":
+        raise ValueError("Unerwarteter Git-Branch")
+
+    return config
+
+
+def git(*args, cwd=None):
+    subprocess.run(
+        ["git", *args],
+        cwd=cwd,
+        check=True,
+        env={
+            **os.environ,
+            "GIT_TERMINAL_PROMPT": "0"
+        }
+    )
+
+
+def sync_repository(owner, repo, branch):
+    if not REPO_PATTERN.fullmatch(repo):
+        raise ValueError(f"Ungültiger Repository-Name: {repo}")
+
+    path = SOURCES_DIR / repo
+    remote = f"{GITEA_URL}/{owner}/{repo}.git"
+
+    if not (path / ".git").is_dir():
+        git(
+            "clone",
+            "--depth", "1",
+            "--branch", branch,
+            remote,
+            str(path)
+        )
+    else:
+        git("fetch", "--depth", "1", "origin", branch, cwd=path)
+        git("reset", "--hard", "FETCH_HEAD", cwd=path)
+
+    result = subprocess.check_output(
+        ["git", "rev-parse", "HEAD"],
+        cwd=path,
+        text=True
+    ).strip()
+
+    return {
+        "repo": repo,
+        "commit": result,
+        "path": str(path)
+    }
+
+
+def main():
+    config = load_sources()
+    SOURCES_DIR.mkdir(parents=True, exist_ok=True)
+
+    result = []
+
+    for source in config["sources"]:
+        repo = source["repo"]
+        print(f"Synchronisiere {repo}", flush=True)
+
+        info = sync_repository(
+            config["owner"],
+            repo,
+            config["branch"]
+        )
+
+        result.append(info)
+
+    manifest = CACHE_ROOT / "source-manifest.json"
+
+    manifest.write_text(
+        json.dumps(result, indent=2),
+        encoding="utf-8"
+    )
+
+    print(f"{len(result)} Repositories synchronisiert")
+
+
+if __name__ == "__main__":
+    main()