Preserve IAP request member modes

ops/gcp_leoclean_nosend_iap_remote.py

@@ -272,15 +272,32 @@ def _acquire_lock(path: Path = LOCK_PATH, *, expected_uid: int = 0) -> int:
 
 
 def _copy_into_request(path: Path, content: bytes, mode: int = 0o600) -> None:
+    _require(mode in {0o600, 0o644, 0o755}, "request file mode was invalid")
     flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
     if hasattr(os, "O_NOFOLLOW"):
         flags |= os.O_NOFOLLOW
     try:
         descriptor = os.open(path, flags, mode)
         with os.fdopen(descriptor, "wb") as handle:
+            # receive_request() deliberately runs under umask 077. Apply the
+            # manifest-bound mode explicitly so public release metadata and
+            # executable helpers retain their reviewed 0644/0755 posture.
+            os.fchmod(handle.fileno(), mode)
             handle.write(content)
             handle.flush()
             os.fsync(handle.fileno())
+            observed = os.fstat(handle.fileno())
+            named = path.stat(follow_symlinks=False)
+            _require(
+                stat.S_ISREG(observed.st_mode)
+                and (observed.st_dev, observed.st_ino) == (named.st_dev, named.st_ino)
+                and observed.st_uid == os.geteuid()
+                and stat.S_IMODE(observed.st_mode) == mode
+                and stat.S_IMODE(named.st_mode) == mode
+                and observed.st_nlink == 1
+                and observed.st_size == len(content),
+                "request file posture differed from its manifest",
+            )
     except OSError as exc:
         raise RemoteError("request file could not be secured") from exc
 

tests/test_gcp_leoclean_nosend_iap_runner.py

@@ -284,6 +284,31 @@ def test_archive_is_deterministic_and_extracts_exact_members(tmp_path: Path) ->
     )
 
 
+def test_archive_extraction_preserves_manifest_modes_under_restrictive_umask(tmp_path: Path) -> None:
+    archive_root = private_directory(tmp_path / "archive")
+    extracted = private_directory(tmp_path / "extracted")
+    allowed_modes = (0o600, 0o644, 0o755)
+    payloads = {
+        name: (f"payload:{name}\n".encode(), allowed_modes[index % len(allowed_modes)])
+        for index, name in enumerate(sorted(remote.EXPECTED_MEMBERS - {"request-manifest.json"}))
+    }
+    manifest = archive_manifest(payloads)
+    archive = archive_root / "request.tar.gz"
+    local.build_archive(archive, manifest=manifest, payloads=payloads)
+
+    previous_umask = os.umask(0o077)
+    try:
+        remote.extract_archive(archive, extracted)
+    finally:
+        os.umask(previous_umask)
+
+    assert {
+        str(path.relative_to(extracted)): stat.S_IMODE(path.stat().st_mode)
+        for path in extracted.rglob("*")
+        if path.is_file()
+    } == {name: mode for name, (_content, mode) in payloads.items()}
+
+
 def write_unsafe_archive(path: Path, *, kind: str) -> None:
     with tarfile.open(path, "w:gz") as archive:
         for index, name in enumerate(sorted(remote.EXPECTED_MEMBERS)):