fix(server): refuse stream ARNs from another account or region

crates/core/src/arn.rs

@@ -14,8 +14,8 @@
 //!
 //! [`parse`] requires the literal `arn:` prefix and five colon-separated fields.
 //! The partition, service, region, and account are captured but not constrained
-//! to any value: a caller that pins them to its deployment does so itself, and a
-//! caller that ignores them (authorization canonicalizes every supplied ARN to
+//! to any value: a caller that pins them to its deployment does so itself
+//! ([`ArnFields::in_deployment`]), and a caller that ignores them (authorization canonicalizes every supplied ARN to
 //! its own deployment) reads only the table and sub-resource.
 
 /// A table sub-resource: a secondary index, a stream, or a backup.
@@ -49,6 +49,18 @@ pub struct TableResource {
     pub sub: Option<SubResource>,
 }
 
+impl ArnFields<'_> {
+    /// True when the partition is `aws`, the service `dynamodb` and the region
+    /// `region`, each compared exactly. Every ARN the server issues carries
+    /// these three values. The server's `StreamArn` scope check and the engine's
+    /// tag `ResourceArn` check both refuse an ARN for which this is false; the
+    /// account and resource are not compared.
+    #[must_use]
+    pub fn in_deployment(&self, region: &str) -> bool {
+        self.partition == "aws" && self.service == "dynamodb" && self.region == region
+    }
+}
+
 /// Split a string into the five ARN fields after the `arn:` prefix.
 ///
 /// The resource field keeps every `:` and `/` it contains, so a stream ARN's
@@ -148,6 +160,34 @@ mod tests {
         assert_eq!(f.resource, "table/T");
     }
 
