209 lines
10 KiB
Python
209 lines
10 KiB
Python
# #region Test.Tooling.DockerBundle [C:3] [TYPE Module] [SEMANTICS bundle,import,fixtures]
|
|
# @TEST_INVARIANT Synthetic real tar/xz files exercise validators; Docker is the only substituted external boundary.
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import lzma
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
import import_docker_bundle as importer
|
|
|
|
TAG = "fixture-v1"
|
|
|
|
# #region Test.Tooling.DockerBundle.Fixture [C:3] [TYPE Function]
|
|
def archive_bytes(image, modern=False, extra=None):
|
|
layer = b"fixture-layer"
|
|
layer_id = hashlib.sha256(layer).hexdigest()
|
|
config = json.dumps({"rootfs": {"type": "layers", "diff_ids": ["sha256:" + layer_id]}}).encode()
|
|
config_id = hashlib.sha256(config).hexdigest()
|
|
config_path = "blobs/sha256/" + config_id if modern else config_id + ".json"
|
|
layer_path = "blobs/sha256/" + layer_id if modern else layer_id + "/layer.tar"
|
|
members = {config_path: config, layer_path: layer}
|
|
compatible = {"Config": config_path, "RepoTags": [image], "Layers": [layer_path]}
|
|
if modern:
|
|
def descriptor(media, raw):
|
|
return {"mediaType": media, "digest": "sha256:" + hashlib.sha256(raw).hexdigest(), "size": len(raw)}
|
|
record = {"schemaVersion": 2, "mediaType": "application/vnd.oci.image.manifest.v1+json",
|
|
"config": descriptor("application/vnd.oci.image.config.v1+json", config),
|
|
"layers": [descriptor("application/vnd.oci.image.layer.v1.tar", layer)]}
|
|
raw = json.dumps(record).encode()
|
|
pointer = descriptor(record["mediaType"], raw)
|
|
pointer["annotations"] = {"io.containerd.image.name": "docker.io/library/" + image,
|
|
"org.opencontainers.image.ref.name": image.split(":")[1]}
|
|
members["blobs/sha256/" + hashlib.sha256(raw).hexdigest()] = raw
|
|
members["oci-layout"] = b'{"imageLayoutVersion":"1.0.0"}'
|
|
members["index.json"] = json.dumps({"schemaVersion": 2, "mediaType": "application/vnd.oci.image.index.v1+json", "manifests": [pointer]}).encode()
|
|
compatible["LayerSources"] = {"sha256:" + layer_id: record["layers"][0]}
|
|
members["repositories"] = json.dumps({image.split(":")[0]: {image.split(":")[1]: layer_id}}).encode()
|
|
members["manifest.json"] = json.dumps([compatible]).encode()
|
|
if extra:
|
|
extra(members)
|
|
target = io.BytesIO()
|
|
with tarfile.open(fileobj=target, mode="w") as archive:
|
|
for name, raw in members.items():
|
|
item = tarfile.TarInfo(name)
|
|
item.size = len(raw)
|
|
archive.addfile(item, io.BytesIO(raw))
|
|
return lzma.compress(target.getvalue()), "sha256:" + config_id
|
|
|
|
|
|
def bundle(root, modern=False):
|
|
images = []
|
|
for role in importer.ROLES:
|
|
image = f"superset-tools-{role}:{TAG}"
|
|
archive, identity = archive_bytes(image, modern)
|
|
name = f"superset-tools-{role}.{TAG}.tar.xz"
|
|
(root / name).write_bytes(archive)
|
|
images.append({"role": role, "image": image, "image_id": identity, "repo_digest": "unavailable", "archive": name})
|
|
compose, env = b"services: {}\n", b"# Synthetic public env template\n"
|
|
(root / "docker-compose.enterprise-clean.yml").write_bytes(compose)
|
|
(root / ".env.example").write_bytes(env)
|
|
value = {"manifest_version": 2, "bundle_kind": "enterprise", "release_tag": TAG,
|
|
"generated_at_utc": "2026-10-23T00:00:00Z", "images": images,
|
|
"compose_file": "docker-compose.enterprise-clean.yml", "env_template": ".env.example",
|
|
"compose_sha256": hashlib.sha256(compose).hexdigest(), "env_template_sha256": hashlib.sha256(env).hexdigest(),
|
|
"env_bootstrap_fields": importer.ENV_FIELDS, "checksums_file": f"sha256sums.{TAG}.txt"}
|
|
write_manifest(root, value)
|
|
return value
|
|
|
|
|
|
def write_manifest(root, value):
|
|
(root / "manifest.json").write_text(json.dumps(value))
|
|
(root / value["checksums_file"]).write_text("".join(hashlib.sha256((root / image["archive"]).read_bytes()).hexdigest() + " " + image["archive"] + "\n" for image in value["images"] if (root / image["archive"]).is_file()))
|
|
# #endregion Test.Tooling.DockerBundle.Fixture
|
|
|
|
# #region Test.Tooling.DockerBundle.ReadOnly [C:3] [TYPE Function]
|
|
@pytest.mark.parametrize("modern", [False, True])
|
|
def test_verify_real_archive_formats_without_docker(tmp_path, modern):
|
|
bundle(tmp_path, modern)
|
|
actual = subprocess.run
|
|
def boundary(argv, **kwargs):
|
|
assert argv[0] != "docker"
|
|
return actual(argv, **kwargs)
|
|
with patch.object(importer.subprocess, "run", side_effect=boundary):
|
|
rc, result = importer.execute(tmp_path, "manifest.json", TAG, True)
|
|
assert rc == 0, result
|
|
assert result["verified_roles"] == ["postgres", "backend", "frontend"]
|
|
assert result["loaded_roles"] == []
|
|
assert result["source_identity"]["status"] == "unavailable"
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["wrong_release", "duplicate_role", "foreign_id", "traversal", "missing", "checksum", "xz", "symlink", "unversioned"])
|
|
def test_negative_bundle_refuses_before_any_docker_or_load(tmp_path, change):
|
|
value = bundle(tmp_path)
|
|
target = tmp_path / value["images"][2]["archive"]
|
|
if change == "wrong_release":
|
|
value["release_tag"] = "foreign"
|
|
elif change == "duplicate_role":
|
|
value["images"][2] = value["images"][0]
|
|
elif change == "foreign_id":
|
|
value["images"][2]["image_id"] = "sha256:" + "f" * 64
|
|
elif change == "traversal":
|
|
value["images"][2]["archive"] = "../foreign.tar.xz"
|
|
elif change == "missing":
|
|
target.unlink()
|
|
elif change == "xz":
|
|
target.write_bytes(target.read_bytes()[:-8])
|
|
elif change == "symlink":
|
|
other = tmp_path / "other"
|
|
target.rename(other)
|
|
target.symlink_to(other)
|
|
elif change == "unversioned":
|
|
value.pop("manifest_version")
|
|
write_manifest(tmp_path, value)
|
|
if change == "checksum":
|
|
target.write_bytes(b"tampered")
|
|
actual = subprocess.run
|
|
actual_popen = subprocess.Popen
|
|
def boundary(argv, **kwargs):
|
|
assert argv[0] != "docker"
|
|
return actual(argv, **kwargs)
|
|
def popen_boundary(argv, **kwargs):
|
|
assert argv[0] != "docker"
|
|
return actual_popen(argv, **kwargs)
|
|
with patch.object(importer.subprocess, "run", side_effect=boundary), patch.object(importer.subprocess, "Popen", side_effect=popen_boundary):
|
|
rc, result = importer.execute(tmp_path, "manifest.json", TAG)
|
|
assert rc == 2, result
|
|
assert result["loaded_roles"] == []
|
|
assert not result["partial_load_possible"]
|
|
# #endregion Test.Tooling.DockerBundle.ReadOnly
|
|
|
|
# #region Test.Tooling.DockerBundle.ClosedMetadata [C:3] [TYPE Function]
|
|
@pytest.mark.parametrize("attack", ["root_annotation", "config_annotation", "config_platform", "layer_sources"])
|
|
def test_foreign_oci_metadata_and_layer_source_substitution_refuse(tmp_path, attack):
|
|
value = bundle(tmp_path, modern=True)
|
|
image = value["images"][2]
|
|
def mutate(members):
|
|
index = json.loads(members["index.json"])
|
|
pointer = index["manifests"][0]
|
|
old = "blobs/sha256/" + pointer["digest"][7:]
|
|
manifest = json.loads(members[old])
|
|
if attack == "root_annotation":
|
|
pointer["annotations"]["org.opencontainers.image.base.name"] = "foreign:latest"
|
|
elif attack == "config_annotation":
|
|
manifest["config"]["annotations"] = {"io.containerd.image.name": "docker.io/library/foreign:latest"}
|
|
elif attack == "config_platform":
|
|
manifest["config"]["platform"] = {"os": "linux", "architecture": "amd64"}
|
|
else:
|
|
compatible = json.loads(members["manifest.json"])
|
|
key = next(iter(compatible[0]["LayerSources"]))
|
|
compatible[0]["LayerSources"][key] = {**manifest["config"], "mediaType": "application/vnd.oci.image.layer.v1.tar"}
|
|
members["manifest.json"] = json.dumps(compatible).encode()
|
|
raw = json.dumps(manifest).encode()
|
|
members.pop(old)
|
|
pointer.update(digest="sha256:" + hashlib.sha256(raw).hexdigest(), size=len(raw))
|
|
members["blobs/sha256/" + pointer["digest"][7:]] = raw
|
|
members["index.json"] = json.dumps(index).encode()
|
|
raw, _ = archive_bytes(image["image"], True, mutate)
|
|
(tmp_path / image["archive"]).write_bytes(raw)
|
|
write_manifest(tmp_path, value)
|
|
rc, result = importer.execute(tmp_path, "manifest.json", TAG, True)
|
|
assert rc == 2, result
|
|
assert result["loaded_roles"] == []
|
|
|
|
|
|
def test_deep_json_is_typed_refusal_without_recursion_traceback(tmp_path):
|
|
(tmp_path / "manifest.json").write_bytes(b"[" * 10000 + b"]" * 10000)
|
|
rc, result = importer.execute(tmp_path, "manifest.json", TAG, True)
|
|
assert rc == 2
|
|
assert result["code"] == "json_depth_limit"
|
|
|
|
|
|
@pytest.mark.parametrize("attack", ["traversal", "symlink", "hardlink", "fifo", "duplicate", "extra_image", "unexpected_file"])
|
|
def test_outer_tar_hostile_members_refuse_before_import(tmp_path, attack):
|
|
value = bundle(tmp_path)
|
|
image = value["images"][-1]
|
|
if attack == "extra_image":
|
|
def extra(members):
|
|
manifest = json.loads(members["manifest.json"])
|
|
manifest.append({**manifest[0], "RepoTags": ["foreign:latest"]})
|
|
members["manifest.json"] = json.dumps(manifest).encode()
|
|
raw, _ = archive_bytes(image["image"], extra=extra)
|
|
else:
|
|
target = io.BytesIO(lzma.decompress((tmp_path / image["archive"]).read_bytes()))
|
|
with tarfile.open(fileobj=target, mode="a") as archive:
|
|
name = "../escape" if attack == "traversal" else ("manifest.json" if attack == "duplicate" else "unexpected")
|
|
entry = tarfile.TarInfo(name)
|
|
if attack in ("symlink", "hardlink", "fifo"):
|
|
entry.type = {"symlink": tarfile.SYMTYPE, "hardlink": tarfile.LNKTYPE, "fifo": tarfile.FIFOTYPE}[attack]
|
|
entry.linkname = "../outside"
|
|
archive.addfile(entry)
|
|
else:
|
|
entry.size = 2
|
|
archive.addfile(entry, io.BytesIO(b"{}"))
|
|
raw = lzma.compress(target.getvalue())
|
|
(tmp_path / image["archive"]).write_bytes(raw)
|
|
write_manifest(tmp_path, value)
|
|
rc, result = importer.execute(tmp_path, "manifest.json", TAG, True)
|
|
assert rc == 2
|
|
assert result["loaded_roles"] == []
|
|
# #endregion Test.Tooling.DockerBundle.ClosedMetadata
|
|
# #endregion Test.Tooling.DockerBundle
|