Harden staging image push authority on current main

docs/gcp-leoclean-nosend-service-package.md

@@ -231,8 +231,16 @@ explicit inputs; inherited `DOCKER_*`, `BUILDKIT_*`, and `BUILDX_*` overrides
 are removed and Docker stdin is closed. The pull does not
 start a container, install a unit, restart a service, or contact Cloud SQL.
 Registry authentication and this pull require a separate live-read approval.
-While holding an advisory lock on the explicit Docker config directory, the
+The package opens every component of the explicit Docker config path without
-command pulls and records the dynamically observed config ID. It intentionally
+following links and retains that full descriptor chain while Docker runs. Every
+component must be root- or current-user-owned and non-writable by group/other;
+a root-owned sticky temporary directory is accepted as a boundary only when the
+operator directory below it is private. Docker starts in the held final
+directory and receives `--config .`, so replacing and restoring the named path
+mid-command cannot redirect credential reads. The package also revalidates the
+named chain before and after each call while holding an advisory lock on the
+final directory. The command pulls and records the dynamically observed config
+ID. It intentionally
 retains the exact repository-digest reference: Docker's image cache belongs to
 the daemon, so a config-directory lock cannot prove that another client did not
 start relying on the same immutable reference after the precheck. Repeated runs

ops/gcp_leoclean_nosend_package.py

@@ -1916,6 +1916,100 @@ def _docker_environment() -> dict[str, str]:
     }
 
 
+def _close_descriptors(descriptors: tuple[int, ...] | list[int]) -> None:
+    for descriptor in reversed(descriptors):
+        os.close(descriptor)
+
+
+def _open_directory_chain_descriptors(
+    directory: Path,
+    label: str,
+) -> tuple[tuple[int, ...], tuple[tuple[int, int], ...]]:
+    """Retain every directory component without following links."""
+
+    _require(directory.is_absolute() and ".." not in directory.parts, f"{label} must be an absolute direct path")
+    flags = os.O_RDONLY
+    if hasattr(os, "O_DIRECTORY"):
+        flags |= os.O_DIRECTORY
+    if hasattr(os, "O_NOFOLLOW"):
+        flags |= os.O_NOFOLLOW
+    descriptors: list[int] = []
+    identities: list[tuple[int, int]] = []
+    try:
+        descriptors.append(os.open("/", flags))
+        observed = os.fstat(descriptors[-1])
+        _require(stat.S_ISDIR(observed.st_mode), f"{label} root is unsafe")
+        identities.append((observed.st_dev, observed.st_ino))
+        for component in directory.parts[1:]:
+            _require(component not in {"", ".", ".."}, f"{label} path is unsafe")
+            descriptors.append(os.open(component, flags, dir_fd=descriptors[-1]))
+            observed = os.fstat(descriptors[-1])
+            _require(stat.S_ISDIR(observed.st_mode), f"{label} path is unsafe")
+            identities.append((observed.st_dev, observed.st_ino))
+        named = os.stat(directory, follow_symlinks=False)
+        _require(
+            stat.S_ISDIR(named.st_mode) and identities[-1] == (named.st_dev, named.st_ino),
+            f"{label} path identity drifted",
+        )
+    except (OSError, PackageError) as exc:
+        _close_descriptors(descriptors)
+        if isinstance(exc, PackageError):
+            raise
+        raise PackageError(f"{label} is missing or contains a symbolic link") from exc
+    return tuple(descriptors), tuple(identities)
+
+
+def _open_directory_chain_nofollow(
+    directory: Path,
+    label: str,
+) -> tuple[int, tuple[tuple[int, int], ...]]:
+    """Open an absolute directory one component at a time without following links."""
+
+    descriptors, identities = _open_directory_chain_descriptors(directory, label)
+    try:
+        descriptor = os.dup(descriptors[-1])
+    except OSError as exc:
+        raise PackageError(f"{label} descriptor cannot be retained") from exc
+    finally:
+        _close_descriptors(descriptors)
+    return descriptor, identities
+
+
+def _validate_docker_config_chain_posture(descriptors: tuple[int, ...]) -> None:
+    _require(bool(descriptors), "Docker config directory chain is empty")
+    for index, descriptor in enumerate(descriptors):
+        observed = os.fstat(descriptor)
+        mode = stat.S_IMODE(observed.st_mode)
+        final = index == len(descriptors) - 1
+        if final:
+            trusted = observed.st_uid == os.geteuid() and mode & 0o022 == 0
+        else:
+            trusted_owner = observed.st_uid in {0, os.geteuid()}
+            nonwritable = mode & 0o022 == 0
+            root_sticky_boundary = observed.st_uid == 0 and bool(mode & stat.S_ISVTX)
+            trusted = trusted_owner and (nonwritable or root_sticky_boundary)
+        _require(
+            stat.S_ISDIR(observed.st_mode) and trusted,
+            "Docker config directory must have only root/current-user trusted ancestors; "
+            "a writable ancestor is allowed only when it is a root-owned sticky boundary",
+        )
+
+
+def _open_trusted_docker_config_chain(
+    docker_config: Path,
+) -> tuple[tuple[int, ...], tuple[tuple[int, int], ...]]:
+    descriptors, identities = _open_directory_chain_descriptors(
+        docker_config,
+        "Docker config directory",
+    )
+    try:
+        _validate_docker_config_chain_posture(descriptors)
+    except (OSError, PackageError):
+        _close_descriptors(descriptors)
+        raise
+    return descriptors, identities
+
+
 def _validate_docker_authority(docker_binary: Path, docker_host: str, docker_config: Path) -> None:
     _require(docker_binary.is_absolute(), "Docker binary must be absolute")
     _require("\x00" not in str(docker_binary) and "\n" not in str(docker_binary), "Docker binary is invalid")