+    #[test]
+    fn in_deployment_compares_partition_service_and_region_exactly() {
+        let arn = "arn:aws:dynamodb:us-east-1:1:table/T/stream/2026-10-20T16:17:40.188";
+        assert!(fields(arn).unwrap().in_deployment("us-east-1"));
+        // The account and the resource are not compared.
+        assert!(
+            fields("arn:aws:dynamodb:us-east-1:2:table/U")
+                .unwrap()
+                .in_deployment("us-east-1")
+        );
+        for other in [
+            arn.replace("arn:aws:", "arn:aws-cn:"),
+            arn.replace("arn:aws:", "arn:aws-us-gov:"),
+            arn.replace("arn:aws:", "arn::"),
+            arn.replace(":dynamodb:", ":DynamoDB:"),
+            arn.replace(":dynamodb:", ":DYNAMODB:"),
+            arn.replace(":dynamodb:", ":s3:"),
+            arn.replace("us-east-1", "us-west-2"),
+            arn.replace("us-east-1", "US-EAST-1"),
+            arn.replace("us-east-1", ""),
+        ] {
+            assert!(
+                !fields(&other).unwrap().in_deployment("us-east-1"),
+                "{other}"
+            );
+        }
+    }
+
     #[test]
     fn parses_table_index_stream_backup() {
         assert_eq!(

crates/engine/src/streams.rs

@@ -314,8 +314,9 @@ pub async fn handle_get_records(
     // Cassandra after reading the table. A forged token that pairs an
     // authorized stream ARN with another stream's shard id reads nothing and
     // gets the "stream does not exist" answer. A token with no stream ARN gets
-    // the answer the authorization layer gives it (`classify_shard_iterator`);
-    // the layer is skipped in developer mode, so the engine answers the same.
+    // the answer the authorization layer gives it (`classify_shard_iterator`)
+    // in every build, developer mode included; the engine answers the same for
+    // a caller that reaches it without the layer.
     let Some(stream_arn) = it.stream_arn.as_deref() else {
         return Err(no_stream_arn_error(&it));
     };

crates/engine/src/tagging.rs

@@ -34,6 +34,43 @@ fn tag_subresource_not_table_or_stream(resource: &str) -> bool {
     !matches!(segs.next(), None | Some("") | Some("stream"))
 }
 
+/// True for a `table/<name>/stream/<label>` resource.
+fn tag_subresource_is_stream(resource: &str) -> bool {
+    resource
+        .strip_prefix("table/")
+        .is_some_and(|tail| tail.split('/').nth(1) == Some("stream"))
+}
+
+/// The refusal for a tag operation's ARN whose partition, service or region is
+/// not this deployment's, or `None` when all three are. The rule is
+/// [`extenddb_core::arn::ArnFields::in_deployment`], the predicate the server's
+/// `StreamArn` scope check uses.
+///
+/// A stream ARN with another partition, service or region reaches this point
+/// for an authorized principal (the server authorizes a tag operation's stream
+/// ARN as sent, and refuses another account before authorization). The service
+/// answers "Invalid StreamArn" here (measured live 2026-10-23,
+/// ListTagsOfResource and TagResource for an admin and for a principal with
+/// Allow * plus a Deny on the canonical ARN). The service's table message
+/// differs (measured live).
+fn deployment_scope_error(
+    fields: &extenddb_core::arn::ArnFields<'_>,
+    arn: &str,
+    region: &str,
+) -> Option<DynamoDbError> {
+    if fields.in_deployment(region) {
+        return None;
+    }
+    if tag_subresource_is_stream(fields.resource) {
+        return Some(DynamoDbError::ValidationException(
+            "Invalid StreamArn".to_owned(),
+        ));
+    }
+    Some(DynamoDbError::ValidationException(format!(
+        "Invalid TableArn: Invalid ResourceArn provided as input {arn}"
+    )))
+}
+
 /// Extract the table name from a `DynamoDB` table ARN, using the one shared
 /// resource parser ([`extenddb_core::arn::fields`]) so this layer splits an ARN
 /// the same way authorization and the engine stream resolver do.
@@ -60,6 +97,7 @@ fn extract_table_name_from_arn(arn: &str) -> Option<&str> {
 /// | Table ARN naming an index or other sub-resource | `ValidationException` (only table or stream is accepted) |
 /// | Account differs from the caller's | `AccessDeniedException` |
 /// | Table ARN with a different partition, service, or region | `ValidationException` (invalid `TableArn`) |
+/// | Stream ARN with a different partition, service, or region | `ValidationException` (`Invalid StreamArn`) |
 /// | Table does not exist | `ResourceNotFoundException` |
 ///
 /// Account is checked before the partition, service, and region, matching the
@@ -76,25 +114,19 @@ async fn validate_resource_arn(arn: &str, ctx: &OperationContext) -> Result<(),
 
     // A string that is not shaped like an ARN (fewer than the five
     // colon-separated fields) is denied rather than reported as malformed.
-    let Some(extenddb_core::arn::ArnFields {
-        partition,
-        service,
-        region,
-        account,
-        resource,
-    }) = extenddb_core::arn::fields(arn)
-    else {
+    let Some(fields) = extenddb_core::arn::fields(arn) else {
         return Err(DynamoDbError::AccessDeniedException(
             "Access is denied".to_owned(),
         ));
     };
+    let resource = fields.resource;
 
     if !resource.starts_with("table/") {
         // A DynamoDB ARN whose resource is not a table (a bare index, an
         // export, and so on) is reported as not a DynamoDB resource. An ARN for
         // another service is denied rather than reported as malformed, as the
         // service does.
-        if service == "dynamodb" {
+        if fields.service == "dynamodb" {
             return Err(DynamoDbError::ValidationException(format!(
                 "One or more parameter values were invalid: \
                  Provided Arn is not a DynamoDB resource arn: {arn}"
@@ -121,16 +153,14 @@ async fn validate_resource_arn(arn: &str, ctx: &OperationContext) -> Result<(),
         )));
     }
 
-    if account != ctx.account_id.as_ref() {
+    if fields.account != ctx.account_id.as_ref() {
         return Err(DynamoDbError::AccessDeniedException(
             "Access is denied".to_owned(),
         ));
     }
 
-    if partition != "aws" || service != "dynamodb" || region != ctx.region.as_ref() {
-        return Err(DynamoDbError::ValidationException(format!(
-            "Invalid TableArn: Invalid ResourceArn provided as input {arn}"
-        )));
+    if let Some(err) = deployment_scope_error(&fields, arn, &ctx.region) {
+        return Err(err);
     }
 
     let table_name = extract_table_name_from_arn(arn).ok_or_else(|| {
@@ -264,7 +294,50 @@ pub async fn handle_list_tags_of_resource(
 
 #[cfg(test)]
 mod tests {
-    use super::{tag_subresource_not_dynamodb_resource, tag_subresource_not_table_or_stream};
+    use super::{
+        deployment_scope_error, tag_subresource_not_dynamodb_resource,
+        tag_subresource_not_table_or_stream,
+    };
+    use extenddb_core::error::DynamoDbError;
+
+    /// The tag handler refuses exactly the ARNs the shared predicate puts
+    /// outside this deployment, the predicate the server's StreamArn scope
+    /// check uses (`stream_arn_scope_uses_the_shared_predicate` in
+    /// `crates/server/src/authz_resource/stream_scope.rs`), with the
+    /// service's message for a stream and for a table.
+    #[test]
+    fn deployment_scope_uses_the_shared_predicate() {
+        const R: &str = "us-east-1";
+        let stream =
+            "arn:aws:dynamodb:us-east-1:444455556666:table/T/stream/2026-10-20T16:17:40.188";
+        let table = "arn:aws:dynamodb:us-east-1:444455556666:table/T";
+        for base in [stream, table] {
+            for arn in [
+                base.to_owned(),
+                base.replace("arn:aws:", "arn:aws-cn:"),
+                base.replace("arn:aws:", "arn:aws-us-gov:"),
+                base.replace(":dynamodb:", ":DynamoDB:"),
+                base.replace(":dynamodb:", ":DYNAMODB:"),
+                base.replace("us-east-1", "us-west-2"),
+                base.replace("us-east-1", "US-EAST-1"),
+                base.replace("us-east-1", ""),
+            ] {
+                let fields = extenddb_core::arn::fields(&arn).unwrap();
+                let err = deployment_scope_error(&fields, &arn, R);
+                assert_eq!(err.is_none(), fields.in_deployment(R), "{arn}");
+                let expected = if base == stream {
+                    "Invalid StreamArn".to_owned()
+                } else {
+                    format!("Invalid TableArn: Invalid ResourceArn provided as input {arn}")
+                };
+                match err {
+                    None => assert_eq!(arn, base, "only the canonical ARN passes"),
+                    Some(DynamoDbError::ValidationException(m)) => assert_eq!(m, expected),
+                    Some(_) => panic!("{arn}: expected ValidationException"),
+                }
+            }
+        }
+    }
 
     #[test]
     fn tag_subresource_classification() {