From 0c3c9cd5843cdc6b140c26457129e2195c7e725e Mon Sep 17 00:00:00 2001 From: Ronny Grobel Date: Wed, 11 Nov 2026 22:18:47 +0100 Subject: [PATCH] =?UTF-8?q?scripts/sync=5Fsources.py=20hinzugef=C3=BCgt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/sync_sources.py | 120 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 120 insertions(+) create mode 100644 scripts/sync_sources.py diff --git a/scripts/sync_sources.py b/scripts/sync_sources.py new file mode 100644 index 0000000..5d2213a --- /dev/null +++ b/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()