#!/usr/bin/env python3 import argparse import base64 import hashlib import importlib.util import json import re import sys import tempfile import time import urllib.error import urllib.parse import urllib.request from datetime import datetime, timezone from pathlib import Path import yaml spec = importlib.util.spec_from_file_location("build_slice", Path(__file__).with_name("build-slice.py")) bs = importlib.util.module_from_spec(spec) spec.loader.exec_module(bs) REGISTRY_HOST = "cr.yandex" SLICE_REGISTRY = "crpct2b2lsmao4nm0cjk" CONFIG_TYPE = "application/vnd.sarex.slice.config.v1+json" LAYER_TYPE = "application/vnd.oci.image.layer.v1.tar" MANIFEST_TYPE = "application/vnd.oci.image.manifest.v1+json" JOB_IMAGE = "${CR}/infra/slice-builder:1.0.0" class PublishError(Exception): pass def digest(data: bytes) -> str: return "sha256:" + hashlib.sha256(data).hexdigest() def list_contours(repo: Path) -> list[str]: return sorted(p.parent.name for p in repo.glob("clusters/*/kustomization.yaml")) def child_pipeline(contours: list[str], release: str, commit: str) -> str: jobs: dict = {"stages": ["slices"]} for c in contours: jobs[f"slice {c}"] = { "stage": "slices", "image": JOB_IMAGE, "tags": ["build"], "variables": {"GIT_DEPTH": "1"}, "script": [ "printf '%s' \"$SLICES_PUSHER_KEY\" | base64 -d > /tmp/slices-key.json", f"python3 ci/publish-slice.py publish --contour {c} --release {release} --commit {commit} --key-file /tmp/slices-key.json", ], "after_script": ["rm -f /tmp/slices-key.json"], } if not contours: jobs["no contours"] = {"stage": "slices", "image": JOB_IMAGE, "tags": ["build"], "script": ["echo 'no clusters/*/kustomization.yaml found'"]} return yaml.safe_dump(jobs, sort_keys=False, allow_unicode=False) def build_manifest(layer: bytes, annotations: dict[str, str]) -> tuple[bytes, bytes]: config = b"{}" manifest = { "schemaVersion": 2, "mediaType": MANIFEST_TYPE, "config": {"mediaType": CONFIG_TYPE, "digest": digest(config), "size": len(config)}, "layers": [{ "mediaType": LAYER_TYPE, "digest": digest(layer), "size": len(layer), "annotations": {"org.opencontainers.image.title": "slice.tar"}, }], "annotations": dict(sorted(annotations.items())), } return config, json.dumps(manifest, sort_keys=True, separators=(",", ":")).encode() class Registry: def __init__(self, username: str, password: str): self.base = f"https://{REGISTRY_HOST}" self.basic = "Basic " + base64.b64encode(f"{username}:{password}".encode()).decode() self.tokens: dict[str, str] = {} def _send(self, method, url, data=None, headers=None): req = urllib.request.Request(url, data=data, method=method, headers=headers or {}) try: with urllib.request.urlopen(req, timeout=60) as r: return r.status, dict(r.headers), r.read() except urllib.error.HTTPError as e: return e.code, dict(e.headers), e.read() def _token(self, challenge: str, scope: str) -> str: params = dict(re.findall(r'(\w+)="([^"]*)"', challenge)) realm = params.pop("realm") params["scope"] = scope status, _, body = self._send("GET", f"{realm}?{urllib.parse.urlencode(params)}", headers={"Authorization": self.basic}) if status != 200: raise PublishError(f"registry auth failed: HTTP {status}") payload = json.loads(body) return payload.get("token") or payload.get("access_token") def request(self, method, repo, path, data=None, headers=None): scope = f"repository:{repo}:pull,push" url = path if path.startswith("http") else f"{self.base}/v2/{repo}/{path}" headers = dict(headers or {}) if scope in self.tokens: headers["Authorization"] = f"Bearer {self.tokens[scope]}" status, h, body = self._send(method, url, data, headers) if status == 401: challenge = {k.lower(): v for k, v in h.items()}.get("www-authenticate", "") self.tokens[scope] = self._token(challenge, scope) headers["Authorization"] = f"Bearer {self.tokens[scope]}" status, h, body = self._send(method, url, data, headers) return status, {k.lower(): v for k, v in h.items()}, body def current_layer(self, repo: str) -> str | None: status, _, body = self.request("GET", repo, "manifests/current", headers={"Accept": MANIFEST_TYPE}) if status == 404: return None if status != 200: raise PublishError(f"{repo}:current: HTTP {status}") return json.loads(body)["layers"][0]["digest"] def put_blob(self, repo: str, data: bytes) -> bool: d = digest(data) if self.request("HEAD", repo, f"blobs/{d}")[0] == 200: return False status, h, _ = self.request("POST", repo, "blobs/uploads/") if status != 202: raise PublishError(f"blob upload to {repo}: HTTP {status}") location = h["location"] if location.startswith("/"): location = self.base + location sep = "&" if "?" in location else "?" status, _, _ = self.request("PUT", repo, f"{location}{sep}digest={d}", data, {"Content-Type": "application/octet-stream"}) if status != 201: raise PublishError(f"blob upload to {repo}: HTTP {status}") return True def push(self, repo: str, tags: list[str], layer: bytes, annotations: dict[str, str]) -> bool: config, manifest = build_manifest(layer, annotations) self.put_blob(repo, config) uploaded = self.put_blob(repo, layer) for tag in tags: status, _, body = self.request("PUT", repo, f"manifests/{tag}", manifest, {"Content-Type": MANIFEST_TYPE}) if status != 201: raise PublishError(f"{repo}:{tag}: HTTP {status} {body[:200]!r}") return uploaded def log(msg: str): print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True) def publish(repo: Path, contour: str, release: str, commit: str, key_file: Path) -> int: target = f"{SLICE_REGISTRY}/{contour}-iac" log(f"contour {contour}, source iac@{commit[:8]}, release {release}") try: with tempfile.TemporaryDirectory() as tmp: out = Path(tmp) / "slice.tar" files, entries = bs.collect(repo, f"clusters/{contour}") log(f"collected {len(files)} files, entry points: {', '.join(entries)}") bs.write_tar(repo, files, out) bs.verify(repo, out, entries) log(f"verified: kustomize build of {len(entries)} entry point(s) identical to full repository") layer = out.read_bytes() new = digest(layer) log(f"slice size {len(layer)} bytes, digest {new}") registry = Registry("json_key", key_file.read_text()) current = registry.current_layer(f"{target}") changed = current != new log(f"registry {REGISTRY_HOST}/{target}, current " + (current or "absent") + (", content changed" if changed else ", content unchanged")) tags = [commit[:8], release] + (["current"] if changed else []) uploaded = registry.push(target, tags, layer, { "org.opencontainers.image.source": "iac", "org.opencontainers.image.revision": commit, "org.opencontainers.image.version": release, "org.opencontainers.image.created": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "io.sarex.slice.contour": contour, "io.sarex.slice.kind": "iac", }) log(f"layer {'uploaded' if uploaded else 'already in registry'}, tags pushed: {', '.join(tags)}") except (bs.SliceError, PublishError) as e: log(f"FAILED: {e}") return 1 log("done") return 0 def main() -> int: ap = argparse.ArgumentParser() sub = ap.add_subparsers(dest="cmd", required=True) gen = sub.add_parser("generate") gen.add_argument("--repo", default=".") gen.add_argument("--release", required=True) gen.add_argument("--commit", required=True) gen.add_argument("--out", required=True) pub = sub.add_parser("publish") pub.add_argument("--repo", default=".") pub.add_argument("--contour", required=True) pub.add_argument("--release", required=True) pub.add_argument("--commit", required=True) pub.add_argument("--key-file", required=True) args = ap.parse_args() repo = Path(args.repo).resolve() if args.cmd == "generate": contours = list_contours(repo) Path(args.out).write_text(child_pipeline(contours, args.release, args.commit)) log(f"{len(contours)} contours: {', '.join(contours)}") return 0 return publish(repo, args.contour, args.release, args.commit, Path(args.key_file)) if __name__ == "__main__": sys.exit(main())