121 lines
2.6 KiB
Python
121 lines
2.6 KiB
Python
|
|
#!/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()
|