Coverage for gco_mcp/local_data.py: 95.24%
222 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""Shared confinement and secure staging for MCP tools that access host data."""
3from __future__ import annotations
5import errno
6import os
7import secrets
8import shutil
9import stat
10from collections.abc import Iterator
11from contextlib import contextmanager, suppress
12from dataclasses import dataclass
13from pathlib import Path
15_LOCAL_ROOT_ENV = "GCO_STORAGE_LOCAL_ROOT"
16_UPLOAD_STAGE_PREFIX = ".gco-mcp-upload-"
19@dataclass(frozen=True)
20class LocalPathContract:
21 """A root-confined path and the filesystem identities checked at use time."""
23 local_argument: str
24 resolved_path: Path
25 root: Path
26 device: int
27 inode: int
28 source_device: int | None
29 source_inode: int | None
30 source_mode: int | None
33@dataclass(frozen=True)
34class StagedUpload:
35 """Descriptor-backed upload argument valid for the context lifetime."""
37 argument: str
38 directory_fd: int
41def _file_identity(metadata: os.stat_result) -> tuple[int, int, int]:
42 """Return the device, inode, and file-type bits for one artifact."""
43 return metadata.st_dev, metadata.st_ino, stat.S_IFMT(metadata.st_mode)
46def resolve_local_path(
47 local_path: str,
48 *,
49 require_exists: bool,
50 purpose: str = "Local data",
51) -> LocalPathContract:
52 """Resolve ``local_path`` beneath the configured local-data root.
54 Relative paths (including short forms such as ``weights`` and
55 ``./weights``) resolve beneath ``GCO_STORAGE_LOCAL_ROOT`` rather than the
56 server process's working directory. Lexical traversal and realpath symlink
57 escapes are rejected. Existing source identity is captured so short upload
58 tools can verify it again while building a private no-follow snapshot.
59 """
60 configured_root = os.environ.get(_LOCAL_ROOT_ENV, "").strip()
61 if not configured_root:
62 raise ValueError(f"{_LOCAL_ROOT_ENV} must be set before enabling local data access")
63 if os.name != "posix" or not hasattr(os, "O_NOFOLLOW") or not hasattr(os, "O_DIRECTORY"):
64 raise ValueError(
65 "Local data access requires descriptor-relative no-follow filesystem support"
66 )
68 try:
69 root = Path(configured_root).expanduser().resolve(strict=True)
70 except OSError as exc:
71 if exc.errno != errno.ELOOP:
72 raise
73 raise ValueError(f"{_LOCAL_ROOT_ENV} could not be resolved safely") from exc
74 except RuntimeError as exc:
75 raise ValueError(f"{_LOCAL_ROOT_ENV} could not be resolved safely") from exc
76 root_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
77 root_fd = os.open(root, root_flags)
78 try:
79 root_stat = os.fstat(root_fd)
80 finally:
81 os.close(root_fd)
82 if not stat.S_ISDIR(root_stat.st_mode): 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true
83 raise ValueError(f"{_LOCAL_ROOT_ENV} is not a directory: {root}")
85 supplied = Path(local_path).expanduser()
86 candidate = supplied if supplied.is_absolute() else root / supplied
87 lexical = Path(os.path.abspath(candidate))
88 try:
89 relative = lexical.relative_to(root)
90 except ValueError as exc:
91 raise ValueError(
92 f"{purpose} path must stay within {_LOCAL_ROOT_ENV}: {local_path}"
93 ) from exc
95 try:
96 resolved = lexical.resolve(strict=require_exists)
97 except FileNotFoundError as exc:
98 raise ValueError(f"{purpose} source does not exist: {local_path}") from exc
99 except OSError as exc:
100 if exc.errno != errno.ELOOP: 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true
101 raise
102 raise ValueError(f"{purpose} path could not be resolved safely: {local_path}") from exc
103 except RuntimeError as exc:
104 raise ValueError(f"{purpose} path could not be resolved safely: {local_path}") from exc
105 if not resolved.is_relative_to(root):
106 raise ValueError(f"{purpose} path must stay within {_LOCAL_ROOT_ENV}: {local_path}")
108 source_stat: os.stat_result | None
109 try:
110 source_stat = os.stat(resolved, follow_symlinks=False)
111 except FileNotFoundError:
112 if require_exists:
113 raise ValueError(f"{purpose} source does not exist: {local_path}") from None
114 source_stat = None
115 if source_stat is not None and not (
116 stat.S_ISREG(source_stat.st_mode) or stat.S_ISDIR(source_stat.st_mode)
117 ):
118 raise ValueError(f"{purpose} source must be a regular file or directory: {local_path}")
120 return LocalPathContract(
121 local_argument=str(relative) if relative.parts else ".",
122 resolved_path=resolved,
123 root=root,
124 device=root_stat.st_dev,
125 inode=root_stat.st_ino,
126 source_device=source_stat.st_dev if source_stat is not None else None,
127 source_inode=source_stat.st_ino if source_stat is not None else None,
128 source_mode=source_stat.st_mode if source_stat is not None else None,
129 )
132def _verified_root_fd(contract: LocalPathContract) -> int:
133 """Open the configured root without following a swapped final symlink."""
134 flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
135 fd = os.open(contract.root, flags)
136 try:
137 metadata = os.fstat(fd)
138 if not stat.S_ISDIR(metadata.st_mode) or (metadata.st_dev, metadata.st_ino) != (
139 contract.device,
140 contract.inode,
141 ):
142 raise ValueError(f"{_LOCAL_ROOT_ENV} changed after validation")
143 return fd
144 except Exception:
145 os.close(fd)
146 raise
149def _open_source(
150 root_fd: int,
151 contract: LocalPathContract,
152) -> tuple[int, int | None, str | None]:
153 """Open the captured source by canonical root-relative components."""
154 relative = contract.resolved_path.relative_to(contract.root)
155 components = relative.parts
156 current_fd = os.dup(root_fd)
157 try:
158 if not components:
159 return current_fd, None, None
160 directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
161 for component in components[:-1]:
162 next_fd = os.open(component, directory_flags, dir_fd=current_fd)
163 os.close(current_fd)
164 current_fd = next_fd
165 source_flags = (
166 os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NONBLOCK", 0)
167 )
168 source_fd = os.open(components[-1], source_flags, dir_fd=current_fd)
169 return source_fd, current_fd, components[-1]
170 except Exception:
171 os.close(current_fd)
172 raise
175def _verify_source_identity(contract: LocalPathContract, metadata: os.stat_result) -> None:
176 """Reject source replacement, mount crossings, links, and special files."""
177 expected = (contract.source_device, contract.source_inode)
178 if None in expected or contract.source_mode is None:
179 raise ValueError("Upload source must exist before secure staging")
180 if _file_identity(metadata) != (
181 contract.source_device,
182 contract.source_inode,
183 stat.S_IFMT(contract.source_mode),
184 ):
185 raise ValueError("Upload source changed after validation")
186 if metadata.st_dev != contract.device:
187 raise ValueError("Upload source crosses a filesystem boundary")
188 if not (stat.S_ISREG(metadata.st_mode) or stat.S_ISDIR(metadata.st_mode)):
189 raise ValueError("Upload source must be a regular file or directory")
190 if stat.S_ISREG(metadata.st_mode) and metadata.st_nlink != 1:
191 raise ValueError("Upload source must not be hard-linked")
194def _link_verified_regular(
195 source_dir_fd: int,
196 source_name: str,
197 source_stat: os.stat_result,
198 destination_dir_fd: int,
199 destination_name: str,
200) -> None:
201 """Hard-link one opened regular file and verify the name did not race."""
202 if not stat.S_ISREG(source_stat.st_mode) or source_stat.st_nlink != 1:
203 raise ValueError(f"Upload entry is not a private regular file: {source_name}")
204 os.link(
205 source_name,
206 destination_name,
207 src_dir_fd=source_dir_fd,
208 dst_dir_fd=destination_dir_fd,
209 follow_symlinks=False,
210 )
211 try:
212 source_after = os.stat(source_name, dir_fd=source_dir_fd, follow_symlinks=False)
213 destination = os.stat(
214 destination_name,
215 dir_fd=destination_dir_fd,
216 follow_symlinks=False,
217 )
218 expected = _file_identity(source_stat)
219 if (
220 _file_identity(source_after) != expected
221 or _file_identity(destination) != expected
222 or source_after.st_nlink != 2
223 or destination.st_nlink != 2
224 ):
225 raise ValueError(f"Upload entry changed while staging: {source_name}")
226 except Exception:
227 with suppress(OSError):
228 os.unlink(destination_name, dir_fd=destination_dir_fd)
229 raise
232def _stage_directory(
233 source_fd: int,
234 destination_fd: int,
235 *,
236 root_device: int,
237 visited: set[tuple[int, int]],
238 skip_internal_stages: bool,
239) -> None:
240 """Recursively snapshot regular files without following any link."""
241 directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
242 regular_flags = (
243 os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NONBLOCK", 0)
244 )
245 for name in sorted(os.listdir(source_fd)):
246 if skip_internal_stages and name.startswith(_UPLOAD_STAGE_PREFIX):
247 continue
248 before = os.stat(name, dir_fd=source_fd, follow_symlinks=False)
249 if before.st_dev != root_device:
250 raise ValueError(f"Upload entry crosses a filesystem boundary: {name}")
251 if stat.S_ISLNK(before.st_mode):
252 raise ValueError(f"Upload entry must not be a symbolic link: {name}")
253 if stat.S_ISREG(before.st_mode):
254 opened_fd = os.open(name, regular_flags, dir_fd=source_fd)
255 try:
256 opened = os.fstat(opened_fd)
257 if _file_identity(opened) != _file_identity(before):
258 raise ValueError(f"Upload entry changed while opening: {name}")
259 _link_verified_regular(source_fd, name, opened, destination_fd, name)
260 finally:
261 os.close(opened_fd)
262 continue
263 if stat.S_ISDIR(before.st_mode):
264 identity = (before.st_dev, before.st_ino)
265 if identity in visited:
266 raise ValueError(f"Upload directory cycle detected: {name}")
267 child_fd = os.open(name, directory_flags, dir_fd=source_fd)
268 try:
269 opened = os.fstat(child_fd)
270 if _file_identity(opened) != _file_identity(before):
271 raise ValueError(f"Upload directory changed while opening: {name}")
272 os.mkdir(name, mode=0o700, dir_fd=destination_fd)
273 child_destination_fd = os.open(name, directory_flags, dir_fd=destination_fd)
274 try:
275 visited.add(identity)
276 _stage_directory(
277 child_fd,
278 child_destination_fd,
279 root_device=root_device,
280 visited=visited,
281 skip_internal_stages=False,
282 )
283 finally:
284 visited.remove(identity)
285 os.close(child_destination_fd)
286 finally:
287 os.close(child_fd)
288 continue
289 raise ValueError(f"Upload entry must be a regular file or directory: {name}")
292def _create_stage_directory(root_fd: int) -> tuple[str, int]:
293 """Create one unpredictable private staging directory under the root."""
294 flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
295 for _ in range(10):
296 name = f"{_UPLOAD_STAGE_PREFIX}{secrets.token_hex(16)}"
297 try:
298 os.mkdir(name, mode=0o700, dir_fd=root_fd)
299 except FileExistsError:
300 continue
301 return name, os.open(name, flags, dir_fd=root_fd)
302 raise OSError("Unable to allocate a private upload staging directory")
305@contextmanager
306def stage_upload_path(contract: LocalPathContract) -> Iterator[StagedUpload]:
307 """Yield a private no-follow snapshot suitable for a short CLI upload.
309 The snapshot contains only directories and hard links to regular,
310 single-link files discovered through descriptor-relative no-follow opens.
311 The CLI receives ``/dev/fd/<dirfd>/<name>`` and that directory descriptor is
312 explicitly inherited by its subprocess, closing validation/use path races.
313 """
314 if not Path("/dev/fd").is_dir():
315 raise ValueError("Secure upload staging requires /dev/fd support")
317 root_fd = _verified_root_fd(contract)
318 stage_name: str | None = None
319 stage_fd: int | None = None
320 source_fd: int | None = None
321 source_parent_fd: int | None = None
322 try:
323 source_fd, source_parent_fd, source_name = _open_source(root_fd, contract)
324 source_stat = os.fstat(source_fd)
325 _verify_source_identity(contract, source_stat)
327 stage_name, stage_fd = _create_stage_directory(root_fd)
328 target_name = contract.resolved_path.name or "upload"
329 if stat.S_ISREG(source_stat.st_mode):
330 if source_parent_fd is None or source_name is None: 330 ↛ 331line 330 didn't jump to line 331 because the condition on line 330 was never true
331 raise ValueError("Regular upload source has no parent directory")
332 _link_verified_regular(
333 source_parent_fd,
334 source_name,
335 source_stat,
336 stage_fd,
337 target_name,
338 )
339 else:
340 os.mkdir(target_name, mode=0o700, dir_fd=stage_fd)
341 flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
342 destination_fd = os.open(target_name, flags, dir_fd=stage_fd)
343 try:
344 source_identity = (source_stat.st_dev, source_stat.st_ino)
345 _stage_directory(
346 source_fd,
347 destination_fd,
348 root_device=contract.device,
349 visited={source_identity},
350 skip_internal_stages=source_identity == (contract.device, contract.inode),
351 )
352 finally:
353 os.close(destination_fd)
355 argument = f"/dev/fd/{stage_fd}/{target_name}"
356 staged_stat = os.stat(argument, follow_symlinks=False)
357 if not (stat.S_ISREG(staged_stat.st_mode) or stat.S_ISDIR(staged_stat.st_mode)):
358 raise ValueError("Secure upload staging produced an invalid source")
359 yield StagedUpload(argument=argument, directory_fd=stage_fd)
360 finally:
361 if source_fd is not None: 361 ↛ 363line 361 didn't jump to line 363 because the condition on line 361 was always true
362 os.close(source_fd)
363 if source_parent_fd is not None:
364 os.close(source_parent_fd)
365 if stage_fd is not None:
366 os.close(stage_fd)
367 if stage_name is not None:
368 with suppress(OSError):
369 shutil.rmtree(stage_name, dir_fd=root_fd)
370 os.close(root_fd)