Coverage for gco_mcp / local_data.py: 100.00%
219 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +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# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
16# Generated at (UTC): 2026-09-29T11:35:23Z
17# Generated from Git commit: a081fbf678ca3f3fa3f7edb74687de923efe6657
18# Flowchart(s) generated from this file:
19# * ``resolve_local_path`` -> ``diagrams/code_diagrams/gco_mcp/local_data.resolve_local_path.html``
20# (PNG: ``diagrams/code_diagrams/gco_mcp/local_data.resolve_local_path.png``)
21# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
22# <pyflowchart-code-diagram> END
25_LOCAL_ROOT_ENV = "GCO_STORAGE_LOCAL_ROOT"
26_UPLOAD_STAGE_PREFIX = ".gco-mcp-upload-"
29@dataclass(frozen=True)
30class LocalPathContract:
31 """A root-confined path and the filesystem identities checked at use time."""
33 local_argument: str
34 resolved_path: Path
35 root: Path
36 device: int
37 inode: int
38 source_device: int | None
39 source_inode: int | None
40 source_mode: int | None
43@dataclass(frozen=True)
44class StagedUpload:
45 """Descriptor-backed upload argument valid for the context lifetime."""
47 argument: str
48 directory_fd: int
51def _file_identity(metadata: os.stat_result) -> tuple[int, int, int]:
52 """Return the device, inode, and file-type bits for one artifact."""
53 return metadata.st_dev, metadata.st_ino, stat.S_IFMT(metadata.st_mode)
56def resolve_local_path(
57 local_path: str,
58 *,
59 require_exists: bool,
60 purpose: str = "Local data",
61) -> LocalPathContract:
62 """Resolve ``local_path`` beneath the configured local-data root.
64 Relative paths (including short forms such as ``weights`` and
65 ``./weights``) resolve beneath ``GCO_STORAGE_LOCAL_ROOT`` rather than the
66 server process's working directory. Lexical traversal and realpath symlink
67 escapes are rejected. Existing source identity is captured so short upload
68 tools can verify it again while building a private no-follow snapshot.
69 """
70 configured_root = os.environ.get(_LOCAL_ROOT_ENV, "").strip()
71 if not configured_root:
72 raise ValueError(f"{_LOCAL_ROOT_ENV} must be set before enabling local data access")
73 if os.name != "posix" or not hasattr(os, "O_NOFOLLOW") or not hasattr(os, "O_DIRECTORY"):
74 raise ValueError(
75 "Local data access requires descriptor-relative no-follow filesystem support"
76 )
78 try:
79 root = Path(configured_root).expanduser().resolve(strict=True)
80 except OSError as exc:
81 if exc.errno != errno.ELOOP:
82 raise
83 raise ValueError(f"{_LOCAL_ROOT_ENV} could not be resolved safely") from exc
84 except RuntimeError as exc:
85 raise ValueError(f"{_LOCAL_ROOT_ENV} could not be resolved safely") from exc
86 root_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
87 root_fd = os.open(root, root_flags)
88 try:
89 root_stat = os.fstat(root_fd)
90 finally:
91 os.close(root_fd)
93 supplied = Path(local_path).expanduser()
94 candidate = supplied if supplied.is_absolute() else root / supplied
95 lexical = Path(os.path.abspath(candidate))
96 try:
97 relative = lexical.relative_to(root)
98 except ValueError as exc:
99 raise ValueError(
100 f"{purpose} path must stay within {_LOCAL_ROOT_ENV}: {local_path}"
101 ) from exc
103 try:
104 resolved = lexical.resolve(strict=require_exists)
105 except FileNotFoundError as exc:
106 raise ValueError(f"{purpose} source does not exist: {local_path}") from exc
107 except OSError as exc:
108 if exc.errno != errno.ELOOP:
109 raise
110 raise ValueError(f"{purpose} path could not be resolved safely: {local_path}") from exc
111 except RuntimeError as exc:
112 raise ValueError(f"{purpose} path could not be resolved safely: {local_path}") from exc
113 if not resolved.is_relative_to(root):
114 raise ValueError(f"{purpose} path must stay within {_LOCAL_ROOT_ENV}: {local_path}")
116 source_stat: os.stat_result | None
117 try:
118 source_stat = os.stat(resolved, follow_symlinks=False)
119 except FileNotFoundError:
120 if require_exists:
121 raise ValueError(f"{purpose} source does not exist: {local_path}") from None
122 source_stat = None
123 if source_stat is not None and not (
124 stat.S_ISREG(source_stat.st_mode) or stat.S_ISDIR(source_stat.st_mode)
125 ):
126 raise ValueError(f"{purpose} source must be a regular file or directory: {local_path}")
128 return LocalPathContract(
129 local_argument=str(relative) if relative.parts else ".",
130 resolved_path=resolved,
131 root=root,
132 device=root_stat.st_dev,
133 inode=root_stat.st_ino,
134 source_device=source_stat.st_dev if source_stat is not None else None,
135 source_inode=source_stat.st_ino if source_stat is not None else None,
136 source_mode=source_stat.st_mode if source_stat is not None else None,
137 )
140def _verified_root_fd(contract: LocalPathContract) -> int:
141 """Open the configured root without following a swapped final symlink."""
142 flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
143 fd = os.open(contract.root, flags)
144 try:
145 metadata = os.fstat(fd)
146 if not stat.S_ISDIR(metadata.st_mode) or (metadata.st_dev, metadata.st_ino) != (
147 contract.device,
148 contract.inode,
149 ):
150 raise ValueError(f"{_LOCAL_ROOT_ENV} changed after validation")
151 return fd
152 except Exception:
153 os.close(fd)
154 raise
157def _open_source(
158 root_fd: int,
159 contract: LocalPathContract,
160) -> tuple[int, int | None, str | None]:
161 """Open the captured source by canonical root-relative components."""
162 relative = contract.resolved_path.relative_to(contract.root)
163 components = relative.parts
164 current_fd = os.dup(root_fd)
165 try:
166 if not components:
167 return current_fd, None, None
168 directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
169 for component in components[:-1]:
170 next_fd = os.open(component, directory_flags, dir_fd=current_fd)
171 os.close(current_fd)
172 current_fd = next_fd
173 source_flags = (
174 os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NONBLOCK", 0)
175 )
176 source_fd = os.open(components[-1], source_flags, dir_fd=current_fd)
177 return source_fd, current_fd, components[-1]
178 except Exception:
179 os.close(current_fd)
180 raise
183def _verify_source_identity(contract: LocalPathContract, metadata: os.stat_result) -> None:
184 """Reject source replacement, mount crossings, links, and special files."""
185 expected = (contract.source_device, contract.source_inode)
186 if None in expected or contract.source_mode is None:
187 raise ValueError("Upload source must exist before secure staging")
188 if _file_identity(metadata) != (
189 contract.source_device,
190 contract.source_inode,
191 stat.S_IFMT(contract.source_mode),
192 ):
193 raise ValueError("Upload source changed after validation")
194 if metadata.st_dev != contract.device:
195 raise ValueError("Upload source crosses a filesystem boundary")
196 if not (stat.S_ISREG(metadata.st_mode) or stat.S_ISDIR(metadata.st_mode)):
197 raise ValueError("Upload source must be a regular file or directory")
198 if stat.S_ISREG(metadata.st_mode) and metadata.st_nlink != 1:
199 raise ValueError("Upload source must not be hard-linked")
202def _link_verified_regular(
203 source_dir_fd: int,
204 source_name: str,
205 source_stat: os.stat_result,
206 destination_dir_fd: int,
207 destination_name: str,
208) -> None:
209 """Hard-link one opened regular file and verify the name did not race."""
210 if not stat.S_ISREG(source_stat.st_mode) or source_stat.st_nlink != 1:
211 raise ValueError(f"Upload entry is not a private regular file: {source_name}")
212 os.link(
213 source_name,
214 destination_name,
215 src_dir_fd=source_dir_fd,
216 dst_dir_fd=destination_dir_fd,
217 follow_symlinks=False,
218 )
219 try:
220 source_after = os.stat(source_name, dir_fd=source_dir_fd, follow_symlinks=False)
221 destination = os.stat(
222 destination_name,
223 dir_fd=destination_dir_fd,
224 follow_symlinks=False,
225 )
226 expected = _file_identity(source_stat)
227 if (
228 _file_identity(source_after) != expected
229 or _file_identity(destination) != expected
230 or source_after.st_nlink != 2
231 or destination.st_nlink != 2
232 ):
233 raise ValueError(f"Upload entry changed while staging: {source_name}")
234 except Exception:
235 with suppress(OSError):
236 os.unlink(destination_name, dir_fd=destination_dir_fd)
237 raise
240def _stage_directory(
241 source_fd: int,
242 destination_fd: int,
243 *,
244 root_device: int,
245 visited: set[tuple[int, int]],
246 skip_internal_stages: bool,
247) -> None:
248 """Recursively snapshot regular files without following any link."""
249 directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
250 regular_flags = (
251 os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NONBLOCK", 0)
252 )
253 for name in sorted(os.listdir(source_fd)):
254 if skip_internal_stages and name.startswith(_UPLOAD_STAGE_PREFIX):
255 continue
256 before = os.stat(name, dir_fd=source_fd, follow_symlinks=False)
257 if before.st_dev != root_device:
258 raise ValueError(f"Upload entry crosses a filesystem boundary: {name}")
259 if stat.S_ISLNK(before.st_mode):
260 raise ValueError(f"Upload entry must not be a symbolic link: {name}")
261 if stat.S_ISREG(before.st_mode):
262 opened_fd = os.open(name, regular_flags, dir_fd=source_fd)
263 try:
264 opened = os.fstat(opened_fd)
265 if _file_identity(opened) != _file_identity(before):
266 raise ValueError(f"Upload entry changed while opening: {name}")
267 _link_verified_regular(source_fd, name, opened, destination_fd, name)
268 finally:
269 os.close(opened_fd)
270 continue
271 if stat.S_ISDIR(before.st_mode):
272 identity = (before.st_dev, before.st_ino)
273 if identity in visited:
274 raise ValueError(f"Upload directory cycle detected: {name}")
275 child_fd = os.open(name, directory_flags, dir_fd=source_fd)
276 try:
277 opened = os.fstat(child_fd)
278 if _file_identity(opened) != _file_identity(before):
279 raise ValueError(f"Upload directory changed while opening: {name}")
280 os.mkdir(name, mode=0o700, dir_fd=destination_fd)
281 child_destination_fd = os.open(name, directory_flags, dir_fd=destination_fd)
282 try:
283 visited.add(identity)
284 _stage_directory(
285 child_fd,
286 child_destination_fd,
287 root_device=root_device,
288 visited=visited,
289 skip_internal_stages=False,
290 )
291 finally:
292 visited.remove(identity)
293 os.close(child_destination_fd)
294 finally:
295 os.close(child_fd)
296 continue
297 raise ValueError(f"Upload entry must be a regular file or directory: {name}")
300def _create_stage_directory(root_fd: int) -> tuple[str, int]:
301 """Create one unpredictable private staging directory under the root."""
302 flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
303 for _ in range(10):
304 name = f"{_UPLOAD_STAGE_PREFIX}{secrets.token_hex(16)}"
305 try:
306 os.mkdir(name, mode=0o700, dir_fd=root_fd)
307 except FileExistsError:
308 continue
309 return name, os.open(name, flags, dir_fd=root_fd)
310 raise OSError("Unable to allocate a private upload staging directory")
313@contextmanager
314def stage_upload_path(contract: LocalPathContract) -> Iterator[StagedUpload]:
315 """Yield a private no-follow snapshot suitable for a short CLI upload.
317 The snapshot contains only directories and hard links to regular,
318 single-link files discovered through descriptor-relative no-follow opens.
319 The CLI receives ``/dev/fd/<dirfd>/<name>`` and that directory descriptor is
320 explicitly inherited by its subprocess, closing validation/use path races.
321 """
322 if not Path("/dev/fd").is_dir():
323 raise ValueError("Secure upload staging requires /dev/fd support")
325 root_fd = _verified_root_fd(contract)
326 stage_name: str | None = None
327 stage_fd: int | None = None
328 source_fd: int | None = None
329 source_parent_fd: int | None = None
330 try:
331 source_fd, source_parent_fd, source_name = _open_source(root_fd, contract)
332 source_stat = os.fstat(source_fd)
333 _verify_source_identity(contract, source_stat)
335 stage_name, stage_fd = _create_stage_directory(root_fd)
336 target_name = contract.resolved_path.name or "upload"
337 if stat.S_ISREG(source_stat.st_mode):
338 # ``_open_source`` returns a null parent/name pair only for the root
339 # directory itself, and the root is opened with ``O_DIRECTORY``, so a
340 # regular source always carries both values. Stating that invariant
341 # here narrows the types and fails loudly if it is ever broken,
342 # rather than carrying an unreachable user-error path.
343 assert source_parent_fd is not None and source_name is not None
344 _link_verified_regular(
345 source_parent_fd,
346 source_name,
347 source_stat,
348 stage_fd,
349 target_name,
350 )
351 else:
352 os.mkdir(target_name, mode=0o700, dir_fd=stage_fd)
353 flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
354 destination_fd = os.open(target_name, flags, dir_fd=stage_fd)
355 try:
356 source_identity = (source_stat.st_dev, source_stat.st_ino)
357 _stage_directory(
358 source_fd,
359 destination_fd,
360 root_device=contract.device,
361 visited={source_identity},
362 skip_internal_stages=source_identity == (contract.device, contract.inode),
363 )
364 finally:
365 os.close(destination_fd)
367 argument = f"/dev/fd/{stage_fd}/{target_name}"
368 staged_stat = os.stat(argument, follow_symlinks=False)
369 if not (stat.S_ISREG(staged_stat.st_mode) or stat.S_ISDIR(staged_stat.st_mode)):
370 raise ValueError("Secure upload staging produced an invalid source")
371 yield StagedUpload(argument=argument, directory_fd=stage_fd)
372 finally:
373 if source_fd is not None:
374 os.close(source_fd)
375 if source_parent_fd is not None:
376 os.close(source_parent_fd)
377 if stage_fd is not None:
378 os.close(stage_fd)
379 if stage_name is not None:
380 with suppress(OSError):
381 shutil.rmtree(stage_name, dir_fd=root_fd)
382 os.close(root_fd)