315 lines
14 KiB
Python
315 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
# #region Tooling.DockerBundle.Import [C:4] [TYPE Module] [SEMANTICS bundle,import,identity]
|
|
# @BRIEF Verify a complete enterprise-v2 image bundle and bound deployment metadata before Docker operations.
|
|
# @INVARIANT All archive/checksum identities pass before Docker mutation; verified open descriptors are reused for load.
|
|
# @RATIONALE Bundle checksums prove byte integrity, not publisher authenticity or absent source provenance.
|
|
# @REJECTED Shell xz pipelines and tag-only verification can hide decompressor failure or load a foreign image/tag.
|
|
import argparse
|
|
from contextlib import ExitStack
|
|
from datetime import datetime
|
|
import hashlib
|
|
import json
|
|
import lzma
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import selectors
|
|
import stat
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import time
|
|
|
|
from docker_bundle_contract import IMAGE_ID, Refusal, decode, validate_archive
|
|
|
|
ROLES = ("postgres", "backend", "frontend")
|
|
TAG = re.compile(r"[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}")
|
|
FIELDS = {"manifest_version", "bundle_kind", "release_tag", "generated_at_utc", "images", "compose_file",
|
|
"env_template", "env_bootstrap_fields", "checksums_file", "compose_sha256", "env_template_sha256"}
|
|
IMAGE_FIELDS = {"role", "image", "image_id", "repo_digest", "archive"}
|
|
ENV_FIELDS = ["INITIAL_ADMIN_CREATE", "INITIAL_ADMIN_USERNAME", "INITIAL_ADMIN_PASSWORD", "INITIAL_ADMIN_EMAIL"]
|
|
DOCKER_QUERY_TIMEOUT = 10
|
|
|
|
# #region Tooling.DockerBundle.Basename [C:2] [TYPE Function]
|
|
def basename(value):
|
|
if (not isinstance(value, str) or not value or len(value) > 255 or value in (".", "..")
|
|
or "/" in value or "\\" in value or "\x00" in value or any(ord(char) < 32 for char in value)):
|
|
raise Refusal("unsafe_bundle_filename")
|
|
return value
|
|
# #endregion Tooling.DockerBundle.Basename
|
|
|
|
# #region Tooling.DockerBundle.File [C:3] [TYPE Class]
|
|
class BundleFile:
|
|
def __init__(self, fd):
|
|
self.stream = os.fdopen(fd, "rb")
|
|
self.identity = self.snapshot()
|
|
if not stat.S_ISREG(os.fstat(fd).st_mode):
|
|
self.stream.close()
|
|
raise Refusal("bundle_file_not_regular")
|
|
|
|
def snapshot(self):
|
|
info = os.fstat(self.stream.fileno())
|
|
return (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns, info.st_ctime_ns)
|
|
|
|
def stable(self):
|
|
if self.snapshot() != self.identity:
|
|
raise Refusal("bundle_file_changed")
|
|
|
|
def read(self, maximum):
|
|
if self.identity[2] > maximum:
|
|
raise Refusal("bundle_metadata_limit")
|
|
self.stream.seek(0)
|
|
raw = self.stream.read(maximum + 1)
|
|
self.stable()
|
|
if len(raw) > maximum:
|
|
raise Refusal("bundle_metadata_limit")
|
|
return raw
|
|
|
|
def checksum(self):
|
|
self.stream.seek(0)
|
|
digest = hashlib.sha256()
|
|
for chunk in iter(lambda: self.stream.read(1 << 20), b""):
|
|
digest.update(chunk)
|
|
self.stable()
|
|
return digest.hexdigest()
|
|
|
|
def close(self):
|
|
self.stream.close()
|
|
# #endregion Tooling.DockerBundle.File
|
|
|
|
# #region Tooling.DockerBundle.Directory [C:3] [TYPE Class]
|
|
class Bundle:
|
|
def __init__(self, directory):
|
|
path = Path(directory).absolute()
|
|
fd = os.open("/", os.O_RDONLY | os.O_DIRECTORY)
|
|
try:
|
|
for part in path.parts[1:]:
|
|
child = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=fd)
|
|
os.close(fd)
|
|
fd = child
|
|
except BaseException:
|
|
os.close(fd)
|
|
raise
|
|
self.fd = fd
|
|
|
|
def open(self, name, stack):
|
|
fd = os.open(basename(name), os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=self.fd)
|
|
result = BundleFile(fd)
|
|
stack.callback(result.close)
|
|
return result
|
|
|
|
def close(self):
|
|
os.close(self.fd)
|
|
# #endregion Tooling.DockerBundle.Directory
|
|
|
|
# #region Tooling.DockerBundle.Manifest [C:3] [TYPE Function]
|
|
def manifest(raw, expected):
|
|
if not isinstance(expected, str) or TAG.fullmatch(expected) is None:
|
|
raise Refusal("invalid_expected_release_tag")
|
|
value = decode(raw)
|
|
if (not isinstance(value, dict) or set(value) != FIELDS or type(value["manifest_version"]) is not int
|
|
or value["manifest_version"] != 2 or value["bundle_kind"] != "enterprise"):
|
|
raise Refusal("unsupported_bundle_manifest")
|
|
if value["release_tag"] != expected:
|
|
raise Refusal("release_tag_conflict")
|
|
try:
|
|
stamp = datetime.strptime(value["generated_at_utc"], "%Y-%m-%dT%H:%M:%SZ")
|
|
if stamp.strftime("%Y-%m-%dT%H:%M:%SZ") != value["generated_at_utc"]:
|
|
raise ValueError
|
|
except (ValueError, TypeError):
|
|
raise Refusal("invalid_manifest_timestamp") from None
|
|
if (value["compose_file"] != "docker-compose.enterprise-clean.yml" or value["env_template"] != ".env.example"
|
|
or value["env_bootstrap_fields"] != ENV_FIELDS or value["checksums_file"] != f"sha256sums.{expected}.txt"):
|
|
raise Refusal("manifest_deployment_contract")
|
|
for field in ("compose_sha256", "env_template_sha256"):
|
|
if not isinstance(value[field], str) or re.fullmatch(r"[0-9a-f]{64}", value[field]) is None:
|
|
raise Refusal("manifest_deployment_checksum")
|
|
if not isinstance(value["images"], list) or len(value["images"]) != 3:
|
|
raise Refusal("manifest_image_roles")
|
|
images = {}
|
|
for image in value["images"]:
|
|
if not isinstance(image, dict) or set(image) != IMAGE_FIELDS:
|
|
raise Refusal("manifest_image_schema")
|
|
role = image["role"]
|
|
if not isinstance(role, str) or role not in ROLES or role in images:
|
|
raise Refusal("manifest_image_roles")
|
|
if (image["image"] != f"superset-tools-{role}:{expected}"
|
|
or not isinstance(image["image_id"], str) or not IMAGE_ID.fullmatch(image["image_id"])
|
|
or image["archive"] != f"superset-tools-{role}.{expected}.tar.xz"):
|
|
raise Refusal("manifest_image_identity")
|
|
digest = image["repo_digest"]
|
|
if (not isinstance(digest, str) or len(digest) > 4096 or
|
|
(digest != "unavailable" and not re.fullmatch(r"[A-Za-z0-9_./:-]+@sha256:[0-9a-f]{64}(?:,[A-Za-z0-9_./:-]+@sha256:[0-9a-f]{64})*", digest))):
|
|
raise Refusal("manifest_repository_digest")
|
|
images[role] = image
|
|
return value, [images[role] for role in ROLES]
|
|
# #endregion Tooling.DockerBundle.Manifest
|
|
|
|
# #region Tooling.DockerBundle.Checksums [C:2] [TYPE Function]
|
|
def checksums(raw, images):
|
|
try:
|
|
lines = raw.decode("ascii").splitlines()
|
|
except UnicodeError:
|
|
raise Refusal("invalid_checksums") from None
|
|
expected, result = {image["archive"] for image in images}, {}
|
|
if len(lines) != 3:
|
|
raise Refusal("checksum_file_set")
|
|
for line in lines:
|
|
match = re.fullmatch(r"([0-9a-f]{64}) [ *]([^/\\\x00]+)", line)
|
|
if not match or match[2] not in expected or match[2] in result:
|
|
raise Refusal("checksum_file_set")
|
|
result[match[2]] = match[1]
|
|
return result
|
|
# #endregion Tooling.DockerBundle.Checksums
|
|
|
|
# #region Tooling.DockerBundle.DockerInspect [C:3] [TYPE Function]
|
|
def docker_output(argv):
|
|
process = subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL, close_fds=True, shell=False)
|
|
output = bytearray()
|
|
deadline = time.monotonic() + DOCKER_QUERY_TIMEOUT
|
|
try:
|
|
with selectors.DefaultSelector() as selected:
|
|
os.set_blocking(process.stdout.fileno(), False)
|
|
selected.register(process.stdout, selectors.EVENT_READ)
|
|
while True:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0 or not selected.select(remaining):
|
|
raise Refusal("docker_query_timeout")
|
|
raw = os.read(process.stdout.fileno(), min(4096, 16385 - len(output)))
|
|
if not raw:
|
|
break
|
|
output.extend(raw)
|
|
if len(output) > 16384:
|
|
raise Refusal("docker_query_output_limit")
|
|
try:
|
|
rc = process.wait(timeout=max(0, deadline - time.monotonic()))
|
|
except subprocess.TimeoutExpired:
|
|
raise Refusal("docker_query_timeout") from None
|
|
if rc:
|
|
raise Refusal("docker_inspection_unavailable")
|
|
finally:
|
|
if process.poll() is None:
|
|
process.kill() # Only this owned read-only query client, never Docker daemon/container.
|
|
process.wait()
|
|
process.stdout.close()
|
|
try:
|
|
return output.decode("ascii").strip()
|
|
except UnicodeError:
|
|
raise Refusal("docker_inspection_invalid") from None
|
|
|
|
def inspect(image):
|
|
actual = docker_output(["docker", "image", "inspect", "--format", "{{.Id}}", image["image"]])
|
|
if actual != image["image_id"]:
|
|
raise Refusal("docker_image_identity_conflict")
|
|
|
|
def existing(image):
|
|
value = docker_output(["docker", "image", "ls", "--no-trunc", "--filter", "reference=" + image["image"],
|
|
"--format", "{{.Repository}}:{{.Tag}} {{.ID}}"])
|
|
if not value:
|
|
return False
|
|
if value != image["image"] + " " + image["image_id"]:
|
|
raise Refusal("existing_tag_conflict")
|
|
inspect(image)
|
|
return True
|
|
# #endregion Tooling.DockerBundle.DockerInspect
|
|
|
|
# #region Tooling.DockerBundle.Load [C:4] [TYPE Function]
|
|
def load(archive):
|
|
archive.stable()
|
|
archive.stream.seek(0)
|
|
with tempfile.TemporaryFile() as output, tempfile.TemporaryFile() as errors:
|
|
decompressor = subprocess.Popen(["xz", "--decompress", "--stdout"], stdin=archive.stream,
|
|
stdout=subprocess.PIPE, stderr=errors, close_fds=True)
|
|
loader = None
|
|
try:
|
|
loader = subprocess.Popen(["docker", "image", "load", "--quiet"], stdin=decompressor.stdout,
|
|
stdout=output, stderr=errors, close_fds=True)
|
|
finally:
|
|
decompressor.stdout.close()
|
|
if loader is not None:
|
|
docker_rc = loader.wait()
|
|
xz_rc = decompressor.wait()
|
|
return xz_rc, docker_rc
|
|
# #endregion Tooling.DockerBundle.Load
|
|
|
|
# #region Tooling.DockerBundle.Execute [C:4] [TYPE Function]
|
|
def execute(directory, manifest_name, expected, verify_only=False):
|
|
result = {"status": "refused", "code": None, "mode": "verify-only" if verify_only else "import",
|
|
"verified_roles": [], "loaded_roles": [], "existing_roles": [], "attempted_roles": [],
|
|
"partial_load_possible": False, "source_identity": {"status": "unavailable", "reason": "manifest_v2_has_no_source_identity"},
|
|
"authenticity": "not_established_by_checksums"}
|
|
try:
|
|
with ExitStack() as stack:
|
|
bundle = Bundle(directory)
|
|
stack.callback(bundle.close)
|
|
value, images = manifest(bundle.open(manifest_name, stack).read(1 << 20), expected)
|
|
metadata = []
|
|
for name, digest in (("compose_file", "compose_sha256"), ("env_template", "env_template_sha256")):
|
|
item = bundle.open(value[name], stack)
|
|
if hashlib.sha256(item.read(1 << 20)).hexdigest() != value[digest]:
|
|
raise Refusal("deployment_metadata_checksum_conflict")
|
|
metadata.append(item)
|
|
sums = checksums(bundle.open(value["checksums_file"], stack).read(65536), images)
|
|
archives = {image["role"]: bundle.open(image["archive"], stack) for image in images}
|
|
for image in images:
|
|
archive = archives[image["role"]]
|
|
if archive.checksum() != sums[image["archive"]]:
|
|
raise Refusal("archive_checksum_conflict")
|
|
archive.stream.seek(0)
|
|
tested = subprocess.run(["xz", "--test"], stdin=archive.stream, stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL, check=False)
|
|
if tested.returncode:
|
|
raise Refusal("xz_integrity_failed")
|
|
validate_archive(archive.stream, image)
|
|
archive.stable()
|
|
result["verified_roles"].append(image["role"])
|
|
for item in metadata:
|
|
item.stable()
|
|
if verify_only:
|
|
result.update(status="verified", target_state="not_checked")
|
|
return 0, result
|
|
present = {image["role"]: existing(image) for image in images}
|
|
result["existing_roles"] = [role for role in ROLES if present[role]]
|
|
for image in images:
|
|
if present[image["role"]]:
|
|
continue
|
|
for item in metadata:
|
|
item.stable()
|
|
result["attempted_roles"].append(image["role"])
|
|
xz_rc, docker_rc = load(archives[image["role"]])
|
|
result.update(xz_rc=xz_rc, docker_load_rc=docker_rc)
|
|
archives[image["role"]].stable()
|
|
if xz_rc or docker_rc:
|
|
raise Refusal("image_load_failed")
|
|
inspect(image)
|
|
result["loaded_roles"].append(image["role"])
|
|
for image in images:
|
|
inspect(image)
|
|
for item in metadata:
|
|
item.stable()
|
|
result.update(status="imported" if result["loaded_roles"] else "already_present")
|
|
return 0, result
|
|
except (Refusal, OSError, EOFError, lzma.LZMAError, tarfile.TarError, ValueError, TypeError, KeyError) as error:
|
|
result["code"] = str(error) if isinstance(error, Refusal) else "bundle_io_or_archive_invalid"
|
|
result["partial_load_possible"] = bool(result["attempted_roles"])
|
|
return (3 if result["partial_load_possible"] else 2), result
|
|
# #endregion Tooling.DockerBundle.Execute
|
|
|
|
# #region Tooling.DockerBundle.Main [C:2] [TYPE Function]
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description="Verify/import a versioned full enterprise Docker image bundle.")
|
|
parser.add_argument("--bundle-dir", required=True)
|
|
parser.add_argument("--manifest", required=True)
|
|
parser.add_argument("--expected-release-tag", required=True)
|
|
parser.add_argument("--verify-only", action="store_true")
|
|
args = parser.parse_args(argv)
|
|
rc, result = execute(args.bundle_dir, args.manifest, args.expected_release_tag, args.verify_only)
|
|
print(json.dumps(result, sort_keys=True))
|
|
return rc
|
|
# #endregion Tooling.DockerBundle.Main
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|
|
# #endregion Tooling.DockerBundle.Import
|