Coverage for cli / storage.py: 100.00%
824 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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 (root_stat.st_dev, root_stat.st_ino) != (contract.device, contract.inode):
68 os.close(self._fd)
69 self._fd = -1
70 raise RuntimeError(
71 "GCO_STORAGE_LOCAL_ROOT changed after the MCP request was validated; retry the call"
72 )
74 def __enter__(self) -> Self:
75 return self
77 def __exit__(
78 self,
79 exc_type: type[BaseException] | None,
80 exc_value: BaseException | None,
81 traceback: TracebackType | None,
82 ) -> None:
83 if self._fd >= 0:
84 os.close(self._fd)
85 self._fd = -1
87 @staticmethod
88 def _directory_flags() -> int:
89 return os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
91 @staticmethod
92 def _file_flags() -> int:
93 # Avoid blocking if a raced source replacement is a FIFO; fstat below
94 # still rejects every opened object that is not a regular file.
95 return (
96 os.O_RDONLY
97 | os.O_NOFOLLOW
98 | getattr(os, "O_NONBLOCK", 0)
99 | getattr(os, "O_BINARY", 0)
100 | getattr(os, "O_CLOEXEC", 0)
101 )
103 def relative_parts(self, local_path: str) -> tuple[str, ...]:
104 """Return a lexical root-relative path; traversal itself stays descriptor-relative."""
105 supplied = Path(local_path).expanduser()
106 candidate = supplied if supplied.is_absolute() else self.root / supplied
107 lexical = Path(os.path.abspath(candidate))
108 try:
109 relative = lexical.relative_to(self.root)
110 except ValueError as exc:
111 raise ValueError(
112 f"Local sync path must stay within GCO_STORAGE_LOCAL_ROOT: {local_path}"
113 ) from exc
115 return tuple(relative.parts)
117 def display_path(self, parts: tuple[str, ...]) -> Path:
118 """Return a human-readable path without using it for confined access."""
119 return self.root.joinpath(*parts)
121 def open_directory(
122 self,
123 parts: tuple[str, ...],
124 *,
125 create: bool = False,
126 allow_missing: bool = False,
127 ) -> int | None:
128 """Open a directory by walking from the pinned root without following links."""
129 current_fd = os.dup(self._fd)
130 try:
131 walked: tuple[str, ...] = ()
132 for part in parts:
133 walked += (part,)
134 try:
135 child_fd = os.open(part, self._directory_flags(), dir_fd=current_fd)
136 except FileNotFoundError:
137 if not create:
138 if allow_missing:
139 os.close(current_fd)
140 return None
141 raise FileNotFoundError(
142 f"Confined local directory does not exist: {self.display_path(walked)}"
143 ) from None
144 with suppress(FileExistsError):
145 os.mkdir(part, dir_fd=current_fd)
146 try:
147 child_fd = os.open(part, self._directory_flags(), dir_fd=current_fd)
148 except OSError as exc:
149 raise ValueError(
150 "Confined local path changed while creating a directory: "
151 f"{self.display_path(walked)}"
152 ) from exc
153 except OSError as exc:
154 raise ValueError(
155 "Confined local path contains a symbolic link or non-directory: "
156 f"{self.display_path(walked)}"
157 ) from exc
158 os.close(current_fd)
159 current_fd = child_fd
160 return current_fd
161 except BaseException:
162 os.close(current_fd)
163 raise
165 def inspect_directory(
166 self,
167 parts: tuple[str, ...],
168 *,
169 create: bool,
170 ) -> bool:
171 """Securely inspect or create a confined directory."""
172 directory_fd = self.open_directory(parts, create=create, allow_missing=not create)
173 if directory_fd is None:
174 return False
175 os.close(directory_fd)
176 return True
178 def lstat(self, parts: tuple[str, ...]) -> os.stat_result | None:
179 """Stat a confined path without following its final component."""
180 if not parts:
181 return os.fstat(self._fd)
182 parent_fd = self.open_directory(parts[:-1], allow_missing=True)
183 if parent_fd is None:
184 return None
185 try:
186 try:
187 return os.stat(parts[-1], dir_fd=parent_fd, follow_symlinks=False)
188 except FileNotFoundError:
189 return None
190 finally:
191 os.close(parent_fd)
193 def open_regular_file(self, parts: tuple[str, ...]) -> int:
194 """Open a confined regular file without following any path component."""
195 if not parts:
196 raise IsADirectoryError(f"Upload source is a directory: {self.root}")
197 parent_fd = self.open_directory(parts[:-1])
198 if parent_fd is None: # pragma: no cover - allow_missing is false
199 raise FileNotFoundError(self.display_path(parts))
200 try:
201 try:
202 file_fd = os.open(parts[-1], self._file_flags(), dir_fd=parent_fd)
203 except OSError as exc:
204 raise ValueError(
205 "Confined upload source is missing, linked, or not a regular file: "
206 f"{self.display_path(parts)}"
207 ) from exc
208 finally:
209 os.close(parent_fd)
211 file_stat = os.fstat(file_fd)
212 if not stat.S_ISREG(file_stat.st_mode):
213 os.close(file_fd)
214 raise ValueError(f"Upload source is not a regular file: {self.display_path(parts)}")
215 return file_fd
217 def open_child_directory(self, parent_fd: int, name: str, display: Path) -> int:
218 """Open an enumerated child directory without following a raced replacement."""
219 try:
220 child_fd = os.open(name, self._directory_flags(), dir_fd=parent_fd)
221 except OSError as exc:
222 raise ValueError(f"Upload source directory changed or is linked: {display}") from exc
223 return child_fd
225 def open_child_regular_file(self, parent_fd: int, name: str, display: Path) -> int:
226 """Open an enumerated child file without following a raced replacement."""
227 try:
228 file_fd = os.open(name, self._file_flags(), dir_fd=parent_fd)
229 except OSError as exc:
230 raise ValueError(f"Upload source file changed or is linked: {display}") from exc
231 if not stat.S_ISREG(os.fstat(file_fd).st_mode):
232 os.close(file_fd)
233 raise ValueError(f"Upload source is not a regular file: {display}")
234 return file_fd
236 def download_target_is_current(
237 self,
238 parts: tuple[str, ...],
239 size: int,
240 modified: datetime | None,
241 *,
242 evaluate_current: bool,
243 ) -> bool:
244 """Securely inspect a prospective destination and optionally test freshness."""
245 parent_fd = self.open_directory(parts[:-1], allow_missing=True)
246 if parent_fd is None:
247 return False
248 try:
249 try:
250 target_stat = os.stat(parts[-1], dir_fd=parent_fd, follow_symlinks=False)
251 except FileNotFoundError:
252 return False
253 finally:
254 os.close(parent_fd)
256 display = self.display_path(parts)
257 if stat.S_ISLNK(target_stat.st_mode):
258 raise ValueError(f"Download destination must not be a symbolic link: {display}")
259 if stat.S_ISDIR(target_stat.st_mode):
260 raise IsADirectoryError(f"S3 object maps to an existing directory: {display}")
261 if not stat.S_ISREG(target_stat.st_mode):
262 raise ValueError(f"Download destination is not a regular file: {display}")
263 return bool(
264 evaluate_current
265 and modified is not None
266 and target_stat.st_size == size
267 and int(target_stat.st_mtime) >= int(modified.timestamp())
268 )
270 def download_object(self, s3: Any, bucket: str, obj: _SyncObject) -> None:
271 """Download to a secure sibling temporary file and atomically install it."""
272 if obj.destination_parts is None: # pragma: no cover - internal invariant
273 raise RuntimeError("Missing confined destination components")
274 parent_fd = self.open_directory(obj.destination_parts[:-1], create=True)
275 if parent_fd is None: # pragma: no cover - create is true
276 raise RuntimeError("Failed to create confined destination directory")
278 temporary_name = ""
279 temporary_created = False
280 try:
281 temporary_flags = (
282 os.O_RDWR
283 | os.O_CREAT
284 | os.O_EXCL
285 | os.O_NOFOLLOW
286 | getattr(os, "O_BINARY", 0)
287 | getattr(os, "O_CLOEXEC", 0)
288 )
289 for _ in range(128):
290 temporary_name = f".gco-sync-{os.getpid()}-{secrets.token_hex(12)}.tmp"
291 try:
292 temporary_fd = os.open(
293 temporary_name,
294 temporary_flags,
295 0o600,
296 dir_fd=parent_fd,
297 )
298 temporary_created = True
299 break
300 except FileExistsError:
301 continue
302 else: # pragma: no cover - cryptographically improbable
303 raise RuntimeError("Could not allocate a unique download temporary file")
305 with os.fdopen(temporary_fd, "w+b") as temporary_file:
306 s3.download_fileobj(bucket, obj.key, temporary_file)
307 temporary_file.flush()
308 downloaded_stat = os.fstat(temporary_file.fileno())
309 if downloaded_stat.st_size != obj.size:
310 raise RuntimeError(
311 f"Downloaded size changed for s3://{bucket}/{obj.key}: "
312 f"expected {obj.size}, received {downloaded_stat.st_size}"
313 )
314 if obj.last_modified is not None:
315 timestamp = obj.last_modified.timestamp()
316 os.utime(temporary_file.fileno(), (timestamp, timestamp))
318 os.replace(
319 temporary_name,
320 obj.destination_parts[-1],
321 src_dir_fd=parent_fd,
322 dst_dir_fd=parent_fd,
323 )
324 temporary_created = False
325 finally:
326 if temporary_created:
327 with suppress(FileNotFoundError):
328 os.unlink(temporary_name, dir_fd=parent_fd)
329 os.close(parent_fd)
332@dataclass(frozen=True)
333class _SyncObject:
334 """One validated object in a bucket-to-local sync plan."""
336 key: str
337 destination: Path
338 destination_parts: tuple[str, ...] | None
339 size: int
340 last_modified: datetime | None
341 current: bool
344@dataclass(frozen=True)
345class _PreparedUpload:
346 """One securely opened and hashed local upload source."""
348 source: Path
349 relative: str
350 source_parts: tuple[str, ...] | None
351 size: int
352 sha256: str
353 signature: _FileSignature
356@dataclass(frozen=True)
357class _UploadObject:
358 """One validated local file in a local-to-bucket sync plan."""
360 source: Path
361 source_parts: tuple[str, ...] | None
362 key: str
363 size: int
364 sha256: str
365 signature: _FileSignature
366 current: bool
369class StorageManager:
370 """Resolve friendly GCO bucket aliases and transfer their contents."""
372 _UPLOAD_DIGEST_METADATA = "gco-sync-sha256"
374 _PURPOSES = {
375 "cluster-shared": "Cross-region cluster job artifacts and shared data",
376 "model-weights": "Central model weights used by inference endpoints",
377 "regional-shared": "General-purpose data for workloads in one region",
378 "analytics-studio": "SageMaker Studio private scratch data and outputs",
379 }
381 def __init__(self, config: GCOConfig | None = None):
382 self.config = config or get_config()
383 # One CloudFormation sweep per (stack, region) serves every access-logs
384 # entry in that stack, so a full inventory stays at one call per stack.
385 self._stack_resource_cache: dict[tuple[str, str], dict[str, str]] = {}
387 def list_buckets(self, region: str | None = None) -> list[dict[str, str]]:
388 """Return deployed user-facing buckets under their stable aliases.
390 Global and analytics buckets are always considered. ``region`` limits
391 regional-bucket discovery to one region; otherwise every regional
392 deployment configured in ``cdk.json`` is considered. A bucket whose
393 stack is not deployed is omitted, while permission and transport
394 errors still surface to the caller.
395 """
396 buckets: list[dict[str, str]] = []
398 for alias in ("cluster-shared", "model-weights"):
399 with suppress(StorageBucketNotFoundError):
400 buckets.append(self.resolve_bucket(alias))
402 regional_regions = [region] if region else self._configured_regional_regions()
403 for regional_region in regional_regions:
404 with suppress(StorageBucketNotFoundError):
405 buckets.append(self.resolve_bucket("regional-shared", region=regional_region))
407 with suppress(StorageBucketNotFoundError):
408 buckets.append(self.resolve_bucket("analytics-studio"))
410 return buckets
412 def s3_inventory(self, region: str | None = None) -> dict[str, Any]:
413 """Describe every S3 bucket this deployment creates, deployed or not.
415 Complements :meth:`list_buckets`, which deliberately reports only the
416 four user-facing buckets addressable by ``storage sync``. This reports
417 the full set — including the server-access-log sinks and the cost-report
418 bucket — with the deployment-contract facts a caller actually needs:
419 which stack owns each bucket, what it is for, whether job pods can reach
420 it and how they discover it, what happens to it on teardown, and which
421 object-key prefixes the platform has already reserved.
423 A bucket whose stack is not deployed is reported with
424 ``status="not-deployed"`` rather than omitted, so the answer to "what
425 buckets does this deployment have?" is complete even before a region is
426 rolled out. Static facts come from :data:`BUCKET_DESCRIPTORS` and need no
427 AWS call; only the physical name, ARN, and status are resolved live.
429 ``region`` limits the regional entries to one region. Global,
430 monitoring, and analytics entries are always included because they exist
431 once per deployment, not once per region.
433 Note this inventories *buckets and their deployment contract*. It is
434 unrelated to the AWS "S3 Inventory" feature, which produces scheduled
435 reports of the objects inside a single bucket.
436 """
437 project = self.config.project_name
438 regional_regions = [region] if region else self._configured_regional_regions()
439 account = self._account_id()
441 # One CloudFormation sweep per stack, shared by every access-logs entry
442 # in that stack — those buckets are CDK-auto-named, so their physical
443 # names exist only as stack resources.
444 log_buckets: dict[tuple[str, str], str | None] = {}
445 stacks_to_sweep: list[tuple[str, str, str]] = [
446 ("global", self.config.global_stack_name, self.config.global_region),
447 ("monitoring", f"{project}-monitoring", self.config.monitoring_region),
448 ("analytics", f"{project}-analytics", self.config.api_gateway_region),
449 ]
450 stacks_to_sweep.extend(
451 (f"regional:{item}", f"{self.config.regional_stack_prefix}-{item}", item)
452 for item in regional_regions
453 )
454 for scope_key, stack_name, stack_region in stacks_to_sweep:
455 for logical_id, physical in self._stack_bucket_resources(
456 stack_name, stack_region
457 ).items():
458 log_buckets[(scope_key, logical_id)] = physical
460 records: list[dict[str, Any]] = []
461 for descriptor in BUCKET_DESCRIPTORS:
462 if descriptor.scope == "regional":
463 for regional_region in regional_regions:
464 records.append(
465 self._s3_inventory_record(
466 descriptor,
467 region=regional_region,
468 stack_name=f"{self.config.regional_stack_prefix}-{regional_region}",
469 account=account,
470 log_buckets=log_buckets,
471 scope_key=f"regional:{regional_region}",
472 )
473 )
474 continue
476 scope_region, stack_name = {
477 "global": (self.config.global_region, self.config.global_stack_name),
478 "monitoring": (self.config.monitoring_region, f"{project}-monitoring"),
479 "analytics": (self.config.api_gateway_region, f"{project}-analytics"),
480 }[descriptor.scope]
481 records.append(
482 self._s3_inventory_record(
483 descriptor,
484 region=scope_region,
485 stack_name=stack_name,
486 account=account,
487 log_buckets=log_buckets,
488 scope_key=descriptor.scope,
489 )
490 )
492 deployed = sum(1 for item in records if item["status"] == "deployed")
493 return {
494 "project_name": project,
495 "account": account,
496 "regions": {
497 "global": self.config.global_region,
498 "monitoring": self.config.monitoring_region,
499 "analytics": self.config.api_gateway_region,
500 "regional": regional_regions,
501 },
502 "buckets": records,
503 "summary": {
504 "total": len(records),
505 "deployed": deployed,
506 "not_deployed": len(records) - deployed,
507 "pod_writable": sorted(
508 item["bucket"]
509 for item in records
510 if item["pod_access"] == "read-write" and item["bucket"]
511 ),
512 },
513 }
515 def _s3_inventory_record(
516 self,
517 descriptor: BucketDescriptor,
518 *,
519 region: str,
520 stack_name: str,
521 account: str | None,
522 log_buckets: dict[tuple[str, str], str | None],
523 scope_key: str,
524 ) -> dict[str, Any]:
525 """Merge one descriptor's static facts with its live name and status."""
526 name: str | None = None
527 arn: str | None = None
528 detail = ""
530 try:
531 if descriptor.role == "access-logs":
532 name = log_buckets.get((scope_key, descriptor.logical_id_prefix))
533 else:
534 name, arn = self._resolve_primary_bucket(descriptor, region, account)
535 except Exception as exc: # noqa: BLE001 - an unresolvable entry is reported, not fatal
536 detail = f"could not resolve: {exc}"
538 if name and not arn:
539 arn = f"arn:{self._partition_for(region)}:s3:::{name}"
540 if not name and not detail:
541 detail = (
542 f"{stack_name} is not deployed"
543 if descriptor.opt_in is None
544 else f"{stack_name} is not deployed (opt-in: {descriptor.opt_in})"
545 )
547 return {
548 "id": descriptor.id if descriptor.scope != "regional" else f"{descriptor.id}:{region}",
549 "role": descriptor.role,
550 "scope": descriptor.scope,
551 "region": region,
552 "owning_stack": stack_name,
553 "bucket": name,
554 "arn": arn,
555 "s3_uri": f"s3://{name}/" if name else None,
556 "status": "deployed" if name else "not-deployed",
557 "detail": detail,
558 "purpose": descriptor.purpose,
559 "pod_access": descriptor.pod_access,
560 "discovery": descriptor.discovery,
561 "removal_policy": _effective_removal_policy(descriptor),
562 "reserved_prefixes": list(descriptor.reserved_prefixes),
563 "sync_alias": (
564 f"{descriptor.sync_alias}:{region}"
565 if descriptor.sync_alias and descriptor.scope == "regional"
566 else descriptor.sync_alias
567 ),
568 "opt_in": descriptor.opt_in,
569 }
571 def _resolve_primary_bucket(
572 self, descriptor: BucketDescriptor, region: str, account: str | None
573 ) -> tuple[str | None, str | None]:
574 """Resolve a primary bucket's physical name and ARN, or ``(None, None)``.
576 Each family publishes its identity differently, so this routes to the
577 contract that family actually uses rather than reconstructing names
578 (every primary bucket is CloudFormation-named): the two shared buckets
579 and the cost bucket publish name+ARN to SSM in their home region, the
580 model bucket publishes its name, and the Studio bucket is only
581 knowable from its stack's resources.
582 """
583 from gco.services.aws_ssm import get_ssm_parameter_optional
584 from gco.stacks.constants import (
585 cluster_shared_ssm_parameter_prefix,
586 cost_report_ssm_parameter_prefix,
587 regional_shared_ssm_parameter_prefix,
588 )
590 project = self.config.project_name
592 if descriptor.id == "cluster-shared":
593 prefix = cluster_shared_ssm_parameter_prefix(project)
594 return (
595 get_ssm_parameter_optional(f"{prefix}/name", region=region),
596 get_ssm_parameter_optional(f"{prefix}/arn", region=region),
597 )
598 if descriptor.id == "regional-shared":
599 prefix = regional_shared_ssm_parameter_prefix(project)
600 return (
601 get_ssm_parameter_optional(f"{prefix}/name", region=region),
602 get_ssm_parameter_optional(f"{prefix}/arn", region=region),
603 )
604 if descriptor.id == "model-weights":
605 return get_ssm_parameter_optional(f"/{project}/model-bucket-name", region=region), None
606 if descriptor.id == "cost-reports":
607 prefix = cost_report_ssm_parameter_prefix(project)
608 return (
609 get_ssm_parameter_optional(f"{prefix}/name", region=region),
610 get_ssm_parameter_optional(f"{prefix}/arn", region=region),
611 )
612 if descriptor.id == "analytics-studio":
613 resources = self._stack_bucket_resources(f"{project}-analytics", region)
614 return resources.get(descriptor.logical_id_prefix), None
615 return None, None
617 def _stack_bucket_resources(self, stack_name: str, region: str) -> dict[str, str]:
618 """Map ``logical-id-prefix -> physical bucket name`` for one stack.
620 CDK appends a hash to logical IDs, so entries are keyed by the stable
621 construct-id prefix the descriptors declare. Returns ``{}`` when the
622 stack is absent — an undeployed stack is an expected state here, not an
623 error — while permission and transport failures propagate so they are
624 never silently reported as "not deployed".
625 """
626 cache_key = (stack_name, region)
627 if cache_key in self._stack_resource_cache:
628 return self._stack_resource_cache[cache_key]
630 prefixes = {item.logical_id_prefix for item in BUCKET_DESCRIPTORS}
631 found: dict[str, str] = {}
632 cfn = boto3.client("cloudformation", region_name=region)
633 token: str | None = None
634 try:
635 while True:
636 kwargs: dict[str, str] = {"StackName": stack_name}
637 if token:
638 kwargs["NextToken"] = token
639 response = cfn.list_stack_resources(**kwargs)
640 for resource in response.get("StackResourceSummaries", []):
641 if resource.get("ResourceType") != "AWS::S3::Bucket":
642 continue
643 logical_id = str(resource.get("LogicalResourceId", ""))
644 physical = resource.get("PhysicalResourceId")
645 if not isinstance(physical, str) or not physical:
646 continue
647 for prefix in prefixes:
648 if logical_id.startswith(prefix):
649 found[prefix] = physical
650 break
651 token_value = response.get("NextToken")
652 token = token_value if isinstance(token_value, str) else None
653 if not token:
654 break
655 except ClientError as exc:
656 error = exc.response.get("Error", {})
657 if error.get("Code") == "ValidationError" and "does not exist" in str(
658 error.get("Message", "")
659 ):
660 found = {}
661 else:
662 raise
664 self._stack_resource_cache[cache_key] = found
665 return found
667 def _account_id(self) -> str | None:
668 """The caller's account id, or ``None`` when it cannot be determined.
670 ``sts:GetCallerIdentity`` needs no IAM permission, so this normally
671 succeeds wherever credentials exist at all.
672 """
673 try:
674 identity = boto3.client("sts").get_caller_identity()
675 value = identity.get("Account")
676 return str(value) if value else None
677 except Exception: # noqa: BLE001 - the inventory degrades without it
678 return None
680 def _partition_for(self, region: str) -> str:
681 """ARN partition for a region (aws, aws-cn, aws-us-gov)."""
682 try:
683 return str(boto3.Session().get_partition_for_region(region))
684 except Exception: # noqa: BLE001 - commercial is the right default
685 return "aws"
687 def resolve_bucket(self, alias: str, region: str | None = None) -> dict[str, str]:
688 """Resolve a stable alias to a physical bucket and home region.
690 Supported aliases are ``cluster-shared``, ``model-weights``,
691 ``analytics-studio``, and either ``regional-shared:<region>`` or
692 ``regional-shared`` with ``region`` supplied. The unqualified regional
693 alias is inferred only when exactly one deployment region is configured.
694 """
695 normalized = alias.strip().lower()
696 embedded_region: str | None = None
698 if normalized.startswith("regional-shared:"):
699 normalized, embedded_region = normalized.split(":", 1)
700 embedded_region = embedded_region.strip()
701 if not embedded_region:
702 raise ValueError("Regional bucket alias must include a region after ':'")
703 if region and region != embedded_region:
704 raise ValueError(
705 f"Alias region '{embedded_region}' conflicts with --region '{region}'"
706 )
707 region = embedded_region
709 if normalized == "regional-shared":
710 target_region = region or self._infer_single_regional_region()
711 return self._resolve_regional_shared(target_region)
713 if region:
714 raise ValueError("--region is only valid with the 'regional-shared' alias")
716 if normalized == "cluster-shared":
717 return self._resolve_cluster_shared()
718 if normalized == "model-weights":
719 return self._resolve_model_weights()
720 if normalized == "analytics-studio":
721 return self._resolve_analytics_studio()
723 raise ValueError(
724 f"Unknown bucket alias '{alias}'. Use one of: cluster-shared, "
725 "model-weights, regional-shared:<region>, analytics-studio"
726 )
728 def sync(
729 self,
730 alias: str,
731 local_dir: str,
732 *,
733 region: str | None = None,
734 prefix: str = "",
735 direction: str = "download",
736 dry_run: bool = False,
737 force: bool = False,
738 confinement_root: str | None = None,
739 confinement_device: int | None = None,
740 confinement_inode: int | None = None,
741 ) -> dict[str, Any]:
742 """Incrementally transfer files in one explicit direction.
744 ``download`` preserves the original S3-to-local behavior. ``upload``
745 transfers a local file or directory into S3. Neither direction deletes
746 destination-only files or objects. The confinement values form an
747 internal MCP-to-CLI contract and are not public CLI options.
748 """
749 normalized_direction = direction.strip().lower()
750 if normalized_direction not in {"download", "upload"}:
751 raise ValueError("Sync direction must be either 'download' or 'upload'")
753 contract = self._make_confinement_contract(
754 confinement_root,
755 confinement_device,
756 confinement_inode,
757 )
758 bucket = self.resolve_bucket(alias, region=region)
759 normalized_prefix = self._normalize_prefix(prefix)
760 s3 = boto3.client("s3", region_name=bucket["region"])
762 if contract is not None:
763 with _PinnedRoot(contract) as confinement:
764 local_parts = confinement.relative_parts(local_dir)
765 local_path = confinement.display_path(local_parts)
766 if normalized_direction == "upload":
767 return self._sync_upload(
768 bucket,
769 s3,
770 local_path,
771 normalized_prefix,
772 dry_run=dry_run,
773 force=force,
774 confinement=confinement,
775 source_parts=local_parts,
776 )
777 return self._sync_download(
778 bucket,
779 s3,
780 local_path,
781 normalized_prefix,
782 dry_run=dry_run,
783 force=force,
784 confinement=confinement,
785 destination_parts=local_parts,
786 )
788 local_path = Path(local_dir).expanduser()
789 if normalized_direction == "upload":
790 return self._sync_upload(
791 bucket,
792 s3,
793 local_path,
794 normalized_prefix,
795 dry_run=dry_run,
796 force=force,
797 )
798 return self._sync_download(
799 bucket,
800 s3,
801 local_path,
802 normalized_prefix,
803 dry_run=dry_run,
804 force=force,
805 )
807 @staticmethod
808 def _make_confinement_contract(
809 root: str | None,
810 device: int | None,
811 inode: int | None,
812 ) -> _ConfinementContract | None:
813 if root is None and device is None and inode is None:
814 return None
815 if not root or device is None or inode is None:
816 raise ValueError("The internal storage confinement contract is incomplete")
817 if device < 0 or inode < 0:
818 raise ValueError("The internal storage confinement identity is invalid")
819 root_path = Path(root).expanduser()
820 if not root_path.is_absolute() or Path(os.path.abspath(root_path)) != root_path:
821 raise ValueError(
822 "The internal storage confinement root must be normalized and absolute"
823 )
824 return _ConfinementContract(root=root_path, device=device, inode=inode)
826 def _sync_download(
827 self,
828 bucket: dict[str, str],
829 s3: Any,
830 destination: Path,
831 prefix: str,
832 *,
833 dry_run: bool,
834 force: bool,
835 confinement: _PinnedRoot | None = None,
836 destination_parts: tuple[str, ...] = (),
837 ) -> dict[str, Any]:
838 """Download the selected bucket prefix into a local directory."""
839 if confinement is None:
840 if destination.exists() and not destination.is_dir():
841 raise NotADirectoryError(f"Sync destination is not a directory: {destination}")
842 if not dry_run:
843 destination.mkdir(parents=True, exist_ok=True)
844 destination = destination.resolve()
845 else:
846 confinement.inspect_directory(destination_parts, create=not dry_run)
848 objects, directory_markers = self._build_sync_plan(
849 s3,
850 bucket["bucket"],
851 prefix,
852 destination,
853 force=force,
854 confinement=confinement,
855 destination_parts=destination_parts,
856 )
857 pending = [obj for obj in objects if not obj.current]
858 current = [obj for obj in objects if obj.current]
859 skipped = len(current)
860 bytes_planned = sum(obj.size for obj in pending)
861 downloaded = 0
862 bytes_downloaded = 0
864 if not dry_run:
865 for obj in pending:
866 try:
867 if confinement is not None:
868 confinement.download_object(s3, bucket["bucket"], obj)
869 else:
870 obj.destination.parent.mkdir(parents=True, exist_ok=True)
871 s3.download_file(bucket["bucket"], obj.key, str(obj.destination))
872 if obj.last_modified is not None:
873 timestamp = obj.last_modified.timestamp()
874 os.utime(obj.destination, (timestamp, timestamp))
875 except Exception as exc:
876 raise RuntimeError(
877 "Sync did not complete: failed to download "
878 f"'s3://{bucket['bucket']}/{obj.key}' to "
879 f"'{obj.destination}': {exc}"
880 ) from exc
881 downloaded += 1
882 bytes_downloaded += obj.size
884 for obj in current:
885 if confinement is not None:
886 if obj.destination_parts is None: # pragma: no cover - internal invariant
887 raise RuntimeError("Missing confined destination components")
888 still_current = confinement.download_target_is_current(
889 obj.destination_parts,
890 obj.size,
891 obj.last_modified,
892 evaluate_current=True,
893 )
894 else:
895 still_current = self._is_current(
896 obj.destination,
897 obj.size,
898 obj.last_modified,
899 )
900 if not still_current:
901 raise RuntimeError(
902 "Sync did not complete: a skipped local file changed after planning: "
903 f"{obj.destination}"
904 )
906 source = f"s3://{bucket['bucket']}/{prefix}"
907 return {
908 "alias": bucket["alias"],
909 "bucket": bucket["bucket"],
910 "region": bucket["region"],
911 "direction": "download",
912 "source": source,
913 "destination": str(destination),
914 "prefix": prefix,
915 "dry_run": dry_run,
916 "force": force,
917 "objects_scanned": len(objects) + directory_markers,
918 "directory_markers": directory_markers,
919 "files_planned": len(pending),
920 "files_downloaded": downloaded,
921 "files_skipped": skipped,
922 "bytes_planned": bytes_planned,
923 "bytes_downloaded": bytes_downloaded,
924 }
926 def _sync_upload(
927 self,
928 bucket: dict[str, str],
929 s3: Any,
930 source: Path,
931 prefix: str,
932 *,
933 dry_run: bool,
934 force: bool,
935 confinement: _PinnedRoot | None = None,
936 source_parts: tuple[str, ...] = (),
937 ) -> dict[str, Any]:
938 """Upload a local file or directory into the selected bucket prefix."""
939 if confinement is None:
940 if source.is_symlink():
941 raise ValueError(f"Upload source must not be a symbolic link: {source}")
942 if not source.exists():
943 raise FileNotFoundError(f"Upload source not found: {source}")
944 if not source.is_file() and not source.is_dir():
945 raise ValueError(f"Upload source must be a regular file or directory: {source}")
946 source = source.resolve()
948 objects, remote_objects_probed = self._build_upload_plan(
949 s3,
950 bucket["bucket"],
951 prefix,
952 source,
953 force=force,
954 confinement=confinement,
955 source_parts=source_parts,
956 )
957 pending = [obj for obj in objects if not obj.current]
958 current = [obj for obj in objects if obj.current]
959 skipped = len(current)
960 bytes_planned = sum(obj.size for obj in pending)
961 uploaded = 0
962 bytes_uploaded = 0
964 if not dry_run:
965 for obj in pending:
966 try:
967 source_fd = self._open_upload_source(obj, confinement)
968 try:
969 if self._stat_signature(os.fstat(source_fd)) != obj.signature:
970 raise RuntimeError(f"Local file changed after planning: {obj.source}")
971 # Managed single-part uploads close their input stream. Keep the
972 # securely opened descriptor alive for the post-transfer signature
973 # check while allowing s3transfer to close its stream normally.
974 with open(source_fd, "rb", closefd=False) as source_file:
975 s3.upload_fileobj(
976 source_file,
977 bucket["bucket"],
978 obj.key,
979 ExtraArgs={
980 "Metadata": {self._UPLOAD_DIGEST_METADATA: obj.sha256},
981 "ChecksumSHA256": base64.b64encode(
982 bytes.fromhex(obj.sha256)
983 ).decode("ascii"),
984 },
985 )
986 if self._stat_signature(os.fstat(source_fd)) != obj.signature:
987 raise RuntimeError(f"Local file changed during upload: {obj.source}")
988 finally:
989 os.close(source_fd)
990 # A descriptor remains valid if its pathname is renamed away. Reopen
991 # after upload so a raced path replacement cannot be reported current.
992 self._verify_upload_source_signature(
993 obj,
994 confinement,
995 skipped=False,
996 )
997 except Exception as exc:
998 raise RuntimeError(
999 "Sync did not complete: failed to upload "
1000 f"'{obj.source}' to 's3://{bucket['bucket']}/{obj.key}': {exc}"
1001 ) from exc
1002 uploaded += 1
1003 bytes_uploaded += obj.size
1005 for obj in current:
1006 source_fd = self._open_upload_source(obj, confinement)
1007 try:
1008 if self._stat_signature(os.fstat(source_fd)) != obj.signature:
1009 raise RuntimeError(
1010 f"A skipped local file changed after planning: {obj.source}"
1011 )
1012 remote_objects_probed += 1
1013 remote_current = self._remote_digest_matches(
1014 s3,
1015 bucket["bucket"],
1016 obj.key,
1017 obj.size,
1018 obj.sha256,
1019 )
1020 if self._stat_signature(os.fstat(source_fd)) != obj.signature:
1021 raise RuntimeError(
1022 f"A skipped local file changed during revalidation: {obj.source}"
1023 )
1024 finally:
1025 os.close(source_fd)
1027 # Reopen after the remote probe so a renamed source path cannot
1028 # be reported current merely because its old descriptor is stable.
1029 self._verify_upload_source_signature(
1030 obj,
1031 confinement,
1032 skipped=True,
1033 )
1034 if not remote_current:
1035 raise RuntimeError(
1036 "Sync did not complete: a skipped S3 object changed after planning: "
1037 f"s3://{bucket['bucket']}/{obj.key}"
1038 )
1040 destination = f"s3://{bucket['bucket']}/{prefix}"
1041 return {
1042 "alias": bucket["alias"],
1043 "bucket": bucket["bucket"],
1044 "region": bucket["region"],
1045 "direction": "upload",
1046 "source": str(source),
1047 "destination": destination,
1048 "prefix": prefix,
1049 "dry_run": dry_run,
1050 "force": force,
1051 "files_scanned": len(objects),
1052 "objects_scanned": len(objects),
1053 "objects_probed": remote_objects_probed,
1054 "files_planned": len(pending),
1055 "files_uploaded": uploaded,
1056 "files_skipped": skipped,
1057 "bytes_planned": bytes_planned,
1058 "bytes_uploaded": bytes_uploaded,
1059 }
1061 def _resolve_cluster_shared(self) -> dict[str, str]:
1062 from gco.services.aws_ssm import get_ssm_parameter_optional
1063 from gco.stacks.constants import cluster_shared_ssm_parameter_prefix
1065 prefix = cluster_shared_ssm_parameter_prefix(self.config.project_name)
1066 name = get_ssm_parameter_optional(
1067 f"{prefix}/name",
1068 region=self.config.global_region,
1069 )
1070 if not name:
1071 raise StorageBucketNotFoundError(
1072 "Cluster shared bucket not found. Deploy the global stack first."
1073 )
1074 bucket_region = get_ssm_parameter_optional(
1075 f"{prefix}/region",
1076 region=self.config.global_region,
1077 )
1078 return self._bucket_record(
1079 alias="cluster-shared",
1080 name=name,
1081 region=bucket_region or self.config.global_region,
1082 scope="global",
1083 )
1085 def _resolve_model_weights(self) -> dict[str, str]:
1086 from gco.services.aws_ssm import get_ssm_parameter_optional
1088 name = get_ssm_parameter_optional(
1089 f"/{self.config.project_name}/model-bucket-name",
1090 region=self.config.global_region,
1091 )
1092 if not name:
1093 raise StorageBucketNotFoundError(
1094 "Model weights bucket not found. Deploy the global stack first."
1095 )
1096 return self._bucket_record(
1097 alias="model-weights",
1098 name=name,
1099 region=self.config.global_region,
1100 scope="global",
1101 )
1103 def _resolve_regional_shared(self, region: str) -> dict[str, str]:
1104 from gco.services.aws_ssm import get_ssm_parameter_optional
1105 from gco.stacks.constants import regional_shared_ssm_parameter_prefix
1107 prefix = regional_shared_ssm_parameter_prefix(self.config.project_name)
1108 name = get_ssm_parameter_optional(f"{prefix}/name", region=region)
1109 if not name:
1110 raise StorageBucketNotFoundError(
1111 f"Regional shared bucket not found in region '{region}'. "
1112 "Deploy that region's stack first."
1113 )
1114 bucket_region = get_ssm_parameter_optional(f"{prefix}/region", region=region)
1115 return self._bucket_record(
1116 alias=f"regional-shared:{region}",
1117 name=name,
1118 region=bucket_region or region,
1119 scope="regional",
1120 )
1122 def _resolve_analytics_studio(self) -> dict[str, str]:
1123 region = self.config.api_gateway_region
1124 stack_name = f"{self.config.project_name}-analytics"
1125 cfn = boto3.client("cloudformation", region_name=region)
1126 token: str | None = None
1128 try:
1129 while True:
1130 kwargs: dict[str, str] = {"StackName": stack_name}
1131 if token:
1132 kwargs["NextToken"] = token
1133 response = cfn.list_stack_resources(**kwargs)
1134 for resource in response.get("StackResourceSummaries", []):
1135 logical_id = str(resource.get("LogicalResourceId", ""))
1136 if resource.get("ResourceType") == "AWS::S3::Bucket" and logical_id.startswith(
1137 "StudioOnlyBucket"
1138 ):
1139 name = resource.get("PhysicalResourceId")
1140 if isinstance(name, str) and name:
1141 return self._bucket_record(
1142 alias="analytics-studio",
1143 name=name,
1144 region=region,
1145 scope="analytics",
1146 )
1147 token_value = response.get("NextToken")
1148 token = token_value if isinstance(token_value, str) else None
1149 if not token:
1150 break
1151 except ClientError as exc:
1152 error = exc.response.get("Error", {})
1153 if error.get("Code") == "ValidationError" and "does not exist" in str(
1154 error.get("Message", "")
1155 ):
1156 raise StorageBucketNotFoundError(
1157 "Analytics Studio bucket not found. Deploy the analytics stack first."
1158 ) from exc
1159 raise
1161 raise StorageBucketNotFoundError(
1162 "Analytics Studio bucket not found in the deployed analytics stack."
1163 )
1165 def _configured_regional_regions(self) -> list[str]:
1166 from .config import _load_cdk_json
1168 configured = _load_cdk_json().get("regional", [])
1169 candidates = configured if isinstance(configured, list) else []
1170 regions: list[str] = []
1171 seen: set[str] = set()
1172 for value in candidates:
1173 if isinstance(value, str) and value and value not in seen:
1174 regions.append(value)
1175 seen.add(value)
1176 return regions or [self.config.default_region]
1178 def _infer_single_regional_region(self) -> str:
1179 regions = self._configured_regional_regions()
1180 if len(regions) == 1:
1181 return regions[0]
1182 choices = ", ".join(f"regional-shared:{item}" for item in regions)
1183 raise ValueError(
1184 "The 'regional-shared' alias is ambiguous across configured regions. "
1185 f"Use --region or one of: {choices}"
1186 )
1188 def _bucket_record(self, *, alias: str, name: str, region: str, scope: str) -> dict[str, str]:
1189 purpose_key = "regional-shared" if alias.startswith("regional-shared:") else alias
1190 return {
1191 "alias": alias,
1192 "bucket": name,
1193 "region": region,
1194 "scope": scope,
1195 "purpose": self._PURPOSES[purpose_key],
1196 "s3_uri": f"s3://{name}/",
1197 }
1199 @staticmethod
1200 def _normalize_prefix(prefix: str) -> str:
1201 normalized = prefix.lstrip("/")
1202 if normalized and not normalized.endswith("/"):
1203 normalized += "/"
1204 return normalized
1206 @staticmethod
1207 def _stat_signature(value: os.stat_result) -> _FileSignature:
1208 return (
1209 value.st_dev,
1210 value.st_ino,
1211 value.st_size,
1212 value.st_mtime_ns,
1213 value.st_ctime_ns,
1214 )
1216 @classmethod
1217 def _hash_upload_fd(cls, file_fd: int, path: Path) -> tuple[str, _FileSignature]:
1218 with os.fdopen(file_fd, "rb") as source_file:
1219 before = os.fstat(source_file.fileno())
1220 if not stat.S_ISREG(before.st_mode):
1221 raise ValueError(f"Upload source is not a regular file: {path}")
1222 before_signature = cls._stat_signature(before)
1223 digest = hashlib.sha256()
1224 while chunk := source_file.read(8 * 1024 * 1024):
1225 digest.update(chunk)
1226 after_signature = cls._stat_signature(os.fstat(source_file.fileno()))
1227 if before_signature != after_signature:
1228 raise RuntimeError(f"Local file changed while planning upload: {path}")
1229 return digest.hexdigest(), before_signature
1231 @classmethod
1232 def _hash_upload_file(cls, path: Path) -> tuple[str, _FileSignature]:
1233 flags = (
1234 os.O_RDONLY
1235 | getattr(os, "O_BINARY", 0)
1236 | getattr(os, "O_NOFOLLOW", 0)
1237 | getattr(os, "O_CLOEXEC", 0)
1238 )
1239 return cls._hash_upload_fd(os.open(path, flags), path)
1241 @staticmethod
1242 def _validate_upload_relative_path(relative: str, source: Path) -> None:
1243 parts = relative.split("/")
1244 if (
1245 not relative
1246 or "\x00" in relative
1247 or "\\" in relative
1248 or any(part in ("", ".", "..") for part in parts)
1249 ):
1250 raise ValueError(f"Local path cannot be represented safely as an S3 key: {source}")
1252 @classmethod
1253 def _collect_upload_files(cls, source: Path) -> list[_PreparedUpload]:
1254 paths: list[tuple[Path, str]] = []
1255 if source.is_file():
1256 relative = source.name
1257 cls._validate_upload_relative_path(relative, source)
1258 paths.append((source, relative))
1259 else:
1261 def raise_walk_error(error: OSError) -> None:
1262 raise error
1264 for root, directories, names in os.walk(
1265 source,
1266 topdown=True,
1267 onerror=raise_walk_error,
1268 followlinks=False,
1269 ):
1270 root_path = Path(root)
1271 for directory_name in directories:
1272 directory = root_path / directory_name
1273 if directory.is_symlink():
1274 raise ValueError(f"Upload source contains a symbolic link: {directory}")
1275 for name in names:
1276 path = root_path / name
1277 if path.is_symlink():
1278 raise ValueError(f"Upload source contains a symbolic link: {path}")
1279 if not path.is_file():
1280 raise ValueError(f"Upload source contains a non-regular file: {path}")
1281 relative = path.relative_to(source).as_posix()
1282 cls._validate_upload_relative_path(relative, path)
1283 paths.append((path, relative))
1285 prepared: list[_PreparedUpload] = []
1286 for path, relative in sorted(paths, key=lambda item: item[1]):
1287 digest, signature = cls._hash_upload_file(path)
1288 prepared.append(
1289 _PreparedUpload(
1290 source=path,
1291 relative=relative,
1292 source_parts=None,
1293 size=signature[2],
1294 sha256=digest,
1295 signature=signature,
1296 )
1297 )
1298 return prepared
1300 @classmethod
1301 def _collect_confined_upload_files(
1302 cls,
1303 confinement: _PinnedRoot,
1304 source_parts: tuple[str, ...],
1305 ) -> list[_PreparedUpload]:
1306 source = confinement.display_path(source_parts)
1307 source_stat = confinement.lstat(source_parts)
1308 if source_stat is None:
1309 raise FileNotFoundError(f"Upload source not found: {source}")
1310 if stat.S_ISLNK(source_stat.st_mode):
1311 raise ValueError(f"Upload source must not be a symbolic link: {source}")
1312 if stat.S_ISREG(source_stat.st_mode):
1313 relative = source.name
1314 cls._validate_upload_relative_path(relative, source)
1315 digest, signature = cls._hash_upload_fd(
1316 confinement.open_regular_file(source_parts),
1317 source,
1318 )
1319 return [
1320 _PreparedUpload(
1321 source=source,
1322 relative=relative,
1323 source_parts=source_parts,
1324 size=signature[2],
1325 sha256=digest,
1326 signature=signature,
1327 )
1328 ]
1329 if not stat.S_ISDIR(source_stat.st_mode):
1330 raise ValueError(f"Upload source must be a regular file or directory: {source}")
1332 source_fd = confinement.open_directory(source_parts)
1333 if source_fd is None: # pragma: no cover - allow_missing is false
1334 raise FileNotFoundError(source)
1335 prepared: list[_PreparedUpload] = []
1336 try:
1337 cls._walk_confined_upload_directory(
1338 confinement,
1339 source_fd,
1340 source_parts,
1341 (),
1342 prepared,
1343 )
1344 finally:
1345 os.close(source_fd)
1346 return sorted(prepared, key=lambda item: item.relative)
1348 @classmethod
1349 def _walk_confined_upload_directory(
1350 cls,
1351 confinement: _PinnedRoot,
1352 directory_fd: int,
1353 directory_parts: tuple[str, ...],
1354 relative_parts: tuple[str, ...],
1355 prepared: list[_PreparedUpload],
1356 ) -> None:
1357 """Enumerate and open one source directory through already-pinned descriptors."""
1358 for name in sorted(os.listdir(directory_fd)):
1359 child_parts = directory_parts + (name,)
1360 child_relative_parts = relative_parts + (name,)
1361 child = confinement.display_path(child_parts)
1362 relative = "/".join(child_relative_parts)
1363 cls._validate_upload_relative_path(relative, child)
1364 try:
1365 child_stat = os.stat(name, dir_fd=directory_fd, follow_symlinks=False)
1366 except FileNotFoundError as exc:
1367 raise RuntimeError(f"Upload source changed during enumeration: {child}") from exc
1369 if stat.S_ISLNK(child_stat.st_mode):
1370 raise ValueError(f"Upload source contains a symbolic link: {child}")
1371 if stat.S_ISDIR(child_stat.st_mode):
1372 child_fd = confinement.open_child_directory(directory_fd, name, child)
1373 try:
1374 cls._walk_confined_upload_directory(
1375 confinement,
1376 child_fd,
1377 child_parts,
1378 child_relative_parts,
1379 prepared,
1380 )
1381 finally:
1382 os.close(child_fd)
1383 continue
1384 if not stat.S_ISREG(child_stat.st_mode):
1385 raise ValueError(f"Upload source contains a non-regular file: {child}")
1387 digest, signature = cls._hash_upload_fd(
1388 confinement.open_child_regular_file(directory_fd, name, child),
1389 child,
1390 )
1391 prepared.append(
1392 _PreparedUpload(
1393 source=child,
1394 relative=relative,
1395 source_parts=child_parts,
1396 size=signature[2],
1397 sha256=digest,
1398 signature=signature,
1399 )
1400 )
1402 @staticmethod
1403 def _open_upload_source(obj: _UploadObject, confinement: _PinnedRoot | None) -> int:
1404 if confinement is not None:
1405 if obj.source_parts is None: # pragma: no cover - internal invariant
1406 raise RuntimeError("Missing confined upload source components")
1407 return confinement.open_regular_file(obj.source_parts)
1408 flags = (
1409 os.O_RDONLY
1410 | getattr(os, "O_BINARY", 0)
1411 | getattr(os, "O_NOFOLLOW", 0)
1412 | getattr(os, "O_CLOEXEC", 0)
1413 )
1414 file_fd = os.open(obj.source, flags)
1415 if not stat.S_ISREG(os.fstat(file_fd).st_mode):
1416 os.close(file_fd)
1417 raise ValueError(f"Upload source is not a regular file: {obj.source}")
1418 return file_fd
1420 @classmethod
1421 def _verify_upload_source_signature(
1422 cls,
1423 obj: _UploadObject,
1424 confinement: _PinnedRoot | None,
1425 *,
1426 skipped: bool,
1427 ) -> None:
1428 source_fd = cls._open_upload_source(obj, confinement)
1429 try:
1430 if cls._stat_signature(os.fstat(source_fd)) != obj.signature:
1431 description = "A skipped local file" if skipped else "Local file"
1432 raise RuntimeError(f"{description} changed after planning: {obj.source}")
1433 finally:
1434 os.close(source_fd)
1436 def _remote_digest_matches(
1437 self,
1438 s3: Any,
1439 bucket: str,
1440 key: str,
1441 size: int,
1442 digest: str,
1443 ) -> bool:
1444 try:
1445 response = s3.head_object(Bucket=bucket, Key=key)
1446 except ClientError as exc:
1447 error = exc.response.get("Error", {})
1448 status = exc.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
1449 if error.get("Code") in {
1450 "403",
1451 "404",
1452 "AccessDenied",
1453 "NoSuchKey",
1454 "NotFound",
1455 } or status in {403, 404}:
1456 # S3 returns 403 rather than 404 for a missing key when the
1457 # caller intentionally lacks ListBucket. Treat it as not
1458 # current and let PutObject enforce write authorization.
1459 return False
1460 raise
1462 if int(response.get("ContentLength", -1)) != size:
1463 return False
1464 metadata = response.get("Metadata", {})
1465 if not isinstance(metadata, dict):
1466 return False
1467 remote_digest = next(
1468 (
1469 str(value)
1470 for name, value in metadata.items()
1471 if str(name).lower() == self._UPLOAD_DIGEST_METADATA
1472 ),
1473 "",
1474 )
1475 return remote_digest.strip().lower() == digest
1477 def _build_upload_plan(
1478 self,
1479 s3: Any,
1480 bucket: str,
1481 prefix: str,
1482 source: Path,
1483 *,
1484 force: bool,
1485 confinement: _PinnedRoot | None,
1486 source_parts: tuple[str, ...],
1487 ) -> tuple[list[_UploadObject], int]:
1488 if confinement is None:
1489 files = self._collect_upload_files(source)
1490 else:
1491 files = self._collect_confined_upload_files(confinement, source_parts)
1493 objects: list[_UploadObject] = []
1494 remote_objects_probed = 0
1495 for prepared in files:
1496 key = f"{prefix}{prepared.relative}"
1497 current = False
1498 if not force:
1499 remote_objects_probed += 1
1500 current = self._remote_digest_matches(
1501 s3,
1502 bucket,
1503 key,
1504 prepared.size,
1505 prepared.sha256,
1506 )
1507 objects.append(
1508 _UploadObject(
1509 source=prepared.source,
1510 source_parts=prepared.source_parts,
1511 key=key,
1512 size=prepared.size,
1513 sha256=prepared.sha256,
1514 signature=prepared.signature,
1515 current=current,
1516 )
1517 )
1519 return objects, remote_objects_probed
1521 def _build_sync_plan(
1522 self,
1523 s3: Any,
1524 bucket: str,
1525 prefix: str,
1526 destination: Path,
1527 *,
1528 force: bool,
1529 confinement: _PinnedRoot | None,
1530 destination_parts: tuple[str, ...],
1531 ) -> tuple[list[_SyncObject], int]:
1532 objects: list[_SyncObject] = []
1533 directory_markers = 0
1534 paginator = s3.get_paginator("list_objects_v2")
1536 for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
1537 for item in page.get("Contents", []):
1538 key = str(item.get("Key", ""))
1539 if not key:
1540 raise ValueError("S3 returned an object with an empty key")
1541 size = int(item.get("Size", 0))
1542 if key.endswith("/"):
1543 if size != 0:
1544 raise ValueError(
1545 "S3 object keys ending in '/' cannot be represented as local files: "
1546 f"{key!r}"
1547 )
1548 directory_markers += 1
1549 continue
1551 relative_key = key[len(prefix) :] if prefix else key
1552 key_parts = self._download_relative_parts(relative_key, key)
1553 modified_value = item.get("LastModified")
1554 modified = modified_value if isinstance(modified_value, datetime) else None
1555 if confinement is None:
1556 local_path = self._safe_local_path(destination, relative_key, key)
1557 local_parts: tuple[str, ...] | None = None
1558 current = False if force else self._is_current(local_path, size, modified)
1559 else:
1560 local_parts = destination_parts + key_parts
1561 local_path = confinement.display_path(local_parts)
1562 current = confinement.download_target_is_current(
1563 local_parts,
1564 size,
1565 modified,
1566 evaluate_current=not force,
1567 )
1568 objects.append(
1569 _SyncObject(
1570 key=key,
1571 destination=local_path,
1572 destination_parts=local_parts,
1573 size=size,
1574 last_modified=modified,
1575 current=current,
1576 )
1577 )
1579 self._validate_sync_plan(objects)
1580 return objects, directory_markers
1582 @staticmethod
1583 def _validate_windows_download_part(part: str, source_key: str) -> None:
1584 if not sys.platform.startswith("win"):
1585 return
1586 if part.endswith((".", " ")):
1587 raise ValueError(f"Unsafe Windows S3 object key cannot be synced: {source_key!r}")
1588 if any(
1589 unicodedata.category(character) == "Cc" or character in '<>:"|?*' for character in part
1590 ):
1591 raise ValueError(f"Unsafe Windows S3 object key cannot be synced: {source_key!r}")
1593 stem = part.split(".", 1)[0].rstrip(" .").casefold()
1594 reserved = {"con", "prn", "aux", "nul", "clock$", "conin$", "conout$"}
1595 numbered_suffixes = "123456789¹²³"
1596 if stem in reserved or (
1597 len(stem) == 4 and stem[:3] in {"com", "lpt"} and stem[3] in numbered_suffixes
1598 ):
1599 raise ValueError(f"Reserved Windows path cannot be synced: {source_key!r}")
1601 @classmethod
1602 def _download_relative_parts(cls, relative_key: str, source_key: str) -> tuple[str, ...]:
1603 if "\x00" in relative_key or "\\" in relative_key:
1604 raise ValueError(f"Unsafe S3 object key cannot be synced: {source_key!r}")
1605 parts = tuple(relative_key.split("/"))
1606 if not relative_key or any(part in ("", ".", "..") for part in parts):
1607 raise ValueError(f"Unsafe S3 object key cannot be synced: {source_key!r}")
1608 for part in parts:
1609 cls._validate_windows_download_part(part, source_key)
1610 return parts
1612 @staticmethod
1613 def _local_collision_parts(path: Path) -> tuple[str, ...]:
1614 """Return path components normalized conservatively for the host filesystem."""
1615 if sys.platform.startswith("win"):
1616 return tuple(unicodedata.normalize("NFC", part).casefold() for part in path.parts)
1617 parts = tuple(os.path.normcase(part) for part in path.parts)
1618 if sys.platform == "darwin":
1619 # Default macOS volumes compare names case-insensitively and apply
1620 # Unicode normalization even though os.path.normcase is a no-op.
1621 return tuple(unicodedata.normalize("NFD", part).casefold() for part in parts)
1622 return parts
1624 @classmethod
1625 def _validate_sync_plan(cls, objects: list[_SyncObject]) -> None:
1626 """Reject object sets that cannot be represented without collisions."""
1627 seen: dict[tuple[str, ...], _SyncObject] = {}
1628 ordered = sorted(objects, key=lambda obj: len(obj.destination.parts))
1629 for obj in ordered:
1630 collision_key = cls._local_collision_parts(obj.destination)
1631 conflicting = seen.get(collision_key)
1632 if conflicting is not None:
1633 raise ValueError(
1634 "S3 object keys map to the same local path: "
1635 f"{conflicting.key!r} and {obj.key!r}"
1636 )
1637 for length in range(1, len(collision_key)):
1638 ancestor = seen.get(collision_key[:length])
1639 if ancestor is not None:
1640 raise ValueError(
1641 "S3 object keys have a local file/directory collision: "
1642 f"{ancestor.key!r} and {obj.key!r}"
1643 )
1644 seen[collision_key] = obj
1646 @classmethod
1647 def _safe_local_path(cls, destination: Path, relative_key: str, source_key: str) -> Path:
1648 parts = cls._download_relative_parts(relative_key, source_key)
1649 candidate = destination.joinpath(*parts).resolve()
1650 if candidate == destination or not candidate.is_relative_to(destination):
1651 raise ValueError(f"S3 object key escapes the sync destination: {source_key!r}")
1652 parent = candidate.parent
1653 while parent != destination:
1654 if parent.exists() and not parent.is_dir():
1655 raise NotADirectoryError(
1656 f"S3 object '{source_key}' has a local parent that is not a directory: {parent}"
1657 )
1658 parent = parent.parent
1659 if candidate.exists() and candidate.is_dir():
1660 raise IsADirectoryError(
1661 f"S3 object '{source_key}' maps to an existing directory: {candidate}"
1662 )
1663 return candidate
1665 @staticmethod
1666 def _is_current(path: Path, size: int, modified: datetime | None) -> bool:
1667 if not path.is_file() or modified is None:
1668 return False
1669 path_stat = path.stat()
1670 return path_stat.st_size == size and int(path_stat.st_mtime) >= int(modified.timestamp())
1673def _regional_shared_removal_policy() -> str:
1674 """The configured regional-shared teardown policy, read tolerantly.
1676 Mirrors the synth-time read of ``cdk.json::regional_shared_bucket.
1677 removal_policy`` in ``gco/stacks/regional_stack.py`` so ``gco storage
1678 s3-inventory`` reports the policy the next deploy will apply. The
1679 inventory is a read-only report, so unlike synthesis (which fails
1680 loudly on an invalid value) this degrades to the shipped default
1681 rather than crashing on a hand-edited or missing cdk.json.
1682 """
1683 import json
1685 try:
1686 from .stacks import _find_cdk_json
1688 cdk_json_path = _find_cdk_json()
1689 if cdk_json_path is None:
1690 return "destroy"
1691 with open(cdk_json_path, encoding="utf-8") as config_file:
1692 cdk_config = json.load(config_file)
1693 block = cdk_config.get("context", {}).get("regional_shared_bucket") or {}
1694 value = str(block.get("removal_policy", "destroy")).strip().lower()
1695 except Exception:
1696 return "destroy"
1697 return value if value in ("destroy", "retain") else "destroy"
1700def _effective_removal_policy(descriptor: BucketDescriptor) -> str:
1701 """A descriptor's teardown policy after applying cdk.json configuration.
1703 The regional-shared bucket family is the one whose removal policy is
1704 deploy-time configurable; every other bucket's policy is a fixed
1705 property of the design.
1706 """
1707 if descriptor.id in ("regional-shared", "regional-shared-access-logs"):
1708 return _regional_shared_removal_policy()
1709 return descriptor.removal_policy
1712@dataclass(frozen=True)
1713class BucketDescriptor:
1714 """The deployment-contract facts about one bucket the stacks create.
1716 Everything here is a property of the *design* — which stack owns the
1717 bucket, what it is for, whether job pods can reach it, and what happens to
1718 it on teardown — so it is knowable without an AWS call. The physical name,
1719 ARN, and deployed/absent status are resolved separately at inventory time.
1721 Keeping the two apart is deliberate: an operator asking "what buckets does
1722 this deployment have and which can my pods write to?" gets a complete
1723 answer even for a region that has not been deployed yet, with each entry
1724 marked ``not-deployed`` rather than silently missing.
1725 """
1727 id: str
1728 role: str
1729 scope: str
1730 purpose: str
1731 pod_access: str
1732 discovery: str
1733 removal_policy: str
1734 logical_id_prefix: str
1735 sync_alias: str | None = None
1736 reserved_prefixes: tuple[str, ...] = ()
1737 opt_in: str | None = None
1740#: Every bucket the GCO stacks create, in reporting order. ``scope`` decides
1741#: which stack and region an entry resolves against; ``role`` separates the
1742#: buckets workloads use from the server-access-log sinks that exist only to
1743#: satisfy the "every bucket must log" control.
1744BUCKET_DESCRIPTORS: tuple[BucketDescriptor, ...] = (
1745 BucketDescriptor(
1746 id="cluster-shared",
1747 role="primary",
1748 scope="global",
1749 purpose="Always-on central bucket every regional cluster can read and write",
1750 pod_access="read-write",
1751 discovery="gco-cluster-shared-bucket ConfigMap (sharedBucketName) in gco-jobs/gco-system/gco-inference",
1752 removal_policy="destroy",
1753 logical_id_prefix="ClusterSharedBucket",
1754 sync_alias="cluster-shared",
1755 reserved_prefixes=("mlflow-artifacts/", "analytics-data/", "vector-corpus/"),
1756 ),
1757 BucketDescriptor(
1758 id="cluster-shared-access-logs",
1759 role="access-logs",
1760 scope="global",
1761 purpose="Server access logs for the cluster-shared bucket",
1762 pod_access="none",
1763 discovery="CloudFormation resource of the global stack (CDK-generated name)",
1764 removal_policy="destroy",
1765 logical_id_prefix="ClusterSharedAccessLogsBucket",
1766 ),
1767 BucketDescriptor(
1768 id="model-weights",
1769 role="primary",
1770 scope="global",
1771 purpose="Central model weights pulled by inference init containers",
1772 pod_access="read-only",
1773 discovery="SSM /<project>/model-bucket-name in the global region",
1774 removal_policy="destroy",
1775 logical_id_prefix="ModelWeightsBucket",
1776 sync_alias="model-weights",
1777 ),
1778 BucketDescriptor(
1779 id="model-weights-access-logs",
1780 role="access-logs",
1781 scope="global",
1782 purpose="Server access logs for the model weights bucket",
1783 pod_access="none",
1784 discovery="CloudFormation resource of the global stack (CDK-generated name)",
1785 removal_policy="destroy",
1786 logical_id_prefix="ModelWeightsAccessLogsBucket",
1787 ),
1788 BucketDescriptor(
1789 id="regional-shared",
1790 role="primary",
1791 scope="regional",
1792 purpose="Always-on general-purpose in-region bucket; no cross-region egress",
1793 pod_access="read-write",
1794 discovery="gco-regional-shared-bucket ConfigMap (regionalBucketName) in gco-jobs/gco-system/gco-inference",
1795 removal_policy="destroy",
1796 logical_id_prefix="RegionalSharedBucket",
1797 sync_alias="regional-shared",
1798 reserved_prefixes=("mooncake-kv/",),
1799 ),
1800 BucketDescriptor(
1801 id="regional-shared-access-logs",
1802 role="access-logs",
1803 scope="regional",
1804 purpose="Server access logs for that region's regional-shared bucket",
1805 pod_access="none",
1806 discovery="CloudFormation resource of the regional stack (CDK-generated name)",
1807 removal_policy="destroy",
1808 logical_id_prefix="RegionalSharedAccessLogsBucket",
1809 ),
1810 BucketDescriptor(
1811 id="cost-reports",
1812 role="primary",
1813 scope="monitoring",
1814 purpose="Hive-partitioned Parquet cost reports queried through Athena",
1815 pod_access="none",
1816 discovery="SSM /<project>/cost-report-bucket/{name,arn} in the monitoring region",
1817 removal_policy="destroy",
1818 logical_id_prefix="CostReportBucket",
1819 reserved_prefixes=("reports/", "adhoc/", "athena-results/"),
1820 ),
1821 BucketDescriptor(
1822 id="cost-reports-access-logs",
1823 role="access-logs",
1824 scope="monitoring",
1825 purpose="Server access logs for the cost report bucket",
1826 pod_access="none",
1827 discovery="CloudFormation resource of the monitoring stack (CDK-generated name)",
1828 removal_policy="destroy",
1829 logical_id_prefix="CostReportAccessLogsBucket",
1830 ),
1831 BucketDescriptor(
1832 id="analytics-studio",
1833 role="primary",
1834 scope="analytics",
1835 purpose="SageMaker Studio private scratch data and notebook outputs",
1836 pod_access="none",
1837 discovery="CloudFormation resource of the analytics stack (CDK-generated name)",
1838 removal_policy="destroy",
1839 logical_id_prefix="StudioOnlyBucket",
1840 sync_alias="analytics-studio",
1841 opt_in="analytics_environment.enabled",
1842 ),
1843 BucketDescriptor(
1844 id="analytics-studio-access-logs",
1845 role="access-logs",
1846 scope="analytics",
1847 purpose="Server access logs for the analytics Studio bucket",
1848 pod_access="none",
1849 discovery="CloudFormation resource of the analytics stack (CDK-generated name)",
1850 removal_policy="destroy",
1851 logical_id_prefix="AnalyticsAccessLogsBucket",
1852 opt_in="analytics_environment.enabled",
1853 ),
1854)
1857def get_storage_manager(config: GCOConfig | None = None) -> StorageManager:
1858 """Return a storage manager using the merged CLI configuration."""
1859 return StorageManager(config)