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