#!/usr/bin/env python3 import argparse import base64 import hashlib import importlib.util import json import os import re import sys import tempfile 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" INSTALLER_GITEA_URL = "http://gitea.gitea.svc:3000/infra/iac.git" 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" class PublishError(Exception): pass def digest(data: bytes) -> str: return "sha256:" + hashlib.sha256(data).hexdigest() def connected_contours(repo: Path) -> list[str]: found = [] for sync in sorted(repo.glob("clusters/*/flux-system/gotk-sync.yaml")): for doc in yaml.safe_load_all(sync.read_text()): if isinstance(doc, dict) and doc.get("kind") == "GitRepository" and doc.get("spec", {}).get("url") == INSTALLER_GITEA_URL: found.append(sync.parents[1].name) return found 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"авторизация в реестре: 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): d = digest(data) if self.request("HEAD", repo, f"blobs/{d}")[0] == 200: return status, h, _ = self.request("POST", repo, "blobs/uploads/") if status != 202: raise PublishError(f"загрузка блоба в {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"загрузка блоба в {repo}: HTTP {status}") def push(self, repo: str, tags: list[str], layer: bytes, annotations: dict[str, str]): config, manifest = build_manifest(layer, annotations) self.put_blob(repo, config) 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}") def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--repo", default=".") ap.add_argument("--release", required=True) ap.add_argument("--commit", required=True) ap.add_argument("--branch", default="") ap.add_argument("--key-file", required=True) args = ap.parse_args() repo = Path(args.repo).resolve() registry = Registry("json_key", Path(args.key_file).read_text()) contours = connected_contours(repo) print(f"подключённые контуры: {', '.join(contours) or 'нет'}") failed = [] for contour in contours: target = f"{SLICE_REGISTRY}/{contour}-iac" try: with tempfile.TemporaryDirectory() as tmp: out = Path(tmp) / "slice.tar" files, entries = bs.collect(repo, f"clusters/{contour}") bs.write_tar(repo, files, out) bs.verify(repo, out, entries) layer = out.read_bytes() changed = registry.current_layer(target) != digest(layer) tags = [args.commit[:8], args.release] + (["current"] if changed else []) registry.push(target, tags, layer, { "org.opencontainers.image.source": "iac", "org.opencontainers.image.revision": args.commit, "org.opencontainers.image.version": args.release, "org.opencontainers.image.created": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "io.sarex.slice.branch": args.branch, "io.sarex.slice.contour": contour, "io.sarex.slice.kind": "iac", }) state = "current переставлен" if changed else "содержимое как в current" print(f"{target}: {len(files)} файлов, {digest(layer)[:19]}, теги {', '.join(tags)}, {state}") except (bs.SliceError, PublishError) as e: print(f"{target}: ОШИБКА {e}", file=sys.stderr) failed.append(contour) return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())