tegwick
5fbd44a703
Some checks failed
Build and publish policy-nexus image / build-and-push (push) Failing after 0s
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a058f3-8ba0-7692-a042-9a870fc3d663
216 lines
8.1 KiB
Python
216 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Fetch exact Forgejo source archives declared by the policy source inventory."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import tarfile
|
|
import tempfile
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
def _url_origin(url: str) -> str:
|
|
parsed = urllib.parse.urlparse(url)
|
|
if parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password:
|
|
raise ValueError(f"authenticated source URL must use an HTTPS origin: {url!r}")
|
|
return f"{parsed.scheme}://{parsed.netloc.lower()}"
|
|
|
|
|
|
def _credential_origin(value: str) -> str:
|
|
parsed = urllib.parse.urlparse(value)
|
|
if parsed.path not in ("", "/") or parsed.params or parsed.query or parsed.fragment:
|
|
raise ValueError("credential origin must contain only an HTTPS scheme and authority")
|
|
return _url_origin(value)
|
|
|
|
|
|
def _request(
|
|
url: str, *, token: str | None = None, token_origin: str | None = None
|
|
) -> urllib.request.Request:
|
|
if bool(token) != bool(token_origin):
|
|
raise ValueError("source token and token origin must be provided together")
|
|
request = urllib.request.Request(url)
|
|
if token:
|
|
expected_origin = _credential_origin(token_origin or "")
|
|
if _url_origin(url) != expected_origin:
|
|
raise ValueError(
|
|
f"refusing to send source credential outside {expected_origin}"
|
|
)
|
|
request.add_header("Authorization", f"token {token}")
|
|
return request
|
|
|
|
|
|
def _revision(
|
|
remote: str,
|
|
branch: str,
|
|
*,
|
|
token: str | None = None,
|
|
token_origin: str | None = None,
|
|
) -> str:
|
|
parsed = urllib.parse.urlparse(remote)
|
|
parts = parsed.path.removesuffix(".git").strip("/").split("/")
|
|
if parsed.scheme != "https" or len(parts) != 2:
|
|
raise ValueError(f"unsupported Forgejo remote {remote!r}")
|
|
owner, repo = (urllib.parse.quote(part, safe="") for part in parts)
|
|
branch_name = urllib.parse.quote(branch, safe="")
|
|
url = f"{parsed.scheme}://{parsed.netloc}/api/v1/repos/{owner}/{repo}/branches/{branch_name}"
|
|
with urllib.request.urlopen(
|
|
_request(url, token=token, token_origin=token_origin), timeout=30
|
|
) as response:
|
|
revision = json.load(response).get("commit", {}).get("id", "")
|
|
if len(revision) != 40 or any(char not in "0123456789abcdef" for char in revision):
|
|
raise ValueError(f"{remote}: could not resolve a clean 40-hex {branch} revision")
|
|
return revision
|
|
|
|
|
|
def _archive_url(remote: str, revision: str) -> str:
|
|
if not remote.startswith("https://") or not remote.endswith(".git"):
|
|
raise ValueError(f"archive source must be an HTTPS .git URL, got {remote!r}")
|
|
return f"{remote[:-4]}/archive/{revision}.tar.gz"
|
|
|
|
|
|
def _extract(archive: Path, target: Path) -> None:
|
|
target.mkdir(parents=True, exist_ok=False)
|
|
with tarfile.open(archive, "r:gz") as bundle:
|
|
members = bundle.getmembers()
|
|
roots = {
|
|
PurePosixPath(member.name).parts[0]
|
|
for member in members
|
|
if PurePosixPath(member.name).parts
|
|
}
|
|
if len(roots) != 1:
|
|
raise ValueError(f"{archive}: expected exactly one archive root")
|
|
root = next(iter(roots))
|
|
for member in members:
|
|
path = PurePosixPath(member.name)
|
|
if not path.parts or path.parts[0] != root:
|
|
raise ValueError(f"{archive}: inconsistent archive root")
|
|
relative = PurePosixPath(*path.parts[1:])
|
|
if not relative.parts:
|
|
continue
|
|
if relative.is_absolute() or ".." in relative.parts:
|
|
raise ValueError(f"{archive}: unsafe member {member.name!r}")
|
|
destination = target.joinpath(*relative.parts)
|
|
if member.isdir():
|
|
destination.mkdir(parents=True, exist_ok=True)
|
|
continue
|
|
if not member.isfile():
|
|
# Links and special files are not part of the publishable source
|
|
# corpus. Skipping them avoids materializing archive links; if a
|
|
# selector ever names one, the inventory audit fails on absence.
|
|
continue
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
source = bundle.extractfile(member)
|
|
if source is None:
|
|
raise ValueError(f"{archive}: could not read {member.name!r}")
|
|
with destination.open("wb") as output:
|
|
while chunk := source.read(1024 * 1024):
|
|
output.write(chunk)
|
|
|
|
|
|
def fetch(
|
|
config_path: Path,
|
|
destination: Path,
|
|
policy_revision: str,
|
|
*,
|
|
token: str | None = None,
|
|
token_origin: str | None = None,
|
|
) -> dict[str, object]:
|
|
if len(policy_revision) != 40 or any(
|
|
char not in "0123456789abcdef" for char in policy_revision
|
|
):
|
|
raise ValueError("--policy-revision must be a clean 40-hex Git commit")
|
|
config = json.loads(config_path.read_text(encoding="utf-8"))
|
|
if config.get("schema_version") != 1:
|
|
raise ValueError("source inventory config schema_version must be 1")
|
|
destination.mkdir(parents=True, exist_ok=True)
|
|
if any(destination.iterdir()):
|
|
raise ValueError(f"destination must be empty: {destination}")
|
|
|
|
revisions: dict[str, str] = {}
|
|
repositories: dict[str, dict[str, str]] = {}
|
|
for name, repository in sorted(config["repositories"].items()):
|
|
if repository.get("local"):
|
|
revision = policy_revision
|
|
repositories[name] = {"revision": revision, "source": "workflow-checkout"}
|
|
revisions[name] = revision
|
|
continue
|
|
remote = repository["remote"]
|
|
branch = repository.get("branch", "main")
|
|
revision = _revision(
|
|
remote, branch, token=token, token_origin=token_origin
|
|
)
|
|
url = _archive_url(remote, revision)
|
|
with tempfile.NamedTemporaryFile(suffix=".tar.gz") as archive:
|
|
with urllib.request.urlopen(
|
|
_request(url, token=token, token_origin=token_origin), timeout=60
|
|
) as response:
|
|
while chunk := response.read(1024 * 1024):
|
|
archive.write(chunk)
|
|
archive.flush()
|
|
_extract(Path(archive.name), destination / name)
|
|
revisions[name] = revision
|
|
repositories[name] = {
|
|
"revision": revision,
|
|
"remote": remote,
|
|
"branch": branch,
|
|
}
|
|
|
|
canonical = json.dumps(revisions, sort_keys=True, separators=(",", ":")).encode()
|
|
lock: dict[str, object] = {
|
|
"schema_version": 1,
|
|
"source_set_digest": hashlib.sha256(canonical).hexdigest(),
|
|
"repositories": repositories,
|
|
}
|
|
(destination / "source-lock.json").write_text(
|
|
json.dumps(lock, indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
|
)
|
|
return lock
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--config", type=Path, default=Path("source-inventory.config.json"))
|
|
parser.add_argument("--destination", type=Path, required=True)
|
|
parser.add_argument("--policy-revision", required=True)
|
|
parser.add_argument(
|
|
"--token-env",
|
|
help="environment variable containing a Forgejo repository-read token",
|
|
)
|
|
parser.add_argument(
|
|
"--token-origin",
|
|
help="sole HTTPS origin to which the source token may be sent",
|
|
)
|
|
args = parser.parse_args()
|
|
token = None
|
|
if args.token_env:
|
|
token = os.environ.get(args.token_env, "").strip()
|
|
if not token:
|
|
parser.error(f"source token environment variable {args.token_env!r} is empty")
|
|
if bool(token) != bool(args.token_origin):
|
|
parser.error("--token-env and --token-origin must be supplied together")
|
|
lock = fetch(
|
|
args.config.resolve(),
|
|
args.destination.resolve(),
|
|
args.policy_revision,
|
|
token=token,
|
|
token_origin=args.token_origin,
|
|
)
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"source_set_digest": lock["source_set_digest"],
|
|
"repositories": len(lock["repositories"]),
|
|
},
|
|
sort_keys=True,
|
|
)
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|