308 lines
16 KiB
Python
308 lines
16 KiB
Python
# #region Tooling.DockerBundle.Archive [C:4] [TYPE Module] [SEMANTICS bundle,archive,identity]
|
|
# @BRIEF Validate image-save contents without extraction or trusting a filename/tag as image identity.
|
|
# @INVARIANT Every archive member is canonical regular/directory data; one exact image/config/layer graph is required.
|
|
# @RATIONALE Actual Docker exports include an OCI graph and Docker-compatible metadata; both can carry image references.
|
|
# @REJECTED Checking only manifest.json RepoTags leaves foreign OCI-index tags and extra images unchecked.
|
|
import hashlib
|
|
import json
|
|
import lzma
|
|
from pathlib import PurePosixPath
|
|
import re
|
|
import tarfile
|
|
|
|
HEX = re.compile(r"[0-9a-f]{64}")
|
|
IMAGE_ID = re.compile(r"sha256:[0-9a-f]{64}")
|
|
BLOB = re.compile(r"blobs/sha256/([0-9a-f]{64})")
|
|
MAX_METADATA = 16 * 1024 * 1024
|
|
MAX_METADATA_TOTAL = 64 * 1024 * 1024
|
|
MAX_MEMBERS = 100_000
|
|
OCI_MANIFEST = "application/vnd.oci.image.manifest.v1+json"
|
|
OCI_INDEX = "application/vnd.oci.image.index.v1+json"
|
|
MANIFEST_TYPES = {OCI_MANIFEST, "application/vnd.docker.distribution.manifest.v2+json"}
|
|
CONFIG_TYPES = {"application/vnd.oci.image.config.v1+json", "application/vnd.docker.container.image.v1+json"}
|
|
LAYER_TYPES = {"application/vnd.oci.image.layer.v1.tar", "application/vnd.docker.image.rootfs.diff.tar"}
|
|
|
|
# #region Tooling.DockerBundle.Refusal [C:1] [TYPE Class]
|
|
class Refusal(Exception):
|
|
pass
|
|
# #endregion Tooling.DockerBundle.Refusal
|
|
|
|
# #region Tooling.DockerBundle.Json [C:2] [TYPE Function]
|
|
def decode(raw):
|
|
def unique(pairs):
|
|
result = {}
|
|
for key, value in pairs:
|
|
if key in result:
|
|
raise Refusal("duplicate_json_key")
|
|
result[key] = value
|
|
return result
|
|
def constant(_):
|
|
raise Refusal("invalid_json")
|
|
try:
|
|
text = raw if isinstance(raw, str) else raw.decode("utf-8")
|
|
depth, quoted, escaped = 0, False, False
|
|
for char in text:
|
|
if quoted:
|
|
if escaped:
|
|
escaped = False
|
|
elif char == "\\":
|
|
escaped = True
|
|
elif char == '"':
|
|
quoted = False
|
|
elif char == '"':
|
|
quoted = True
|
|
elif char in "[{":
|
|
depth += 1
|
|
if depth > 64:
|
|
raise Refusal("json_depth_limit")
|
|
elif char in "]}":
|
|
depth -= 1
|
|
return json.loads(text, object_pairs_hook=unique, parse_constant=constant)
|
|
except (ValueError, UnicodeError, RecursionError):
|
|
raise Refusal("invalid_json") from None
|
|
# #endregion Tooling.DockerBundle.Json
|
|
|
|
# #region Tooling.DockerBundle.MemberPath [C:2] [TYPE Function]
|
|
def member_name(value, directory=False):
|
|
if not isinstance(value, str) or not value or len(value) > 512 or "\\" in value or "\x00" in value:
|
|
raise Refusal("unsafe_archive_path")
|
|
name = value[:-1] if directory and value.endswith("/") else value
|
|
path = PurePosixPath(name)
|
|
if path.is_absolute() or any(part in ("", ".", "..") for part in name.split("/")):
|
|
raise Refusal("unsafe_archive_path")
|
|
return name
|
|
# #endregion Tooling.DockerBundle.MemberPath
|
|
|
|
# #region Tooling.DockerBundle.Reader [C:3] [TYPE Class]
|
|
class Reader:
|
|
def __init__(self, stream):
|
|
self.stream, self.count, self.tail = stream, 0, b""
|
|
|
|
def read(self, size=-1):
|
|
raw = self.stream.read(size)
|
|
self.count += len(raw)
|
|
self.tail = (self.tail + raw)[-32768:]
|
|
return raw
|
|
# #endregion Tooling.DockerBundle.Reader
|
|
|
|
# #region Tooling.DockerBundle.Inventory [C:4] [TYPE Function]
|
|
def inventory(source):
|
|
source.seek(0)
|
|
entries, directories, metadata_size, end = {}, set(), 0, 0
|
|
with lzma.LZMAFile(source, "rb") as compressed:
|
|
reader = Reader(compressed)
|
|
with tarfile.open(fileobj=reader, mode="r|") as archive:
|
|
for item in archive:
|
|
name = member_name(item.name, item.isdir())
|
|
if name in entries or name in directories or len(entries) + len(directories) >= MAX_MEMBERS:
|
|
raise Refusal("duplicate_or_excess_archive_members")
|
|
if item.pax_headers or item.type not in (tarfile.REGTYPE, tarfile.AREGTYPE, tarfile.DIRTYPE) or item.size < 0:
|
|
raise Refusal("unsupported_archive_member")
|
|
end = item.offset_data + ((item.size + 511) // 512) * 512
|
|
if item.isdir():
|
|
if item.size:
|
|
raise Refusal("invalid_archive_directory")
|
|
directories.add(name)
|
|
continue
|
|
stream = archive.extractfile(item)
|
|
digest, data = hashlib.sha256(), bytearray()
|
|
first = stream.read(1 << 20)
|
|
retain = item.size <= MAX_METADATA and (name in ("manifest.json", "index.json", "oci-layout", "repositories")
|
|
or name.endswith("/VERSION") or first.lstrip().startswith((b"{", b"[")))
|
|
for chunk in iter(lambda: stream.read(1 << 20), b""):
|
|
digest.update(first)
|
|
if retain:
|
|
data.extend(first)
|
|
first = chunk
|
|
digest.update(first)
|
|
if retain:
|
|
data.extend(first)
|
|
metadata_size += len(data)
|
|
if metadata_size > MAX_METADATA_TOTAL:
|
|
raise Refusal("metadata_limit")
|
|
value = None
|
|
if data and not name.endswith("/VERSION"):
|
|
try:
|
|
value = decode(data)
|
|
except Refusal:
|
|
if name in ("manifest.json", "index.json", "oci-layout", "repositories"):
|
|
raise
|
|
entries[name] = {"size": item.size, "sha256": digest.hexdigest(), "json": value,
|
|
"version": bytes(data) if name.endswith("/VERSION") else None}
|
|
buffered = reader.count - end
|
|
if buffered < 0 or buffered > len(reader.tail) or (buffered and any(reader.tail[-buffered:])):
|
|
raise Refusal("trailing_tar_data")
|
|
while True:
|
|
chunk = reader.read(1 << 20)
|
|
if not chunk:
|
|
break
|
|
if any(chunk):
|
|
raise Refusal("trailing_tar_data")
|
|
if reader.count - end < 1024:
|
|
raise Refusal("incomplete_tar_terminator")
|
|
return entries, directories
|
|
# #endregion Tooling.DockerBundle.Inventory
|
|
|
|
# #region Tooling.DockerBundle.Record [C:2] [TYPE Function]
|
|
def record(entries, name, limit=MAX_METADATA):
|
|
name = member_name(name)
|
|
item = entries.get(name)
|
|
if item is None or item["size"] > limit or item["json"] is None:
|
|
raise Refusal("missing_or_invalid_archive_metadata")
|
|
return item["json"]
|
|
# #endregion Tooling.DockerBundle.Record
|
|
|
|
# #region Tooling.DockerBundle.Descriptor [C:3] [TYPE Function]
|
|
def descriptor(entries, value, media, root=False):
|
|
required = {"mediaType", "digest", "size"}
|
|
if (not isinstance(value, dict) or not required <= set(value)
|
|
or set(value) - required - ({"annotations", "platform"} if root else set())
|
|
or not isinstance(value["mediaType"], str) or value["mediaType"] not in media
|
|
or not isinstance(value["digest"], str) or not IMAGE_ID.fullmatch(value["digest"])
|
|
or type(value["size"]) is not int or value["size"] < 0):
|
|
raise Refusal("invalid_oci_descriptor")
|
|
name = "blobs/sha256/" + value["digest"][7:]
|
|
if "platform" in value:
|
|
platform = value["platform"]
|
|
if (not isinstance(platform, dict) or not {"os", "architecture"} <= set(platform)
|
|
or set(platform) - {"os", "architecture", "variant"}
|
|
or any(not isinstance(item, str) or re.fullmatch(r"[a-z0-9_+-]{1,64}", item) is None for item in platform.values())):
|
|
raise Refusal("invalid_oci_platform")
|
|
if name not in entries or entries[name]["sha256"] != value["digest"][7:] or entries[name]["size"] != value["size"]:
|
|
raise Refusal("oci_descriptor_content_conflict")
|
|
return name
|
|
# #endregion Tooling.DockerBundle.Descriptor
|
|
|
|
# #region Tooling.DockerBundle.Oci [C:4] [TYPE Function]
|
|
def oci_graph(entries, image, config, layers):
|
|
if record(entries, "oci-layout", 1024) != {"imageLayoutVersion": "1.0.0"}:
|
|
raise Refusal("unsupported_oci_layout")
|
|
index = record(entries, "index.json", 1 << 20)
|
|
if (not isinstance(index, dict) or set(index) != {"schemaVersion", "mediaType", "manifests"}
|
|
or type(index["schemaVersion"]) is not int or index["schemaVersion"] != 2
|
|
or index["mediaType"] != OCI_INDEX or not isinstance(index["manifests"], list)
|
|
or len(index["manifests"]) != 1):
|
|
raise Refusal("unsupported_or_multi_image_oci_index")
|
|
root = index["manifests"][0]
|
|
manifest_path = descriptor(entries, root, MANIFEST_TYPES, root=True)
|
|
annotations = root.get("annotations")
|
|
if (not isinstance(annotations, dict)
|
|
or set(annotations) != {"io.containerd.image.name", "org.opencontainers.image.ref.name"}
|
|
or annotations.get("io.containerd.image.name") != "docker.io/library/" + image["image"]
|
|
or annotations.get("org.opencontainers.image.ref.name") != image["image"].split(":", 1)[1]):
|
|
raise Refusal("foreign_oci_tag")
|
|
manifest = record(entries, manifest_path, 1 << 20)
|
|
if (not isinstance(manifest, dict) or set(manifest) != {"schemaVersion", "mediaType", "config", "layers"}
|
|
or type(manifest["schemaVersion"]) is not int or manifest["schemaVersion"] != 2
|
|
or manifest["mediaType"] not in MANIFEST_TYPES or manifest["mediaType"] != root["mediaType"]
|
|
or not isinstance(manifest["layers"], list)):
|
|
raise Refusal("invalid_oci_manifest")
|
|
if descriptor(entries, manifest["config"], CONFIG_TYPES) != config:
|
|
raise Refusal("oci_config_conflict")
|
|
paths = [descriptor(entries, item, LAYER_TYPES) for item in manifest["layers"]]
|
|
if paths != layers:
|
|
raise Refusal("oci_layer_conflict")
|
|
return {"oci-layout", "index.json", manifest_path}
|
|
# #endregion Tooling.DockerBundle.Oci
|
|
|
|
# #region Tooling.DockerBundle.Ancillary [C:3] [TYPE Function]
|
|
def ancillary(entries, remaining, layers, modern):
|
|
if not modern:
|
|
allowed = {str(PurePosixPath(layer).parent / name) for layer in layers for name in ("json", "VERSION")}
|
|
if not remaining <= allowed:
|
|
raise Refusal("extra_archive_image_or_file")
|
|
for name in remaining:
|
|
if name.endswith("/VERSION") and entries[name]["version"] not in (b"1.0", b"1.0\n"):
|
|
raise Refusal("legacy_layer_metadata_invalid")
|
|
if name.endswith("/json"):
|
|
value = record(entries, name)
|
|
if (not isinstance(value, dict) or value.get("id") != str(PurePosixPath(name).parent)
|
|
or set(value) - {"id", "parent", "created", "container_config", "container", "docker_version",
|
|
"author", "config", "architecture", "os", "size"}):
|
|
raise Refusal("legacy_layer_metadata_invalid")
|
|
return
|
|
nodes = {}
|
|
allowed_keys = {"id", "parent", "created", "container_config", "container", "docker_version",
|
|
"author", "config", "architecture", "os", "size"}
|
|
for name in remaining:
|
|
value = record(entries, name)
|
|
if (not BLOB.fullmatch(name) or not isinstance(value, dict) or not {"id", "created", "container_config", "os"} <= set(value)
|
|
or set(value) - allowed_keys or not isinstance(value["id"], str) or not HEX.fullmatch(value["id"])
|
|
or value["id"] in nodes):
|
|
raise Refusal("extra_archive_image_or_file")
|
|
nodes[value["id"]] = value.get("parent")
|
|
if nodes:
|
|
roots = [key for key, parent in nodes.items() if parent in (None, "")]
|
|
if len(nodes) != len(layers) or len(roots) != 1:
|
|
raise Refusal("legacy_layer_chain_invalid")
|
|
visited, current = set(), roots[0]
|
|
while current not in visited:
|
|
visited.add(current)
|
|
children = [key for key, parent in nodes.items() if parent == current]
|
|
if not children:
|
|
break
|
|
if len(children) != 1:
|
|
raise Refusal("legacy_layer_chain_invalid")
|
|
current = children[0]
|
|
if len(visited) != len(nodes):
|
|
raise Refusal("legacy_layer_chain_invalid")
|
|
# #endregion Tooling.DockerBundle.Ancillary
|
|
|
|
# #region Tooling.DockerBundle.ValidateArchive [C:4] [TYPE Function]
|
|
def validate_archive(source, image):
|
|
entries, directories = inventory(source)
|
|
manifest = record(entries, "manifest.json", 1 << 20)
|
|
if not isinstance(manifest, list) or len(manifest) != 1 or not isinstance(manifest[0], dict):
|
|
raise Refusal("extra_or_missing_archive_image")
|
|
item = manifest[0]
|
|
if set(item) not in ({"Config", "RepoTags", "Layers"}, {"Config", "RepoTags", "Layers", "LayerSources"}):
|
|
raise Refusal("unsupported_docker_save_manifest")
|
|
if item["RepoTags"] != [image["image"]] or not isinstance(item["Layers"], list):
|
|
raise Refusal("foreign_archive_tag")
|
|
config = member_name(item["Config"])
|
|
modern = BLOB.fullmatch(config) is not None
|
|
expected = image["image_id"][7:]
|
|
if config != ("blobs/sha256/" + expected if modern else expected + ".json"):
|
|
raise Refusal("config_identity_conflict")
|
|
value = record(entries, config)
|
|
if entries[config]["sha256"] != expected or not isinstance(value, dict):
|
|
raise Refusal("config_digest_conflict")
|
|
layers = [member_name(name) for name in item["Layers"]]
|
|
if len(layers) != len(set(layers)) or any(name not in entries for name in layers):
|
|
raise Refusal("missing_or_duplicate_layers")
|
|
if any(not (BLOB.fullmatch(name) if modern else re.fullmatch(r"[0-9a-f]{64}/layer\.tar", name)) for name in layers):
|
|
raise Refusal("invalid_layer_path")
|
|
rootfs = value.get("rootfs")
|
|
if not isinstance(rootfs, dict) or rootfs.get("type") != "layers" or rootfs.get("diff_ids") != ["sha256:" + entries[name]["sha256"] for name in layers]:
|
|
raise Refusal("layer_content_conflict")
|
|
for name in entries:
|
|
match = BLOB.fullmatch(name)
|
|
if match and entries[name]["sha256"] != match[1]:
|
|
raise Refusal("blob_digest_conflict")
|
|
used = {"manifest.json", config, *layers}
|
|
if modern:
|
|
used |= oci_graph(entries, image, config, layers)
|
|
elif "index.json" in entries or "oci-layout" in entries:
|
|
raise Refusal("mixed_archive_layout")
|
|
if "LayerSources" in item:
|
|
sources = item["LayerSources"]
|
|
if not modern or not isinstance(sources, dict) or set(sources) != {"sha256:" + entries[name]["sha256"] for name in layers}:
|
|
raise Refusal("layer_sources_conflict")
|
|
for key, value in sources.items():
|
|
path = descriptor(entries, value, LAYER_TYPES)
|
|
if value["digest"] != key or path not in layers:
|
|
raise Refusal("layer_sources_conflict")
|
|
if "repositories" in entries:
|
|
repository, tag = image["image"].split(":", 1)
|
|
top = PurePosixPath(layers[-1]).name if modern and layers else (str(PurePosixPath(layers[-1]).parent) if layers else None)
|
|
if record(entries, "repositories", 1 << 20) != {repository: {tag: top}}:
|
|
raise Refusal("foreign_repository_tag")
|
|
used.add("repositories")
|
|
ancillary(entries, set(entries) - used, layers, modern)
|
|
permitted_dirs = {str(parent) for name in entries for parent in PurePosixPath(name).parents if str(parent) != "."}
|
|
if not directories <= permitted_dirs:
|
|
raise Refusal("extra_archive_directory")
|
|
return {"format": "docker-save-oci" if modern else "docker-save-classic", "members": len(entries), "layers": len(layers)}
|
|
# #endregion Tooling.DockerBundle.ValidateArchive
|
|
# #endregion Tooling.DockerBundle.Archive
|