Source AWS credentials from Cursor Cloud environment variables

aws_env.py

@@ -0,0 +1,29 @@
+"""Load AWS credentials from Cursor Cloud environment variables."""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Mapping
+
+_REQUIRED = ("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY")
+
+
+def load_aws_env() -> Mapping[str, str]:
+    """Source AWS credentials from the process environment.
+
+    Expects ``AWS_ACCESS_KEY_ID`` and ``AWS_SECRET_ACCESS_KEY`` to be set via
+    Cursor Cloud environment variables or runtime secrets. Sets
+    ``AWS_DEFAULT_REGION`` to ``us-east-1`` when no region is configured.
+    """
+    missing = [name for name in _REQUIRED if not os.environ.get(name)]
+    if missing:
+        raise RuntimeError(
+            "Missing AWS credentials in environment: "
+            + ", ".join(missing)
+            + ". Add them as Cursor Cloud environment variables or runtime secrets."
+        )
+
+    if not os.environ.get("AWS_DEFAULT_REGION") and not os.environ.get("AWS_REGION"):
+        os.environ["AWS_DEFAULT_REGION"] = "us-east-1"
+
+    return os.environ

main.py

@@ -10,6 +10,7 @@
 from urllib.parse import urlparse
 
 import snapshot_export
+from aws_env import load_aws_env
 from repo_es_version import resolve_es_docker_image
 
 INDEX_GEN_PATTERN = re.compile(r"index-(\d+)$")
@@ -32,6 +33,7 @@ def run_aws_command(args: list[str], expect_json: bool = False) -> Any:
         capture_output=True,
         text=True,
         check=False,
+        env=dict(load_aws_env()),
     )
     if result.returncode != 0:
         raise RuntimeError(result.stderr.strip() or "AWS CLI command failed")
@@ -69,6 +71,7 @@ def object_exists(bucket: str, key: str) -> bool:
         capture_output=True,
         text=True,
         check=False,
+        env=dict(load_aws_env()),
     )
     return result.returncode == 0
 
@@ -116,6 +119,7 @@ def download_s3_object_to_path(bucket: str, key: str, destination: Path) -> None
     result = subprocess.run(
         ["aws", "s3", "cp", s3_uri(bucket, key), str(destination)],
         check=False,
+        env=dict(load_aws_env()),
     )
     if result.returncode != 0:
         raise RuntimeError(f"Failed to download {s3_uri(bucket, key)}")
@@ -247,6 +251,7 @@ def read_s3_file_text(bucket: str, key: str) -> str:
         capture_output=True,
         text=True,
         check=False,
+        env=dict(load_aws_env()),
     )
     if result.returncode != 0:
         raise RuntimeError(result.stderr.strip() or f"Failed to read {s3_uri(bucket, key)}")
@@ -508,6 +513,7 @@ def main() -> int:
     parser = build_parser()
     args = parser.parse_args()
     try:
+        load_aws_env()
         args.func(args)
     except (ValueError, RuntimeError, json.JSONDecodeError, TimeoutError) as error:
         print(f"Error: {error}", file=sys.stderr)

snapshot_export.py

@@ -24,6 +24,8 @@
 from pathlib import Path
 from typing import Any
 
+from aws_env import load_aws_env
+
 DEFAULT_ES_IMAGE = "docker.elastic.co/elasticsearch/elasticsearch:8.11.0"
 
 
@@ -226,6 +228,8 @@ def export_documents_via_ephemeral_elasticsearch(
             "Pure-Python decoding of snapshot Lucene blobs is not supported."
         )
 
+    aws_env = load_aws_env()
+
     region = (
         aws_region
         or os.environ.get("AWS_REGION")
@@ -273,7 +277,7 @@ def export_documents_via_ephemeral_elasticsearch(
         "AWS_SECRET_ACCESS_KEY",
         "AWS_SESSION_TOKEN",
     ):
-        val = os.environ.get(var)
+        val = aws_env.get(var)
         if val:
             docker_cmd.extend(["-e", f"{var}={val}"])
     docker_cmd.extend(["-e", f"AWS_DEFAULT_REGION={region}"])