Coverage for cli/storage.py: 94.97%
700 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""Discover GCO S3 buckets and safely sync data with local storage.
3The physical names of GCO buckets contain deployment-generated account and
4region components. This module keeps those names out of the user interface by
5resolving stable aliases from the SSM and CloudFormation contracts published by
6the stacks.
7"""
9from __future__ import annotations
11import base64
12import hashlib
13import os
14import secrets
15import stat
16import sys
17import unicodedata
18from contextlib import suppress
19from dataclasses import dataclass
20from datetime import datetime
21from pathlib import Path
22from types import TracebackType
23from typing import Any, Self
25import boto3
26from botocore.exceptions import ClientError
28from .config import GCOConfig, get_config
30type _FileSignature = tuple[int, int, int, int, int]
33class StorageBucketNotFoundError(RuntimeError):
34 """Raised when a friendly alias has no deployed backing bucket."""
37@dataclass(frozen=True)
38class _ConfinementContract:
39 """Identity-bound MCP local-root contract propagated to the CLI."""
41 root: Path
42 device: int
43 inode: int
46class _PinnedRoot:
47 """Descriptor-pinned root for race-resistant confined filesystem access."""
49 def __init__(self, contract: _ConfinementContract):
50 if os.name != "posix" or not hasattr(os, "O_NOFOLLOW") or not hasattr(os, "O_DIRECTORY"):
51 raise RuntimeError(
52 "Confined storage sync requires descriptor-relative no-follow filesystem support"
53 )
54 if not contract.root.is_absolute():
55 raise ValueError("The internal storage confinement root must be absolute")
57 self.root = contract.root
58 flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
59 try:
60 self._fd = os.open(self.root, flags)
61 except OSError as exc:
62 raise ValueError(
63 f"Cannot open the configured storage confinement root securely: {self.root}"
64 ) from exc
66 root_stat = os.fstat(self._fd)
67 if not stat.S_ISDIR(root_stat.st_mode): 67 ↛ 68line 67 didn't jump to line 68 because the condition on line 67 was never true
68 os.close(self._fd)
69 self._fd = -1
70 raise NotADirectoryError(
71 f"The configured storage confinement root is not a directory: {self.root}"
72 )
73 if (root_stat.st_dev, root_stat.st_ino) != (contract.device, contract.inode):
74 os.close(self._fd)
75 self._fd = -1
76 raise RuntimeError(
77 "GCO_STORAGE_LOCAL_ROOT changed after the MCP request was validated; retry the call"
78 )
80 def __enter__(self) -> Self:
81 return self
83 def __exit__(
84 self,
85 exc_type: type[BaseException] | None,
86 exc_value: BaseException | None,
87 traceback: TracebackType | None,
88 ) -> None:
89 if self._fd >= 0: 89 ↛ exitline 89 didn't return from function '__exit__' because the condition on line 89 was always true
90 os.close(self._fd)
91 self._fd = -1
93 @staticmethod
94 def _directory_flags() -> int:
95 return os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
97 @staticmethod
98 def _file_flags() -> int:
99 # Avoid blocking if a raced source replacement is a FIFO; fstat below
100 # still rejects every opened object that is not a regular file.
101 return (
102 os.O_RDONLY
103 | os.O_NOFOLLOW
104 | getattr(os, "O_NONBLOCK", 0)
105 | getattr(os, "O_BINARY", 0)
106 | getattr(os, "O_CLOEXEC", 0)
107 )
109 def relative_parts(self, local_path: str) -> tuple[str, ...]:
110 """Return a lexical root-relative path; traversal itself stays descriptor-relative."""
111 supplied = Path(local_path).expanduser()
112 candidate = supplied if supplied.is_absolute() else self.root / supplied
113 lexical = Path(os.path.abspath(candidate))
114 try:
115 relative = lexical.relative_to(self.root)
116 except ValueError as exc:
117 raise ValueError(
118 f"Local sync path must stay within GCO_STORAGE_LOCAL_ROOT: {local_path}"
119 ) from exc
121 parts = tuple(relative.parts)
122 if any(part in ("", ".", "..") for part in parts): 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true
123 raise ValueError(
124 f"Local sync path must stay within GCO_STORAGE_LOCAL_ROOT: {local_path}"
125 )
126 return parts
128 def display_path(self, parts: tuple[str, ...]) -> Path:
129 """Return a human-readable path without using it for confined access."""
130 return self.root.joinpath(*parts)
132 def open_directory(
133 self,
134 parts: tuple[str, ...],
135 *,
136 create: bool = False,
137 allow_missing: bool = False,
138 ) -> int | None:
139 """Open a directory by walking from the pinned root without following links."""
140 current_fd = os.dup(self._fd)
141 try:
142 walked: tuple[str, ...] = ()
143 for part in parts:
144 walked += (part,)
145 try:
146 child_fd = os.open(part, self._directory_flags(), dir_fd=current_fd)
147 except FileNotFoundError:
148 if not create:
149 if allow_missing:
150 os.close(current_fd)
151 return None
152 raise FileNotFoundError(
153 f"Confined local directory does not exist: {self.display_path(walked)}"
154 ) from None
155 with suppress(FileExistsError):
156 os.mkdir(part, dir_fd=current_fd)
157 try:
158 child_fd = os.open(part, self._directory_flags(), dir_fd=current_fd)
159 except OSError as exc:
160 raise ValueError(
161 "Confined local path changed while creating a directory: "
162 f"{self.display_path(walked)}"
163 ) from exc
164 except OSError as exc:
165 raise ValueError(
166 "Confined local path contains a symbolic link or non-directory: "
167 f"{self.display_path(walked)}"
168 ) from exc
169 os.close(current_fd)
170 current_fd = child_fd
171 return current_fd
172 except BaseException:
173 os.close(current_fd)
174 raise
176 def inspect_directory(
177 self,
178 parts: tuple[str, ...],
179 *,
180 create: bool,
181 ) -> bool:
182 """Securely inspect or create a confined directory."""
183 directory_fd = self.open_directory(parts, create=create, allow_missing=not create)
184 if directory_fd is None:
185 return False
186 os.close(directory_fd)
187 return True
189 def lstat(self, parts: tuple[str, ...]) -> os.stat_result | None:
190 """Stat a confined path without following its final component."""
191 if not parts: 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true
192 return os.fstat(self._fd)
193 parent_fd = self.open_directory(parts[:-1], allow_missing=True)
194 if parent_fd is None: 194 ↛ 195line 194 didn't jump to line 195 because the condition on line 194 was never true
195 return None
196 try:
197 try:
198 return os.stat(parts[-1], dir_fd=parent_fd, follow_symlinks=False)
199 except FileNotFoundError:
200 return None
201 finally:
202 os.close(parent_fd)
204 def open_regular_file(self, parts: tuple[str, ...]) -> int:
205 """Open a confined regular file without following any path component."""
206 if not parts:
207 raise IsADirectoryError(f"Upload source is a directory: {self.root}")
208 parent_fd = self.open_directory(parts[:-1])
209 if parent_fd is None: # pragma: no cover - allow_missing is false
210 raise FileNotFoundError(self.display_path(parts))
211 try:
212 try:
213 file_fd = os.open(parts[-1], self._file_flags(), dir_fd=parent_fd)
214 except OSError as exc:
215 raise ValueError(
216 "Confined upload source is missing, linked, or not a regular file: "
217 f"{self.display_path(parts)}"
218 ) from exc
219 finally:
220 os.close(parent_fd)
222 file_stat = os.fstat(file_fd)
223 if not stat.S_ISREG(file_stat.st_mode):
224 os.close(file_fd)
225 raise ValueError(f"Upload source is not a regular file: {self.display_path(parts)}")
226 return file_fd
228 def open_child_directory(self, parent_fd: int, name: str, display: Path) -> int:
229 """Open an enumerated child directory without following a raced replacement."""
230 try:
231 child_fd = os.open(name, self._directory_flags(), dir_fd=parent_fd)
232 except OSError as exc:
233 raise ValueError(f"Upload source directory changed or is linked: {display}") from exc
234 if not stat.S_ISDIR(os.fstat(child_fd).st_mode):
235 os.close(child_fd)
236 raise ValueError(f"Upload source is not a directory: {display}")
237 return child_fd
239 def open_child_regular_file(self, parent_fd: int, name: str, display: Path) -> int:
240 """Open an enumerated child file without following a raced replacement."""
241 try:
242 file_fd = os.open(name, self._file_flags(), dir_fd=parent_fd)
243 except OSError as exc:
244 raise ValueError(f"Upload source file changed or is linked: {display}") from exc
245 if not stat.S_ISREG(os.fstat(file_fd).st_mode):
246 os.close(file_fd)
247 raise ValueError(f"Upload source is not a regular file: {display}")
248 return file_fd
250 def download_target_is_current(
251 self,
252 parts: tuple[str, ...],
253 size: int,
254 modified: datetime | None,
255 *,
256 evaluate_current: bool,
257 ) -> bool:
258 """Securely inspect a prospective destination and optionally test freshness."""
259 parent_fd = self.open_directory(parts[:-1], allow_missing=True)
260 if parent_fd is None:
261 return False
262 try:
263 try:
264 target_stat = os.stat(parts[-1], dir_fd=parent_fd, follow_symlinks=False)
265 except FileNotFoundError:
266 return False
267 finally:
268 os.close(parent_fd)
270 display = self.display_path(parts)
271 if stat.S_ISLNK(target_stat.st_mode):
272 raise ValueError(f"Download destination must not be a symbolic link: {display}")
273 if stat.S_ISDIR(target_stat.st_mode):
274 raise IsADirectoryError(f"S3 object maps to an existing directory: {display}")
275 if not stat.S_ISREG(target_stat.st_mode): 275 ↛ 277line 275 didn't jump to line 277 because the condition on line 275 was always true
276 raise ValueError(f"Download destination is not a regular file: {display}")
277 return bool(
278 evaluate_current
279 and modified is not None
280 and target_stat.st_size == size
281 and int(target_stat.st_mtime) >= int(modified.timestamp())
282 )
284 def download_object(self, s3: Any, bucket: str, obj: _SyncObject) -> None:
285 """Download to a secure sibling temporary file and atomically install it."""
286 if obj.destination_parts is None: # pragma: no cover - internal invariant
287 raise RuntimeError("Missing confined destination components")
288 parent_fd = self.open_directory(obj.destination_parts[:-1], create=True)
289 if parent_fd is None: # pragma: no cover - create is true
290 raise RuntimeError("Failed to create confined destination directory")
292 temporary_name = ""
293 temporary_created = False
294 try:
295 temporary_flags = (
296 os.O_RDWR
297 | os.O_CREAT
298 | os.O_EXCL
299 | os.O_NOFOLLOW
300 | getattr(os, "O_BINARY", 0)
301 | getattr(os, "O_CLOEXEC", 0)
302 )
303 for _ in range(128):
304 temporary_name = f".gco-sync-{os.getpid()}-{secrets.token_hex(12)}.tmp"
305 try:
306 temporary_fd = os.open(
307 temporary_name,
308 temporary_flags,
309 0o600,
310 dir_fd=parent_fd,
311 )
312 temporary_created = True
313 break
314 except FileExistsError:
315 continue
316 else: # pragma: no cover - cryptographically improbable
317 raise RuntimeError("Could not allocate a unique download temporary file")
319 with os.fdopen(temporary_fd, "w+b") as temporary_file:
320 s3.download_fileobj(bucket, obj.key, temporary_file)
321 temporary_file.flush()
322 downloaded_stat = os.fstat(temporary_file.fileno())
323 if downloaded_stat.st_size != obj.size:
324 raise RuntimeError(
325 f"Downloaded size changed for s3://{bucket}/{obj.key}: "
326 f"expected {obj.size}, received {downloaded_stat.st_size}"
327 )
328 if obj.last_modified is not None: 328 ↛ 332line 328 didn't jump to line 332
329 timestamp = obj.last_modified.timestamp()
330 os.utime(temporary_file.fileno(), (timestamp, timestamp))
332 os.replace(
333 temporary_name,
334 obj.destination_parts[-1],
335 src_dir_fd=parent_fd,
336 dst_dir_fd=parent_fd,
337 )
338 temporary_created = False
339 finally:
340 if temporary_created:
341 with suppress(FileNotFoundError):
342 os.unlink(temporary_name, dir_fd=parent_fd)
343 os.close(parent_fd)
346@dataclass(frozen=True)
347class _SyncObject:
348 """One validated object in a bucket-to-local sync plan."""
350 key: str
351 destination: Path
352 destination_parts: tuple[str, ...] | None
353 size: int
354 last_modified: datetime | None
355 current: bool
358@dataclass(frozen=True)
359class _PreparedUpload:
360 """One securely opened and hashed local upload source."""
362 source: Path
363 relative: str
364 source_parts: tuple[str, ...] | None
365 size: int
366 sha256: str
367 signature: _FileSignature
370@dataclass(frozen=True)
371class _UploadObject:
372 """One validated local file in a local-to-bucket sync plan."""
374 source: Path
375 source_parts: tuple[str, ...] | None
376 key: str
377 size: int
378 sha256: str
379 signature: _FileSignature
380 current: bool
383class StorageManager:
384 """Resolve friendly GCO bucket aliases and transfer their contents."""
386 _UPLOAD_DIGEST_METADATA = "gco-sync-sha256"
388 _PURPOSES = {
389 "cluster-shared": "Cross-region cluster job artifacts and shared data",
390 "model-weights": "Central model weights used by inference endpoints",
391 "regional-shared": "General-purpose data for workloads in one region",
392 "analytics-studio": "SageMaker Studio private scratch data and outputs",
393 }
395 def __init__(self, config: GCOConfig | None = None):
396 self.config = config or get_config()
398 def list_buckets(self, region: str | None = None) -> list[dict[str, str]]:
399 """Return deployed user-facing buckets under their stable aliases.
401 Global and analytics buckets are always considered. ``region`` limits
402 regional-bucket discovery to one region; otherwise every regional
403 deployment configured in ``cdk.json`` is considered. A bucket whose
404 stack is not deployed is omitted, while permission and transport
405 errors still surface to the caller.
406 """
407 buckets: list[dict[str, str]] = []
409 for alias in ("cluster-shared", "model-weights"):
410 with suppress(StorageBucketNotFoundError):
411 buckets.append(self.resolve_bucket(alias))
413 regional_regions = [region] if region else self._configured_regional_regions()
414 for regional_region in regional_regions:
415 with suppress(StorageBucketNotFoundError):
416 buckets.append(self.resolve_bucket("regional-shared", region=regional_region))
418 with suppress(StorageBucketNotFoundError):
419 buckets.append(self.resolve_bucket("analytics-studio"))
421 return buckets
423 def resolve_bucket(self, alias: str, region: str | None = None) -> dict[str, str]:
424 """Resolve a stable alias to a physical bucket and home region.
426 Supported aliases are ``cluster-shared``, ``model-weights``,
427 ``analytics-studio``, and either ``regional-shared:<region>`` or
428 ``regional-shared`` with ``region`` supplied. The unqualified regional
429 alias is inferred only when exactly one deployment region is configured.
430 """
431 normalized = alias.strip().lower()
432 embedded_region: str | None = None
434 if normalized.startswith("regional-shared:"):
435 normalized, embedded_region = normalized.split(":", 1)
436 embedded_region = embedded_region.strip()
437 if not embedded_region:
438 raise ValueError("Regional bucket alias must include a region after ':'")
439 if region and region != embedded_region:
440 raise ValueError(
441 f"Alias region '{embedded_region}' conflicts with --region '{region}'"
442 )
443 region = embedded_region
445 if normalized == "regional-shared":
446 target_region = region or self._infer_single_regional_region()
447 return self._resolve_regional_shared(target_region)
449 if region:
450 raise ValueError("--region is only valid with the 'regional-shared' alias")
452 if normalized == "cluster-shared":
453 return self._resolve_cluster_shared()
454 if normalized == "model-weights":
455 return self._resolve_model_weights()
456 if normalized == "analytics-studio":
457 return self._resolve_analytics_studio()
459 raise ValueError(
460 f"Unknown bucket alias '{alias}'. Use one of: cluster-shared, "
461 "model-weights, regional-shared:<region>, analytics-studio"
462 )
464 def sync(
465 self,
466 alias: str,
467 local_dir: str,
468 *,
469 region: str | None = None,
470 prefix: str = "",
471 direction: str = "download",
472 dry_run: bool = False,
473 force: bool = False,
474 confinement_root: str | None = None,
475 confinement_device: int | None = None,
476 confinement_inode: int | None = None,
477 ) -> dict[str, Any]:
478 """Incrementally transfer files in one explicit direction.
480 ``download`` preserves the original S3-to-local behavior. ``upload``
481 transfers a local file or directory into S3. Neither direction deletes
482 destination-only files or objects. The confinement values form an
483 internal MCP-to-CLI contract and are not public CLI options.
484 """
485 normalized_direction = direction.strip().lower()
486 if normalized_direction not in {"download", "upload"}:
487 raise ValueError("Sync direction must be either 'download' or 'upload'")
489 contract = self._make_confinement_contract(
490 confinement_root,
491 confinement_device,
492 confinement_inode,
493 )
494 bucket = self.resolve_bucket(alias, region=region)
495 normalized_prefix = self._normalize_prefix(prefix)
496 s3 = boto3.client("s3", region_name=bucket["region"])
498 if contract is not None:
499 with _PinnedRoot(contract) as confinement:
500 local_parts = confinement.relative_parts(local_dir)
501 local_path = confinement.display_path(local_parts)
502 if normalized_direction == "upload":
503 return self._sync_upload(
504 bucket,
505 s3,
506 local_path,
507 normalized_prefix,
508 dry_run=dry_run,
509 force=force,
510 confinement=confinement,
511 source_parts=local_parts,
512 )
513 return self._sync_download(
514 bucket,
515 s3,
516 local_path,
517 normalized_prefix,
518 dry_run=dry_run,
519 force=force,
520 confinement=confinement,
521 destination_parts=local_parts,
522 )
524 local_path = Path(local_dir).expanduser()
525 if normalized_direction == "upload":
526 return self._sync_upload(
527 bucket,
528 s3,
529 local_path,
530 normalized_prefix,
531 dry_run=dry_run,
532 force=force,
533 )
534 return self._sync_download(
535 bucket,
536 s3,
537 local_path,
538 normalized_prefix,
539 dry_run=dry_run,
540 force=force,
541 )
543 @staticmethod
544 def _make_confinement_contract(
545 root: str | None,
546 device: int | None,
547 inode: int | None,
548 ) -> _ConfinementContract | None:
549 supplied = (root is not None, device is not None, inode is not None)
550 if not any(supplied):
551 return None
552 if not all(supplied) or not root:
553 raise ValueError("The internal storage confinement contract is incomplete")
554 if device is None or inode is None: # pragma: no cover - narrowed by all()
555 raise ValueError("The internal storage confinement identity is incomplete")
556 if device < 0 or inode < 0:
557 raise ValueError("The internal storage confinement identity is invalid")
558 root_path = Path(root).expanduser()
559 if not root_path.is_absolute() or Path(os.path.abspath(root_path)) != root_path:
560 raise ValueError(
561 "The internal storage confinement root must be normalized and absolute"
562 )
563 return _ConfinementContract(root=root_path, device=device, inode=inode)
565 def _sync_download(
566 self,
567 bucket: dict[str, str],
568 s3: Any,
569 destination: Path,
570 prefix: str,
571 *,
572 dry_run: bool,
573 force: bool,
574 confinement: _PinnedRoot | None = None,
575 destination_parts: tuple[str, ...] = (),
576 ) -> dict[str, Any]:
577 """Download the selected bucket prefix into a local directory."""
578 if confinement is None:
579 if destination.exists() and not destination.is_dir():
580 raise NotADirectoryError(f"Sync destination is not a directory: {destination}")
581 if not dry_run:
582 destination.mkdir(parents=True, exist_ok=True)
583 destination = destination.resolve()
584 else:
585 confinement.inspect_directory(destination_parts, create=not dry_run)
587 objects, directory_markers = self._build_sync_plan(
588 s3,
589 bucket["bucket"],
590 prefix,
591 destination,
592 force=force,
593 confinement=confinement,
594 destination_parts=destination_parts,
595 )
596 pending = [obj for obj in objects if not obj.current]
597 current = [obj for obj in objects if obj.current]
598 skipped = len(current)
599 bytes_planned = sum(obj.size for obj in pending)
600 downloaded = 0
601 bytes_downloaded = 0
603 if not dry_run:
604 for obj in pending:
605 try:
606 if confinement is not None:
607 confinement.download_object(s3, bucket["bucket"], obj)
608 else:
609 obj.destination.parent.mkdir(parents=True, exist_ok=True)
610 s3.download_file(bucket["bucket"], obj.key, str(obj.destination))
611 if obj.last_modified is not None: 611 ↛ 620line 611 didn't jump to line 620 because the condition on line 611 was always true
612 timestamp = obj.last_modified.timestamp()
613 os.utime(obj.destination, (timestamp, timestamp))
614 except Exception as exc:
615 raise RuntimeError(
616 "Sync did not complete: failed to download "
617 f"'s3://{bucket['bucket']}/{obj.key}' to "
618 f"'{obj.destination}': {exc}"
619 ) from exc
620 downloaded += 1
621 bytes_downloaded += obj.size
623 for obj in current:
624 if confinement is not None:
625 if obj.destination_parts is None: # pragma: no cover - internal invariant
626 raise RuntimeError("Missing confined destination components")
627 still_current = confinement.download_target_is_current(
628 obj.destination_parts,
629 obj.size,
630 obj.last_modified,
631 evaluate_current=True,
632 )
633 else:
634 still_current = self._is_current(
635 obj.destination,
636 obj.size,
637 obj.last_modified,
638 )
639 if not still_current:
640 raise RuntimeError(
641 "Sync did not complete: a skipped local file changed after planning: "
642 f"{obj.destination}"
643 )
645 source = f"s3://{bucket['bucket']}/{prefix}"
646 return {
647 "alias": bucket["alias"],
648 "bucket": bucket["bucket"],
649 "region": bucket["region"],
650 "direction": "download",
651 "source": source,
652 "destination": str(destination),
653 "prefix": prefix,
654 "dry_run": dry_run,
655 "force": force,
656 "objects_scanned": len(objects) + directory_markers,
657 "directory_markers": directory_markers,
658 "files_planned": len(pending),
659 "files_downloaded": downloaded,
660 "files_skipped": skipped,
661 "bytes_planned": bytes_planned,
662 "bytes_downloaded": bytes_downloaded,
663 }
665 def _sync_upload(
666 self,
667 bucket: dict[str, str],
668 s3: Any,
669 source: Path,
670 prefix: str,
671 *,
672 dry_run: bool,
673 force: bool,
674 confinement: _PinnedRoot | None = None,
675 source_parts: tuple[str, ...] = (),
676 ) -> dict[str, Any]:
677 """Upload a local file or directory into the selected bucket prefix."""
678 if confinement is None:
679 if source.is_symlink():
680 raise ValueError(f"Upload source must not be a symbolic link: {source}")
681 if not source.exists():
682 raise FileNotFoundError(f"Upload source not found: {source}")
683 if not source.is_file() and not source.is_dir():
684 raise ValueError(f"Upload source must be a regular file or directory: {source}")
685 source = source.resolve()
687 objects, remote_objects_probed = self._build_upload_plan(
688 s3,
689 bucket["bucket"],
690 prefix,
691 source,
692 force=force,
693 confinement=confinement,
694 source_parts=source_parts,
695 )
696 pending = [obj for obj in objects if not obj.current]
697 current = [obj for obj in objects if obj.current]
698 skipped = len(current)
699 bytes_planned = sum(obj.size for obj in pending)
700 uploaded = 0
701 bytes_uploaded = 0
703 if not dry_run:
704 for obj in pending:
705 try:
706 source_fd = self._open_upload_source(obj, confinement)
707 try:
708 if self._stat_signature(os.fstat(source_fd)) != obj.signature: 708 ↛ 709line 708 didn't jump to line 709 because the condition on line 708 was never true
709 raise RuntimeError(f"Local file changed after planning: {obj.source}")
710 # Managed single-part uploads close their input stream. Keep the
711 # securely opened descriptor alive for the post-transfer signature
712 # check while allowing s3transfer to close its stream normally.
713 with open(source_fd, "rb", closefd=False) as source_file:
714 s3.upload_fileobj(
715 source_file,
716 bucket["bucket"],
717 obj.key,
718 ExtraArgs={
719 "Metadata": {self._UPLOAD_DIGEST_METADATA: obj.sha256},
720 "ChecksumSHA256": base64.b64encode(
721 bytes.fromhex(obj.sha256)
722 ).decode("ascii"),
723 },
724 )
725 if self._stat_signature(os.fstat(source_fd)) != obj.signature:
726 raise RuntimeError(f"Local file changed during upload: {obj.source}")
727 finally:
728 os.close(source_fd)
729 # A descriptor remains valid if its pathname is renamed away. Reopen
730 # after upload so a raced path replacement cannot be reported current.
731 self._verify_upload_source_signature(
732 obj,
733 confinement,
734 skipped=False,
735 )
736 except Exception as exc:
737 raise RuntimeError(
738 "Sync did not complete: failed to upload "
739 f"'{obj.source}' to 's3://{bucket['bucket']}/{obj.key}': {exc}"
740 ) from exc
741 uploaded += 1
742 bytes_uploaded += obj.size
744 for obj in current:
745 source_fd = self._open_upload_source(obj, confinement)
746 try:
747 if self._stat_signature(os.fstat(source_fd)) != obj.signature:
748 raise RuntimeError(
749 f"A skipped local file changed after planning: {obj.source}"
750 )
751 remote_objects_probed += 1
752 remote_current = self._remote_digest_matches(
753 s3,
754 bucket["bucket"],
755 obj.key,
756 obj.size,
757 obj.sha256,
758 )
759 if self._stat_signature(os.fstat(source_fd)) != obj.signature:
760 raise RuntimeError(
761 f"A skipped local file changed during revalidation: {obj.source}"
762 )
763 finally:
764 os.close(source_fd)
766 # Reopen after the remote probe so a renamed source path cannot
767 # be reported current merely because its old descriptor is stable.
768 self._verify_upload_source_signature(
769 obj,
770 confinement,
771 skipped=True,
772 )
773 if not remote_current:
774 raise RuntimeError(
775 "Sync did not complete: a skipped S3 object changed after planning: "
776 f"s3://{bucket['bucket']}/{obj.key}"
777 )
779 destination = f"s3://{bucket['bucket']}/{prefix}"
780 return {
781 "alias": bucket["alias"],
782 "bucket": bucket["bucket"],
783 "region": bucket["region"],
784 "direction": "upload",
785 "source": str(source),
786 "destination": destination,
787 "prefix": prefix,
788 "dry_run": dry_run,
789 "force": force,
790 "files_scanned": len(objects),
791 "objects_scanned": len(objects),
792 "objects_probed": remote_objects_probed,
793 "files_planned": len(pending),
794 "files_uploaded": uploaded,
795 "files_skipped": skipped,
796 "bytes_planned": bytes_planned,
797 "bytes_uploaded": bytes_uploaded,
798 }
800 def _resolve_cluster_shared(self) -> dict[str, str]:
801 from gco.services.aws_ssm import get_ssm_parameter_optional
802 from gco.stacks.constants import cluster_shared_ssm_parameter_prefix
804 prefix = cluster_shared_ssm_parameter_prefix(self.config.project_name)
805 name = get_ssm_parameter_optional(
806 f"{prefix}/name",
807 region=self.config.global_region,
808 )
809 if not name:
810 raise StorageBucketNotFoundError(
811 "Cluster shared bucket not found. Deploy the global stack first."
812 )
813 bucket_region = get_ssm_parameter_optional(
814 f"{prefix}/region",
815 region=self.config.global_region,
816 )
817 return self._bucket_record(
818 alias="cluster-shared",
819 name=name,
820 region=bucket_region or self.config.global_region,
821 scope="global",
822 )
824 def _resolve_model_weights(self) -> dict[str, str]:
825 from gco.services.aws_ssm import get_ssm_parameter_optional
827 name = get_ssm_parameter_optional(
828 f"/{self.config.project_name}/model-bucket-name",
829 region=self.config.global_region,
830 )
831 if not name:
832 raise StorageBucketNotFoundError(
833 "Model weights bucket not found. Deploy the global stack first."
834 )
835 return self._bucket_record(
836 alias="model-weights",
837 name=name,
838 region=self.config.global_region,
839 scope="global",
840 )
842 def _resolve_regional_shared(self, region: str) -> dict[str, str]:
843 from gco.services.aws_ssm import get_ssm_parameter_optional
844 from gco.stacks.constants import regional_shared_ssm_parameter_prefix
846 prefix = regional_shared_ssm_parameter_prefix(self.config.project_name)
847 name = get_ssm_parameter_optional(f"{prefix}/name", region=region)
848 if not name:
849 raise StorageBucketNotFoundError(
850 f"Regional shared bucket not found in region '{region}'. "
851 "Deploy that region's stack first."
852 )
853 bucket_region = get_ssm_parameter_optional(f"{prefix}/region", region=region)
854 return self._bucket_record(
855 alias=f"regional-shared:{region}",
856 name=name,
857 region=bucket_region or region,
858 scope="regional",
859 )
861 def _resolve_analytics_studio(self) -> dict[str, str]:
862 region = self.config.api_gateway_region
863 stack_name = f"{self.config.project_name}-analytics"
864 cfn = boto3.client("cloudformation", region_name=region)
865 token: str | None = None
867 try:
868 while True:
869 kwargs: dict[str, str] = {"StackName": stack_name}
870 if token:
871 kwargs["NextToken"] = token
872 response = cfn.list_stack_resources(**kwargs)
873 for resource in response.get("StackResourceSummaries", []):
874 logical_id = str(resource.get("LogicalResourceId", ""))
875 if resource.get("ResourceType") == "AWS::S3::Bucket" and logical_id.startswith(
876 "StudioOnlyBucket"
877 ):
878 name = resource.get("PhysicalResourceId")
879 if isinstance(name, str) and name:
880 return self._bucket_record(
881 alias="analytics-studio",
882 name=name,
883 region=region,
884 scope="analytics",
885 )
886 token_value = response.get("NextToken")
887 token = token_value if isinstance(token_value, str) else None
888 if not token:
889 break
890 except ClientError as exc:
891 error = exc.response.get("Error", {})
892 if error.get("Code") == "ValidationError" and "does not exist" in str(
893 error.get("Message", "")
894 ):
895 raise StorageBucketNotFoundError(
896 "Analytics Studio bucket not found. Deploy the analytics stack first."
897 ) from exc
898 raise
900 raise StorageBucketNotFoundError(
901 "Analytics Studio bucket not found in the deployed analytics stack."
902 )
904 def _configured_regional_regions(self) -> list[str]:
905 from .config import _load_cdk_json
907 configured = _load_cdk_json().get("regional", [])
908 candidates = configured if isinstance(configured, list) else []
909 regions: list[str] = []
910 seen: set[str] = set()
911 for value in candidates:
912 if isinstance(value, str) and value and value not in seen:
913 regions.append(value)
914 seen.add(value)
915 return regions or [self.config.default_region]
917 def _infer_single_regional_region(self) -> str:
918 regions = self._configured_regional_regions()
919 if len(regions) == 1:
920 return regions[0]
921 choices = ", ".join(f"regional-shared:{item}" for item in regions)
922 raise ValueError(
923 "The 'regional-shared' alias is ambiguous across configured regions. "
924 f"Use --region or one of: {choices}"
925 )
927 def _bucket_record(self, *, alias: str, name: str, region: str, scope: str) -> dict[str, str]:
928 purpose_key = "regional-shared" if alias.startswith("regional-shared:") else alias
929 return {
930 "alias": alias,
931 "bucket": name,
932 "region": region,
933 "scope": scope,
934 "purpose": self._PURPOSES[purpose_key],
935 "s3_uri": f"s3://{name}/",
936 }
938 @staticmethod
939 def _normalize_prefix(prefix: str) -> str:
940 normalized = prefix.lstrip("/")
941 if normalized and not normalized.endswith("/"):
942 normalized += "/"
943 return normalized
945 @staticmethod
946 def _stat_signature(value: os.stat_result) -> _FileSignature:
947 return (
948 value.st_dev,
949 value.st_ino,
950 value.st_size,
951 value.st_mtime_ns,
952 value.st_ctime_ns,
953 )
955 @classmethod
956 def _hash_upload_fd(cls, file_fd: int, path: Path) -> tuple[str, _FileSignature]:
957 with os.fdopen(file_fd, "rb") as source_file:
958 before = os.fstat(source_file.fileno())
959 if not stat.S_ISREG(before.st_mode):
960 raise ValueError(f"Upload source is not a regular file: {path}")
961 before_signature = cls._stat_signature(before)
962 digest = hashlib.sha256()
963 while chunk := source_file.read(8 * 1024 * 1024):
964 digest.update(chunk)
965 after_signature = cls._stat_signature(os.fstat(source_file.fileno()))
966 if before_signature != after_signature:
967 raise RuntimeError(f"Local file changed while planning upload: {path}")
968 return digest.hexdigest(), before_signature
970 @classmethod
971 def _hash_upload_file(cls, path: Path) -> tuple[str, _FileSignature]:
972 flags = (
973 os.O_RDONLY
974 | getattr(os, "O_BINARY", 0)
975 | getattr(os, "O_NOFOLLOW", 0)
976 | getattr(os, "O_CLOEXEC", 0)
977 )
978 return cls._hash_upload_fd(os.open(path, flags), path)
980 @staticmethod
981 def _validate_upload_relative_path(relative: str, source: Path) -> None:
982 parts = relative.split("/")
983 if (
984 not relative
985 or "\x00" in relative
986 or "\\" in relative
987 or any(part in ("", ".", "..") for part in parts)
988 ):
989 raise ValueError(f"Local path cannot be represented safely as an S3 key: {source}")
991 @classmethod
992 def _collect_upload_files(cls, source: Path) -> list[_PreparedUpload]:
993 paths: list[tuple[Path, str]] = []
994 if source.is_file():
995 relative = source.name
996 cls._validate_upload_relative_path(relative, source)
997 paths.append((source, relative))
998 else:
1000 def raise_walk_error(error: OSError) -> None:
1001 raise error
1003 for root, directories, names in os.walk(
1004 source,
1005 topdown=True,
1006 onerror=raise_walk_error,
1007 followlinks=False,
1008 ):
1009 root_path = Path(root)
1010 for directory_name in directories:
1011 directory = root_path / directory_name
1012 if directory.is_symlink(): 1012 ↛ 1013line 1012 didn't jump to line 1013 because the condition on line 1012 was never true
1013 raise ValueError(f"Upload source contains a symbolic link: {directory}")
1014 for name in names:
1015 path = root_path / name
1016 if path.is_symlink():
1017 raise ValueError(f"Upload source contains a symbolic link: {path}")
1018 if not path.is_file():
1019 raise ValueError(f"Upload source contains a non-regular file: {path}")
1020 relative = path.relative_to(source).as_posix()
1021 cls._validate_upload_relative_path(relative, path)
1022 paths.append((path, relative))
1024 prepared: list[_PreparedUpload] = []
1025 for path, relative in sorted(paths, key=lambda item: item[1]):
1026 digest, signature = cls._hash_upload_file(path)
1027 prepared.append(
1028 _PreparedUpload(
1029 source=path,
1030 relative=relative,
1031 source_parts=None,
1032 size=signature[2],
1033 sha256=digest,
1034 signature=signature,
1035 )
1036 )
1037 return prepared
1039 @classmethod
1040 def _collect_confined_upload_files(
1041 cls,
1042 confinement: _PinnedRoot,
1043 source_parts: tuple[str, ...],
1044 ) -> list[_PreparedUpload]:
1045 source = confinement.display_path(source_parts)
1046 source_stat = confinement.lstat(source_parts)
1047 if source_stat is None:
1048 raise FileNotFoundError(f"Upload source not found: {source}")
1049 if stat.S_ISLNK(source_stat.st_mode):
1050 raise ValueError(f"Upload source must not be a symbolic link: {source}")
1051 if stat.S_ISREG(source_stat.st_mode):
1052 relative = source.name
1053 cls._validate_upload_relative_path(relative, source)
1054 digest, signature = cls._hash_upload_fd(
1055 confinement.open_regular_file(source_parts),
1056 source,
1057 )
1058 return [
1059 _PreparedUpload(
1060 source=source,
1061 relative=relative,
1062 source_parts=source_parts,
1063 size=signature[2],
1064 sha256=digest,
1065 signature=signature,
1066 )
1067 ]
1068 if not stat.S_ISDIR(source_stat.st_mode):
1069 raise ValueError(f"Upload source must be a regular file or directory: {source}")
1071 source_fd = confinement.open_directory(source_parts)
1072 if source_fd is None: # pragma: no cover - allow_missing is false
1073 raise FileNotFoundError(source)
1074 prepared: list[_PreparedUpload] = []
1075 try:
1076 cls._walk_confined_upload_directory(
1077 confinement,
1078 source_fd,
1079 source_parts,
1080 (),
1081 prepared,
1082 )
1083 finally:
1084 os.close(source_fd)
1085 return sorted(prepared, key=lambda item: item.relative)
1087 @classmethod
1088 def _walk_confined_upload_directory(
1089 cls,
1090 confinement: _PinnedRoot,
1091 directory_fd: int,
1092 directory_parts: tuple[str, ...],
1093 relative_parts: tuple[str, ...],
1094 prepared: list[_PreparedUpload],
1095 ) -> None:
1096 """Enumerate and open one source directory through already-pinned descriptors."""
1097 for name in sorted(os.listdir(directory_fd)):
1098 child_parts = directory_parts + (name,)
1099 child_relative_parts = relative_parts + (name,)
1100 child = confinement.display_path(child_parts)
1101 relative = "/".join(child_relative_parts)
1102 cls._validate_upload_relative_path(relative, child)
1103 try:
1104 child_stat = os.stat(name, dir_fd=directory_fd, follow_symlinks=False)
1105 except FileNotFoundError as exc:
1106 raise RuntimeError(f"Upload source changed during enumeration: {child}") from exc
1108 if stat.S_ISLNK(child_stat.st_mode):
1109 raise ValueError(f"Upload source contains a symbolic link: {child}")
1110 if stat.S_ISDIR(child_stat.st_mode): 1110 ↛ 1111line 1110 didn't jump to line 1111 because the condition on line 1110 was never true
1111 child_fd = confinement.open_child_directory(directory_fd, name, child)
1112 try:
1113 cls._walk_confined_upload_directory(
1114 confinement,
1115 child_fd,
1116 child_parts,
1117 child_relative_parts,
1118 prepared,
1119 )
1120 finally:
1121 os.close(child_fd)
1122 continue
1123 if not stat.S_ISREG(child_stat.st_mode):
1124 raise ValueError(f"Upload source contains a non-regular file: {child}")
1126 digest, signature = cls._hash_upload_fd(
1127 confinement.open_child_regular_file(directory_fd, name, child),
1128 child,
1129 )
1130 prepared.append(
1131 _PreparedUpload(
1132 source=child,
1133 relative=relative,
1134 source_parts=child_parts,
1135 size=signature[2],
1136 sha256=digest,
1137 signature=signature,
1138 )
1139 )
1141 @staticmethod
1142 def _open_upload_source(obj: _UploadObject, confinement: _PinnedRoot | None) -> int:
1143 if confinement is not None:
1144 if obj.source_parts is None: # pragma: no cover - internal invariant
1145 raise RuntimeError("Missing confined upload source components")
1146 return confinement.open_regular_file(obj.source_parts)
1147 flags = (
1148 os.O_RDONLY
1149 | getattr(os, "O_BINARY", 0)
1150 | getattr(os, "O_NOFOLLOW", 0)
1151 | getattr(os, "O_CLOEXEC", 0)
1152 )
1153 file_fd = os.open(obj.source, flags)
1154 if not stat.S_ISREG(os.fstat(file_fd).st_mode): 1154 ↛ 1155line 1154 didn't jump to line 1155 because the condition on line 1154 was never true
1155 os.close(file_fd)
1156 raise ValueError(f"Upload source is not a regular file: {obj.source}")
1157 return file_fd
1159 @classmethod
1160 def _verify_upload_source_signature(
1161 cls,
1162 obj: _UploadObject,
1163 confinement: _PinnedRoot | None,
1164 *,
1165 skipped: bool,
1166 ) -> None:
1167 source_fd = cls._open_upload_source(obj, confinement)
1168 try:
1169 if cls._stat_signature(os.fstat(source_fd)) != obj.signature: 1169 ↛ 1170line 1169 didn't jump to line 1170 because the condition on line 1169 was never true
1170 description = "A skipped local file" if skipped else "Local file"
1171 raise RuntimeError(f"{description} changed after planning: {obj.source}")
1172 finally:
1173 os.close(source_fd)
1175 def _remote_digest_matches(
1176 self,
1177 s3: Any,
1178 bucket: str,
1179 key: str,
1180 size: int,
1181 digest: str,
1182 ) -> bool:
1183 try:
1184 response = s3.head_object(Bucket=bucket, Key=key)
1185 except ClientError as exc:
1186 error = exc.response.get("Error", {})
1187 status = exc.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
1188 if error.get("Code") in {
1189 "403",
1190 "404",
1191 "AccessDenied",
1192 "NoSuchKey",
1193 "NotFound",
1194 } or status in {403, 404}:
1195 # S3 returns 403 rather than 404 for a missing key when the
1196 # caller intentionally lacks ListBucket. Treat it as not
1197 # current and let PutObject enforce write authorization.
1198 return False
1199 raise
1201 if int(response.get("ContentLength", -1)) != size:
1202 return False
1203 metadata = response.get("Metadata", {})
1204 if not isinstance(metadata, dict):
1205 return False
1206 remote_digest = next(
1207 (
1208 str(value)
1209 for name, value in metadata.items()
1210 if str(name).lower() == self._UPLOAD_DIGEST_METADATA
1211 ),
1212 "",
1213 )
1214 return remote_digest.strip().lower() == digest
1216 def _build_upload_plan(
1217 self,
1218 s3: Any,
1219 bucket: str,
1220 prefix: str,
1221 source: Path,
1222 *,
1223 force: bool,
1224 confinement: _PinnedRoot | None,
1225 source_parts: tuple[str, ...],
1226 ) -> tuple[list[_UploadObject], int]:
1227 if confinement is None:
1228 files = self._collect_upload_files(source)
1229 else:
1230 files = self._collect_confined_upload_files(confinement, source_parts)
1232 objects: list[_UploadObject] = []
1233 remote_objects_probed = 0
1234 for prepared in files:
1235 key = f"{prefix}{prepared.relative}"
1236 current = False
1237 if not force:
1238 remote_objects_probed += 1
1239 current = self._remote_digest_matches(
1240 s3,
1241 bucket,
1242 key,
1243 prepared.size,
1244 prepared.sha256,
1245 )
1246 objects.append(
1247 _UploadObject(
1248 source=prepared.source,
1249 source_parts=prepared.source_parts,
1250 key=key,
1251 size=prepared.size,
1252 sha256=prepared.sha256,
1253 signature=prepared.signature,
1254 current=current,
1255 )
1256 )
1258 return objects, remote_objects_probed
1260 def _build_sync_plan(
1261 self,
1262 s3: Any,
1263 bucket: str,
1264 prefix: str,
1265 destination: Path,
1266 *,
1267 force: bool,
1268 confinement: _PinnedRoot | None,
1269 destination_parts: tuple[str, ...],
1270 ) -> tuple[list[_SyncObject], int]:
1271 objects: list[_SyncObject] = []
1272 directory_markers = 0
1273 paginator = s3.get_paginator("list_objects_v2")
1275 for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
1276 for item in page.get("Contents", []):
1277 key = str(item.get("Key", ""))
1278 if not key:
1279 raise ValueError("S3 returned an object with an empty key")
1280 size = int(item.get("Size", 0))
1281 if key.endswith("/"):
1282 if size != 0:
1283 raise ValueError(
1284 "S3 object keys ending in '/' cannot be represented as local files: "
1285 f"{key!r}"
1286 )
1287 directory_markers += 1
1288 continue
1290 relative_key = key[len(prefix) :] if prefix else key
1291 key_parts = self._download_relative_parts(relative_key, key)
1292 modified_value = item.get("LastModified")
1293 modified = modified_value if isinstance(modified_value, datetime) else None
1294 if confinement is None:
1295 local_path = self._safe_local_path(destination, relative_key, key)
1296 local_parts: tuple[str, ...] | None = None
1297 current = False if force else self._is_current(local_path, size, modified)
1298 else:
1299 local_parts = destination_parts + key_parts
1300 local_path = confinement.display_path(local_parts)
1301 current = confinement.download_target_is_current(
1302 local_parts,
1303 size,
1304 modified,
1305 evaluate_current=not force,
1306 )
1307 objects.append(
1308 _SyncObject(
1309 key=key,
1310 destination=local_path,
1311 destination_parts=local_parts,
1312 size=size,
1313 last_modified=modified,
1314 current=current,
1315 )
1316 )
1318 self._validate_sync_plan(objects)
1319 return objects, directory_markers
1321 @staticmethod
1322 def _validate_windows_download_part(part: str, source_key: str) -> None:
1323 if not sys.platform.startswith("win"):
1324 return
1325 if part.endswith((".", " ")):
1326 raise ValueError(f"Unsafe Windows S3 object key cannot be synced: {source_key!r}")
1327 if any(
1328 unicodedata.category(character) == "Cc" or character in '<>:"|?*' for character in part
1329 ):
1330 raise ValueError(f"Unsafe Windows S3 object key cannot be synced: {source_key!r}")
1332 stem = part.split(".", 1)[0].rstrip(" .").casefold()
1333 reserved = {"con", "prn", "aux", "nul", "clock$", "conin$", "conout$"}
1334 numbered_suffixes = "123456789¹²³"
1335 if stem in reserved or (
1336 len(stem) == 4 and stem[:3] in {"com", "lpt"} and stem[3] in numbered_suffixes
1337 ):
1338 raise ValueError(f"Reserved Windows path cannot be synced: {source_key!r}")
1340 @classmethod
1341 def _download_relative_parts(cls, relative_key: str, source_key: str) -> tuple[str, ...]:
1342 if "\x00" in relative_key or "\\" in relative_key:
1343 raise ValueError(f"Unsafe S3 object key cannot be synced: {source_key!r}")
1344 parts = tuple(relative_key.split("/"))
1345 if not relative_key or any(part in ("", ".", "..") for part in parts):
1346 raise ValueError(f"Unsafe S3 object key cannot be synced: {source_key!r}")
1347 for part in parts:
1348 cls._validate_windows_download_part(part, source_key)
1349 return parts
1351 @staticmethod
1352 def _local_collision_parts(path: Path) -> tuple[str, ...]:
1353 """Return path components normalized conservatively for the host filesystem."""
1354 if sys.platform.startswith("win"):
1355 return tuple(unicodedata.normalize("NFC", part).casefold() for part in path.parts)
1356 parts = tuple(os.path.normcase(part) for part in path.parts)
1357 if sys.platform == "darwin":
1358 # Default macOS volumes compare names case-insensitively and apply
1359 # Unicode normalization even though os.path.normcase is a no-op.
1360 return tuple(unicodedata.normalize("NFD", part).casefold() for part in parts)
1361 return parts
1363 @classmethod
1364 def _validate_sync_plan(cls, objects: list[_SyncObject]) -> None:
1365 """Reject object sets that cannot be represented without collisions."""
1366 seen: dict[tuple[str, ...], _SyncObject] = {}
1367 ordered = sorted(objects, key=lambda obj: len(obj.destination.parts))
1368 for obj in ordered:
1369 collision_key = cls._local_collision_parts(obj.destination)
1370 conflicting = seen.get(collision_key)
1371 if conflicting is not None:
1372 raise ValueError(
1373 "S3 object keys map to the same local path: "
1374 f"{conflicting.key!r} and {obj.key!r}"
1375 )
1376 for length in range(1, len(collision_key)):
1377 ancestor = seen.get(collision_key[:length])
1378 if ancestor is not None:
1379 raise ValueError(
1380 "S3 object keys have a local file/directory collision: "
1381 f"{ancestor.key!r} and {obj.key!r}"
1382 )
1383 seen[collision_key] = obj
1385 @classmethod
1386 def _safe_local_path(cls, destination: Path, relative_key: str, source_key: str) -> Path:
1387 parts = cls._download_relative_parts(relative_key, source_key)
1388 candidate = destination.joinpath(*parts).resolve()
1389 if candidate == destination or not candidate.is_relative_to(destination):
1390 raise ValueError(f"S3 object key escapes the sync destination: {source_key!r}")
1391 for parent in candidate.parents: 1391 ↛ 1398line 1391 didn't jump to line 1398 because the loop on line 1391 didn't complete
1392 if parent == destination:
1393 break
1394 if parent.exists() and not parent.is_dir():
1395 raise NotADirectoryError(
1396 f"S3 object '{source_key}' has a local parent that is not a directory: {parent}"
1397 )
1398 if candidate.exists() and candidate.is_dir():
1399 raise IsADirectoryError(
1400 f"S3 object '{source_key}' maps to an existing directory: {candidate}"
1401 )
1402 return candidate
1404 @staticmethod
1405 def _is_current(path: Path, size: int, modified: datetime | None) -> bool:
1406 if not path.is_file() or modified is None:
1407 return False
1408 path_stat = path.stat()
1409 return path_stat.st_size == size and int(path_stat.st_mtime) >= int(modified.timestamp())
1412def get_storage_manager(config: GCOConfig | None = None) -> StorageManager:
1413 """Return a storage manager using the merged CLI configuration."""
1414 return StorageManager(config)