Fix S3 storage key limits and permission guidance

changelog/fixed/marshmallow-moonwalk.md

@@ -0,0 +1 @@
+Fix S3 external storage object names by preserving empty namespace segments and using namespace-scoped hash keys when names exceed 1,024 bytes. Document the read and bucket-listing permissions required for uploads.

temporalio/contrib/aws/s3driver/README.md

@@ -73,8 +73,10 @@ Fallback key (used when no namespace, workflow, or activity information is avail
     v0/d/{hash-algorithm}/{hex-digest}
 
 - Missing values (including a missing run ID) are encoded as the literal `null`.
+- An empty namespace on a workflow or activity target is written as `/ns/null`.
 - `hex-digest` is the lower-case SHA-256 hex digest (64 characters).
 - Dynamic path segments are percent-encoded: any byte outside S3's [safe character set](https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-keys.html) (`0-9`, `a-z`, `A-Z`, and `! - _ . * ' ( )`) is escaped as `%XX` over its UTF-8 bytes.
+- Keys exceeding S3's 1,024-byte limit after encoding fall back to `v0/ns/{namespace}/d/sha256/{hex-digest}`, retaining the namespace for access policies. Existing saved references remain readable, including references with no namespace segment.
 
 ### Notes
 
@@ -103,17 +105,26 @@ driver = S3StorageDriver(
 
 ### Required IAM permissions
 
-The AWS credentials used by your S3 client must have the following S3 permissions on the target bucket and its objects:
+Components that store payloads with the built-in aioboto3 adapter require `s3:PutObject`, `s3:GetObject`, and `s3:ListBucket`:
 
 ```json
 {
-  "Effect": "Allow",
-  "Action": [
-    "s3:PutObject",
-    "s3:GetObject"
-  ],
-  "Resource": "arn:aws:s3:::my-temporal-payloads/*"
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["s3:PutObject", "s3:GetObject"],
+      "Resource": "arn:aws:s3:::my-temporal-payloads/*"
+    },
+    {
+      "Effect": "Allow",
+      "Action": "s3:ListBucket",
+      "Resource": "arn:aws:s3:::my-temporal-payloads"
+    }
+  ]
 }
 ```
 
-`s3:PutObject` is required by components that store payloads (typically the Temporal client and worker sending workflow/activity inputs), and `s3:GetObject` is required by components that retrieve them (typically workers and clients reading results). Components that only retrieve payloads do not need `s3:PutObject`, and vice versa.
+Uploads check whether an object already exists using `HeadObject`, which requires read permission even for components that only store payloads. Without `s3:ListBucket`, S3 returns 403 instead of 404 for a missing object, so the driver cannot distinguish it from an access denial and stops the upload. See the [AWS HeadObject permissions documentation](https://docs.aws.amazon.com/AmazonS3/latest/API/API_HeadObject.html).
+
+Components that only retrieve payloads require `s3:GetObject`; they do not need `s3:PutObject` or `s3:ListBucket`.

temporalio/contrib/aws/s3driver/_driver.py

@@ -146,12 +146,15 @@ async def store(
         Python value may serialize differently across payload converter versions
         (e.g. proto binary). The returned list is the same length as
         ``payloads``.
+
+        Object keys exceeding 1,024 bytes fall back to a namespace-scoped hash
+        key. Empty namespaces on workflow and activity targets use ``"null"``.
         """
         # Build context segments from the target identity.
         context_segments = ""
         target = context.target
         namespace = _percent_encode(target.namespace) if target is not None else None
-        namespace_segment = f"/ns/{namespace}" if namespace else ""
+        namespace_segment = f"/ns/{namespace or 'null'}" if target is not None else ""
         if isinstance(target, StorageDriverWorkflowInfo):
             wf_type = _percent_encode(target.type) or "null"
             wf_id = _percent_encode(target.id) or "null"
@@ -178,6 +181,9 @@ async def _upload(payload: Payload) -> StorageDriverClaim:
             digest_segments = f"/d/sha256/{hash_digest}"
 
             key = f"v0{namespace_segment}{context_segments}{digest_segments}"
+            if len(key.encode("utf-8")) > 1024:
+                # Retain namespace scope for authorization policies on object prefixes.
+                key = f"v0{namespace_segment}{digest_segments}"
 
             try:
                 if not await self._client.object_exists(bucket=bucket, key=key):

temporalio/contrib/aws/s3driver/aioboto3.py

@@ -43,7 +43,11 @@ def describe(self) -> Mapping[str, str]:
         return {"client_region": region} if region else {}
 
     async def object_exists(self, *, bucket: str, key: str) -> bool:
-        """Check existence via aioboto3's ``head_object``."""
+        """Check existence via aioboto3's ``head_object``.
+
+        Requires ``s3:GetObject`` and ``s3:ListBucket`` permissions so missing
+        objects return 404 rather than an ambiguous 403.
+        """
         try:
             await self._client.head_object(Bucket=bucket, Key=key)
             return True

tests/contrib/aws/s3driver/test_s3driver.py

@@ -15,7 +15,7 @@
 from collections.abc import Callable, Coroutine
 from functools import wraps
 from typing import Any
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 from botocore.exceptions import ClientError
@@ -203,6 +203,100 @@ def test_type(self) -> None:
 
 
 class TestS3StorageDriverKeyConstruction:
+    @pytest.mark.parametrize(
+        "target_type,identity",
+        [
+            (StorageDriverWorkflowInfo, "wt/null/wi"),
+            (StorageDriverActivityInfo, "at/null/ai"),
+        ],
+    )
+    async def test_empty_namespace(
+        self,
+        driver_client: S3StorageDriverClient,
+        target_type: type[StorageDriverWorkflowInfo] | type[StorageDriverActivityInfo],
+        identity: str,
+    ) -> None:
+        driver = S3StorageDriver(client=driver_client, bucket=BUCKET)
+        payload = make_payload()
+        [claim] = await driver.store(
+            make_store_context(target_type(namespace="")), [payload]
+        )
+        digest = hashlib.sha256(payload.SerializeToString()).hexdigest()
+        assert claim.claim_data["key"] == (
+            f"v0/ns/null/{identity}/null/ri/null/d/sha256/{digest}"
+        )
+        assert await driver.retrieve(StorageDriverRetrieveContext(), [claim]) == [
+            payload
+        ]
+
+    @pytest.mark.parametrize(
+        "target_type", [StorageDriverWorkflowInfo, StorageDriverActivityInfo]
+    )
+    @pytest.mark.parametrize("size", [1023, 1024, 1025])
+    @pytest.mark.parametrize("char", ["x", "日"])
+    async def test_key_length_boundary(
+        self,
+        driver_client: S3StorageDriverClient,
+        target_type: type[StorageDriverWorkflowInfo] | type[StorageDriverActivityInfo],
+        size: int,
+        char: str,
+    ) -> None:
+        driver = S3StorageDriver(client=driver_client, bucket=BUCKET)
+        payload = make_payload()
+        digest = hashlib.sha256(payload.SerializeToString()).hexdigest()
+        identity = (
+            "wt/Type/wi" if target_type is StorageDriverWorkflowInfo else "at/Type/ai"
+        )
+        prefix = f"v0/ns/ns/{identity}/"
+        suffix = f"/ri/run/d/sha256/{digest}"
+        remaining = size - len(prefix) - len(suffix)
+        encoded_char = "x" if char == "x" else "%E6%97%A5"
+        count, padding = divmod(remaining, len(encoded_char))
+        target_id = char * count + "x" * padding
+        [claim] = await driver.store(
+            make_store_context(
+                target_type(namespace="ns", type="Type", id=target_id, run_id="run")
+            ),
+            [payload],
+        )
+        assert claim.claim_data["key"] == (
+            f"v0/ns/ns/d/sha256/{digest}"
+            if size > 1024
+            else f"{prefix}{encoded_char * count}{'x' * padding}{suffix}"
+        )
+        assert await driver.retrieve(StorageDriverRetrieveContext(), [claim]) == [
+            payload
+        ]
+
+    @pytest.mark.parametrize(
+        "target_type", [StorageDriverWorkflowInfo, StorageDriverActivityInfo]
+    )
+    async def test_long_keys_preserve_namespace(
+        self,
+        driver_client: S3StorageDriverClient,
+        target_type: type[StorageDriverWorkflowInfo] | type[StorageDriverActivityInfo],
+    ) -> None:
+        driver = S3StorageDriver(client=driver_client, bucket=BUCKET)
+        payload = make_payload()
+        digest = hashlib.sha256(payload.SerializeToString()).hexdigest()
+        claims = []
+        for namespace, encoded in [
+            ("first", "first"),
+            ("second /%", "second%20%2F%25"),
+            ("", "null"),
+        ]:
+            [claim] = await driver.store(
+                make_store_context(target_type(namespace=namespace, id="日" * 110)),
+                [payload],
+            )
+            assert claim.claim_data["key"] == f"v0/ns/{encoded}/d/sha256/{digest}"
+            claims.append(claim)
+        assert len({claim.claim_data["key"] for claim in claims}) == 3
+        assert (
+            await driver.retrieve(StorageDriverRetrieveContext(), claims)
+            == [payload] * 3
+        )
+
     async def test_key_context_none(self, driver_client: S3StorageDriverClient) -> None:
         driver = S3StorageDriver(client=driver_client, bucket=BUCKET)
         payload = make_payload()
@@ -436,6 +530,32 @@ async def test_key_urlencoded_roundtrip(
 
 
 class TestS3StorageDriverStoreRetrieve:
+    @pytest.mark.parametrize("identity", ["wt", "at"])
+    async def test_retrieve_legacy_empty_namespace_key(
+        self,
+        aioboto3_client: S3Client,
+        driver_client: S3StorageDriverClient,
+        identity: str,
+    ) -> None:
+        payload = make_payload()
+        data = payload.SerializeToString()
+        digest = hashlib.sha256(data).hexdigest()
+        identifier = "wi" if identity == "wt" else "ai"
+        key = f"v0/{identity}/null/{identifier}/null/ri/null/d/sha256/{digest}"
+        await aioboto3_client.put_object(Bucket=BUCKET, Key=key, Body=data)
+        claim = StorageDriverClaim(
+            {
+                "bucket": BUCKET,
+                "key": key,
+                "hash_algorithm": "sha256",
+                "hash_value": digest,
+            }
+        )
+        driver = S3StorageDriver(client=driver_client, bucket=BUCKET)
+        assert await driver.retrieve(StorageDriverRetrieveContext(), [claim]) == [
+            payload
+        ]
+
     async def test_store_returns_claim_with_bucket_key_and_hash(
         self, driver_client: S3StorageDriverClient
     ) -> None:
@@ -690,6 +810,19 @@ def capturing_selector(ctx: StorageDriverStoreContext, p: Payload) -> str:
 
 
 class TestS3StorageDriverErrors:
+    async def test_store_access_denied_does_not_upload(self) -> None:
+        aws_client = MagicMock()
+        cause = ClientError(
+            {"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject"
+        )
+        aws_client.head_object = AsyncMock(side_effect=cause)
+        aws_client.upload_fileobj = AsyncMock()
+        driver = S3StorageDriver(_Aioboto3StorageDriverClient(aws_client), BUCKET)
+        with pytest.raises(RuntimeError) as error:
+            await driver.store(make_store_context(), [make_payload()])
+        assert error.value.__cause__ is cause
+        aws_client.upload_fileobj.assert_not_awaited()
+
     async def test_store_nonexistent_bucket_raises(
         self, driver_client: S3StorageDriverClient
     ) -> None: