126 lines
3.0 KiB
Python
Executable File
126 lines
3.0 KiB
Python
Executable File
import os
|
|
import subprocess
|
|
import tempfile
|
|
import hashlib
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Tuple
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
# ----------------------------
|
|
# utility: run command
|
|
# ----------------------------
|
|
def run(cmd: str, cwd: str = "/") -> Tuple[str, int]:
|
|
"""Run ``cmd`` (a shell string) and return (stdout+stderr, returncode).
|
|
|
|
Note: shell=True is intentional here — snapshot commands are
|
|
operator-authored config strings (git clone, hg clone, svn checkout),
|
|
NOT user input. The operator controls what gets run.
|
|
"""
|
|
process = subprocess.Popen(
|
|
cmd,
|
|
shell=True,
|
|
cwd=cwd,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True
|
|
)
|
|
|
|
out = []
|
|
for line in process.stdout:
|
|
out.append(line)
|
|
|
|
process.wait()
|
|
return "".join(out), process.returncode
|
|
|
|
|
|
# ----------------------------
|
|
# git (Forgejo primary)
|
|
# ----------------------------
|
|
def snapshot_git(repo_url):
|
|
workdir = tempfile.mkdtemp(prefix="fester-snap-git-")
|
|
|
|
run(f"git clone --depth 1 {repo_url} {workdir}")
|
|
|
|
return workdir
|
|
|
|
|
|
# ----------------------------
|
|
# hg adapter (snapshot only)
|
|
# ----------------------------
|
|
def snapshot_hg(repo_url):
|
|
workdir = tempfile.mkdtemp(prefix="fester-snap-hg-")
|
|
|
|
run(f"hg clone {repo_url} {workdir}")
|
|
|
|
return workdir
|
|
|
|
|
|
# ----------------------------
|
|
# svn adapter (export only)
|
|
# ----------------------------
|
|
def snapshot_svn(repo_url):
|
|
workdir = tempfile.mkdtemp(prefix="fester-snap-svn-")
|
|
|
|
run(f"svn checkout {repo_url} {workdir}")
|
|
|
|
return workdir
|
|
|
|
|
|
# ----------------------------
|
|
# cvs adapter (legacy dump)
|
|
# ----------------------------
|
|
def snapshot_cvs(repo_url):
|
|
workdir = tempfile.mkdtemp(prefix="fester-snap-cvs-")
|
|
|
|
# best-effort export only
|
|
run(f"cvs export -d {workdir} {repo_url}")
|
|
|
|
return workdir
|
|
|
|
|
|
# ----------------------------
|
|
# unified entry point
|
|
# ----------------------------
|
|
def create_snapshot(source):
|
|
kind = source["type"]
|
|
url = source["url"]
|
|
|
|
if kind == "git":
|
|
path = snapshot_git(url)
|
|
elif kind == "hg":
|
|
path = snapshot_hg(url)
|
|
elif kind == "svn":
|
|
path = snapshot_svn(url)
|
|
elif kind == "cvs":
|
|
path = snapshot_cvs(url)
|
|
else:
|
|
raise Exception(f"Unsupported VCS type: {kind}")
|
|
|
|
return fingerprint(path)
|
|
|
|
|
|
# ----------------------------
|
|
# deterministic fingerprint
|
|
# ----------------------------
|
|
def fingerprint(path: str) -> dict:
|
|
sha = hashlib.sha256()
|
|
|
|
for root, dirs, files in os.walk(path):
|
|
for f in sorted(files):
|
|
fp = os.path.join(root, f)
|
|
try:
|
|
with open(fp, "rb") as fh:
|
|
sha.update(fh.read())
|
|
except (OSError, PermissionError) as e:
|
|
# Best-effort — log and skip unreadable files
|
|
log.debug("could not read %s for fingerprint: %s", fp, e)
|
|
|
|
return {
|
|
"path": path,
|
|
"hash": sha.hexdigest(),
|
|
"timestamp": datetime.utcnow().isoformat() + "Z"
|
|
}
|