|
1 | 1 | from itertools import product |
2 | 2 | import subprocess |
3 | 3 | import json |
| 4 | +import os |
| 5 | +import tempfile |
4 | 6 | import time |
5 | 7 | from typing import Any |
6 | 8 | import re |
7 | 9 | from collections import defaultdict |
8 | 10 |
|
9 | | -REGISTRY = "docker://ghcr.io/ublue-os/" |
| 11 | +REGISTRY = "ghcr.io/ublue-os/" |
10 | 12 |
|
11 | 13 | IMAGE_MATRIX_LATEST = { |
12 | 14 | "experience": ["base", "dx"], |
|
22 | 24 | RETRIES = 3 |
23 | 25 | RETRY_WAIT = 5 |
24 | 26 | FEDORA_PATTERN = re.compile(r"\.fc\d\d") |
| 27 | +EPOCH_PATTERN = re.compile(r"^\d+:") |
25 | 28 | START_PATTERN = lambda target: re.compile(rf"{target}-\d\d\d+") |
26 | 29 |
|
27 | 30 | PATTERN_ADD = "\n| ✨ | {name} | | {version} |" |
@@ -125,7 +128,7 @@ def get_manifests(target: str): |
125 | 128 | for i in range(RETRIES): |
126 | 129 | try: |
127 | 130 | output = subprocess.run( |
128 | | - ["skopeo", "inspect", REGISTRY + img + ":" + target], |
| 131 | + ["skopeo", "inspect", f"docker://{REGISTRY}{img}:{target}"], |
129 | 132 | check=True, |
130 | 133 | stdout=subprocess.PIPE, |
131 | 134 | ).stdout |
@@ -165,24 +168,104 @@ def get_tags(target: str, manifests: dict[str, Any]): |
165 | 168 | return tags[-2], tags[-1] |
166 | 169 |
|
167 | 170 |
|
168 | | -def get_packages(manifests: dict[str, Any]): |
| 171 | +def get_image_digest(image: str, tag: str) -> str: |
| 172 | + """Get image digest using skopeo.""" |
| 173 | + result = subprocess.run( |
| 174 | + ["skopeo", "inspect", f"docker://{image}:{tag}"], |
| 175 | + capture_output=True, |
| 176 | + text=True, |
| 177 | + check=True |
| 178 | + ) |
| 179 | + return json.loads(result.stdout)["Digest"] |
| 180 | + |
| 181 | + |
| 182 | +def get_sbom(image: str, digest: str) -> dict: |
| 183 | + """Fetch SBOM using ORAS.""" |
| 184 | + full_ref = f"{image}@{digest}" |
| 185 | + |
| 186 | + # Find the SBOM referrer attached to this image |
| 187 | + result = subprocess.run( |
| 188 | + ["oras", "discover", "--format", "json", full_ref], |
| 189 | + capture_output=True, |
| 190 | + text=True, |
| 191 | + check=True, |
| 192 | + ) |
| 193 | + discovered = json.loads(result.stdout) |
| 194 | + |
| 195 | + sbom_digest = None |
| 196 | + for referrer in discovered.get("referrers", []): |
| 197 | + if "spdx+json" in referrer.get("artifactType", ""): |
| 198 | + sbom_digest = referrer["digest"] |
| 199 | + break |
| 200 | + |
| 201 | + if sbom_digest is None: |
| 202 | + raise RuntimeError(f"No SBOM referrer found for {full_ref}") |
| 203 | + |
| 204 | + sbom_ref = f"{image}@{sbom_digest}" |
| 205 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 206 | + subprocess.run( |
| 207 | + ["oras", "pull", sbom_ref], |
| 208 | + capture_output=True, |
| 209 | + check=True, |
| 210 | + cwd=tmpdir, |
| 211 | + ) |
| 212 | + for fname in os.listdir(tmpdir): |
| 213 | + fpath = os.path.join(tmpdir, fname) |
| 214 | + if fname.endswith(".zst"): |
| 215 | + result = subprocess.run( |
| 216 | + ["zstd", "-d", fpath, "--stdout"], |
| 217 | + capture_output=True, |
| 218 | + check=True, |
| 219 | + ) |
| 220 | + return json.loads(result.stdout) |
| 221 | + elif fname.endswith(".json"): |
| 222 | + with open(fpath) as f: |
| 223 | + return json.load(f) |
| 224 | + |
| 225 | + raise RuntimeError(f"No SBOM file found after pulling {sbom_ref}") |
| 226 | + |
| 227 | + |
| 228 | +def parse_sbom_packages(sbom: dict) -> dict[str, str]: |
169 | 229 | packages = {} |
170 | | - for img, manifest in manifests.items(): |
| 230 | + for artifact in sbom.get("artifacts", []): |
| 231 | + # Only process RPM packages |
| 232 | + if artifact.get("type") != "rpm": |
| 233 | + continue |
| 234 | + name = artifact.get("name") |
| 235 | + version = artifact.get("version") |
| 236 | + if name and version: |
| 237 | + # If we see the same package, keep the one with epoch (more specific) |
| 238 | + if name not in packages or (":" in version and ":" not in packages[name]): |
| 239 | + packages[name] = version |
| 240 | + return packages |
| 241 | + |
| 242 | + |
| 243 | +def get_packages(target: str, images: list[tuple[str, str, str, str]]): |
| 244 | + packages = {} |
| 245 | + for j, (img, _, _, _) in enumerate(images): |
| 246 | + print(f"Getting packages for {img}:{target} via SBOM ({j+1}/{len(images)})") |
171 | 247 | try: |
172 | | - packages[img] = json.loads(manifest["Labels"]["dev.hhd.rechunk.info"])[ |
173 | | - "packages" |
174 | | - ] |
| 248 | + full_image = f"{REGISTRY}{img}" |
| 249 | + digest = get_image_digest(full_image, target) |
| 250 | + sbom = get_sbom(full_image, digest) |
| 251 | + packages[img] = parse_sbom_packages(sbom) |
| 252 | + print(f" Found {len(packages[img])} packages") |
175 | 253 | except Exception as e: |
176 | | - print(f"Failed to get packages for {img}:\n{e}") |
| 254 | + print(f" Failed to get packages for {img}:{target}: {e}") |
| 255 | + raise e |
177 | 256 | return packages |
178 | 257 |
|
179 | 258 |
|
180 | | -def get_package_groups(target: str, prev: dict[str, Any], manifests: dict[str, Any]): |
| 259 | +def get_package_groups(target: str, prev_tag: str, curr_tag: str): |
181 | 260 | common = set() |
182 | 261 | others = {k: set() for k in OTHER_NAMES.keys()} |
183 | 262 |
|
184 | | - npkg = get_packages(manifests) |
185 | | - ppkg = get_packages(prev) |
| 263 | + images = list(get_images(target)) |
| 264 | + |
| 265 | + print(f"\nFetching current packages for {curr_tag}...") |
| 266 | + npkg = get_packages(curr_tag, images) |
| 267 | + print(f"\nFetching previous packages for {prev_tag}...") |
| 268 | + ppkg = get_packages(prev_tag, images) |
186 | 269 |
|
187 | 270 | keys = set(npkg.keys()) | set(ppkg.keys()) |
188 | 271 | pkg = defaultdict(set) |
@@ -232,16 +315,16 @@ def get_package_groups(target: str, prev: dict[str, Any], manifests: dict[str, A |
232 | 315 |
|
233 | 316 | first = False |
234 | 317 |
|
235 | | - return sorted(common), {k: sorted(v) for k, v in others.items()} |
| 318 | + return sorted(common), {k: sorted(v) for k, v in others.items()}, npkg, ppkg |
236 | 319 |
|
237 | 320 |
|
238 | | -def get_versions(manifests: dict[str, Any]): |
| 321 | +def get_versions(packages: dict[str, dict[str, str]]): |
| 322 | + """Extract version info from packages dict, stripping epoch prefix and Fedora suffix.""" |
239 | 323 | versions = {} |
240 | | - pkgs = get_packages(manifests) |
241 | | - for img_pkgs in pkgs.values(): |
| 324 | + for img_pkgs in packages.values(): |
242 | 325 | for pkg, v in img_pkgs.items(): |
| 326 | + v = re.sub(EPOCH_PATTERN, "", v) |
243 | 327 | v = re.sub(FEDORA_PATTERN, "", v) |
244 | | - v = re.sub(r"\.switcheroo", "", v) |
245 | 328 | versions[pkg] = v |
246 | 329 | return versions |
247 | 330 |
|
@@ -338,14 +421,16 @@ def generate_changelog( |
338 | 421 | target: str, |
339 | 422 | pretty: str | None, |
340 | 423 | workdir: str, |
| 424 | + prev_tag: str, |
| 425 | + curr_tag: str, |
341 | 426 | prev_manifests, |
342 | 427 | manifests, |
343 | 428 | ): |
344 | | - common, others = get_package_groups(target, prev_manifests, manifests) |
345 | | - versions = get_versions(manifests) |
346 | | - prev_versions = get_versions(prev_manifests) |
| 429 | + common, others, curr_packages, prev_packages = get_package_groups(target, prev_tag, curr_tag) |
| 430 | + versions = get_versions(curr_packages) |
| 431 | + prev_versions = get_versions(prev_packages) |
347 | 432 |
|
348 | | - prev, curr = get_tags(target, manifests) |
| 433 | + prev, curr = prev_tag, curr_tag |
349 | 434 |
|
350 | 435 | if not pretty: |
351 | 436 | # Generate pretty version since we dont have it |
@@ -447,6 +532,8 @@ def main(): |
447 | 532 | target, |
448 | 533 | args.pretty, |
449 | 534 | args.workdir, |
| 535 | + prev, |
| 536 | + curr, |
450 | 537 | prev_manifests, |
451 | 538 | manifests, |
452 | 539 | ) |
|
0 commit comments