scripts/serve-sandbox-oci-archive.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Serve one owned Docker-save archive as a read-only loopback OCI registry.
No archive member is extracted to the host. The existing Firecrab OCI API still
downloads and verifies normal manifest/config/layer digests. This development
fixture has no remote registry connection, upload, authentication or deletion.
"""
import argparse
import hashlib
import gzip
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
from pathlib import Path, PurePosixPath
import re
import tarfile
from urllib.parse import unquote, urlsplit
MANIFEST_TYPE = "application/vnd.oci.image.manifest.v1+json"
CONFIG_TYPE = "application/vnd.oci.image.config.v1+json"
LAYER_TYPE = "application/vnd.oci.image.layer.v1.tar"
def encoded(value):
return json.dumps(value, separators=(",", ":"), sort_keys=True).encode()
def member(archive, name, limit):
if not isinstance(name, str) or name.startswith("/") or ".." in PurePosixPath(name).parts:
raise ValueError("unsafe archive member name")
entry = archive.getmember(name)
if not entry.isfile() or entry.size > limit:
raise ValueError("archive member must be a bounded regular file")
return entry
class Image:
def __init__(self, path, repository="firecrab/sandbox-seed"):
if not re.fullmatch(r"[a-z0-9]+(?:[._/-][a-z0-9]+)*", repository) or len(repository) > 128:
raise ValueError("invalid owned fixture repository")
self.path = Path(path).resolve(strict=True)
self.identity = self.signature()
self.repository = repository
self.blobs = {}
with tarfile.open(self.path, mode="r:") as archive:
saved = json.load(archive.extractfile(member(archive, "manifest.json", 4 * 1024 * 1024)))
if not isinstance(saved, list) or len(saved) != 1:
raise ValueError("docker save must contain exactly one selected image")
image = saved[0]
config_entry = member(archive, image["Config"], 4 * 1024 * 1024)
config_bytes = archive.extractfile(config_entry).read()
config = json.loads(config_bytes)
self.architecture = config.get("architecture")
if config.get("os") != "linux" or self.architecture not in ("arm64", "amd64"):
raise ValueError("fixture requires a Linux amd64/arm64 image")
config_digest = "sha256:" + hashlib.sha256(config_bytes).hexdigest()
self.blobs[config_digest] = config_entry.name, config_entry.size
layers = []
diff_ids = []
for name in image["Layers"]:
entry = member(archive, name, 1024 * 1024 * 1024)
digest = hashlib.sha256()
with archive.extractfile(entry) as stream:
while chunk := stream.read(1024 * 1024):
digest.update(chunk)
identity = "sha256:" + digest.hexdigest()
self.blobs[identity] = entry.name, entry.size
with archive.extractfile(entry) as stream:
magic=stream.read(4)
media_type=LAYER_TYPE
diff_id=identity
if magic.startswith(b"\x1f\x8b"):
media_type += "+gzip"
decoded=hashlib.sha256();total=0
with archive.extractfile(entry) as stream:
with gzip.GzipFile(fileobj=stream) as plain:
while chunk:=plain.read(1024*1024):
total+=len(chunk)
if total>1024*1024*1024:raise ValueError("decoded fixture layer exceeds 1 GiB")
decoded.update(chunk)
diff_id="sha256:"+decoded.hexdigest()
elif magic==b"\x28\xb5\x2f\xfd":
raise ValueError("zstd Docker-save layers are unsupported by this stdlib fixture; use gzip/plain layers")
diff_ids.append(diff_id)
layers.append({"mediaType": media_type, "size": entry.size, "digest": identity})
if config.get("rootfs", {}).get("diff_ids") != diff_ids:
raise ValueError("saved layer bytes do not match the image's uncompressed diff IDs")
self.manifest = encoded({"schemaVersion": 2, "mediaType": MANIFEST_TYPE,
"config": {"mediaType": CONFIG_TYPE, "size": config_entry.size, "digest": config_digest},
"layers": layers})
self.digest = "sha256:" + hashlib.sha256(self.manifest).hexdigest()
def signature(self):
status = self.path.stat()
return status.st_dev, status.st_ino, status.st_size, status.st_mtime_ns
def unchanged(self):
return self.signature() == self.identity
class RegistryHandler(BaseHTTPRequestHandler):
def log_message(self, *_):
pass
def do_GET(self):
self.respond(True)
def do_HEAD(self):
self.respond(False)
def respond(self, body):
image = self.server.image
if not image.unchanged():
self.send_error(503, "owned archive changed")
return
request = urlsplit(self.path)
path = unquote(request.path)
prefix = f"/v2/{image.repository}/"
data, content_type, digest = None, "application/json", None
blob = None
if request.query:
self.send_error(404)
return
if path in ("/v2/", "/v2"):
data = b"{}"
elif path in (prefix + "manifests/qa", prefix + "manifests/" + image.digest):
data, content_type, digest = image.manifest, MANIFEST_TYPE, image.digest
elif path == prefix + "tags/list":
data = encoded({"name": image.repository, "tags": ["qa"]})
elif path.startswith(prefix + "blobs/"):
digest = path.removeprefix(prefix + "blobs/")
blob = image.blobs.get(digest)
content_type = "application/octet-stream"
if data is None and blob is None:
self.send_error(404)
return
size = len(data) if data is not None else blob[1]
self.send_response(200)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(size))
self.send_header("Docker-Distribution-API-Version", "registry/2.0")
if digest:
self.send_header("Docker-Content-Digest", digest)
self.end_headers()
if not body:
return
try:
if data is not None:
self.wfile.write(data)
else:
with tarfile.open(image.path, mode="r:") as archive:
with archive.extractfile(blob[0]) as stream:
while chunk := stream.read(1024 * 1024):
self.wfile.write(chunk)
except (BrokenPipeError, ConnectionResetError):
return
class Registry(ThreadingHTTPServer):
daemon_threads = True
def __init__(self, image, port=0):
self.image = image
super().__init__(("127.0.0.1", port), RegistryHandler)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("archive", type=Path, help="owned uncompressed docker save archive")
parser.add_argument("--repository", default="firecrab/sandbox-seed")
parser.add_argument("--port", type=int, default=0)
args = parser.parse_args()
image = Image(args.archive, args.repository)
with Registry(image, args.port) as registry:
reference = f"127.0.0.1:{registry.server_port}/{image.repository}@{image.digest}"
print(json.dumps({"kind": "owned_loopback_oci_fixture", "reference": reference,
"digest": image.digest, "architecture": image.architecture,
"archiveBytes": image.identity[2], "manifestBytes": len(image.manifest)}, separators=(",", ":")), flush=True)
registry.serve_forever()
if __name__ == "__main__":
main()