1200 lines
42 KiB
Python
1200 lines
42 KiB
Python
from __future__ import annotations
|
|
|
|
import gzip
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import sys
|
|
import tarfile
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
MODULE_PATH = (
|
|
Path(__file__).resolve().parents[1]
|
|
/ "scripts"
|
|
/ "build-infrastructure-image-bundle.py"
|
|
)
|
|
SPEC = importlib.util.spec_from_file_location("build_infrastructure_image_bundle", MODULE_PATH)
|
|
assert SPEC is not None and SPEC.loader is not None
|
|
bundle = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = bundle
|
|
SPEC.loader.exec_module(bundle)
|
|
POLICY_PATH = (
|
|
MODULE_PATH.parents[1]
|
|
/ "release"
|
|
/ "linux-release-security-policy.v1.json"
|
|
)
|
|
|
|
|
|
def _sha(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def _record(path: Path, root: Path) -> dict[str, object]:
|
|
return {
|
|
"path": path.relative_to(root).as_posix(),
|
|
"sha256": _sha(path),
|
|
"size_bytes": path.stat().st_size,
|
|
}
|
|
|
|
|
|
def _write_canonical(path: Path, payload: object) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(bundle.canonical_json_bytes(payload))
|
|
|
|
|
|
def _with_hash(payload: dict[str, object], field: str) -> dict[str, object]:
|
|
result = dict(payload)
|
|
result[field] = bundle.canonical_sha256(result)
|
|
return result
|
|
|
|
|
|
def _make_image_artifacts(
|
|
root: Path,
|
|
*,
|
|
role: str,
|
|
repo_tag: str,
|
|
marker: str,
|
|
) -> tuple[
|
|
str,
|
|
str,
|
|
dict[str, object],
|
|
dict[str, object],
|
|
list[dict[str, object]],
|
|
]:
|
|
layer_payloads: list[bytes] = []
|
|
for sequence in (1, 2):
|
|
buffer = io.BytesIO()
|
|
content = f"{role}-layer-{sequence}-{marker}".encode("utf-8")
|
|
with tarfile.open(fileobj=buffer, mode="w") as layer:
|
|
item = tarfile.TarInfo(f"usr/share/{role}-{sequence}.txt")
|
|
item.size = len(content)
|
|
item.mode = 0o644
|
|
item.mtime = 0
|
|
item.uid = 0
|
|
item.gid = 0
|
|
item.uname = ""
|
|
item.gname = ""
|
|
layer.addfile(item, io.BytesIO(content))
|
|
layer_payloads.append(buffer.getvalue())
|
|
config = {
|
|
"architecture": "amd64",
|
|
"os": "linux",
|
|
"rootfs": {
|
|
"type": "layers",
|
|
"diff_ids": [
|
|
"sha256:" + hashlib.sha256(layer).hexdigest()
|
|
for layer in layer_payloads
|
|
],
|
|
},
|
|
}
|
|
config_raw = bundle.canonical_json_bytes(config)
|
|
config_digest = "sha256:" + hashlib.sha256(config_raw).hexdigest()
|
|
blob_records: list[dict[str, object]] = []
|
|
registry_layers: list[dict[str, object]] = []
|
|
for sequence, layer_raw in enumerate(layer_payloads, start=1):
|
|
compressed = gzip.compress(layer_raw, compresslevel=9, mtime=0)
|
|
blob_path = root / "registry-blobs" / f"{role}-{sequence}.tar.gz"
|
|
blob_path.parent.mkdir(parents=True, exist_ok=True)
|
|
blob_path.write_bytes(compressed)
|
|
blob_records.append(_record(blob_path, root))
|
|
registry_layers.append(
|
|
{
|
|
"mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
|
|
"digest": "sha256:" + hashlib.sha256(compressed).hexdigest(),
|
|
"size": len(compressed),
|
|
}
|
|
)
|
|
registry_manifest = {
|
|
"schemaVersion": 2,
|
|
"mediaType": "application/vnd.oci.image.manifest.v1+json",
|
|
"config": {
|
|
"mediaType": "application/vnd.oci.image.config.v1+json",
|
|
"digest": config_digest,
|
|
"size": len(config_raw),
|
|
},
|
|
"layers": registry_layers,
|
|
}
|
|
manifest_path = root / "registry-manifests" / f"{role}.json"
|
|
_write_canonical(manifest_path, registry_manifest)
|
|
manifest_digest = "sha256:" + _sha(manifest_path)
|
|
|
|
archive_path = root / "docker-archives" / f"{role}.docker.tar"
|
|
archive_path.parent.mkdir(parents=True, exist_ok=True)
|
|
config_name = config_digest.removeprefix("sha256:") + ".json"
|
|
layer_names = [
|
|
hashlib.sha256(layer).hexdigest() + "/layer.tar"
|
|
for layer in layer_payloads
|
|
]
|
|
docker_manifest = [
|
|
{"Config": config_name, "RepoTags": [repo_tag], "Layers": layer_names}
|
|
]
|
|
with tarfile.open(archive_path, mode="w") as archive:
|
|
members = [(config_name, config_raw), *zip(layer_names, layer_payloads, strict=True), (
|
|
"manifest.json",
|
|
bundle.canonical_json_bytes(docker_manifest),
|
|
)]
|
|
for name, raw in members:
|
|
item = tarfile.TarInfo(name)
|
|
item.size = len(raw)
|
|
item.mode = 0o644
|
|
archive.addfile(item, io.BytesIO(raw))
|
|
return (
|
|
manifest_digest,
|
|
config_digest,
|
|
_record(manifest_path, root),
|
|
_record(archive_path, root),
|
|
blob_records,
|
|
)
|
|
|
|
|
|
def _make_security_artifacts(
|
|
root: Path,
|
|
*,
|
|
role: str,
|
|
manifest_digest: str,
|
|
config_digest: str,
|
|
) -> tuple[dict[str, object], dict[str, object], dict[str, object]]:
|
|
sbom_path = root / "sbom" / f"{role}.spdx.json"
|
|
root_package = f"SPDXRef-Package-{role}"
|
|
_write_canonical(
|
|
sbom_path,
|
|
{
|
|
"spdxVersion": "SPDX-2.3",
|
|
"SPDXID": "SPDXRef-DOCUMENT",
|
|
"name": f"{role}-sbom",
|
|
"documentNamespace": f"https://example.invalid/spdx/{role}",
|
|
"creationInfo": {"creators": ["Tool: syft-1.18.1"]},
|
|
"documentDescribes": [root_package],
|
|
"packages": [
|
|
{
|
|
"SPDXID": root_package,
|
|
"name": role,
|
|
"versionInfo": manifest_digest,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
vulnerability_path = root / "scans" / f"{role}.vulnerability.json"
|
|
license_path = root / "scans" / f"{role}.license.json"
|
|
common = {
|
|
"SchemaVersion": 2,
|
|
"ArtifactName": f"{role}.docker.tar",
|
|
"ArtifactType": "container_image",
|
|
"Metadata": {"ImageID": config_digest},
|
|
}
|
|
_write_canonical(
|
|
vulnerability_path,
|
|
{
|
|
**common,
|
|
"Results": [
|
|
{
|
|
"Target": role,
|
|
"Class": "os-pkgs",
|
|
"Type": "alpine",
|
|
"Vulnerabilities": [],
|
|
}
|
|
],
|
|
},
|
|
)
|
|
_write_canonical(
|
|
license_path,
|
|
{
|
|
**common,
|
|
"Results": [
|
|
{
|
|
"Target": role,
|
|
"Class": "license",
|
|
"Type": "image",
|
|
"Licenses": [
|
|
{
|
|
"Name": "Apache-2.0",
|
|
"Category": "notice",
|
|
"Severity": "LOW",
|
|
"PkgName": f"fixture-{role}",
|
|
"FilePath": f"usr/share/doc/{role}/copyright",
|
|
}
|
|
],
|
|
}
|
|
],
|
|
},
|
|
)
|
|
return (
|
|
_record(sbom_path, root),
|
|
{
|
|
**_record(vulnerability_path, root),
|
|
"scanner": "trivy",
|
|
"scanner_version": bundle.REQUIRED_TOOL_VERSIONS["trivy"],
|
|
"waiver_evaluated_at": "2026-09-27T00:05:00Z",
|
|
"passed": True,
|
|
},
|
|
{
|
|
**_record(license_path, root),
|
|
"scanner": "trivy",
|
|
"scanner_version": bundle.REQUIRED_TOOL_VERSIONS["trivy"],
|
|
},
|
|
)
|
|
|
|
|
|
def _make_security_gate(root: Path) -> dict[str, object]:
|
|
policy_path = root / "policy/linux-release-security-policy.v1.json"
|
|
policy_path.parent.mkdir(parents=True, exist_ok=True)
|
|
policy_path.write_bytes(POLICY_PATH.read_bytes())
|
|
database_path = root / "security/trivy-db-metadata.json"
|
|
_write_canonical(
|
|
database_path,
|
|
{
|
|
"Version": bundle.REQUIRED_TOOL_VERSIONS["trivy"],
|
|
"VulnerabilityDB": {
|
|
"Version": 2,
|
|
"UpdatedAt": "2026-09-27T00:00:00Z",
|
|
"NextUpdate": "2026-09-27T06:00:00Z",
|
|
"DownloadedAt": "2026-09-27T00:01:00Z",
|
|
},
|
|
},
|
|
)
|
|
return {
|
|
"policy": _record(policy_path, root),
|
|
"database": {
|
|
**_record(database_path, root),
|
|
"validated_at": "2026-09-27T00:02:00Z",
|
|
"repository_transport": {
|
|
"reference": bundle.TRIVY_DB_REPOSITORY_TRANSPORT["reference"],
|
|
"transport": bundle.TRIVY_DB_REPOSITORY_TRANSPORT["transport"],
|
|
"schema_version": bundle.TRIVY_DB_REPOSITORY_TRANSPORT[
|
|
"schema_version"
|
|
],
|
|
"resolved_digest": "sha256:" + "a" * 64,
|
|
"media_type": bundle.TRIVY_DB_REPOSITORY_TRANSPORT[
|
|
"allowed_media_types"
|
|
][0],
|
|
"stable_during_download": True,
|
|
"before_download_inspect_sha256": "b" * 64,
|
|
"after_download_inspect_sha256": "c" * 64,
|
|
},
|
|
},
|
|
"scanner": "trivy",
|
|
"scanner_version": bundle.REQUIRED_TOOL_VERSIONS["trivy"],
|
|
}
|
|
|
|
|
|
def _fixture(root: Path) -> tuple[dict[str, object], dict[str, object]]:
|
|
commit = "1" * 40
|
|
tag = "v2-test-1"
|
|
security_gate = _make_security_gate(root)
|
|
app_images: list[dict[str, object]] = []
|
|
for index, (role, environment_key) in enumerate(bundle.APPLICATION_SPECS, start=1):
|
|
repo_tag = f"yuqei/{role}:{commit}"
|
|
manifest_digest, config_digest, _manifest, archive, _blobs = _make_image_artifacts(
|
|
root, role=role, repo_tag=repo_tag, marker=hex(index)[-1]
|
|
)
|
|
app_images.append(
|
|
{
|
|
"role": role,
|
|
"environment_key": environment_key,
|
|
"platform": "linux/amd64",
|
|
"manifest_digest": manifest_digest,
|
|
"config_digest": config_digest,
|
|
"image_ref": repo_tag,
|
|
"docker_archive": archive,
|
|
}
|
|
)
|
|
application = _with_hash(
|
|
{
|
|
"schema_version": bundle.APPLICATION_MANIFEST_SCHEMA,
|
|
"candidate": {"commit": commit, "tag": tag},
|
|
"policy": security_gate["policy"],
|
|
"images": app_images,
|
|
},
|
|
"manifest_payload_sha256",
|
|
)
|
|
|
|
repositories = (
|
|
"docker.io/pgvector/pgvector",
|
|
"docker.io/library/postgres",
|
|
"docker.io/library/redis",
|
|
"docker.io/minio/minio",
|
|
"docker.io/minio/mc",
|
|
"docker.io/library/nginx",
|
|
"docker.io/library/alpine",
|
|
)
|
|
tags = (
|
|
"pg16",
|
|
"16-alpine",
|
|
"7-alpine",
|
|
"RELEASE.2023-06-23T20-26-00Z",
|
|
"RELEASE.2023-06-23T18-12-07Z",
|
|
"1.27-alpine",
|
|
"3.21",
|
|
)
|
|
infra_images: list[dict[str, object]] = []
|
|
for index, ((role, environment_key, services), repository, image_tag) in enumerate(
|
|
zip(bundle.INFRASTRUCTURE_SPECS, repositories, tags, strict=True), start=8
|
|
):
|
|
offline_tag = f"yuqei-infra/{role}:offline"
|
|
manifest_digest, config_digest, manifest, archive, registry_blobs = _make_image_artifacts(
|
|
root, role=role, repo_tag=offline_tag, marker=hex(index)[-1]
|
|
)
|
|
sbom, vulnerability_scan, license_scan = _make_security_artifacts(
|
|
root,
|
|
role=role,
|
|
manifest_digest=manifest_digest,
|
|
config_digest=config_digest,
|
|
)
|
|
infra_images.append(
|
|
{
|
|
"role": role,
|
|
"environment_key": environment_key,
|
|
"services": list(services),
|
|
"canonical_source_tag": f"{repository}:{image_tag}",
|
|
"canonical_reference": f"{repository}@{manifest_digest}",
|
|
"transport_reference": (
|
|
f"docker.m.daocloud.io/{repository.removeprefix('docker.io/')}@{manifest_digest}"
|
|
),
|
|
"manifest_digest": manifest_digest,
|
|
"config_digest": config_digest,
|
|
"registry_manifest": manifest,
|
|
"registry_blobs": registry_blobs,
|
|
"docker_archive": archive,
|
|
"expected_repo_tag": offline_tag,
|
|
"sbom": sbom,
|
|
"vulnerability_scan": vulnerability_scan,
|
|
"license_scan": license_scan,
|
|
}
|
|
)
|
|
source_set = _with_hash(
|
|
{
|
|
"schema_version": bundle.SOURCE_SET_SCHEMA,
|
|
"candidate": {"commit": commit, "tag": tag},
|
|
"target": bundle.TARGET,
|
|
"security_gate": security_gate,
|
|
"images": infra_images,
|
|
},
|
|
"source_set_payload_sha256",
|
|
)
|
|
return source_set, application
|
|
|
|
|
|
def _rehash_source(source: dict[str, object]) -> dict[str, object]:
|
|
source = dict(source)
|
|
source.pop("source_set_payload_sha256", None)
|
|
return _with_hash(source, "source_set_payload_sha256")
|
|
|
|
|
|
def _valid_layer_bytes(marker: str) -> bytes:
|
|
buffer = io.BytesIO()
|
|
content = marker.encode("utf-8")
|
|
with tarfile.open(fileobj=buffer, mode="w") as layer:
|
|
item = tarfile.TarInfo("usr/share/changed.txt")
|
|
item.size = len(content)
|
|
item.mode = 0o644
|
|
item.mtime = 0
|
|
layer.addfile(item, io.BytesIO(content))
|
|
return buffer.getvalue()
|
|
|
|
|
|
def _layer_bytes_with_links() -> bytes:
|
|
buffer = io.BytesIO()
|
|
with tarfile.open(fileobj=buffer, mode="w") as layer:
|
|
directory = tarfile.TarInfo("bin")
|
|
directory.type = tarfile.DIRTYPE
|
|
directory.mode = 0o755
|
|
layer.addfile(directory)
|
|
|
|
executable = tarfile.TarInfo("bin/busybox")
|
|
payload = b"synthetic-busybox"
|
|
executable.size = len(payload)
|
|
executable.mode = 0o755
|
|
layer.addfile(executable, io.BytesIO(payload))
|
|
|
|
symbolic = tarfile.TarInfo("bin/sh")
|
|
symbolic.type = tarfile.SYMTYPE
|
|
symbolic.linkname = "/bin/busybox"
|
|
layer.addfile(symbolic)
|
|
|
|
hard = tarfile.TarInfo("bin/ash")
|
|
hard.type = tarfile.LNKTYPE
|
|
hard.linkname = "bin/busybox"
|
|
layer.addfile(hard)
|
|
return buffer.getvalue()
|
|
|
|
|
|
def _rewrite_archive_member(path: Path, member_name: str, replacement: bytes) -> None:
|
|
records: list[tuple[str, bytes]] = []
|
|
with tarfile.open(path, mode="r") as archive:
|
|
for item in archive.getmembers():
|
|
if not item.isfile():
|
|
continue
|
|
stream = archive.extractfile(item)
|
|
assert stream is not None
|
|
records.append(
|
|
(item.name, replacement if item.name == member_name else stream.read())
|
|
)
|
|
with tarfile.open(path, mode="w") as archive:
|
|
for name, raw in records:
|
|
item = tarfile.TarInfo(name)
|
|
item.size = len(raw)
|
|
item.mode = 0o644
|
|
archive.addfile(item, io.BytesIO(raw))
|
|
|
|
|
|
def _docker_manifest(path: Path) -> list[dict[str, object]]:
|
|
with tarfile.open(path, mode="r") as archive:
|
|
stream = archive.extractfile("manifest.json")
|
|
assert stream is not None
|
|
return json.loads(stream.read().decode("utf-8"))
|
|
|
|
|
|
def _archive_files(path: Path) -> dict[str, bytes]:
|
|
result: dict[str, bytes] = {}
|
|
with tarfile.open(path, mode="r") as archive:
|
|
for member in archive.getmembers():
|
|
if not member.isfile():
|
|
continue
|
|
stream = archive.extractfile(member)
|
|
assert stream is not None
|
|
result[member.name] = stream.read()
|
|
return result
|
|
|
|
|
|
def _write_archive_files(path: Path, files: dict[str, bytes]) -> None:
|
|
with tarfile.open(path, mode="w") as archive:
|
|
for name, raw in files.items():
|
|
item = tarfile.TarInfo(name)
|
|
item.size = len(raw)
|
|
item.mode = 0o644
|
|
archive.addfile(item, io.BytesIO(raw))
|
|
|
|
|
|
def _convert_archive_to_content_addressed(path: Path) -> tuple[list[dict[str, object]], list[bytes]]:
|
|
files = _archive_files(path)
|
|
manifest = json.loads(files["manifest.json"].decode("utf-8"))
|
|
item = manifest[0]
|
|
legacy_config = item["Config"]
|
|
config_raw = files.pop(legacy_config)
|
|
item["Config"] = "blobs/sha256/" + hashlib.sha256(config_raw).hexdigest()
|
|
files[item["Config"]] = config_raw
|
|
compressed_layers: list[bytes] = []
|
|
content_addressed_layers: list[str] = []
|
|
for legacy_layer in item["Layers"]:
|
|
layer_raw = files.pop(legacy_layer)
|
|
compressed = gzip.compress(layer_raw, compresslevel=9, mtime=0)
|
|
content_addressed_name = "blobs/sha256/" + hashlib.sha256(compressed).hexdigest()
|
|
files[content_addressed_name] = compressed
|
|
compressed_layers.append(compressed)
|
|
content_addressed_layers.append(content_addressed_name)
|
|
item["Layers"] = content_addressed_layers
|
|
files["manifest.json"] = bundle.canonical_json_bytes(manifest)
|
|
_write_archive_files(path, files)
|
|
return manifest, compressed_layers
|
|
|
|
|
|
def _refresh_registry_manifest_identity(
|
|
source: dict[str, object], root: Path, image_index: int
|
|
) -> dict[str, object]:
|
|
image = source["images"][image_index]
|
|
path = root / image["registry_manifest"]["path"]
|
|
digest = "sha256:" + _sha(path)
|
|
image["registry_manifest"] = _record(path, root)
|
|
image["manifest_digest"] = digest
|
|
for field in ("canonical_reference", "transport_reference"):
|
|
image[field] = image[field].split("@", 1)[0] + "@" + digest
|
|
return _rehash_source(source)
|
|
|
|
|
|
def test_bundle_is_deterministic_file_only_and_covers_all_eleven_images(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
first = bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
second = bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
assert first == second
|
|
assert first["target"] == bundle.TARGET
|
|
assert first["all_release_image_keys"] == list(bundle.ALL_IMAGE_KEYS)
|
|
assert list(first["offline_compose_overrides"]) == list(bundle.ALL_IMAGE_KEYS)
|
|
assert all(
|
|
value.startswith("sha256:")
|
|
for value in first["offline_compose_overrides"].values()
|
|
)
|
|
plan = first["docker_import_plan"]
|
|
assert plan["network_access_required"] is False
|
|
assert plan["docker_required_for_validation"] is False
|
|
assert plan["import_verified"] is False
|
|
assert plan["controlled_docker_load_and_inspect_receipt_required"] is True
|
|
assert len(plan["images"]) == 7
|
|
assert [item["role"] for item in plan["images"]] == [
|
|
role for role, _key, _services in bundle.INFRASTRUCTURE_SPECS
|
|
]
|
|
assert all(item["load_command_argv"][0] == "/usr/bin/docker" for item in plan["images"])
|
|
assert all(
|
|
item["controlled_docker_load_and_inspect_receipt_required"] is True
|
|
for item in plan["images"]
|
|
)
|
|
assert all(
|
|
item["archive_import_validation"]
|
|
== "requires_controlled_docker_load_and_inspect_receipt"
|
|
for item in first["infrastructure_images"]
|
|
)
|
|
assert all(
|
|
item["archive_profile"] == bundle.LEGACY_ARCHIVE_PROFILE
|
|
and item["archive_inspection"]["archive_profile"]
|
|
== bundle.LEGACY_ARCHIVE_PROFILE
|
|
for item in first["application_images"] + first["infrastructure_images"]
|
|
)
|
|
assert all(
|
|
item["archive_profile"] == bundle.LEGACY_ARCHIVE_PROFILE
|
|
for item in plan["images"]
|
|
)
|
|
assert all(
|
|
set(item) >= {
|
|
"sbom",
|
|
"sbom_summary",
|
|
"vulnerability_scan",
|
|
"vulnerability_scan_summary",
|
|
"license_scan",
|
|
"license_scan_summary",
|
|
}
|
|
for item in first["infrastructure_images"]
|
|
)
|
|
assert all(
|
|
item["sbom_summary"]["root_manifest_digest"] == item["manifest_digest"]
|
|
and item["vulnerability_scan_summary"]["config_digest"] == item["config_digest"]
|
|
and item["license_scan_summary"]["config_digest"] == item["config_digest"]
|
|
for item in first["infrastructure_images"]
|
|
)
|
|
assert first["bundle_payload_sha256"] == bundle.canonical_sha256(
|
|
{key: value for key, value in first.items() if key != "bundle_payload_sha256"}
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mutation", "message"),
|
|
(
|
|
(lambda source: source["images"].pop(), "exactly seven"),
|
|
(lambda source: source.update({"target": {"os": "linux", "architecture": "arm64", "platform": "linux/arm64"}}), "linux/amd64"),
|
|
(lambda source: source["images"][4].update({"canonical_source_tag": "docker.io/minio/mc:latest"}), "latest"),
|
|
(lambda source: source["images"][0].update({"services": ["postgres", "unexpected"]}), "mapping"),
|
|
),
|
|
)
|
|
def test_source_set_shape_and_platform_fail_closed(
|
|
tmp_path: Path, mutation, message: str
|
|
) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
mutation(source)
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match=message):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_archive_content_change_is_rejected_even_if_outer_source_hash_is_refreshed(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
record = source["images"][2]["docker_archive"]
|
|
archive = tmp_path / record["path"]
|
|
archive.write_bytes(archive.read_bytes() + b"tampered")
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match="size or SHA-256 mismatch"):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_registry_manifest_and_docker_archive_config_identity_must_match(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
source["images"][0]["config_digest"] = "sha256:" + "f" * 64
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match="config digest mismatch"):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_candidate_binding_between_application_and_infrastructure_is_required(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
application = dict(application)
|
|
application.pop("manifest_payload_sha256")
|
|
application["candidate"] = {"commit": "2" * 40, "tag": "v2-other"}
|
|
application = _with_hash(application, "manifest_payload_sha256")
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match="candidate bindings differ"):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("field", ("sbom", "vulnerability_scan", "license_scan"))
|
|
def test_every_infrastructure_security_artifact_is_required(
|
|
tmp_path: Path,
|
|
field: str,
|
|
) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
source["images"][0].pop(field)
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match="fields are missing or unknown"):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "message"),
|
|
(
|
|
("sbom", "SBOM root does not bind"),
|
|
("vulnerability_scan", "vulnerability gate failed"),
|
|
("license_scan", "license gate failed"),
|
|
),
|
|
)
|
|
def test_security_artifact_image_identity_drift_is_rejected(
|
|
tmp_path: Path,
|
|
field: str,
|
|
message: str,
|
|
) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
record = source["images"][0][field]
|
|
path = tmp_path / record["path"]
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
if field == "sbom":
|
|
payload["packages"][0]["versionInfo"] = "sha256:" + "f" * 64
|
|
else:
|
|
payload["Metadata"]["ImageID"] = "sha256:" + "f" * 64
|
|
_write_canonical(path, payload)
|
|
refreshed = _record(path, tmp_path)
|
|
source["images"][0][field] = {
|
|
**record,
|
|
**refreshed,
|
|
}
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match=message):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_infrastructure_vulnerability_gate_rejects_high_finding_even_when_claimed_pass(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
record = source["images"][0]["vulnerability_scan"]
|
|
path = tmp_path / record["path"]
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
payload["Results"][0]["Vulnerabilities"] = [
|
|
{
|
|
"VulnerabilityID": "CVE-2099-0001",
|
|
"PkgName": "unwaived-package",
|
|
"InstalledVersion": "1.0.0",
|
|
"FixedVersion": "1.0.1",
|
|
"Severity": "HIGH",
|
|
}
|
|
]
|
|
_write_canonical(path, payload)
|
|
source["images"][0]["vulnerability_scan"] = {
|
|
**record,
|
|
**_record(path, tmp_path),
|
|
}
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match="vulnerability gate failed"):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mutation", "message"),
|
|
(
|
|
(
|
|
lambda source: source["security_gate"].update(
|
|
{"scanner_version": "0.0.0"}
|
|
),
|
|
"scanner identity is not pinned",
|
|
),
|
|
(
|
|
lambda source: source["images"][0]["vulnerability_scan"].update(
|
|
{"waiver_evaluated_at": "2026-09-29T00:05:00Z"}
|
|
),
|
|
"database/freshness evidence is invalid",
|
|
),
|
|
),
|
|
)
|
|
def test_infrastructure_security_gate_rejects_scanner_or_database_drift(
|
|
tmp_path: Path,
|
|
mutation,
|
|
message: str,
|
|
) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
mutation(source)
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match=message):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_infrastructure_license_inventory_cannot_be_empty(tmp_path: Path) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
record = source["images"][0]["license_scan"]
|
|
path = tmp_path / record["path"]
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
payload["Results"][0]["Licenses"] = []
|
|
_write_canonical(path, payload)
|
|
source["images"][0]["license_scan"] = {
|
|
**record,
|
|
**_record(path, tmp_path),
|
|
}
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match="license inventory is empty"):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mutation", "message"),
|
|
(
|
|
(
|
|
lambda payload: payload.update({"SchemaVersion": 3}),
|
|
"SchemaVersion must be exactly 2",
|
|
),
|
|
(
|
|
lambda payload: payload["Results"][0].pop("Target"),
|
|
"Target is missing",
|
|
),
|
|
(
|
|
lambda payload: payload["Results"][0].pop("Class"),
|
|
"Class is missing",
|
|
),
|
|
(
|
|
lambda payload: payload["Results"][0].pop("Licenses"),
|
|
"lacks an explicit Licenses inventory",
|
|
),
|
|
(
|
|
lambda payload: payload["Results"][0]["Licenses"][0].pop(
|
|
"Category"
|
|
),
|
|
"has no category",
|
|
),
|
|
(
|
|
lambda payload: payload["Results"][0]["Licenses"][0].pop(
|
|
"Severity"
|
|
),
|
|
"has no severity",
|
|
),
|
|
(
|
|
lambda payload: (
|
|
payload["Results"][0]["Licenses"][0].pop("PkgName"),
|
|
payload["Results"][0]["Licenses"][0].pop("FilePath"),
|
|
),
|
|
"has no package or file identity",
|
|
),
|
|
),
|
|
)
|
|
def test_infrastructure_license_report_shape_fails_closed(
|
|
tmp_path: Path,
|
|
mutation,
|
|
message: str,
|
|
) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
record = source["images"][0]["license_scan"]
|
|
path = tmp_path / record["path"]
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
mutation(payload)
|
|
_write_canonical(path, payload)
|
|
source["images"][0]["license_scan"] = {
|
|
**record,
|
|
**_record(path, tmp_path),
|
|
}
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match=message):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_docker_layer_must_be_a_valid_tar_stream(tmp_path: Path) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
archive_record = source["images"][0]["docker_archive"]
|
|
archive_path = tmp_path / archive_record["path"]
|
|
layer_name = _docker_manifest(archive_path)[0]["Layers"][0]
|
|
_rewrite_archive_member(archive_path, layer_name, b"not-a-tar-layer")
|
|
source["images"][0]["docker_archive"] = _record(archive_path, tmp_path)
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match="valid uncompressed tar layer"):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_legacy_archive_rejects_gzip_encoded_layer_even_when_diff_id_matches(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
archive_record = source["images"][0]["docker_archive"]
|
|
archive_path = tmp_path / archive_record["path"]
|
|
layer_name = _docker_manifest(archive_path)[0]["Layers"][0]
|
|
with tarfile.open(archive_path, mode="r") as archive:
|
|
member = archive.getmember(layer_name)
|
|
stream = archive.extractfile(member)
|
|
assert stream is not None
|
|
raw_layer = stream.read()
|
|
_rewrite_archive_member(
|
|
archive_path,
|
|
layer_name,
|
|
gzip.compress(raw_layer, compresslevel=9, mtime=0),
|
|
)
|
|
source["images"][0]["docker_archive"] = _record(archive_path, tmp_path)
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(
|
|
bundle.BundleValidationError,
|
|
match="legacy Docker archive layers must be uncompressed",
|
|
):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_content_addressed_gzip_layer_binds_compressed_blob_digest(tmp_path: Path) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
archive_record = source["images"][0]["docker_archive"]
|
|
archive_path = tmp_path / archive_record["path"]
|
|
manifest, compressed_layers = _convert_archive_to_content_addressed(archive_path)
|
|
compressed = compressed_layers[0]
|
|
new_layer_name = manifest[0]["Layers"][0]
|
|
source["images"][0]["docker_archive"] = _record(archive_path, tmp_path)
|
|
source = _rehash_source(source)
|
|
|
|
result = bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
assert (
|
|
result["infrastructure_images"][0]["archive_profile"]
|
|
== bundle.CONTENT_ADDRESSED_ARCHIVE_PROFILE
|
|
)
|
|
assert (
|
|
result["infrastructure_images"][0]["archive_inspection"]["archive_profile"]
|
|
== bundle.CONTENT_ADDRESSED_ARCHIVE_PROFILE
|
|
)
|
|
inspection = result["infrastructure_images"][0]["archive_inspection"]["layers"][0]
|
|
assert inspection["content_addressed"] is True
|
|
assert inspection["archive_blob_digest"] == "sha256:" + hashlib.sha256(compressed).hexdigest()
|
|
|
|
# Re-encode the same uncompressed tar with different gzip metadata. The
|
|
# diff ID remains valid, but the content-addressed compressed blob digest
|
|
# no longer matches its member name and must be rejected.
|
|
tampered_name = manifest[0]["Layers"][1]
|
|
tampered = gzip.compress(
|
|
gzip.decompress(compressed_layers[1]), compresslevel=9, mtime=1
|
|
)
|
|
files = _archive_files(archive_path)
|
|
files[tampered_name] = tampered
|
|
_write_archive_files(archive_path, files)
|
|
source["images"][0]["docker_archive"] = _record(archive_path, tmp_path)
|
|
source = _rehash_source(source)
|
|
with pytest.raises(bundle.BundleValidationError, match="blob digest mismatch"):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_legacy_config_cannot_mix_in_a_content_addressed_layer(tmp_path: Path) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
archive_path = tmp_path / source["images"][0]["docker_archive"]["path"]
|
|
files = _archive_files(archive_path)
|
|
manifest = json.loads(files["manifest.json"].decode("utf-8"))
|
|
old_name = manifest[0]["Layers"][0]
|
|
compressed = gzip.compress(files.pop(old_name), compresslevel=9, mtime=0)
|
|
new_name = "blobs/sha256/" + hashlib.sha256(compressed).hexdigest()
|
|
manifest[0]["Layers"][0] = new_name
|
|
files[new_name] = compressed
|
|
files["manifest.json"] = bundle.canonical_json_bytes(manifest)
|
|
_write_archive_files(archive_path, files)
|
|
source["images"][0]["docker_archive"] = _record(archive_path, tmp_path)
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(
|
|
bundle.BundleValidationError,
|
|
match="legacy Docker archive config requires every layer",
|
|
):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_content_addressed_config_cannot_mix_in_a_legacy_layer(tmp_path: Path) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
archive_path = tmp_path / source["images"][0]["docker_archive"]["path"]
|
|
manifest, _compressed_layers = _convert_archive_to_content_addressed(archive_path)
|
|
files = _archive_files(archive_path)
|
|
old_name = manifest[0]["Layers"][1]
|
|
legacy_name = hashlib.sha256(files[old_name]).hexdigest() + "/layer.tar"
|
|
manifest[0]["Layers"][1] = legacy_name
|
|
files[legacy_name] = files.pop(old_name)
|
|
files["manifest.json"] = bundle.canonical_json_bytes(manifest)
|
|
_write_archive_files(archive_path, files)
|
|
source["images"][0]["docker_archive"] = _record(archive_path, tmp_path)
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(
|
|
bundle.BundleValidationError,
|
|
match="content-addressed Docker archive config requires every layer",
|
|
):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_content_addressed_profile_rejects_uncompressed_rename_downgrade(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
archive_path = tmp_path / source["images"][0]["docker_archive"]["path"]
|
|
files = _archive_files(archive_path)
|
|
manifest = json.loads(files["manifest.json"].decode("utf-8"))
|
|
legacy_config = manifest[0]["Config"]
|
|
config_raw = files.pop(legacy_config)
|
|
modern_config = "blobs/sha256/" + hashlib.sha256(config_raw).hexdigest()
|
|
manifest[0]["Config"] = modern_config
|
|
files[modern_config] = config_raw
|
|
modern_layers: list[str] = []
|
|
for legacy_layer in manifest[0]["Layers"]:
|
|
raw_layer = files.pop(legacy_layer)
|
|
modern_layer = "blobs/sha256/" + hashlib.sha256(raw_layer).hexdigest()
|
|
modern_layers.append(modern_layer)
|
|
files[modern_layer] = raw_layer
|
|
manifest[0]["Layers"] = modern_layers
|
|
files["manifest.json"] = bundle.canonical_json_bytes(manifest)
|
|
_write_archive_files(archive_path, files)
|
|
source["images"][0]["docker_archive"] = _record(archive_path, tmp_path)
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(
|
|
bundle.BundleValidationError,
|
|
match="content-addressed Docker archive layers must be gzip-compressed",
|
|
):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_legacy_profile_rejects_noncanonical_layer_layout(tmp_path: Path) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
archive_path = tmp_path / source["images"][0]["docker_archive"]["path"]
|
|
files = _archive_files(archive_path)
|
|
manifest = json.loads(files["manifest.json"].decode("utf-8"))
|
|
old_name = manifest[0]["Layers"][0]
|
|
renamed = "layers/" + old_name
|
|
manifest[0]["Layers"][0] = renamed
|
|
files[renamed] = files.pop(old_name)
|
|
files["manifest.json"] = bundle.canonical_json_bytes(manifest)
|
|
_write_archive_files(archive_path, files)
|
|
source["images"][0]["docker_archive"] = _record(archive_path, tmp_path)
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(
|
|
bundle.BundleValidationError,
|
|
match="legacy Docker archive config requires every layer",
|
|
):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_config_member_layout_must_select_a_supported_archive_profile(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
archive_path = tmp_path / source["images"][0]["docker_archive"]["path"]
|
|
files = _archive_files(archive_path)
|
|
manifest = json.loads(files["manifest.json"].decode("utf-8"))
|
|
old_name = manifest[0]["Config"]
|
|
renamed = "configs/" + old_name
|
|
manifest[0]["Config"] = renamed
|
|
files[renamed] = files.pop(old_name)
|
|
files["manifest.json"] = bundle.canonical_json_bytes(manifest)
|
|
_write_archive_files(archive_path, files)
|
|
source["images"][0]["docker_archive"] = _record(archive_path, tmp_path)
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(
|
|
bundle.BundleValidationError,
|
|
match="config member does not match a supported archive profile",
|
|
):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_linux_layer_symlinks_and_hardlinks_are_accepted_and_counted(tmp_path: Path) -> None:
|
|
link_layer = _layer_bytes_with_links()
|
|
inspection = bundle._inspect_uncompressed_layer(
|
|
io.BytesIO(link_layer),
|
|
expected_diff_id="sha256:" + hashlib.sha256(link_layer).hexdigest(),
|
|
label="synthetic Linux layer",
|
|
)
|
|
assert inspection["link_count"] == 2
|
|
|
|
|
|
def test_layer_link_target_with_control_character_is_rejected(tmp_path: Path) -> None:
|
|
buffer = io.BytesIO()
|
|
with tarfile.open(fileobj=buffer, mode="w") as layer:
|
|
link = tarfile.TarInfo("bin/sh")
|
|
link.type = tarfile.SYMTYPE
|
|
link.linkname = "bin/busybox\ninvalid"
|
|
layer.addfile(link)
|
|
invalid_layer = buffer.getvalue()
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match="invalid link target"):
|
|
bundle._inspect_uncompressed_layer(
|
|
io.BytesIO(invalid_layer),
|
|
expected_diff_id="sha256:" + hashlib.sha256(invalid_layer).hexdigest(),
|
|
label="synthetic Linux layer",
|
|
)
|
|
|
|
|
|
def test_docker_layer_bytes_must_match_config_diff_id(tmp_path: Path) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
archive_record = source["images"][0]["docker_archive"]
|
|
archive_path = tmp_path / archive_record["path"]
|
|
layer_name = _docker_manifest(archive_path)[0]["Layers"][0]
|
|
_rewrite_archive_member(archive_path, layer_name, _valid_layer_bytes("tampered"))
|
|
source["images"][0]["docker_archive"] = _record(archive_path, tmp_path)
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match="diff ID mismatch"):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_docker_layer_order_must_match_config_diff_id_order(tmp_path: Path) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
archive_record = source["images"][0]["docker_archive"]
|
|
archive_path = tmp_path / archive_record["path"]
|
|
manifest = _docker_manifest(archive_path)
|
|
manifest[0]["Layers"].reverse()
|
|
_rewrite_archive_member(
|
|
archive_path,
|
|
"manifest.json",
|
|
bundle.canonical_json_bytes(manifest),
|
|
)
|
|
source["images"][0]["docker_archive"] = _record(archive_path, tmp_path)
|
|
source = _rehash_source(source)
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match="diff ID mismatch"):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_registry_blob_tampering_and_order_are_rejected(tmp_path: Path) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
image = source["images"][0]
|
|
blob_path = tmp_path / image["registry_blobs"][0]["path"]
|
|
blob_path.write_bytes(blob_path.read_bytes() + b"tampered")
|
|
image["registry_blobs"][0] = _record(blob_path, tmp_path)
|
|
source = _rehash_source(source)
|
|
with pytest.raises(bundle.BundleValidationError, match="digest/size differs"):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
source, application = _fixture(tmp_path / "ordered")
|
|
source["images"][0]["registry_blobs"].reverse()
|
|
source = _rehash_source(source)
|
|
with pytest.raises(bundle.BundleValidationError, match="digest/size differs"):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path / "ordered",
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|
|
|
|
|
|
def test_registry_manifest_layer_size_must_match_staged_blob(tmp_path: Path) -> None:
|
|
source, application = _fixture(tmp_path)
|
|
image = source["images"][0]
|
|
manifest_path = tmp_path / image["registry_manifest"]["path"]
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
manifest["layers"][0]["size"] += 1
|
|
_write_canonical(manifest_path, manifest)
|
|
source = _refresh_registry_manifest_identity(source, tmp_path, 0)
|
|
|
|
with pytest.raises(bundle.BundleValidationError, match="digest/size differs"):
|
|
bundle.build_bundle(
|
|
artifact_root=tmp_path,
|
|
source_set=source,
|
|
application_manifest=application,
|
|
)
|