@@ -1928,31 +2022,57 @@ def _validate_docker_authority(docker_binary: Path, docker_host: str, docker_con
         and "\n" not in docker_host,
         "Docker host must be a local Unix endpoint",
     )
-    _require(
+    descriptors, _identities = _open_trusted_docker_config_chain(docker_config)
-        docker_config.is_absolute() and docker_config.is_dir() and not docker_config.is_symlink(),
+    _close_descriptors(descriptors)
-        "Docker config directory is missing or unsafe",
-    )
 
 
 class _DockerCacheLock:
     def __init__(self, docker_config: Path) -> None:
         self._docker_config = docker_config
         self._descriptor: int | None = None
+        self._descriptors: tuple[int, ...] = ()
+        self._identities: tuple[tuple[int, int], ...] = ()
 
-    def __enter__(self) -> None:
+    def __enter__(self) -> _DockerCacheLock:
-        flags = os.O_RDONLY
+        _require(self._descriptor is None, "Docker cache authority lock is already active")
-        if hasattr(os, "O_DIRECTORY"):
+        descriptors: tuple[int, ...] = ()
-            flags |= os.O_DIRECTORY
-        if hasattr(os, "O_NOFOLLOW"):
-            flags |= os.O_NOFOLLOW
         try:
-            descriptor = os.open(self._docker_config, flags)
+            descriptors, identities = _open_trusted_docker_config_chain(self._docker_config)
+            descriptor = descriptors[-1]
             fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
-        except OSError as exc:
+        except (OSError, PackageError) as exc:
-            if "descriptor" in locals():
+            _close_descriptors(descriptors)
-                os.close(descriptor)
             raise PackageError("Docker cache authority is already in use or unsafe") from exc
+        self._descriptors = descriptors
+        self._identities = identities
         self._descriptor = descriptor
+        return self
+
+    def revalidate(self, docker_config: Path) -> None:
+        _require(
+            self._descriptor is not None
+            and bool(self._descriptors)
+            and bool(self._identities)
+            and docker_config == self._docker_config,
+            "Docker cache authority is not active for this config",
+        )
+        _validate_docker_config_chain_posture(self._descriptors)
+        named_descriptors, named_identities = _open_trusted_docker_config_chain(docker_config)
+        try:
+            _require(
+                named_identities == self._identities,
+                "Docker config directory ancestor identity or posture drifted",
+            )
+        finally:
+            _close_descriptors(named_descriptors)
+
+    def enter_subprocess_config_directory(self) -> None:
+        _require(self._descriptor is not None, "Docker cache authority is not active")
+        os.fchdir(self._descriptor)
+
+    def subprocess_config_authority(self) -> tuple[str, tuple[int, ...]]:
+        _require(self._descriptor is not None, "Docker cache authority is not active")
+        return ".", (self._descriptor,)
 
     def __exit__(self, _kind: object, _value: object, _traceback: object) -> None:
         if self._descriptor is None:
@@ -1960,8 +2080,10 @@ class _DockerCacheLock:
         try:
             fcntl.flock(self._descriptor, fcntl.LOCK_UN)
         finally:
-            os.close(self._descriptor)
+            _close_descriptors(self._descriptors)
             self._descriptor = None
