Coverage for scripts / live_release_validation / artifact_io.py: 100.00%
125 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Owner-only artifact directory and atomic text I/O for live validation."""
3from __future__ import annotations
5import os
6import secrets
7import stat
8import tempfile
9from collections.abc import Iterator
10from contextlib import contextmanager, suppress
11from pathlib import Path
12from typing import TextIO
14_PRIVATE_DIRECTORY_MODE = 0o700
15_PRIVATE_FILE_MODE = 0o600
16REPORT_FILENAMES = frozenset(
17 {
18 "live-release-validation.json",
19 "live-release-validation.md",
20 "example-job-validation.json",
21 "example-job-validation.md",
22 "kubeconfig",
23 }
24)
27def _validate_private_regular_metadata(metadata: os.stat_result, path: Path) -> None:
28 """Reject links, special files, foreign owners, and non-private POSIX modes."""
29 if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
30 raise ValueError(f"Live-validation output must be a regular file, not {path}")
31 if hasattr(os, "geteuid") and metadata.st_uid != os.geteuid():
32 raise PermissionError(f"Live-validation output is not owned by this user: {path}")
33 if os.name != "nt" and stat.S_IMODE(metadata.st_mode) != _PRIVATE_FILE_MODE:
34 raise PermissionError(f"Live-validation output must have mode 0600: {path}")
37def _validate_private_regular_file(path: Path) -> None:
38 _validate_private_regular_metadata(path.lstat(), path)
41def _validate_private_directory_metadata(metadata: os.stat_result, directory: Path) -> None:
42 """Require a real, current-user-owned, owner-only run directory."""
43 if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
44 raise ValueError(f"Live-validation output directory must be real: {directory}")
45 if hasattr(os, "geteuid") and metadata.st_uid != os.geteuid():
46 raise PermissionError(
47 f"Live-validation output directory is not owned by this user: {directory}"
48 )
49 if os.name != "nt" and stat.S_IMODE(metadata.st_mode) != _PRIVATE_DIRECTORY_MODE:
50 raise PermissionError(
51 f"Live-validation output directory must already have mode 0700: {directory}"
52 )
55def ensure_private_directory(directory: Path) -> None:
56 """Create a private directory or validate it without changing permissions."""
57 directory = Path(directory)
58 with suppress(FileExistsError):
59 directory.mkdir(parents=True, mode=_PRIVATE_DIRECTORY_MODE, exist_ok=False)
60 _validate_private_directory_metadata(directory.lstat(), directory)
63def _assert_directory_binding(directory: Path, descriptor: int) -> None:
64 """Fail if the verified pathname no longer names the pinned directory."""
65 try:
66 current = directory.lstat()
67 except OSError as exc:
68 raise RuntimeError(
69 f"Live-validation output directory was rebound while open: {directory}"
70 ) from exc
71 if not os.path.samestat(current, os.fstat(descriptor)):
72 raise RuntimeError(f"Live-validation output directory was rebound while open: {directory}")
75@contextmanager
76def _open_private_directory(directory: Path) -> Iterator[int | None]:
77 """Pin a validated directory so artifact I/O cannot follow a rebound pathname."""
78 directory = Path(directory)
79 ensure_private_directory(directory)
80 if os.name == "nt":
81 yield None
82 return
84 no_follow = getattr(os, "O_NOFOLLOW", 0)
85 directory_only = getattr(os, "O_DIRECTORY", 0)
86 if not no_follow or not directory_only:
87 raise RuntimeError("Secure directory-descriptor operations are unavailable")
89 before = directory.lstat()
90 descriptor = os.open(
91 directory,
92 os.O_RDONLY | no_follow | directory_only | getattr(os, "O_CLOEXEC", 0),
93 )
94 try:
95 opened = os.fstat(descriptor)
96 if not os.path.samestat(before, opened):
97 raise RuntimeError(
98 f"Live-validation output directory changed while opening: {directory}"
99 )
100 _validate_private_directory_metadata(opened, directory)
101 yield descriptor
102 _assert_directory_binding(directory, descriptor)
103 finally:
104 os.close(descriptor)
107def ensure_private_run_directory(directory: Path, checkpoint_path: Path) -> None:
108 """Validate that an existing private directory is dedicated to one harness run."""
109 directory = Path(directory)
110 allowed_names = {*REPORT_FILENAMES, checkpoint_path.name}
111 with _open_private_directory(directory) as descriptor:
112 if descriptor is None:
113 entries = [(entry.name, entry.lstat()) for entry in directory.iterdir()]
114 else:
115 entries = [
116 (name, os.stat(name, dir_fd=descriptor, follow_symlinks=False))
117 for name in os.listdir(descriptor)
118 ]
119 for name, metadata in entries:
120 is_temporary = any(
121 name.startswith(f".{allowed_name}.") and name.endswith(".tmp")
122 for allowed_name in allowed_names
123 )
124 entry = directory / name
125 if name not in allowed_names and not is_temporary:
126 raise ValueError(
127 "Live-validation output directory contains an unrelated entry and is not "
128 f"dedicated to this run: {entry}"
129 )
130 _validate_private_regular_metadata(metadata, entry)
133def read_private_text(path: Path) -> str:
134 """Read one owner-only regular file relative to a pinned directory."""
135 with _open_private_directory(path.parent) as descriptor:
136 if descriptor is None:
137 _validate_private_regular_file(path)
138 return path.read_text(encoding="utf-8")
140 opened_descriptor = os.open(
141 path.name,
142 os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0),
143 dir_fd=descriptor,
144 )
145 descriptor_to_close: int | None = opened_descriptor
146 try:
147 _validate_private_regular_metadata(os.fstat(opened_descriptor), path)
148 text_handle: TextIO = os.fdopen(opened_descriptor, mode="r", encoding="utf-8")
149 descriptor_to_close = None
150 with text_handle:
151 return text_handle.read()
152 finally:
153 if descriptor_to_close is not None:
154 os.close(descriptor_to_close)
157def atomic_write_text(path: Path, content: str) -> None:
158 """Atomically persist owner-only text relative to a pinned private directory."""
159 with _open_private_directory(path.parent) as descriptor:
160 if descriptor is None:
161 temporary_path_to_unlink: Path | None = None
162 try:
163 with tempfile.NamedTemporaryFile(
164 mode="w",
165 encoding="utf-8",
166 dir=path.parent,
167 prefix=f".{path.name}.",
168 suffix=".tmp",
169 delete=False,
170 ) as named_handle:
171 temporary_path = Path(named_handle.name)
172 temporary_path_to_unlink = temporary_path
173 named_handle.write(content)
174 named_handle.flush()
175 os.fsync(named_handle.fileno())
176 os.replace(temporary_path, path)
177 temporary_path_to_unlink = None
178 finally:
179 if temporary_path_to_unlink is not None:
180 temporary_path_to_unlink.unlink(missing_ok=True)
181 return
183 temporary_name = f".{path.name}.{secrets.token_hex(16)}.tmp"
184 temporary_name_to_unlink: str | None = temporary_name
185 descriptor_to_close: int | None = None
186 try:
187 opened_descriptor = os.open(
188 temporary_name,
189 os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0),
190 _PRIVATE_FILE_MODE,
191 dir_fd=descriptor,
192 )
193 descriptor_to_close = opened_descriptor
194 os.fchmod(opened_descriptor, _PRIVATE_FILE_MODE)
195 text_handle: TextIO = os.fdopen(opened_descriptor, mode="w", encoding="utf-8")
196 descriptor_to_close = None
197 with text_handle:
198 text_handle.write(content)
199 text_handle.flush()
200 os.fsync(text_handle.fileno())
202 os.replace(
203 temporary_name,
204 path.name,
205 src_dir_fd=descriptor,
206 dst_dir_fd=descriptor,
207 )
208 temporary_name_to_unlink = None
209 finally:
210 if descriptor_to_close is not None:
211 os.close(descriptor_to_close)
212 if temporary_name_to_unlink is not None:
213 with suppress(FileNotFoundError):
214 os.unlink(temporary_name_to_unlink, dir_fd=descriptor)