feat: publish custodian fleet standards batch

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a058f3-8ba0-7692-a042-9a870fc3d663

tools/fetch_sources.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 import argparse
 import hashlib
 import json
+import os
 from pathlib import Path, PurePosixPath
 import tarfile
 import tempfile
@@ -13,7 +14,43 @@ import urllib.parse
 import urllib.request
 
 
-def _revision(remote: str, branch: str) -> str:
+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:
@@ -21,7 +58,9 @@ def _revision(remote: str, branch: str) -> str:
     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(url, timeout=30) as response:
+    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")
@@ -31,7 +70,7 @@ def _revision(remote: str, branch: str) -> str:
 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[:7]}.tar.gz"
+    return f"{remote[:-4]}/archive/{revision}.tar.gz"
 
 
 def _extract(archive: Path, target: Path) -> None:
@@ -73,7 +112,14 @@ def _extract(archive: Path, target: Path) -> None:
                     output.write(chunk)
 
 
-def fetch(config_path: Path, destination: Path, policy_revision: str) -> dict[str, object]:
+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
     ):
@@ -95,10 +141,14 @@ def fetch(config_path: Path, destination: Path, policy_revision: str) -> dict[st
             continue
         remote = repository["remote"]
         branch = repository.get("branch", "main")
-        revision = _revision(remote, branch)
+        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(url, timeout=60) as response:
+            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()
@@ -127,8 +177,29 @@ def main() -> int:
     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()
-    lock = fetch(args.config.resolve(), args.destination.resolve(), args.policy_revision)
+    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(
             {