+            self._descriptors = ()
+            self._identities = ()
 
 
 def _run_docker(
@@ -1972,16 +2094,36 @@ def _run_docker(
     *,
     check: bool,
     timeout: int,
+    authority: _DockerCacheLock | None = None,
 ) -> subprocess.CompletedProcess[bytes]:
     _validate_docker_authority(docker_binary, docker_host, docker_config)
+    _require("--config" not in arguments, "Docker arguments cannot override the retained config authority")
+    if authority is None:
+        with _DockerCacheLock(docker_config) as owned_authority:
+            return _run_docker(
+                docker_binary,
+                docker_host,
+                docker_config,
+                arguments,
+                check=check,
+                timeout=timeout,
+                authority=owned_authority,
+            )
+    authority.revalidate(docker_config)
+    subprocess_config, pass_fds = authority.subprocess_config_authority()
+    try:
         return subprocess.run(
-        [str(docker_binary), "--host", docker_host, "--config", str(docker_config), *arguments],
+            [str(docker_binary), "--host", docker_host, "--config", subprocess_config, *arguments],
             check=check,
             capture_output=True,
             stdin=subprocess.DEVNULL,
             timeout=timeout,
             env=_docker_environment(),
+            pass_fds=pass_fds,
+            preexec_fn=authority.enter_subprocess_config_directory,
         )
+    finally:
+        authority.revalidate(docker_config)
 
 
 def _bounded_docker_output(result: subprocess.CompletedProcess[bytes], label: str) -> None:
@@ -1996,6 +2138,8 @@ def _inspect_local_image(
     docker_host: str,
     docker_config: Path,
     image_reference: str,
+    *,
+    authority: _DockerCacheLock | None = None,
 ) -> dict[str, Any]:
     result = _run_docker(
         docker_binary,
@@ -2004,6 +2148,7 @@ def _inspect_local_image(
         ["image", "inspect", image_reference],
         check=True,
         timeout=60,
+        authority=authority,
     )
     _bounded_docker_output(result, "local image inspection")
     inspected = _load_docker_inspection(result.stdout)
@@ -2019,6 +2164,7 @@ def _local_tag_config_id(
     *,
     repository: str,
     tag: str,
+    authority: _DockerCacheLock | None = None,
 ) -> str | None:
     _require(
         repository in {LOCAL_BUILD_REPOSITORY, IMAGE_REPOSITORY}
@@ -2034,6 +2180,7 @@ def _local_tag_config_id(
         ["image", "ls", "--no-trunc", "--format", "{{json .}}", repository],
         check=True,
         timeout=60,
+        authority=authority,
     )
     _bounded_docker_output(inventory, "local image inventory")
     matching_ids: set[str] = set()
@@ -2059,7 +2206,13 @@ def _local_tag_config_id(
     if not matching_ids:
         return None
     inventory_config_id = next(iter(matching_ids))
-    inspected = _inspect_local_image(docker_binary, docker_host, docker_config, reference)
+    inspected = _inspect_local_image(
+        docker_binary,
+        docker_host,
+        docker_config,
+        reference,
+        authority=authority,
+    )
     config_id = inspected.get("Id")
     _require(config_id == inventory_config_id, "local image inventory and inspection disagree")
     return inventory_config_id
@@ -2084,6 +2237,44 @@ def _target_repository_digests(inspected: dict[str, Any]) -> list[str]:
     return matching
 
 
+def _validate_receipt_context_separation(
+    context: Path,
+    output_receipt: Path,
+    *,
+    output_parent_descriptor: int | None = None,
+) -> None:
+    _require(
+        context.is_absolute() and output_receipt.is_absolute() and output_receipt.name not in {"", ".", ".."},
+        "build/push context and receipt paths must be absolute",
+    )
+    context_descriptor, _context_chain = _open_directory_chain_nofollow(
+        context,
+        "prepared image context",
+    )
+    output_descriptor: int | None = None
+    try:
+        output_descriptor, output_chain = _open_directory_chain_nofollow(
+            output_receipt.parent,
+            "build/push receipt parent",
+        )
+        context_stat = os.fstat(context_descriptor)
+        context_identity = (context_stat.st_dev, context_stat.st_ino)
+        if output_parent_descriptor is not None:
+            held_parent = os.fstat(output_parent_descriptor)
+            _require(
+                stat.S_ISDIR(held_parent.st_mode) and (held_parent.st_dev, held_parent.st_ino) == output_chain[-1],
+                "build/push receipt parent identity drifted",
+            )
+        _require(
+            context_identity not in output_chain,
+            "build/push receipt must be outside the image context",
+        )
+    finally:
+        if output_descriptor is not None:
+            os.close(output_descriptor)
+        os.close(context_descriptor)
+
+
 def build_and_push_staging_image(
     repo_root: Path,
     context: Path,
@@ -2101,15 +2292,12 @@ def build_and_push_staging_image(
     validate_reviewed_context_source(repo_root, context, image_input)
     _validate_docker_authority(docker_binary, docker_host, docker_config)
     _validate_receipt_output_path(output_receipt)
-    _require(
+    _validate_receipt_context_separation(context, output_receipt)
-        context != output_receipt and context not in output_receipt.parents,
-        "build/push receipt must be outside the image context",
-    )
 
     build_tag = f"input-{image_input['input_sha256']}"
     build_reference = f"{LOCAL_BUILD_REPOSITORY}:{build_tag}"
     try:
-        with _DockerCacheLock(docker_config):
+        with _DockerCacheLock(docker_config) as authority:
             _require(
                 _local_tag_config_id(
                     docker_binary,
@@ -2117,6 +2305,7 @@ def build_and_push_staging_image(
                     docker_config,
                     repository=LOCAL_BUILD_REPOSITORY,
                     tag=build_tag,
+                    authority=authority,
                 )
                 is None,
                 "local build tag must be absent before build",
@@ -2142,9 +2331,16 @@ def build_and_push_staging_image(
                 command,
                 check=True,
                 timeout=1800,
+                authority=authority,
             )
             _bounded_docker_output(built, "staging image build")
-            built_image = _inspect_local_image(docker_binary, docker_host, docker_config, build_reference)
+            built_image = _inspect_local_image(
+                docker_binary,
+                docker_host,
+                docker_config,
+                build_reference,
+                authority=authority,
+            )
             observed = validate_image_configuration(image_input, built_image)
             build_config_id = observed["config_digest"]
             _require(
@@ -2154,6 +2350,7 @@ def build_and_push_staging_image(
                     docker_config,
                     repository=LOCAL_BUILD_REPOSITORY,
                     tag=build_tag,
+                    authority=authority,
                 )
                 == build_config_id,
                 "local build tag differs from the inspected image",
@@ -2178,6 +2375,7 @@ def build_and_push_staging_image(
                     docker_config,
                     repository=IMAGE_REPOSITORY,
                     tag=candidate_tag,
+                    authority=authority,
                 )
                 is None,
                 "content-bound staging candidate tag already exists locally",
@@ -2189,6 +2387,7 @@ def build_and_push_staging_image(
                 ["image", "tag", build_reference, candidate_reference],
                 check=True,
                 timeout=60,
+                authority=authority,
             )
             _bounded_docker_output(tagged, "staging candidate tag")
             _require(
@@ -2198,6 +2397,7 @@ def build_and_push_staging_image(
                     docker_config,
                     repository=IMAGE_REPOSITORY,
                     tag=candidate_tag,
+                    authority=authority,
                 )
                 == build_config_id,
                 "staging candidate tag differs from the built image",
@@ -2207,6 +2407,7 @@ def build_and_push_staging_image(
                 docker_host,
                 docker_config,
                 candidate_reference,
+                authority=authority,
             )
             _require(
                 validate_image_configuration(image_input, pre_push_image) == observed,
@@ -2223,6 +2424,7 @@ def build_and_push_staging_image(
                 ["image", "push", "--quiet", candidate_reference],
                 check=True,
                 timeout=1800,
+                authority=authority,
             )
             _bounded_docker_output(pushed, "staging image push")
             post_push_image = _inspect_local_image(
@@ -2230,6 +2432,7 @@ def build_and_push_staging_image(
                 docker_host,
                 docker_config,
                 candidate_reference,
+                authority=authority,
             )
             _require(
                 validate_image_configuration(image_input, post_push_image) == observed,
@@ -2243,7 +2446,12 @@ def build_and_push_staging_image(
             receipt = build_build_push_receipt(image_input, post_push_image=post_push_image)
     except (OSError, subprocess.SubprocessError, PackageError) as exc:
         raise PackageError("staging image build/push failed") from exc
-    publish_build_push_receipt(output_receipt, receipt, image_input=image_input)
+    publish_build_push_receipt(
+        output_receipt,
+        receipt,
+        image_input=image_input,
+        outside_context=context,
+    )
     return receipt
 
 
@@ -2252,6 +2460,8 @@ def _local_image_reference_config_id(
     docker_host: str,
     docker_config: Path,
     image_reference: str,
+    *,
+    authority: _DockerCacheLock | None = None,
 ) -> str | None:
     inventory = _run_docker(
         docker_binary,
@@ -2260,6 +2470,7 @@ def _local_image_reference_config_id(
         ["image", "ls", "--digests", "--no-trunc", "--format", "{{json .}}", IMAGE_REPOSITORY],
         check=True,
         timeout=60,
+        authority=authority,
     )
     _require(len(inventory.stdout) <= MAX_DOCKER_INSPECTION_BYTES, "local image inventory output is too large")
     match = IMAGE_REFERENCE.fullmatch(image_reference)
@@ -2295,6 +2506,7 @@ def _local_image_reference_config_id(
         ["image", "inspect", image_reference],
         check=True,
         timeout=60,
+        authority=authority,
     )
     inspected = _load_docker_inspection(result.stdout)
     _require(isinstance(inspected, list) and len(inspected) == 1, "local image reference inspection is not exact")
@@ -2329,7 +2541,7 @@ def inspect_registry_image(
     _validate_docker_authority(docker_binary, docker_host, docker_config)
     inspection: dict[str, Any] | None = None
     try:
-        with _DockerCacheLock(docker_config):
+        with _DockerCacheLock(docker_config) as authority:
             _run_docker(
                 docker_binary,
                 docker_host,
@@ -2337,12 +2549,14 @@ def inspect_registry_image(
                 ["pull", "--quiet", "--platform", PLATFORM, image_reference],
                 check=True,
                 timeout=600,
+                authority=authority,
             )
             pulled_config_id = _local_image_reference_config_id(
                 docker_binary,
                 docker_host,
                 docker_config,
                 image_reference,
+                authority=authority,
             )
             _require(pulled_config_id is not None, "pulled registry image reference is missing")
             result = _run_docker(
@@ -2352,6 +2566,7 @@ def inspect_registry_image(
                 ["image", "inspect", image_reference],
                 check=True,
                 timeout=60,
+                authority=authority,
             )
             inspected = _load_docker_inspection(result.stdout)
             _require(isinstance(inspected, list) and len(inspected) == 1, "registry image inspection is not exact")
@@ -2625,16 +2840,7 @@ def _rename_noreplace(source: Path, destination: Path) -> None:
 
 
 def _open_private_operator_directory(directory: Path, label: str) -> int:
-    _require(directory.is_absolute(), f"{label} must be absolute")
+    descriptor, _identities = _open_directory_chain_nofollow(directory, label)
-    flags = os.O_RDONLY
-    if hasattr(os, "O_DIRECTORY"):
-        flags |= os.O_DIRECTORY
-    if hasattr(os, "O_NOFOLLOW"):
-        flags |= os.O_NOFOLLOW
-    try:
-        descriptor = os.open(directory, flags)
-    except OSError as exc:
-        raise PackageError(f"{label} is missing or unsafe") from exc
     try:
         observed = os.fstat(descriptor)
         named = os.stat(directory, follow_symlinks=False)
@@ -2767,11 +2973,14 @@ def publish_build_push_receipt(
     receipt: dict[str, Any],
     *,
     image_input: dict[str, Any],
+    outside_context: Path | None = None,
 ) -> None:
     """Atomically publish one validated, self-hashed build/push receipt."""
 
     validate_build_push_receipt(receipt, image_input=image_input)
     _validate_receipt_output_path(output_receipt)
+    if outside_context is not None:
+        _validate_receipt_context_separation(outside_context, output_receipt)
     try:
         payload = (json.dumps(receipt, allow_nan=False, indent=2, sort_keys=True) + "\n").encode("utf-8")
     except (TypeError, ValueError) as exc:
@@ -2784,6 +2993,12 @@ def publish_build_push_receipt(
     descriptor: int | None = None
     identity: tuple[int, int] | None = None
     try:
+        if outside_context is not None:
+            _validate_receipt_context_separation(
+                outside_context,
+                output_receipt,
+                output_parent_descriptor=parent_descriptor,
+            )
         _require(
             _directory_entry_stat(parent_descriptor, output_receipt.name) is None,
             "build/push receipt output already exists",
@@ -2834,6 +3049,12 @@ def publish_build_push_receipt(
             label="retained build/push receipt link",
         )
         _fsync_directory_descriptor(parent_descriptor)
+        if outside_context is not None:
+            _validate_receipt_context_separation(
+                outside_context,
+                output_receipt,
+                output_parent_descriptor=parent_descriptor,
+            )
     except (OSError, PackageError) as exc:
         if isinstance(exc, PackageError):
             raise

tests/test_gcp_leoclean_nosend_package.py

@@ -9,6 +9,7 @@ import os
 import shutil
 import subprocess
 import sys
+import tempfile
 from pathlib import Path
 
 import pytest
@@ -1138,6 +1139,277 @@ def test_build_push_requires_explicit_execution_flag_before_validation(tmp_path:
         )
 
 
+def test_build_push_rejects_writable_docker_config_before_docker(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    image_input = image_input_fixture(tmp_path)
+    build_args = {
+        "ARTIFACT_SHA256": image_input["runtime"]["artifact_sha256"],
+        "IDENTITY_SHA256": image_input["identity"]["bundle_sha256"],
+        "INPUT_SHA256": image_input["input_sha256"],
+        "TELEO_REVISION": image_input["runtime"]["teleo_git_head"],
+    }
+    context = tmp_path / "context"
+    context.mkdir()
+    repo_root = tmp_path / "repo"
+    repo_root.mkdir()
+    output = tmp_path / "writable-config-build-push-receipt.json"
+    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
+    docker_config.chmod(0o777)
+    docker_calls: list[list[str]] = []
+    monkeypatch.setattr(package, "validate_prepared_image_context", lambda _context: (image_input, build_args))
+    monkeypatch.setattr(package, "validate_reviewed_context_source", lambda *_args: None)
+    monkeypatch.setattr(
+        package.subprocess,
+        "run",
+        lambda arguments, **_kwargs: docker_calls.append(arguments),
+    )
+
+    with pytest.raises(package.PackageError, match="Docker config directory"):
+        package.build_and_push_staging_image(
+            repo_root,
+            context,
+            docker_binary=docker_binary,
+            docker_host=docker_host,
+            docker_config=docker_config,
+            output_receipt=output,
+            execute_staging_push=True,
+        )
+
+    assert docker_calls == []
+    assert not output.exists()
+
+
+def test_docker_authority_rejects_nonsticky_writable_ancestor(tmp_path: Path) -> None:
+    docker_binary, docker_host, _docker_config = docker_authority(tmp_path)
+    shared = tmp_path / "shared"
+    shared.mkdir(mode=0o777)
+    shared.chmod(0o777)
+    operator = shared / "operator"
+    operator.mkdir(mode=0o700)
+    docker_config = operator / "docker-config"
+    docker_config.mkdir(mode=0o700)
+
+    with pytest.raises(package.PackageError, match="trusted ancestors"):
+        package._validate_docker_authority(docker_binary, docker_host, docker_config)
+
+
+def test_docker_authority_accepts_sticky_root_with_private_operator_directory() -> None:
+    sticky_root = Path("/private/tmp") if sys.platform == "darwin" else Path("/tmp")
+    operator = Path(tempfile.mkdtemp(prefix="teleo-docker-authority-", dir=sticky_root))
+    operator.chmod(0o700)
+    try:
+        docker_binary = operator / "docker"
+        docker_binary.write_text("", encoding="utf-8")
+        docker_binary.chmod(0o755)
+        docker_config = operator / "docker-config"
+        docker_config.mkdir(mode=0o700)
+
+        package._validate_docker_authority(
+            docker_binary,
+            "unix:///tmp/livingip-docker.sock",
+            docker_config,
+        )
+        with package._DockerCacheLock(docker_config) as authority:
+            authority.revalidate(docker_config)
+            subprocess_config, pass_fds = authority.subprocess_config_authority()
+            assert subprocess_config == "."
+            assert len(pass_fds) == 1
+    finally:
+        shutil.rmtree(operator)
+
+
+def test_run_docker_uses_retained_config_during_mid_subprocess_swap_restore(
+    tmp_path: Path,
+) -> None:
+    docker_binary = tmp_path / "bin" / "docker"
+    docker_binary.parent.mkdir()
+    docker_binary.write_text(
+        """#!/usr/bin/env python3
+import sys
+from pathlib import Path
+
+config = Path(sys.argv[sys.argv.index("--config") + 1])
+original = Path(sys.argv[-2])
+retained = Path(sys.argv[-1])
+original.rename(retained)
+original.mkdir(mode=0o700)
+(original / "marker").write_text("attacker", encoding="utf-8")
+try:
+    print((config / "marker").read_text(encoding="utf-8"))
+finally:
+    (original / "marker").unlink()
+    original.rmdir()
+    retained.rename(original)
+""",
+        encoding="utf-8",
+    )
+    docker_binary.chmod(0o755)
+    docker_config = tmp_path / "docker-config"
+    docker_config.mkdir(mode=0o700)
+    (docker_config / "marker").write_text("trusted", encoding="utf-8")
+    retained_config = tmp_path / "retained-docker-config"
+
+    result = package._run_docker(
+        docker_binary,
+        "unix:///tmp/livingip-docker.sock",
+        docker_config,
+        ["probe", str(docker_config), str(retained_config)],
+        check=True,
+        timeout=30,
+    )
+
+    assert result.stdout == b"trusted\n"
+    assert (docker_config / "marker").read_text(encoding="utf-8") == "trusted"
+    assert not retained_config.exists()
+
+
+def test_build_push_rejects_replaced_docker_config_before_mutation(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    image_input = image_input_fixture(tmp_path)
+    build_args = {
+        "ARTIFACT_SHA256": image_input["runtime"]["artifact_sha256"],
+        "IDENTITY_SHA256": image_input["identity"]["bundle_sha256"],
+        "INPUT_SHA256": image_input["input_sha256"],
+        "TELEO_REVISION": image_input["runtime"]["teleo_git_head"],
+    }
+    context = tmp_path / "context"
+    context.mkdir()
+    repo_root = tmp_path / "repo"
+    repo_root.mkdir()
+    output = tmp_path / "replaced-config-build-push-receipt.json"
+    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
+    retained_config = tmp_path / "retained-docker-config"
+    docker_calls: list[list[str]] = []
+    monkeypatch.setattr(package, "validate_prepared_image_context", lambda _context: (image_input, build_args))
+    monkeypatch.setattr(package, "validate_reviewed_context_source", lambda *_args: None)
+
+    def replace_config_before_build(*_args: object, **_kwargs: object) -> None:
+        docker_config.rename(retained_config)
+        docker_config.mkdir(mode=0o700)
+        return None
+
+    monkeypatch.setattr(package, "_local_tag_config_id", replace_config_before_build)
+    monkeypatch.setattr(
+        package.subprocess,
+        "run",
+        lambda arguments, **_kwargs: docker_calls.append(arguments),
+    )
+
+    with pytest.raises(package.PackageError, match="staging image build/push failed") as raised:
+        package.build_and_push_staging_image(
+            repo_root,
+            context,
+            docker_binary=docker_binary,
+            docker_host=docker_host,
+            docker_config=docker_config,
+            output_receipt=output,
+            execute_staging_push=True,
+        )
+
+    assert "identity or posture drifted" in str(raised.value.__cause__)
+    assert docker_calls == []
+    assert not output.exists()
+
+
+def test_build_push_rejects_receipt_parent_symlink_alias_before_docker(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    image_input = image_input_fixture(tmp_path)
+    build_args = {
+        "ARTIFACT_SHA256": image_input["runtime"]["artifact_sha256"],
+        "IDENTITY_SHA256": image_input["identity"]["bundle_sha256"],
+        "INPUT_SHA256": image_input["input_sha256"],
+        "TELEO_REVISION": image_input["runtime"]["teleo_git_head"],
+    }
+    context = tmp_path / "context"
+    context.mkdir()
+    (context / "receipts").mkdir()
+    alias = tmp_path / "context-alias"
+    alias.symlink_to(context, target_is_directory=True)
+    output = alias / "receipts" / "receipt.json"
+    repo_root = tmp_path / "repo"
+    repo_root.mkdir()
+    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
+    docker_calls: list[list[str]] = []
+    monkeypatch.setattr(package, "validate_prepared_image_context", lambda _context: (image_input, build_args))
+    monkeypatch.setattr(package, "validate_reviewed_context_source", lambda *_args: None)
+    monkeypatch.setattr(
+        package.subprocess,
+        "run",
+        lambda arguments, **_kwargs: docker_calls.append(arguments),
+    )
+
+    with pytest.raises(package.PackageError, match="symbolic link"):
+        package.build_and_push_staging_image(
+            repo_root,
+            context,
+            docker_binary=docker_binary,
+            docker_host=docker_host,
+            docker_config=docker_config,
+            output_receipt=output,
+            execute_staging_push=True,
+        )
+
+    assert docker_calls == []
+    assert not output.exists()
+
+
+def test_build_push_rechecks_receipt_context_separation_at_publication(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    image_input = image_input_fixture(tmp_path)
+    build_args = {
+        "ARTIFACT_SHA256": image_input["runtime"]["artifact_sha256"],
+        "IDENTITY_SHA256": image_input["identity"]["bundle_sha256"],
+        "INPUT_SHA256": image_input["input_sha256"],
+        "TELEO_REVISION": image_input["runtime"]["teleo_git_head"],
+    }
+    context = tmp_path / "context"
+    context.mkdir()
+    (context / "receipts").mkdir()
+    repo_root = tmp_path / "repo"
+    repo_root.mkdir()
+    output_parent = tmp_path / "receipts"
+    output_parent.mkdir()
+    retained_parent = tmp_path / "retained-receipts"
+    output = output_parent / "receipt.json"
+    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
+    fake_run, state, calls = mocked_build_push_docker(image_input)
+    original_publish = package.publish_build_push_receipt
+    monkeypatch.setattr(package, "validate_prepared_image_context", lambda _context: (image_input, build_args))
+    monkeypatch.setattr(package, "validate_reviewed_context_source", lambda *_args: None)
+    monkeypatch.setattr(package.subprocess, "run", fake_run)
+
+    def alias_parent_before_publish(*args: object, **kwargs: object) -> None:
+        output_parent.rename(retained_parent)
+        output_parent.symlink_to(context / "receipts", target_is_directory=True)
+        original_publish(*args, **kwargs)
+
+    monkeypatch.setattr(package, "publish_build_push_receipt", alias_parent_before_publish)
+
+    with pytest.raises(package.PackageError, match="symbolic link"):
+        package.build_and_push_staging_image(
+            repo_root,
+            context,
+            docker_binary=docker_binary,
+            docker_host=docker_host,
+            docker_config=docker_config,
+            output_receipt=output,
+            execute_staging_push=True,
+        )
+
+    assert state == {"build": True, "candidate": True, "digests": True, "pushes": 1}
+    assert calls
+    assert not output.exists()
+    assert not (context / "receipts" / "receipt.json").exists()
+
+
 @pytest.mark.parametrize(
     "references",
     [
@@ -1437,6 +1709,41 @@ def test_build_push_receipt_publication_rolls_back_failed_parent_fsync(
     assert retained[0].stat().st_ino == output.stat().st_ino
 
 
+def test_build_push_receipt_detects_parent_replacement_after_publication(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    image_input = image_input_fixture(tmp_path)
+    receipt = build_push_receipt_fixture(image_input)
+    context = tmp_path / "prepared-context"
+    context.mkdir()
+    output_parent = tmp_path / "operator-receipts"
+    output_parent.mkdir()
+    retained_parent = tmp_path / "retained-operator-receipts"
+    output = output_parent / "receipt.json"
+    original_fsync = package._fsync_directory_descriptor
+
+    def replace_parent_after_fsync(descriptor: int) -> None:
+        original_fsync(descriptor)
+        output_parent.rename(retained_parent)
+        output_parent.mkdir(mode=0o700)
+
+    monkeypatch.setattr(package, "_fsync_directory_descriptor", replace_parent_after_fsync)
+
+    with pytest.raises(package.PackageError, match="parent identity drifted"):
+        package.publish_build_push_receipt(
+            output,
+            receipt,
+            image_input=image_input,
+            outside_context=context,
+        )
+
+    assert not output.exists()
+    retained_output = retained_parent / output.name
+    assert retained_output.exists()
+    assert retained_output.stat().st_nlink == 2
+
+
 def test_release_cross_binds_receipt_registry_and_independent_inspection(tmp_path: Path) -> None:
     image_input = image_input_fixture(tmp_path)
     reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
@@ -1818,7 +2125,7 @@ def test_registry_inspection_absent_pull_dependent_retains_new_reference(
     image_input = image_input_fixture(tmp_path)
     reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
     docker_binary, docker_host, docker_config = docker_authority(tmp_path)
-    prefix = [str(docker_binary), "--host", docker_host, "--config", str(docker_config)]
+    prefix = [str(docker_binary), "--host", docker_host, "--config"]
     calls: list[list[str]] = []
     environments: list[dict[str, str]] = []
     state = {"present": False, "dependent": False, "inspection_calls": 0}
@@ -1829,7 +2136,10 @@ def test_registry_inspection_absent_pull_dependent_retains_new_reference(
     def fake_run(arguments: list[str], **kwargs):
         calls.append(arguments)
         environments.append(kwargs["env"])
-        assert arguments[:5] == prefix
+        assert arguments[:4] == prefix
+        assert arguments[4] == "."
+        assert len(kwargs["pass_fds"]) == 1
+        assert kwargs["preexec_fn"] is not None
         operation = arguments[5:]
         if operation[:2] == ["image", "ls"]:
             return subprocess.CompletedProcess(