From 9eb7eb2e5d13592b7a02bd10724cde0995eb94de Mon Sep 17 00:00:00 2001 From: yukkop Date: Wed, 9 Sep 2026 23:35:45 +0000 Subject: [PATCH] fix!: attic chaces with hatzner limites --- infra/attic-migration/.gitignore | 2 + infra/attic-migration/README.md | 125 ++++ infra/attic-migration/default.nix | 48 ++ infra/attic-migration/repack.py | 1022 ++++++++++++++++++++++++++ infra/attic-migration/test_repack.py | 474 ++++++++++++ nixos/system/hectic-lab/attic.nix | 81 +- 6 files changed, 1750 insertions(+), 2 deletions(-) create mode 100644 infra/attic-migration/.gitignore create mode 100644 infra/attic-migration/README.md create mode 100644 infra/attic-migration/default.nix create mode 100644 infra/attic-migration/repack.py create mode 100644 infra/attic-migration/test_repack.py diff --git a/infra/attic-migration/.gitignore b/infra/attic-migration/.gitignore new file mode 100644 index 00000000..43ae0e2a --- /dev/null +++ b/infra/attic-migration/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.py[cod] diff --git a/infra/attic-migration/README.md b/infra/attic-migration/README.md new file mode 100644 index 00000000..ee4958b3 --- /dev/null +++ b/infra/attic-migration/README.md @@ -0,0 +1,125 @@ +# Attic repack migration helper + +Local-only operator tool for safe resumable Attic cache repack/migration. Parent automation starts old/new services, supplies secrets, seeds spool, and runs this CLI. + +## Deployment layout + +- Original backend: `atticd`, port 8081, `/var/lib/atticd/server.db`, bucket + `cache-hectic-lab` in HEL1. +- During the write freeze and after cutover the original backend runs in + `api-server` mode, without its garbage collector, to preserve the comparison + dataset. Public write methods remain blocked by nginx. +- Repacked backend: `atticd-repacked`, port 8082, + `/var/lib/atticd-repacked/server.db`, bucket `nix-cache-hectic-lab` in HEL1. +- Both use the same `hectic` signing key and existing JWT verification secret; + clients do not need a new trusted public key or token. +- New chunk settings: threshold/minimum 1 MiB, average 2 MiB, maximum 4 MiB. +- `https://cache.hectic-lab.com/next/hectic` selects the new backend. +- `https://cache.hectic-lab.com/previous/hectic` selects the original backend; + nginx permits GET/HEAD only there. +- `repackedActive` in `nixos/system/hectic-lab/attic.nix` selects which backend + owns the original `/hectic` URL. Keep it false until all cutover gates pass. + +## Cutover and rollback gates + +1. Finish all migration partitions, then run an unfiltered migration/delta pass. +2. Confirm no CI writers remain. Set `migrationWriteFreeze = true` while + `repackedActive = false`, apply the small NixOS change, and briefly stop the + original Attic to drain/cancel any prior in-flight writes. +3. Take a SQLite backup with SQLite's backup API, not a raw live-file copy. + Keep backups and manifests under private `/var/lib/attic-repack`; the SQLite + backup includes the cache's private signing key. +4. Restart the original backend for reads only, refresh the complete inventory, + migrate any final delta, then run unfiltered `verify`. Its exit status must be + zero; independently compare old/new store-path, NAR hash, size and metadata + inventories. `status` alone is not a cutover certificate. +5. Pin the old/staging NixOS generation as a GC root, set `repackedActive = true`, + build, inspect dry activation, and switch. `/hectic` now reaches the new + backend; old data and `/previous/hectic` remain available. +6. Test public reads, signatures, and an authenticated upload at the original + URL. Do not remove the old bucket or database as part of this procedure. + +Rollback reapplies the pinned staging generation. The new backend and its data +must remain preserved: paths first uploaded after cutover may exist only there. +When editing the flags manually, clear `migrationWriteFreeze` explicitly if +writes to the original backend are intended after rollback. + +## Throughput comparison + +Compare the same store-path hashes at `/next/hectic` and `/previous/hectic` with +the same request concurrency. For example, fetch +`https://cache.hectic-lab.com/next/hectic/nar/.nar` with +`curl --fail --location --output /dev/null --write-out 'bytes=%{size_download} seconds=%{time_total}\n'`. +Do not print effective redirect URLs: S3 redirects contain temporary signatures. +Compare wall time and error rate as well as bytes/second because compressed sizes +can differ after rechunking. Do not use a build with source fallback as a pure +cache throughput measurement. The two endpoints share the VPS and nginx, so run +the comparison sequentially or account for shared-resource contention. + +## Spool/state convention + +Default state dir: `/var/lib/attic-repack` (`0700`). Raw NAR spool path: + +```text +/var/lib/attic-repack/raw/{sha256hex}.nar +``` + +Parent may seed this file directly. Tool always verifies SHA-256 and byte length before upload. Checkpoints live under `checkpoints/{store_path_hash}.json` and contain no keypair/token. + +## Commands + +```sh +attic-repack init \ + --old-db file:/var/lib/atticd/server.db?mode=ro \ + --old-url http://127.0.0.1:8081 \ + --new-url http://127.0.0.1:8082 \ + --host cache.hectic-lab.com \ + --atticadm /run/current-system/sw/bin/atticadm \ + --server-config /etc/atticd/server.toml + +attic-repack inventory --state-dir /var/lib/attic-repack > inventory.json +attic-repack status --state-dir /var/lib/attic-repack +attic-repack migrate --state-dir /var/lib/attic-repack --workers 2 --limit 20 +attic-repack verify --state-dir /var/lib/attic-repack --workers 2 +``` + +`ATTIC_MIGRATION_TOKEN` may be set for manual/tests. Otherwise token is minted in memory with `atticadm make-token` for hectic pull/push/create-cache/configure-cache. Token/keypair are never printed. + +## Inventory JSON + +`inventory` writes `attic-repack-inventory-v1`: + +```json +{ + "format": "attic-repack-inventory-v1", + "cache": "hectic", + "spool_dir": "/var/lib/attic-repack/raw", + "raw_nar_filename": "{sha256hex}.nar", + "records": [ + {"nar_hash":"sha256:...","nar_size":123,"store_path":"/nix/store/...","metadata_fingerprint":"..."} + ] +} +``` + +Records also include upload metadata: `store_path_hash`, `references`, `system`, `deriver`, `sigs`, `ca`. + +## Safety + +- Checkpoints and `status` are progress information, not a final cutover proof. + After stopping old writers and taking a consistent snapshot, run an unfiltered + `verify` (no `--paths-file` or `--limit`) to reread every new NAR and reconcile + all paths, metadata, hashes, and sizes before switching the primary endpoint. +- A local store path can differ from the historical cached NAR. Such a local + copy is rejected and recovered from the original S3 chunks instead. +- Old DB is opened readonly; old SQL NAR/chunk tables are never copied. +- Missing local raw NARs are reconstructed from old S3 chunkrefs with per-object retries and chunk/full hash checks. +- Upload uses Attic `PUT /_api/v1/upload-path` with JSON preamble plus raw uncompressed NAR. +- New cache verification compares immutable metadata against old rendered narinfo and reads/decompresses one payload per verified path invocation. +- Authenticated HTTP is refused unless URL host is loopback. + +## Local build/test + +```sh +nix build --option eval-cache false --impure --expr "let flake = builtins.getFlake \"git+file://$PWD\"; pkgs = import flake.inputs.nixpkgs { system = builtins.currentSystem; }; in pkgs.callPackage ./infra/attic-migration {}" +nix build --option eval-cache false --impure --expr "let flake = builtins.getFlake \"git+file://$PWD\"; pkgs = import flake.inputs.nixpkgs { system = builtins.currentSystem; }; p = pkgs.callPackage ./infra/attic-migration {}; in p.passthru.tests.unittest" +``` diff --git a/infra/attic-migration/default.nix b/infra/attic-migration/default.nix new file mode 100644 index 00000000..3429e96b --- /dev/null +++ b/infra/attic-migration/default.nix @@ -0,0 +1,48 @@ +{ pkgs }: + +let + source = pkgs.lib.cleanSourceWith { + src = ./.; + filter = path: type: + builtins.baseNameOf path != "__pycache__" + && !(pkgs.lib.hasSuffix ".pyc" path); + }; + pythonEnv = pkgs.python3.withPackages (ps: [ + ps.requests + ps.boto3 + ps.zstandard + ]); +in +pkgs.stdenv.mkDerivation { + pname = "attic-repack"; + version = "0.1.0"; + src = source; + + nativeBuildInputs = [ pkgs.makeWrapper ]; + + installPhase = '' + mkdir -p $out/bin $out/libexec/attic-repack + cp $src/repack.py $out/libexec/attic-repack/repack.py + chmod +x $out/libexec/attic-repack/repack.py + makeWrapper ${pythonEnv}/bin/python3 $out/bin/attic-repack \ + --add-flags $out/libexec/attic-repack/repack.py \ + --prefix PATH : ${pkgs.lib.makeBinPath [ pkgs.nix ]} + ''; + + doCheck = true; + checkPhase = '' + ${pythonEnv}/bin/python3 -m unittest discover -s $src -p 'test_*.py' + ''; + + passthru = { + inherit pythonEnv; + tests.unittest = pkgs.runCommand "attic-repack-unittest" { + nativeBuildInputs = [ pythonEnv pkgs.nix ]; + } '' + cp -r ${source} ./src + chmod -R u+w ./src + ${pythonEnv}/bin/python3 -m unittest discover -s ./src -p 'test_*.py' + mkdir -p $out + ''; + }; +} diff --git a/infra/attic-migration/repack.py b/infra/attic-migration/repack.py new file mode 100644 index 00000000..8c6ab9f5 --- /dev/null +++ b/infra/attic-migration/repack.py @@ -0,0 +1,1022 @@ +#!/usr/bin/env python3 +# pyright: reportMissingImports=false, reportMissingModuleSource=false +"""Safe local Attic cache repack/migration helper. + +Remote deployment, systemd service management, and secret provisioning are owned +outside this tool. This CLI only reads old Attic metadata/chunks, consumes or +creates raw NAR spool files, uploads through Attic HTTP API, and verifies the new +cache before marking checkpoint entries complete. +""" + +from __future__ import annotations + +import argparse +import concurrent.futures +from collections import deque +import contextlib +import gzip +import hashlib +import itertools +import io +import json +import lzma +import os +import pathlib +import queue +import sqlite3 +import subprocess +import sys +import tempfile +import threading +import time +import urllib.parse +from typing import Any, BinaryIO, Iterator + + +DEFAULT_OLD_DB = "file:/var/lib/atticd/server.db?mode=ro" +DEFAULT_STATE_DIR = "/var/lib/attic-repack" +DEFAULT_OLD_URL = "http://127.0.0.1:8081" +DEFAULT_NEW_URL = "http://127.0.0.1:8082" +DEFAULT_HOST = "cache.hectic-lab.com" +DEFAULT_CACHE = "hectic" +DEFAULT_OLD_BUCKET = "cache-hectic-lab" +DEFAULT_OLD_REGION = "hel1" +DEFAULT_OLD_ENDPOINT = "https://hel1.your-objectstorage.com" +IMMUTABLE_FIELDS = ("StorePath", "NarHash", "NarSize", "References", "Deriver", "System", "CA", "Sig") +TOKEN_ENV = "ATTIC_MIGRATION_TOKEN" +CHUNK = 1024 * 1024 + + +class RepackError(RuntimeError): + """Expected operational failure with sanitized message.""" + + +def eprint(*args: object) -> None: + print(*args, file=sys.stderr, flush=True) + + +def now() -> float: + return time.time() + + +def require_private(path: pathlib.Path, directory: bool) -> None: + mode = 0o700 if directory else 0o600 + if directory: + path.mkdir(parents=True, exist_ok=True, mode=mode) + else: + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + path.touch(mode=mode, exist_ok=True) + os.chmod(path, mode) + + +@contextlib.contextmanager +def private_umask() -> Iterator[None]: + old = os.umask(0o077) + try: + yield + finally: + os.umask(old) + + +def atomic_write(path: pathlib.Path, data: bytes) -> None: + require_private(path.parent, True) + with private_umask(): + fd, name = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent)) + try: + with os.fdopen(fd, "wb") as handle: + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + os.chmod(name, 0o600) + os.replace(name, path) + finally: + with contextlib.suppress(FileNotFoundError): + os.unlink(name) + + +def atomic_json(path: pathlib.Path, value: Any) -> None: + atomic_write(path, (json.dumps(value, sort_keys=True, indent=2) + "\n").encode()) + + +def load_json(path: pathlib.Path, default: Any) -> Any: + try: + return json.loads(path.read_text()) + except FileNotFoundError: + return default + + +def sha256_file(path: pathlib.Path) -> tuple[str, int]: + h = hashlib.sha256() + size = 0 + with path.open("rb") as handle: + while True: + data = handle.read(CHUNK) + if not data: + break + h.update(data) + size += len(data) + return h.hexdigest(), size + + +def sha256_bytes(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def nar_hash_hex(nar_hash: str) -> str: + if not nar_hash.startswith("sha256:"): + raise RepackError("bad nar_hash scheme") + value = nar_hash.split(":", 1)[1] + if len(value) != 64 or any(c not in "0123456789abcdef" for c in value.lower()): + raise RepackError("bad nar_hash hex") + return value.lower() + + +def store_hash(store_path: str) -> str: + name = pathlib.PurePosixPath(store_path).name + if "-" not in name: + raise RepackError(f"bad store path: {store_path}") + return name.split("-", 1)[0] + + +def json_list(value: Any) -> list[str]: + if value in (None, ""): + return [] + if isinstance(value, list): + return [str(v) for v in value] + return [str(v) for v in json.loads(value)] + + +def retention_from_db(value: Any) -> Any: + if value in (None, "", "Global", "global"): + return "Global" + if isinstance(value, int): + return {"Period": value} + with contextlib.suppress(Exception): + loaded = json.loads(value) + if loaded in (None, "Global"): + return "Global" + if isinstance(loaded, int): + return {"Period": loaded} + return loaded + return {"Period": int(value)} + + +def upload_metadata(record: dict[str, Any]) -> dict[str, Any]: + return { + "cache": record["cache"], + "store_path_hash": record["store_path_hash"], + "store_path": record["store_path"], + "references": sorted(record.get("references") or []), + "system": record.get("system"), + "deriver": record.get("deriver"), + "sigs": sorted(record.get("sigs") or []), + "ca": record.get("ca"), + "nar_hash": record["nar_hash"], + "nar_size": int(record["nar_size"]), + } + + +def metadata_fingerprint(record: dict[str, Any]) -> str: + return sha256_bytes(json.dumps(upload_metadata(record), sort_keys=True, separators=(",", ":")).encode()) + + +def parse_narinfo(data: bytes | str) -> dict[str, Any]: + text = data.decode() if isinstance(data, bytes) else data + result: dict[str, Any] = {"Sig": []} + refs: list[str] | None = None + for line in text.splitlines(): + if not line or ": " not in line: + continue + key, value = line.split(": ", 1) + if key == "References": + refs = [v for v in value.split() if v] + elif key == "Sig": + result.setdefault("Sig", []).append(value) + else: + result[key] = value + result["References"] = sorted(refs or []) + result["Sig"] = sorted(result.get("Sig", [])) + return result + + +def expected_narinfo(record: dict[str, Any]) -> dict[str, Any]: + meta = upload_metadata(record) + expect = { + "StorePath": meta["store_path"], + "NarHash": meta["nar_hash"], + "NarSize": str(meta["nar_size"]), + "References": sorted(meta["references"]), + "Sig": sorted(meta["sigs"]), + } + optional = (("Deriver", meta.get("deriver")), ("System", meta.get("system")), ("CA", meta.get("ca"))) + for key, value in optional: + if value not in (None, ""): + expect[key] = str(value) + return expect + + +def normalize_narinfo_for_compare(narinfo: dict[str, Any]) -> dict[str, Any]: + out = dict(narinfo) + refs = out.get("References") or [] + out["References"] = sorted(store_basename(v) for v in refs) + if out.get("Deriver"): + out["Deriver"] = store_basename(str(out["Deriver"])) + out["Sig"] = sorted(out.get("Sig") or []) + return out + + +def store_basename(value: str) -> str: + if value.startswith("/nix/store/"): + return pathlib.PurePosixPath(value).name + return value + + +def compare_narinfo(expected: dict[str, Any], narinfo: dict[str, Any], fields: tuple[str, ...] = IMMUTABLE_FIELDS) -> list[str]: + expect = normalize_narinfo_for_compare(expected) + got_info = normalize_narinfo_for_compare(narinfo) + diffs: list[str] = [] + for key in fields: + if key not in expect and key not in got_info: + continue + value = expect.get(key, [] if key in ("References", "Sig") else None) + got = got_info.get(key, [] if key in ("References", "Sig") else None) + if isinstance(value, list): + got = sorted(got or []) + if got != value: + diffs.append(key) + return sorted(set(diffs)) + + +def sanitized_error(exc: BaseException) -> str: + if isinstance(exc, RepackError): + msg = str(exc) + allowed = [] + for ch in msg[:220]: + allowed.append(ch if ch.isalnum() or ch in " ._:/,-" else "_") + return "RepackError: " + "".join(allowed) + return exc.__class__.__name__ + + +def safe_headers(host: str | None = None, token: str | None = None) -> dict[str, str]: + headers: dict[str, str] = {} + if host: + headers["Host"] = host + if token: + headers["Authorization"] = f"Bearer {token}" + return headers + + +def assert_secret_url_safe(url: str) -> None: + parsed = urllib.parse.urlsplit(url) + if parsed.scheme != "http": + return + host = parsed.hostname or "" + if host not in ("127.0.0.1", "::1", "localhost"): + raise RepackError("refusing authenticated HTTP to non-loopback host") + + +def join_url(base: str, path: str) -> str: + return base.rstrip("/") + "/" + path.lstrip("/") + + +def requests_module() -> Any: + import requests + + return requests + + +def zstd_module() -> Any: + import zstandard + + return zstandard + + + +class TokenProvider: + def __init__(self, atticadm: str | None, server_config: str | None, cache: str) -> None: + self.atticadm = atticadm + self.server_config = server_config + self.cache = cache + self._token = os.environ.get(TOKEN_ENV) + self._expires = now() + 60 * 50 if self._token else 0.0 + self._lock = threading.Lock() + + def get(self) -> str: + with self._lock: + if self._token and now() < self._expires - 300: + return self._token + if not self.atticadm or not self.server_config: + raise RepackError(f"{TOKEN_ENV} or --atticadm/--server-config required") + cmd = [ + self.atticadm, + "--config", + self.server_config, + "make-token", + "--sub", + "attic-repack", + "--validity", + "2h", + "--pull", + self.cache, + "--push", + self.cache, + "--create-cache", + self.cache, + "--configure-cache", + self.cache, + "--configure-cache-retention", + self.cache, + ] + env = {k: v for k, v in os.environ.items() if k != TOKEN_ENV} + proc = subprocess.run(cmd, check=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) + token = proc.stdout.strip().splitlines()[-1] + if not token: + raise RepackError("atticadm returned empty token") + self._token = token + self._expires = now() + 60 * 110 + return token + + +class AtticClient: + def __init__(self, base_url: str, cache: str, host: str | None, token_provider: TokenProvider | None = None) -> None: + self.base_url = base_url.rstrip("/") + self.cache = cache + self.host = host + self.token_provider = token_provider + self._local = threading.local() + + def session(self) -> Any: + sess = getattr(self._local, "session", None) + if sess is None: + sess = requests_module().Session() + self._local.session = sess + return sess + + def api_headers(self) -> dict[str, str]: + token = self.token_provider.get() if self.token_provider else None + if token: + assert_secret_url_safe(self.base_url) + return safe_headers(self.host, token) + + def public_headers(self) -> dict[str, str]: + return safe_headers(self.host, None) + + def _request(self, method: str, url: str, *, auth: bool, **kwargs: Any) -> Any: + headers = kwargs.pop("headers", {}) or {} + headers = {**(self.api_headers() if auth else self.public_headers()), **headers} + allow_redirects = False if auth else kwargs.pop("allow_redirects", True) + try: + resp = self.session().request(method, url, headers=headers, allow_redirects=allow_redirects, **kwargs) + if auth and 300 <= resp.status_code < 400: + raise RepackError(f"HTTP redirect refused {method} {urllib.parse.urlsplit(url).path}") + if resp.status_code >= 400: + raise RepackError(f"HTTP {resp.status_code} {method} {urllib.parse.urlsplit(url).path}") + return resp + except Exception as exc: + if isinstance(exc, RepackError): + raise + raise RepackError(exc.__class__.__name__) from exc + + def _public_stream(self, url: str) -> Any: + sess = self.session() + headers = self.public_headers() + current = url + for _ in range(5): + resp = sess.request("GET", current, headers=headers, allow_redirects=False, stream=True, timeout=(10, 600)) + if 300 <= resp.status_code < 400: + loc = resp.headers.get("Location") + with contextlib.suppress(Exception): + resp.close() + if not loc: + raise RepackError("public redirect missing location") + next_url = urllib.parse.urljoin(current, loc) + if urllib.parse.urlsplit(next_url).netloc != urllib.parse.urlsplit(current).netloc: + headers = {} + current = next_url + continue + if resp.status_code >= 400: + raise RepackError(f"HTTP {resp.status_code} GET {urllib.parse.urlsplit(current).path}") + return resp + raise RepackError("too many public redirects") + + def get_cache_config(self) -> dict[str, Any] | None: + url = join_url(self.base_url, f"_api/v1/cache-config/{self.cache}") + try: + return self._request("GET", url, auth=True, timeout=(10, 60)).json() + except RepackError as exc: + if "HTTP 404" in str(exc): + return None + raise + + def narinfo_url(self, store_path_hash: str) -> str: + return join_url(self.base_url, f"{self.cache}/{store_path_hash}.narinfo") + + def create_cache(self, config: dict[str, Any]) -> None: + url = join_url(self.base_url, f"_api/v1/cache-config/{self.cache}") + self._request("POST", url, auth=True, json=config, timeout=(10, 60)) + + def patch_retention(self, retention: Any) -> None: + if retention is None: + return + url = join_url(self.base_url, f"_api/v1/cache-config/{self.cache}") + self._request("PATCH", url, auth=True, json={"retention_period": retention}, timeout=(10, 60)) + + def get_narinfo(self, store_path_hash: str) -> dict[str, Any] | None: + url = self.narinfo_url(store_path_hash) + try: + resp = self._public_stream(url) + try: + return parse_narinfo(resp.content) + finally: + with contextlib.suppress(Exception): + resp.close() + except RepackError as exc: + if "HTTP 404" in str(exc): + return None + raise + + def upload(self, record: dict[str, Any], nar_path: pathlib.Path) -> None: + meta = upload_metadata(record) + prefix = json.dumps(meta, sort_keys=True, separators=(",", ":")).encode() + nar_size = nar_path.stat().st_size + headers = {"X-Attic-Nar-Info-Preamble-Size": str(len(prefix)), "Content-Length": str(len(prefix) + nar_size)} + url = join_url(self.base_url, "_api/v1/upload-path") + + for attempt in range(3): + with PrefixFileBody(prefix, nar_path) as body: + try: + self._request("PUT", url, auth=True, data=body, headers=headers, timeout=(10, 600)) + return + except RepackError: + if attempt == 2: + raise + time.sleep(0.5 * (2**attempt)) + + def verify_payload(self, narinfo: dict[str, Any], expected_hash: str, expected_size: int, narinfo_url: str) -> None: + raw_url = narinfo.get("URL") + if not raw_url: + raise RepackError("new narinfo missing URL") + url = urllib.parse.urljoin(narinfo_url, raw_url) + resp = self._public_stream(url) + compression = (narinfo.get("Compression") or pathlib.PurePosixPath(raw_url).suffix.lstrip(".")).lower() + h = hashlib.sha256() + size = 0 + source = resp.raw + if compression in ("zstd", "zst"): + reader = zstd_module().ZstdDecompressor().stream_reader(source, read_across_frames=True) + elif compression in ("", "none"): + reader = source + elif compression == "gzip" or compression == "gz": + reader = gzip.GzipFile(fileobj=source) + elif compression == "xz": + reader = lzma.LZMAFile(source) + else: + raise RepackError(f"unsupported new nar compression {compression}") + try: + with contextlib.closing(reader): + while True: + data = reader.read(CHUNK) + if not data: + break + h.update(data) + size += len(data) + finally: + with contextlib.suppress(Exception): + resp.close() + if h.hexdigest() != expected_hash or size != expected_size: + raise RepackError("new NAR payload hash/size mismatch") + + +class PrefixFileBody: + def __init__(self, prefix: bytes, path: pathlib.Path) -> None: + self.prefix = prefix + self.path = path + self.file: BinaryIO | None = None + self.pos = 0 + self.size = len(prefix) + path.stat().st_size + + def __enter__(self) -> "PrefixFileBody": + self.file = self.path.open("rb") + return self + + def __exit__(self, *_args: object) -> None: + if self.file: + self.file.close() + + def __len__(self) -> int: + return self.size + + def tell(self) -> int: + return self.pos + + def read(self, n: int = -1) -> bytes: + if self.file is None: + raise RepackError("upload body not open") + if self.pos >= self.size: + return b"" + want = self.size - self.pos if n is None or n < 0 else n + parts: list[bytes] = [] + if self.pos < len(self.prefix) and want > 0: + chunk = self.prefix[self.pos : min(len(self.prefix), self.pos + want)] + parts.append(chunk) + self.pos += len(chunk) + want -= len(chunk) + if want > 0: + data = self.file.read(want) + parts.append(data) + self.pos += len(data) + return b"".join(parts) + + +class InventoryDB: + def __init__(self, uri: str, cache: str) -> None: + self.uri = readonly_sqlite_uri(uri) + self.cache = cache + + def connect(self) -> sqlite3.Connection: + con = sqlite3.connect(self.uri, uri=True) + con.row_factory = sqlite3.Row + return con + + def cache_row(self) -> dict[str, Any]: + with self.connect() as con: + row = con.execute( + "select name,keypair,is_public,store_dir,priority,upstream_cache_key_names,retention_period " + "from cache where name=? and deleted_at is null", + (self.cache,), + ).fetchone() + if row is None: + raise RepackError(f"cache not found in old DB: {self.cache}") + return dict(row) + + def records(self, paths: set[str] | None = None, limit: int | None = None) -> list[dict[str, Any]]: + sql = ( + 'select c.name as cache,o.nar_id,o.store_path_hash,o.store_path,o."references",o.system,o.deriver,o.sigs,o.ca,' + "n.nar_hash,n.nar_size,n.state as nar_state " + "from object o join cache c on c.id=o.cache_id join nar n on n.id=o.nar_id " + "where c.name=? and c.deleted_at is null and (n.state='V' or n.state='valid') " + "order by o.store_path" + ) + args: list[Any] = [self.cache] + with self.connect() as con: + rows = con.execute(sql, args).fetchall() + out: list[dict[str, Any]] = [] + for row in rows: + record = dict(row) + record["references"] = json_list(record.get("references")) + record["sigs"] = json_list(record.get("sigs")) + record["nar_size"] = int(record["nar_size"]) + record["nar_hash"] = record["nar_hash"] if str(record["nar_hash"]).startswith("sha256:") else f"sha256:{record['nar_hash']}" + nar_hash_hex(record["nar_hash"]) + if paths and record["store_path"] not in paths: + continue + out.append(record) + if limit and len(out) >= limit: + break + return out + + def chunk_rows(self, nar_id: int) -> list[dict[str, Any]]: + sql = ( + "select cr.seq,ch.chunk_hash,ch.chunk_size,ch.file_hash,ch.file_size,ch.compression,ch.remote_file,ch.state " + "from chunkref cr join chunk ch on ch.id=cr.chunk_id where cr.nar_id=? order by cr.seq" + ) + with self.connect() as con: + return [dict(r) for r in con.execute(sql, (nar_id,)).fetchall()] + + +def readonly_sqlite_uri(value: str) -> str: + if value == ":memory:": + return "file::memory:?mode=ro" + if value.startswith("file:"): + parsed = urllib.parse.urlsplit(value) + qs = urllib.parse.parse_qs(parsed.query, keep_blank_values=True) + modes = qs.get("mode") + if modes and modes != ["ro"]: + raise RepackError("old DB URI must use mode=ro") + qs["mode"] = ["ro"] + query = urllib.parse.urlencode(qs, doseq=True) + return urllib.parse.urlunsplit(parsed._replace(query=query)) + path = pathlib.Path(value) + if not path.exists(): + raise RepackError("old DB path missing") + return "file:" + urllib.parse.quote(str(path.resolve())) + "?mode=ro" + + +class State: + def __init__(self, root: pathlib.Path) -> None: + self.root = root + self.raw = root / "raw" + self.chunks = root / "chunks" + self.checkpoints = root / "checkpoints" + for path in (root, self.raw, self.chunks, self.checkpoints): + require_private(path, True) + + def raw_path(self, nar_hash: str) -> pathlib.Path: + return self.raw / f"{nar_hash}.nar" + + def checkpoint_path(self, store_path_hash: str) -> pathlib.Path: + return self.checkpoints / f"{store_path_hash}.json" + + def get_checkpoint(self, record: dict[str, Any]) -> dict[str, Any]: + return load_json(self.checkpoint_path(record["store_path_hash"]), {}) + + def set_checkpoint(self, record: dict[str, Any], status: str, tries: int, error: str | None = None) -> None: + value = { + "store_path": record["store_path"], + "store_path_hash": record["store_path_hash"], + "nar_hash": record["nar_hash"], + "metadata_fingerprint": metadata_fingerprint(record), + "status": status, + "tries": tries, + "updated_at": int(now()), + } + if error: + value["error"] = sanitized_error(RepackError(error)) + atomic_json(self.checkpoint_path(record["store_path_hash"]), value) + + +class NarLocks: + def __init__(self) -> None: + self._lock = threading.Lock() + self._locks: dict[str, threading.Lock] = {} + + @contextlib.contextmanager + def hold(self, key: str) -> Iterator[None]: + with self._lock: + lock = self._locks.setdefault(key, threading.Lock()) + lock.acquire() + try: + yield + finally: + lock.release() + + +class OldS3Assembler: + def __init__(self, db: InventoryDB, state: State, endpoint: str, bucket: str, region: str) -> None: + self.db = db + self.state = state + self.endpoint = endpoint + self.bucket = bucket + self.region = region + self._local = threading.local() + + def client(self) -> Any: + client = getattr(self._local, "client", None) + if client is None: + import boto3 + from botocore.config import Config + + cfg = Config(retries={"mode": "standard", "max_attempts": 3}, max_pool_connections=8, connect_timeout=10, read_timeout=60) + client = boto3.session.Session().client("s3", endpoint_url=self.endpoint, region_name=self.region, config=cfg) + self._local.client = client + return client + + def assemble(self, record: dict[str, Any], out_path: pathlib.Path) -> None: + rows = self.db.chunk_rows(int(record["nar_id"])) + if not rows: + raise RepackError("old S3 chunkrefs missing") + seqs = [int(r["seq"]) for r in rows] + if seqs != list(range(seqs[0], seqs[0] + len(seqs))): + raise RepackError("old S3 chunk sequence has gaps") + expected_hash = nar_hash_hex(record["nar_hash"]) + expected_size = int(record["nar_size"]) + h = hashlib.sha256() + size = 0 + tmp = out_path.with_suffix(".tmp") + def read_chunk(row: dict[str, Any]) -> bytes: + data = self._compressed_chunk(row) + plain = decompress_chunk(data, str(row.get("compression") or "none")) + verify_hash_size(plain, row.get("chunk_hash"), row.get("chunk_size"), "chunk") + return plain + + with private_umask(), tmp.open("wb") as dst, concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool: + remaining = iter(rows) + pending = deque(pool.submit(read_chunk, row) for row in itertools.islice(remaining, 4)) + try: + while pending: + plain = pending.popleft().result() + dst.write(plain) + h.update(plain) + size += len(plain) + row = next(remaining, None) + if row is not None: + pending.append(pool.submit(read_chunk, row)) + finally: + for future in pending: + future.cancel() + dst.flush() + os.fsync(dst.fileno()) + os.chmod(tmp, 0o600) + if h.hexdigest() != expected_hash or size != expected_size: + tmp.unlink(missing_ok=True) + raise RepackError("assembled old NAR hash/size mismatch") + os.replace(tmp, out_path) + + def _compressed_chunk(self, row: dict[str, Any]) -> bytes: + remote = json.loads(row["remote_file"]) + s3 = remote.get("S3") if isinstance(remote, dict) else None + if not s3 or s3.get("bucket") != self.bucket or s3.get("region") != self.region: + raise RepackError("old S3 remote_file bucket/region mismatch") + if row.get("state") not in ("V", "valid", None): + raise RepackError("old S3 chunk state not valid") + key = s3.get("key") + if not key: + raise RepackError("old S3 key missing") + cache_key = sha256_bytes(key.encode()) + cached = self.state.chunks / f"{cache_key}.chunk" + if cached.exists(): + data = cached.read_bytes() + verify_hash_size(data, row.get("file_hash"), row.get("file_size"), "compressed chunk") + return data + last: BaseException | None = None + for attempt in range(5): + try: + obj = self.client().get_object(Bucket=self.bucket, Key=key) + body = obj["Body"] + try: + data = body.read() + finally: + with contextlib.suppress(Exception): + body.close() + verify_hash_size(data, row.get("file_hash"), row.get("file_size"), "compressed chunk") + atomic_write(cached, data) + return data + except Exception as exc: # boto exceptions sanitized at caller + last = exc + time.sleep(min(8, 0.5 * (2**attempt))) + raise RepackError("old S3 get failed") + + +def verify_hash_size(data: bytes, hash_value: Any, size_value: Any, label: str) -> None: + if size_value not in (None, "") and len(data) != int(size_value): + raise RepackError(f"{label} size mismatch") + if hash_value in (None, ""): + return + text = str(hash_value) + if text.startswith("sha256:"): + text = text.split(":", 1)[1] + if len(text) != 64 or any(c not in "0123456789abcdefABCDEF" for c in text): + raise RepackError(f"{label} hash format invalid") + if sha256_bytes(data) != text.lower(): + raise RepackError(f"{label} hash mismatch") + + +def decompress_chunk(data: bytes, compression: str) -> bytes: + c = compression.lower() + if c in ("none", "", "null"): + return data + if c in ("zstd", "zst"): + reader = zstd_module().ZstdDecompressor().stream_reader(io.BytesIO(data), read_across_frames=True) + with contextlib.closing(reader): + return reader.read() + if c in ("gzip", "gz"): + return gzip.decompress(data) + if c == "xz": + return lzma.decompress(data) + raise RepackError(f"unsupported old chunk compression {compression}") + + +class Migrator: + def __init__(self, args: argparse.Namespace) -> None: + self.args = args + self.db = InventoryDB(args.old_db, args.cache) + self.state = State(pathlib.Path(args.state_dir)) + self.tokens = TokenProvider(args.atticadm, args.server_config, args.cache) + self.old_client = AtticClient(args.old_url, args.cache, args.host, self.tokens) + self.new_client = AtticClient(args.new_url, args.cache, args.host, self.tokens) + self.s3 = OldS3Assembler(self.db, self.state, args.old_storage_endpoint, args.old_bucket, args.old_region) + self.locks = NarLocks() + + def selected_records(self) -> list[dict[str, Any]]: + paths = None + if self.args.paths_file: + paths = {p.strip() for p in pathlib.Path(self.args.paths_file).read_text().splitlines() if p.strip()} + return self.db.records(paths=paths, limit=self.args.limit) + + def init_cache(self) -> None: + row = self.db.cache_row() + keypair = row.get("keypair") + if not keypair: + raise RepackError("old cache keypair missing") + create = { + "keypair": {"Keypair": keypair}, + "is_public": bool(row["is_public"]), + "store_dir": row["store_dir"] or "/nix/store", + "priority": int(row["priority"]), + "upstream_cache_key_names": json_list(row.get("upstream_cache_key_names")), + } + old_cfg = self.old_client.get_cache_config() or {} + new_cfg = self.new_client.get_cache_config() + old_public = old_cfg.get("public_key") + if not old_public: + raise RepackError("old cache public_key missing") + if new_cfg: + mismatch = [] + if new_cfg.get("public_key") != old_public: + mismatch.append("public_key") + for key in ("is_public", "store_dir", "priority", "upstream_cache_key_names"): + if key in new_cfg and new_cfg.get(key) != create[key]: + mismatch.append(key) + if mismatch: + raise RepackError("new cache exists with mismatched settings: " + ",".join(sorted(mismatch))) + self.new_client.patch_retention(retention_from_db(row.get("retention_period"))) + print(json.dumps({"exists": True, "public_key_matches": True}, sort_keys=True)) + else: + self.new_client.create_cache(create) + new_cfg = self.new_client.get_cache_config() + if not new_cfg or new_cfg.get("public_key") != old_public: + raise RepackError("created cache public_key mismatch") + self.new_client.patch_retention(retention_from_db(row.get("retention_period"))) + print(json.dumps({"created": True, "public_key_matches": True}, sort_keys=True)) + + def inventory(self) -> None: + records = self.selected_records() + manifest = { + "format": "attic-repack-inventory-v1", + "cache": self.args.cache, + "generated_at": int(now()), + "spool_dir": str(self.state.raw), + "raw_nar_filename": "{sha256hex}.nar", + "records": [{**upload_metadata(r), "metadata_fingerprint": metadata_fingerprint(r)} for r in records], + } + print(json.dumps(manifest, sort_keys=True, indent=2)) + + def status(self) -> None: + records = self.selected_records() + unique: dict[str, int] = {} + counts = {"verified": 0, "failed": 0, "pending": 0} + total_bytes = 0 + for r in records: + h = nar_hash_hex(r["nar_hash"]) + unique[h] = int(r["nar_size"]) + total_bytes += int(r["nar_size"]) + cp = self.state.get_checkpoint(r) + if cp.get("metadata_fingerprint") != metadata_fingerprint(r): + counts["pending"] += 1 + elif cp.get("status") == "verified": + counts["verified"] += 1 + elif cp.get("status") == "failed": + counts["failed"] += 1 + else: + counts["pending"] += 1 + print(json.dumps({"inventory_total": len(records), "migrated_verified": counts["verified"], "missing_failed": counts["failed"], "pending": counts["pending"], "unique_nar": len(unique), "total_bytes": total_bytes}, sort_keys=True)) + + def migrate(self, verify_only: bool = False) -> int: + records = self.selected_records() + q: queue.Queue[tuple[str, str]] = queue.Queue() + failures = 0 + + def work(record: dict[str, Any]) -> None: + tries = int(self.state.get_checkpoint(record).get("tries", 0)) + 1 + try: + if verify_only: + if not self.verify_record(record, force_payload=True): + raise RepackError("new narinfo missing") + else: + self.migrate_record(record) + self.state.set_checkpoint(record, "verified", tries) + q.put(("ok", record["store_path"])) + except Exception as exc: + self.state.set_checkpoint(record, "failed", tries, sanitized_error(exc)) + q.put(("failed", f"{record['store_path']} {sanitized_error(exc)}")) + + with concurrent.futures.ThreadPoolExecutor(max_workers=max(1, int(self.args.workers))) as pool: + futures = [pool.submit(work, r) for r in records] + done = 0 + while done < len(futures): + kind, msg = q.get() + done += 1 + if kind == "failed": + failures += 1 + print(json.dumps({"done": done, "total": len(futures), "status": kind, "path": msg}, sort_keys=True), flush=True) + for fut in futures: + fut.result() + return failures + + def migrate_record(self, record: dict[str, Any]) -> None: + cp = self.state.get_checkpoint(record) + if cp.get("status") == "verified" and cp.get("metadata_fingerprint") == metadata_fingerprint(record): + if self.verify_record(record, force_payload=False): + return + nar_path = self.ensure_raw_nar(record) + actual_hash, actual_size = sha256_file(nar_path) + if actual_hash != nar_hash_hex(record["nar_hash"]) or actual_size != int(record["nar_size"]): + raise RepackError("raw NAR spool hash/size mismatch") + self.new_client.upload(record, nar_path) + if not self.verify_record(record, force_payload=True): + raise RepackError("new narinfo missing after upload") + + def verify_record(self, record: dict[str, Any], force_payload: bool) -> bool: + narinfo = self.new_client.get_narinfo(record["store_path_hash"]) + if narinfo is None: + return False + old_narinfo = self.old_client.get_narinfo(record["store_path_hash"]) + if old_narinfo is None: + raise RepackError("old narinfo missing") + db_guard_fields = tuple(f for f in IMMUTABLE_FIELDS if f != "Sig") + db_diffs = compare_narinfo(expected_narinfo(record), old_narinfo, db_guard_fields) + if db_diffs: + raise RepackError("old narinfo differs from DB snapshot: " + ",".join(db_diffs)) + diffs = compare_narinfo(old_narinfo, narinfo) + if diffs: + raise RepackError("new narinfo immutable metadata mismatch: " + ",".join(diffs)) + if force_payload: + self.new_client.verify_payload(narinfo, nar_hash_hex(record["nar_hash"]), int(record["nar_size"]), self.new_client.narinfo_url(record["store_path_hash"])) + return True + + def ensure_raw_nar(self, record: dict[str, Any]) -> pathlib.Path: + h = nar_hash_hex(record["nar_hash"]) + path = self.state.raw_path(h) + with self.locks.hold(h): + if path.exists(): + actual_hash, actual_size = sha256_file(path) + if actual_hash == h and actual_size == int(record["nar_size"]): + return path + raise RepackError("existing raw NAR spool hash/size mismatch") + store_path = pathlib.Path(record["store_path"]) + if store_path.exists(): + try: + self.dump_local_store_path(record, path) + return path + except (RepackError, subprocess.CalledProcessError, OSError): + path.with_suffix(".tmp").unlink(missing_ok=True) + eprint(json.dumps({"local_source_rejected": record["store_path"], "recovery": "old S3 chunks"})) + self.s3.assemble(record, path) + return path + + def dump_local_store_path(self, record: dict[str, Any], out_path: pathlib.Path) -> None: + tmp = out_path.with_suffix(".tmp") + env = {k: v for k, v in os.environ.items() if k != TOKEN_ENV} + env.pop("NIX_CONFIG", None) + cmd = [self.args.nix, "nar", "pack", record["store_path"]] + with private_umask(), tmp.open("wb") as handle: + subprocess.run(cmd, check=True, stdout=handle, stderr=subprocess.PIPE, env=env) + handle.flush() + os.fsync(handle.fileno()) + os.chmod(tmp, 0o600) + actual_hash, actual_size = sha256_file(tmp) + if actual_hash != nar_hash_hex(record["nar_hash"]) or actual_size != int(record["nar_size"]): + tmp.unlink(missing_ok=True) + raise RepackError("nix dump-path NAR hash/size mismatch") + os.replace(tmp, out_path) + + +def add_common(parser: argparse.ArgumentParser, inherit: bool = False) -> None: + def add(*args: Any, **kwargs: Any) -> None: + if inherit: + kwargs["default"] = argparse.SUPPRESS + parser.add_argument(*args, **kwargs) + + add("--old-db", default=DEFAULT_OLD_DB) + add("--state-dir", default=DEFAULT_STATE_DIR) + add("--old-url", default=DEFAULT_OLD_URL) + add("--new-url", default=DEFAULT_NEW_URL) + add("--host", default=DEFAULT_HOST) + add("--cache", default=DEFAULT_CACHE) + add("--atticadm") + add("--server-config") + add("--nix", default="nix") + add("--old-storage-endpoint", default=DEFAULT_OLD_ENDPOINT) + add("--old-bucket", default=DEFAULT_OLD_BUCKET) + add("--old-region", default=DEFAULT_OLD_REGION) + add("--workers", type=int, default=2) + add("--limit", type=int) + add("--paths-file") + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Safe resumable Attic repack/migration helper") + add_common(parser) + sub = parser.add_subparsers(dest="command", required=True) + for name in ("init", "inventory", "migrate", "verify", "status"): + child = sub.add_parser(name) + add_common(child, inherit=True) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + migrator = Migrator(args) + if args.command == "init": + migrator.init_cache() + elif args.command == "inventory": + migrator.inventory() + elif args.command == "migrate": + if migrator.migrate(False): + return 1 + elif args.command == "verify": + if migrator.migrate(True): + return 1 + elif args.command == "status": + migrator.status() + else: + raise RepackError("unknown command") + return 0 + except Exception as exc: + eprint(sanitized_error(exc)) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/infra/attic-migration/test_repack.py b/infra/attic-migration/test_repack.py new file mode 100644 index 00000000..5b4439db --- /dev/null +++ b/infra/attic-migration/test_repack.py @@ -0,0 +1,474 @@ +import argparse +import hashlib +import http.server +import io +import json +import os +import pathlib +import sqlite3 +import tempfile +import threading +import time +import unittest +from unittest import mock + +import repack + + +def sha(data): + return hashlib.sha256(data).hexdigest() + + +class FakeResponse: + def __init__(self, status_code=200, content=b"", json_data=None, raw=None, headers=None): + self.status_code = status_code + self.content = content + self._json = json_data + self.raw = raw or io.BytesIO(content) + self.headers = headers or {} + + def json(self): + return self._json + + def close(self): + pass + + +class FakeSession: + def __init__(self): + self.calls = [] + self.routes = {} + + def request(self, method, url, **kwargs): + self.calls.append((method, url, kwargs)) + key = (method, pathlib.PurePosixPath(url.split("?", 1)[0]).as_posix()) + response = self.routes.get(key) or self.routes.get((method, url)) + if callable(response): + return response(method, url, kwargs) + return response or FakeResponse(404) + + +class ThreadedHTTP: + def __init__(self, handler): + self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler) + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + + @property + def url(self): + host, port = self.server.server_address[:2] + return f"http://{host}:{port}" + + def __enter__(self): + self.thread.start() + return self + + def __exit__(self, *_args): + self.server.shutdown() + self.thread.join(timeout=5) + self.server.server_close() + + +class RepackTests(unittest.TestCase): + def test_options_before_subcommand_are_preserved(self): + parser = repack.build_parser() + args = parser.parse_args(["--atticadm", "/safe/atticadm", "--server-config", "/safe/config", "--workers", "1", "init"]) + self.assertEqual(args.atticadm, "/safe/atticadm") + self.assertEqual(args.server_config, "/safe/config") + self.assertEqual(args.workers, 1) + args = parser.parse_args(["--workers", "1", "migrate", "--workers", "2"]) + self.assertEqual(args.workers, 2) + + def test_inventory_reads_sqlite_reserved_references_column(self): + with tempfile.TemporaryDirectory() as td: + db = pathlib.Path(td) / "old.db" + con = sqlite3.connect(db) + con.executescript(''' + CREATE TABLE cache(id INTEGER, name TEXT, deleted_at TEXT); + CREATE TABLE nar(id INTEGER, nar_hash TEXT, nar_size INTEGER, state TEXT); + CREATE TABLE object(cache_id INTEGER, nar_id INTEGER, + store_path_hash TEXT, store_path TEXT, "references" TEXT, + system TEXT, deriver TEXT, sigs TEXT, ca TEXT); + INSERT INTO cache VALUES(1, 'hectic', NULL); + ''') + con.execute("INSERT INTO nar VALUES(1, ?, 7, 'V')", ("sha256:" + "a" * 64,)) + con.execute("INSERT INTO object VALUES(1, 1, ?, ?, ?, NULL, NULL, ?, NULL)", + ("b" * 32, "/nix/store/" + "b" * 32 + "-test", '["dependency"]', '[]')) + con.commit() + con.close() + rows = repack.InventoryDB(str(db), "hectic").records() + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["references"], ["dependency"]) + + def test_metadata_import_keypair_capital(self): + with tempfile.TemporaryDirectory() as td: + db = pathlib.Path(td) / "old.db" + con = sqlite3.connect(db) + con.executescript(""" + create table cache(id integer primary key,name text,keypair text,is_public integer,store_dir text,priority integer,upstream_cache_key_names text,retention_period integer,deleted_at text); + insert into cache values(1,'hectic','priv',1,'/nix/store',30,'["up"]',3600,null); + """) + con.close() + args = self.args(td, old_db=str(db)) + mig = repack.Migrator(args) + old = mock.Mock() + old.get_cache_config.return_value = {"public_key": "pub"} + new = mock.Mock() + new.get_cache_config.side_effect = [None, {"public_key": "pub"}] + mig.old_client = old + mig.new_client = new + mig.init_cache() + body = new.create_cache.call_args.args[0] + self.assertEqual(body["keypair"], {"Keypair": "priv"}) + new.patch_retention.assert_called_with({"Period": 3600}) + + def test_bad_hash_rejects_raw_spool(self): + with tempfile.TemporaryDirectory() as td: + args = self.args(td) + mig = repack.Migrator(args) + record = self.record(b"good") + raw = mig.state.raw_path(repack.nar_hash_hex(record["nar_hash"])) + raw.write_bytes(b"bad") + with self.assertRaises(repack.RepackError): + mig.ensure_raw_nar(record) + + def test_local_mismatch_recovers_original_from_old_s3(self): + with tempfile.TemporaryDirectory() as td: + original = b"original cached NAR" + record = self.record(original) + local = pathlib.Path(td) / "different-local-copy" + local.write_bytes(b"different") + record["store_path"] = str(local) + mig = repack.Migrator(self.args(td)) + + def recover(_record, path): + path.write_bytes(original) + + with mock.patch.object(mig, "dump_local_store_path", side_effect=repack.RepackError("nix dump-path NAR hash/size mismatch")), \ + mock.patch.object(mig.s3, "assemble", side_effect=recover) as assemble: + result = mig.ensure_raw_nar(record) + self.assertEqual(result.read_bytes(), original) + assemble.assert_called_once() + + def test_no_compile_subprocess_commands(self): + with tempfile.TemporaryDirectory() as td: + args = self.args(td) + mig = repack.Migrator(args) + data = b"nar" + record = self.record(data) + store = pathlib.Path(record["store_path"]) + with mock.patch("subprocess.run") as run: + def fake_run(cmd, check, stdout, stderr, env): + self.assertEqual(cmd[:3], ["nix", "nar", "pack"]) + self.assertNotIn("build", cmd) + self.assertNotIn(repack.TOKEN_ENV, env) + stdout.write(data) + return mock.Mock() + run.side_effect = fake_run + path = mig.state.raw_path(repack.nar_hash_hex(record["nar_hash"])) + mig.dump_local_store_path(record, path) + self.assertEqual(path.read_bytes(), data) + self.assertTrue(str(store).startswith("/nix/store/")) + + def test_resumable_checkpoint(self): + with tempfile.TemporaryDirectory() as td: + state = repack.State(pathlib.Path(td)) + record = self.record(b"abc") + state.set_checkpoint(record, "verified", 2) + cp = state.get_checkpoint(record) + self.assertEqual(cp["status"], "verified") + self.assertEqual(cp["tries"], 2) + self.assertEqual(cp["metadata_fingerprint"], repack.metadata_fingerprint(record)) + + def test_root_backend_upload_preamble(self): + data = b"abc" + record = self.record(data) + with tempfile.TemporaryDirectory() as td: + nar = pathlib.Path(td) / "x.nar" + nar.write_bytes(data) + client = repack.AtticClient("http://127.0.0.1:8082", "hectic", "cache.hectic-lab.com", repack.TokenProvider(None, None, "hectic")) + assert client.token_provider is not None + client.token_provider._token = "tok" + client.token_provider._expires = repack.now() + 3600 + sess = FakeSession() + client._local.session = sess + def put(method, url, kwargs): + self.assertTrue(url.endswith("/_api/v1/upload-path")) + self.assertIn("X-Attic-Nar-Info-Preamble-Size", kwargs["headers"]) + self.assertEqual(len(kwargs["data"]), int(kwargs["headers"]["Content-Length"])) + body = kwargs["data"].read() + pre = int(kwargs["headers"]["X-Attic-Nar-Info-Preamble-Size"]) + meta = json.loads(body[:pre]) + self.assertEqual(meta["store_path"], record["store_path"]) + self.assertEqual(body[pre:], data) + return FakeResponse(200) + sess.routes[("PUT", "http://127.0.0.1:8082/_api/v1/upload-path")] = put + client.upload(record, nar) + + def test_real_http_upload_has_content_length_no_chunked(self): + try: + repack.requests_module() + except ModuleNotFoundError: + self.skipTest("requests not installed outside Nix test env") + data = b"nar-bytes" + record = self.record(data) + seen = {} + + class Handler(http.server.BaseHTTPRequestHandler): + def do_PUT(self): + seen["path"] = self.path + seen["host"] = self.headers.get("Host") + seen["te"] = self.headers.get("Transfer-Encoding") + length = int(self.headers["Content-Length"]) + body = self.rfile.read(length) + pre = int(self.headers["X-Attic-Nar-Info-Preamble-Size"]) + seen["meta"] = json.loads(body[:pre]) + seen["nar"] = body[pre:] + self.send_response(200); self.end_headers() + + def log_message(self, format, *args): + pass + + with tempfile.TemporaryDirectory() as td, ThreadedHTTP(Handler) as srv: + nar = pathlib.Path(td) / "x.nar" + nar.write_bytes(data) + tp = repack.TokenProvider(None, None, "hectic") + tp._token = "tok"; tp._expires = repack.now() + 3600 + repack.AtticClient(srv.url, "hectic", "cache.hectic-lab.com", tp).upload(record, nar) + self.assertEqual(seen["path"], "/_api/v1/upload-path") + self.assertEqual(seen["host"], "cache.hectic-lab.com") + self.assertIsNone(seen["te"]) + self.assertEqual(seen["meta"]["store_path"], record["store_path"]) + self.assertEqual(seen["nar"], data) + + def test_payload_relative_url_and_redirect_strips_host(self): + try: + repack.requests_module() + except ModuleNotFoundError: + self.skipTest("requests not installed outside Nix test env") + data = b"nar" + seen = {} + + class S3Handler(http.server.BaseHTTPRequestHandler): + def do_GET(self): + seen["s3_path"] = self.path + seen["s3_host"] = self.headers.get("Host") + seen["s3_auth"] = self.headers.get("Authorization") + self.send_response(200); self.end_headers(); self.wfile.write(data) + def log_message(self, format, *args): + pass + + with ThreadedHTTP(S3Handler) as s3: + class CacheHandler(http.server.BaseHTTPRequestHandler): + def do_GET(self): + seen["cache_path"] = self.path + seen["cache_host"] = self.headers.get("Host") + seen["cache_auth"] = self.headers.get("Authorization") + self.send_response(302) + self.send_header("Location", s3.url + "/object") + self.end_headers() + def log_message(self, format, *args): + pass + + with ThreadedHTTP(CacheHandler) as cache: + client = repack.AtticClient(cache.url, "hectic", "cache.hectic-lab.com", None) + client.verify_payload({"URL":"nar/x","Compression":"none"}, sha(data), len(data), cache.url + "/hectic/abcd.narinfo") + self.assertEqual(seen["cache_path"], "/hectic/nar/x") + self.assertEqual(seen["cache_host"], "cache.hectic-lab.com") + self.assertIsNone(seen["cache_auth"]) + self.assertEqual(seen["s3_path"], "/object") + self.assertNotEqual(seen["s3_host"], "cache.hectic-lab.com") + self.assertIsNone(seen["s3_auth"]) + + def test_auth_api_redirect_refused(self): + tp = repack.TokenProvider(None, None, "hectic") + tp._token = "tok"; tp._expires = repack.now() + 3600 + client = repack.AtticClient("http://127.0.0.1:8082", "hectic", None, tp) + sess = FakeSession(); client._local.session = sess + sess.routes[("GET", "http://127.0.0.1:8082/_api/v1/cache-config/hectic")] = FakeResponse(302, headers={"Location":"http://evil/"}) + with self.assertRaises(repack.RepackError): + client.get_cache_config() + + def test_concatenated_zstd_correct(self): + try: + zstd = repack.zstd_module() + except ModuleNotFoundError: + self.skipTest("zstandard not installed outside Nix test env") + plain = b"a" * 100 + b"b" * 100 + cctx = zstd.ZstdCompressor() + payload = cctx.compress(plain[:100]) + cctx.compress(plain[100:]) + narinfo = {"URL": "nar/x.nar.zst", "Compression": "zstd"} + client = repack.AtticClient("http://127.0.0.1:8082", "hectic", None, None) + sess = FakeSession() + client._local.session = sess + sess.routes[("GET", "http://127.0.0.1:8082/hectic/nar/x.nar.zst")] = FakeResponse(200, raw=io.BytesIO(payload)) + client.verify_payload(narinfo, sha(plain), len(plain), "http://127.0.0.1:8082/hectic/abcd.narinfo") + + def test_perchunk_retry_cache(self): + try: + zstd = repack.zstd_module() + except ModuleNotFoundError: + self.skipTest("zstandard not installed outside Nix test env") + with tempfile.TemporaryDirectory() as td: + db = pathlib.Path(td) / "old.db" + plain = b"chunk" + comp = zstd.ZstdCompressor().compress(plain) + con = sqlite3.connect(db) + con.executescript(""" + create table chunkref(nar_id integer,seq integer,chunk_id integer); + create table chunk(id integer primary key,state text,chunk_hash text,chunk_size integer,file_hash text,file_size integer,compression text,remote_file text); + """) + con.execute("insert into chunkref values(1,0,1)") + con.execute("insert into chunk values(1,'V',?,?,?,?,?,?)", (sha(plain), len(plain), sha(comp), len(comp), "zstd", json.dumps({"S3":{"region":"hel1","bucket":"cache-hectic-lab","key":"k"}}))) + con.commit(); con.close() + state = repack.State(pathlib.Path(td) / "state") + asm = repack.OldS3Assembler(repack.InventoryDB(str(db), "hectic"), state, "https://example", "cache-hectic-lab", "hel1") + fake_client = mock.Mock() + fake_client.get_object.side_effect = [Exception("once"), {"Body": io.BytesIO(comp)}] + out = pathlib.Path(td) / "out.nar" + with mock.patch.object(asm, "client", return_value=fake_client): + asm.assemble({"nar_id": 1, "nar_hash": "sha256:" + sha(plain), "nar_size": len(plain)}, out) + self.assertEqual(out.read_bytes(), plain) + self.assertEqual(fake_client.get_object.call_count, 2) + fake_client.get_object.reset_mock() + self.assertEqual(asm._compressed_chunk(asm.db.chunk_rows(1)[0]), comp) + fake_client.get_object.assert_not_called() + + def test_chunk_prefetch_is_bounded_and_preserves_order(self): + with tempfile.TemporaryDirectory() as td: + pieces = [f"chunk-{i}\n".encode() for i in range(12)] + rows = [{"seq": i, "compression": "none", "chunk_hash": sha(data), "chunk_size": len(data)} + for i, data in enumerate(pieces)] + db = mock.Mock() + db.chunk_rows.return_value = rows + asm = repack.OldS3Assembler(db, repack.State(pathlib.Path(td) / "state"), "https://example", "cache-hectic-lab", "hel1") + lock = threading.Lock() + active = 0 + peak = 0 + + def fetch(row): + nonlocal active, peak + with lock: + active += 1 + peak = max(peak, active) + time.sleep(0.04 if row["seq"] == 0 else 0.01) + with lock: + active -= 1 + return pieces[row["seq"]] + + whole = b"".join(pieces) + out = pathlib.Path(td) / "result.nar" + with mock.patch.object(asm, "_compressed_chunk", side_effect=fetch): + asm.assemble({"nar_id": 1, "nar_hash": "sha256:" + sha(whole), "nar_size": len(whole)}, out) + self.assertEqual(out.read_bytes(), whole) + self.assertGreater(peak, 1) + self.assertLessEqual(peak, 4) + + def test_zstd_chunk_no_content_size_concat(self): + try: + zstd = repack.zstd_module() + except ModuleNotFoundError: + self.skipTest("zstandard not installed outside Nix test env") + cctx = zstd.ZstdCompressor(write_content_size=False) + payload = cctx.compress(b"aa") + cctx.compress(b"bb") + self.assertEqual(repack.decompress_chunk(payload, "zstd"), b"aabb") + + def test_mismatch_new_key_fails(self): + with tempfile.TemporaryDirectory() as td: + db = pathlib.Path(td) / "old.db" + con = sqlite3.connect(db) + con.executescript(""" + create table cache(id integer primary key,name text,keypair text,is_public integer,store_dir text,priority integer,upstream_cache_key_names text,retention_period text,deleted_at text); + insert into cache values(1,'hectic','priv',1,'/nix/store',30,'[]',null,null); + """) + con.close() + mig = repack.Migrator(self.args(td, old_db=str(db))) + mig.old_client = mock.Mock(); mig.old_client.get_cache_config.return_value = {"public_key":"old"} + mig.new_client = mock.Mock(); mig.new_client.get_cache_config.return_value = {"public_key":"new","is_public":True,"store_dir":"/nix/store","priority":30,"upstream_cache_key_names":[]} + with self.assertRaises(repack.RepackError): + mig.init_cache() + + def test_init_requires_old_public_key(self): + with tempfile.TemporaryDirectory() as td: + db = pathlib.Path(td) / "old.db" + con = sqlite3.connect(db) + con.executescript(""" + create table cache(id integer primary key,name text,keypair text,is_public integer,store_dir text,priority integer,upstream_cache_key_names text,retention_period text,deleted_at text); + insert into cache values(1,'hectic','priv',1,'/nix/store',30,'[]',null,null); + """) + con.close() + mig = repack.Migrator(self.args(td, old_db=str(db))) + mig.old_client = mock.Mock(); mig.old_client.get_cache_config.return_value = {} + mig.new_client = mock.Mock(); mig.new_client.get_cache_config.return_value = None + with self.assertRaises(repack.RepackError): + mig.init_cache() + + def test_verify_readonly_no_put(self): + with tempfile.TemporaryDirectory() as td: + mig = repack.Migrator(self.args(td)) + record = self.record(b"abc") + mig.selected_records = lambda: [record] + mig.new_client = mock.Mock() + mig.new_client.get_narinfo.return_value = {**repack.expected_narinfo(record), "URL": "nar/x", "Compression": "none"} + mig.new_client.verify_payload.return_value = None + mig.new_client.narinfo_url.return_value = "http://127.0.0.1:8082/hectic/abcd.narinfo" + mig.old_client = mock.Mock() + mig.old_client.get_narinfo.return_value = {**repack.expected_narinfo(record)} + mig.migrate(True) + mig.new_client.upload.assert_not_called() + mig.new_client.verify_payload.assert_called_once() + + def test_key_redaction(self): + secret = "eyJhbGciOiPRIVATEKEYX-Amz-Signature=abc" + msg = repack.sanitized_error(RuntimeError("https://x/y?" + secret)) + self.assertEqual(msg, "RuntimeError") + self.assertNotIn(secret, msg) + + def test_old_new_narinfo_sig_compare_allows_db_sigs_empty(self): + with tempfile.TemporaryDirectory() as td: + mig = repack.Migrator(self.args(td)) + record = self.record(b"abc") + record["sigs"] = [] + old_info = {**repack.expected_narinfo(record), "Sig": ["hectic:sig"]} + new_info = dict(old_info) + mig.old_client = mock.Mock(); mig.old_client.get_narinfo.return_value = old_info + mig.new_client = mock.Mock(); mig.new_client.get_narinfo.return_value = new_info; mig.new_client.narinfo_url.return_value = "http://127.0.0.1/hectic/abcd.narinfo" + mig.new_client.verify_payload.return_value = None + self.assertTrue(mig.verify_record(record, False)) + + def test_migrate_record_missing_narinfo_raises(self): + with tempfile.TemporaryDirectory() as td: + mig = repack.Migrator(self.args(td)) + record = self.record(b"abc") + raw = mig.state.raw_path(repack.nar_hash_hex(record["nar_hash"])) + raw.write_bytes(b"abc") + mig.new_client = mock.Mock(); mig.new_client.get_narinfo.return_value = None + with self.assertRaises(repack.RepackError): + mig.migrate_record(record) + + def test_migrate_returns_failed_count(self): + with tempfile.TemporaryDirectory() as td: + mig = repack.Migrator(self.args(td)) + mig.selected_records = lambda: [self.record(b"abc")] + mig.migrate_record = mock.Mock(side_effect=repack.RepackError("new narinfo missing")) + self.assertEqual(mig.migrate(False), 1) + + def test_readonly_old_db_rejects_rw_and_missing(self): + with self.assertRaises(repack.RepackError): + repack.InventoryDB("file:/tmp/x.db?mode=rwc", "hectic") + with self.assertRaises(repack.RepackError): + repack.InventoryDB("/tmp/definitely-missing-attic.db", "hectic") + + def test_verify_hash_size_fail_closed_unknown_hash(self): + with self.assertRaises(repack.RepackError): + repack.verify_hash_size(b"x", "sha1:abc", 1, "chunk") + + def args(self, td, old_db=":memory:"): + return argparse.Namespace(old_db=old_db, state_dir=str(pathlib.Path(td) / "state"), old_url="http://127.0.0.1:8081", new_url="http://127.0.0.1:8082", host="cache.hectic-lab.com", cache="hectic", atticadm=None, server_config=None, nix="nix", old_storage_endpoint="https://hel1.your-objectstorage.com", old_bucket="cache-hectic-lab", old_region="hel1", workers=2, limit=None, paths_file=None) + + def record(self, data): + h = sha(data) + return {"cache":"hectic","nar_id":1,"store_path_hash":"abcd","store_path":"/nix/store/abcd-name","references":["/nix/store/ref-ref"],"system":"x86_64-linux","deriver":None,"sigs":["cache:sig"],"ca":None,"nar_hash":"sha256:" + h,"nar_size":len(data)} + + +if __name__ == "__main__": + unittest.main() diff --git a/nixos/system/hectic-lab/attic.nix b/nixos/system/hectic-lab/attic.nix index b6895226..5fe8cd59 100644 --- a/nixos/system/hectic-lab/attic.nix +++ b/nixos/system/hectic-lab/attic.nix @@ -3,9 +3,43 @@ ... }: { config, + lib, pkgs, ... -}: { +}: let + repackedActive = false; + migrationWriteFreeze = true; + + repackedSettings = config.services.atticd.settings // { + listen = "127.0.0.1:8082"; + allowed-hosts = [ "cache.${domain}" ]; + api-endpoint = if repackedActive then "https://cache.${domain}/" else "https://cache.${domain}/next/"; + substituter-endpoint = if repackedActive then "https://cache.${domain}/" else "https://cache.${domain}/next/"; + database.url = "sqlite:///var/lib/atticd-repacked/server.db?mode=rwc"; + storage = { + type = "s3"; + bucket = "nix-cache-hectic-lab"; + endpoint = "https://hel1.your-objectstorage.com"; + region = "hel1"; + }; + chunking = { + nar-size-threshold = 1048576; + min-size = 1048576; + avg-size = 2097152; + max-size = 4194304; + }; + compression.type = "zstd"; + }; + + repackedConfigFile = pkgs.runCommand "checked-atticd-repacked.toml" { + configFile = (pkgs.formats.toml { }).generate "server-repacked.toml" repackedSettings; + } '' + export ATTIC_SERVER_TOKEN_RS256_SECRET_BASE64="$(${lib.getExe pkgs.openssl} genrsa -traditional 4096 | ${pkgs.coreutils}/bin/base64 -w0)" + export ATTIC_SERVER_DATABASE_URL="sqlite://:memory:" + ${lib.getExe config.services.atticd.package} --mode check-config -f $configFile + cat <$configFile >$out + ''; +in { hectic.services.attic = { enable = true; hostName = "cache.${domain}"; @@ -32,6 +66,26 @@ ''; }); + services.atticd.settings = lib.mkIf repackedActive { + api-endpoint = lib.mkForce "https://cache.${domain}/previous/"; + substituter-endpoint = "https://cache.${domain}/previous/"; + }; + services.atticd.mode = if migrationWriteFreeze || repackedActive then "api-server" else "monolithic"; + + systemd.services.atticd-repacked = { + wantedBy = [ "multi-user.target" ]; + after = [ "network-online.target" ]; + wants = [ "network-online.target" ]; + + serviceConfig = config.systemd.services.atticd.serviceConfig // { + ExecStart = "${lib.getExe config.services.atticd.package} -f ${repackedConfigFile} --mode monolithic"; + EnvironmentFile = config.sops.secrets."atticd/environment".path; + StateDirectory = "atticd-repacked"; + User = "atticd-repacked"; + Group = "atticd-repacked"; + }; + }; + services.nginx.virtualHosts."cache.${domain}" = { enableACME = true; forceSSL = true; @@ -39,10 +93,33 @@ client_max_body_size 0; ''; locations."/" = { - proxyPass = "http://127.0.0.1:8081"; + proxyPass = if repackedActive then "http://127.0.0.1:8082" else "http://127.0.0.1:8081"; extraConfig = '' # Allow quiet periods while Attic fetches NAR chunks from object storage. proxy_read_timeout 300s; + '' + lib.optionalString (migrationWriteFreeze && !repackedActive) '' + # Quiesce the old writer during the final snapshot and verification. + limit_except GET { + deny all; + } + ''; + }; + locations."/next/" = { + proxyPass = "http://127.0.0.1:8082/"; + extraConfig = '' + # Allow quiet periods while Attic fetches NAR chunks from object storage. + proxy_read_timeout 300s; + ''; + }; + locations."/previous/" = { + proxyPass = "http://127.0.0.1:8081/"; + extraConfig = '' + # Legacy backend is exposed for read-only migration checks. + limit_except GET { + deny all; + } + # Allow quiet periods while Attic fetches NAR chunks from object storage. + proxy_read_timeout 300s; ''; }; };