Coverage for gco_mcp / local_data.py: 100.00%
219 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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)
83 supplied = Path(local_path).expanduser()
84 candidate = supplied if supplied.is_absolute() else root / supplied
85 lexical = Path(os.path.abspath(candidate))
86 try:
87 relative = lexical.relative_to(root)
88 except ValueError as exc:
89 raise ValueError(
90 f"{purpose} path must stay within {_LOCAL_ROOT_ENV}: {local_path}"
91 ) from exc
93 try:
94 resolved = lexical.resolve(strict=require_exists)
95 except FileNotFoundError as exc:
96 raise ValueError(f"{purpose} source does not exist: {local_path}") from exc
97 except OSError as exc:
98 if exc.errno != errno.ELOOP:
99 raise
100 raise ValueError(f"{purpose} path could not be resolved safely: {local_path}") from exc
101 except RuntimeError as exc:
102 raise ValueError(f"{purpose} path could not be resolved safely: {local_path}") from exc
103 if not resolved.is_relative_to(root):
104 raise ValueError(f"{purpose} path must stay within {_LOCAL_ROOT_ENV}: {local_path}")
106 source_stat: os.stat_result | None
107 try:
108 source_stat = os.stat(resolved, follow_symlinks=False)
109 except FileNotFoundError:
110 if require_exists:
111 raise ValueError(f"{purpose} source does not exist: {local_path}") from None
112 source_stat = None
113 if source_stat is not None and not (
114 stat.S_ISREG(source_stat.st_mode) or stat.S_ISDIR(source_stat.st_mode)
115 ):
116 raise ValueError(f"{purpose} source must be a regular file or directory: {local_path}")
118 return LocalPathContract(
119 local_argument=str(relative) if relative.parts else ".",
120 resolved_path=resolved,
121 root=root,
122 device=root_stat.st_dev,
123 inode=root_stat.st_ino,
124 source_device=source_stat.st_dev if source_stat is not None else None,
125 source_inode=source_stat.st_ino if source_stat is not None else None,
126 source_mode=source_stat.st_mode if source_stat is not None else None,
127 )
130def _verified_root_fd(contract: LocalPathContract) -> int:
131 """Open the configured root without following a swapped final symlink."""
132 flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
133 fd = os.open(contract.root, flags)
134 try:
135 metadata = os.fstat(fd)
136 if not stat.S_ISDIR(metadata.st_mode) or (metadata.st_dev, metadata.st_ino) != (
137 contract.device,
138 contract.inode,
139 ):
140 raise ValueError(f"{_LOCAL_ROOT_ENV} changed after validation")
141 return fd
142 except Exception:
143 os.close(fd)
144 raise
147def _open_source(
148 root_fd: int,
149 contract: LocalPathContract,
150) -> tuple[int, int | None, str | None]:
151 """Open the captured source by canonical root-relative components."""
152 relative = contract.resolved_path.relative_to(contract.root)
153 components = relative.parts
154 current_fd = os.dup(root_fd)
155 try:
156 if not components:
157 return current_fd, None, None
158 directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
159 for component in components[:-1]:
160 next_fd = os.open(component, directory_flags, dir_fd=current_fd)
161 os.close(current_fd)
162 current_fd = next_fd
163 source_flags = (
164 os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NONBLOCK", 0)
165 )
166 source_fd = os.open(components[-1], source_flags, dir_fd=current_fd)
167 return source_fd, current_fd, components[-1]
168 except Exception:
169 os.close(current_fd)
170 raise
173def _verify_source_identity(contract: LocalPathContract, metadata: os.stat_result) -> None:
174 """Reject source replacement, mount crossings, links, and special files."""
175 expected = (contract.source_device, contract.source_inode)
176 if None in expected or contract.source_mode is None:
177 raise ValueError("Upload source must exist before secure staging")
178 if _file_identity(metadata) != (
179 contract.source_device,
180 contract.source_inode,
181 stat.S_IFMT(contract.source_mode),
182 ):
183 raise ValueError("Upload source changed after validation")
184 if metadata.st_dev != contract.device:
185 raise ValueError("Upload source crosses a filesystem boundary")
186 if not (stat.S_ISREG(metadata.st_mode) or stat.S_ISDIR(metadata.st_mode)):
187 raise ValueError("Upload source must be a regular file or directory")
188 if stat.S_ISREG(metadata.st_mode) and metadata.st_nlink != 1:
189 raise ValueError("Upload source must not be hard-linked")
192def _link_verified_regular(
193 source_dir_fd: int,
194 source_name: str,
195 source_stat: os.stat_result,
196 destination_dir_fd: int,
197 destination_name: str,
198) -> None:
199 """Hard-link one opened regular file and verify the name did not race."""
200 if not stat.S_ISREG(source_stat.st_mode) or source_stat.st_nlink != 1:
201 raise ValueError(f"Upload entry is not a private regular file: {source_name}")
202 os.link(
203 source_name,
204 destination_name,
205 src_dir_fd=source_dir_fd,
206 dst_dir_fd=destination_dir_fd,
207 follow_symlinks=False,
208 )
209 try:
210 source_after = os.stat(source_name, dir_fd=source_dir_fd, follow_symlinks=False)
211 destination = os.stat(
212 destination_name,
213 dir_fd=destination_dir_fd,
214 follow_symlinks=False,
215 )
216 expected = _file_identity(source_stat)
217 if (
218 _file_identity(source_after) != expected
219 or _file_identity(destination) != expected
220 or source_after.st_nlink != 2
221 or destination.st_nlink != 2
222 ):
223 raise ValueError(f"Upload entry changed while staging: {source_name}")
224 except Exception:
225 with suppress(OSError):
226 os.unlink(destination_name, dir_fd=destination_dir_fd)
227 raise
230def _stage_directory(
231 source_fd: int,
232 destination_fd: int,
233 *,
234 root_device: int,
235 visited: set[tuple[int, int]],
236 skip_internal_stages: bool,
237) -> None:
238 """Recursively snapshot regular files without following any link."""
239 directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
240 regular_flags = (
241 os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NONBLOCK", 0)
242 )
243 for name in sorted(os.listdir(source_fd)):
244 if skip_internal_stages and name.startswith(_UPLOAD_STAGE_PREFIX):
245 continue
246 before = os.stat(name, dir_fd=source_fd, follow_symlinks=False)
247 if before.st_dev != root_device:
248 raise ValueError(f"Upload entry crosses a filesystem boundary: {name}")
249 if stat.S_ISLNK(before.st_mode):
250 raise ValueError(f"Upload entry must not be a symbolic link: {name}")
251 if stat.S_ISREG(before.st_mode):
252 opened_fd = os.open(name, regular_flags, dir_fd=source_fd)
253 try:
254 opened = os.fstat(opened_fd)
255 if _file_identity(opened) != _file_identity(before):
256 raise ValueError(f"Upload entry changed while opening: {name}")
257 _link_verified_regular(source_fd, name, opened, destination_fd, name)
258 finally:
259 os.close(opened_fd)
260 continue
261 if stat.S_ISDIR(before.st_mode):
262 identity = (before.st_dev, before.st_ino)
263 if identity in visited:
264 raise ValueError(f"Upload directory cycle detected: {name}")
265 child_fd = os.open(name, directory_flags, dir_fd=source_fd)
266 try:
267 opened = os.fstat(child_fd)
268 if _file_identity(opened) != _file_identity(before):
269 raise ValueError(f"Upload directory changed while opening: {name}")
270 os.mkdir(name, mode=0o700, dir_fd=destination_fd)
271 child_destination_fd = os.open(name, directory_flags, dir_fd=destination_fd)
272 try:
273 visited.add(identity)
274 _stage_directory(
275 child_fd,
276 child_destination_fd,
277 root_device=root_device,
278 visited=visited,
279 skip_internal_stages=False,
280 )
281 finally:
282 visited.remove(identity)
283 os.close(child_destination_fd)
284 finally:
285 os.close(child_fd)
286 continue
287 raise ValueError(f"Upload entry must be a regular file or directory: {name}")
290def _create_stage_directory(root_fd: int) -> tuple[str, int]:
291 """Create one unpredictable private staging directory under the root."""
292 flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
293 for _ in range(10):
294 name = f"{_UPLOAD_STAGE_PREFIX}{secrets.token_hex(16)}"
295 try:
296 os.mkdir(name, mode=0o700, dir_fd=root_fd)
297 except FileExistsError:
298 continue
299 return name, os.open(name, flags, dir_fd=root_fd)
300 raise OSError("Unable to allocate a private upload staging directory")
303@contextmanager
304def stage_upload_path(contract: LocalPathContract) -> Iterator[StagedUpload]:
305 """Yield a private no-follow snapshot suitable for a short CLI upload.
307 The snapshot contains only directories and hard links to regular,
308 single-link files discovered through descriptor-relative no-follow opens.
309 The CLI receives ``/dev/fd/<dirfd>/<name>`` and that directory descriptor is
310 explicitly inherited by its subprocess, closing validation/use path races.
311 """
312 if not Path("/dev/fd").is_dir():
313 raise ValueError("Secure upload staging requires /dev/fd support")
315 root_fd = _verified_root_fd(contract)
316 stage_name: str | None = None
317 stage_fd: int | None = None
318 source_fd: int | None = None
319 source_parent_fd: int | None = None
320 try:
321 source_fd, source_parent_fd, source_name = _open_source(root_fd, contract)
322 source_stat = os.fstat(source_fd)
323 _verify_source_identity(contract, source_stat)
325 stage_name, stage_fd = _create_stage_directory(root_fd)
326 target_name = contract.resolved_path.name or "upload"
327 if stat.S_ISREG(source_stat.st_mode):
328 # ``_open_source`` returns a null parent/name pair only for the root
329 # directory itself, and the root is opened with ``O_DIRECTORY``, so a
330 # regular source always carries both values. Stating that invariant
331 # here narrows the types and fails loudly if it is ever broken,
332 # rather than carrying an unreachable user-error path.
333 assert source_parent_fd is not None and source_name is not None
334 _link_verified_regular(
335 source_parent_fd,
336 source_name,
337 source_stat,
338 stage_fd,
339 target_name,
340 )
341 else:
342 os.mkdir(target_name, mode=0o700, dir_fd=stage_fd)
343 flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
344 destination_fd = os.open(target_name, flags, dir_fd=stage_fd)
345 try:
346 source_identity = (source_stat.st_dev, source_stat.st_ino)
347 _stage_directory(
348 source_fd,
349 destination_fd,
350 root_device=contract.device,
351 visited={source_identity},
352 skip_internal_stages=source_identity == (contract.device, contract.inode),
353 )
354 finally:
355 os.close(destination_fd)
357 argument = f"/dev/fd/{stage_fd}/{target_name}"
358 staged_stat = os.stat(argument, follow_symlinks=False)
359 if not (stat.S_ISREG(staged_stat.st_mode) or stat.S_ISDIR(staged_stat.st_mode)):
360 raise ValueError("Secure upload staging produced an invalid source")
361 yield StagedUpload(argument=argument, directory_fd=stage_fd)
362 finally:
363 if source_fd is not None:
364 os.close(source_fd)
365 if source_parent_fd is not None:
366 os.close(source_parent_fd)
367 if stage_fd is not None:
368 os.close(stage_fd)
369 if stage_name is not None:
370 with suppress(OSError):
371 shutil.rmtree(stage_name, dir_fd=root_fd)
372 os.close(root_fd)