diff --git a/crates/core/src/arn.rs b/crates/core/src/arn.rs
index 408d368cc..cdcbf0f2c 100644
--- a/crates/core/src/arn.rs
+++ b/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!(
diff --git a/crates/engine/src/streams.rs b/crates/engine/src/streams.rs
index c8ba419bc..e57328de0 100755
--- a/crates/engine/src/streams.rs
+++ b/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));
     };
diff --git a/crates/engine/src/tagging.rs b/crates/engine/src/tagging.rs
index 432d75dd4..d9b674386 100755
--- a/crates/engine/src/tagging.rs
+++ b/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:964157134968:table/T/stream/2026-10-20T16:17:40.188";
+        let table = "arn:aws:dynamodb:us-east-1:964157134968: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() {
diff --git a/crates/server/src/authz_resource.rs b/crates/server/src/authz_resource.rs
index a4439dff1..09fde7c11 100644
--- a/crates/server/src/authz_resource.rs
+++ b/crates/server/src/authz_resource.rs
@@ -29,14 +29,40 @@
 //! | `ListTables` | `ListTables` | `table/*` |
 //! | item/table ops | same | the table ARN (from `TableName`, name or ARN) |
 //!
-//! Every resource ARN produced here is CANONICAL for this deployment: the
-//! partition, service, region, and account are fixed to this server's, and only
-//! the resource identity (table name and optional sub-resource) is taken from
-//! the request. ExtendDB's stream and backup resolvers ignore the partition,
-//! service, region, and account of a supplied ARN and act within this
-//! deployment's account, so a caller who respells those fields would otherwise
-//! evade a Deny written against the canonical ARN. Canonicalizing here closes
-//! that bypass: a Deny on the real resource always matches.
+//! Every resource ARN produced here, except a tag operation's stream ARN, is
+//! CANONICAL for this deployment: the partition, service, region, and account
+//! are fixed to this server's, and only the resource identity (table name and
+//! optional sub-resource) is taken from the request. ExtendDB's stream and
+//! backup resolvers ignore the partition, service, region, and account of a
+//! supplied ARN and act within this deployment's account, so a caller who
+//! respells those fields would otherwise evade a Deny written against the
+//! canonical ARN. Canonicalizing here closes that bypass: a Deny on the real
+//! resource matches every canonical ARN. A tag operation's stream ARN is
+//! authorized as sent, so a Deny on the real stream ARN does not match a
+//! respelled one; the engine's `validate_resource_arn` refuses it after the IAM
+//! decision, before any tag row is read or written, unless its partition,
+//! service, region and account are this deployment's.
+//!
+//! A stream ARN is held to a stricter rule than canonicalization (issues #450
+//! and #451; measured live 2026-10-23 and 2026-10-28, tabled in the module
+//! docstring of `tests/test_stream_arn_scope.py`). Its account must be the
+//! caller's: a well-formed stream ARN in another account is
+//! `AccessDeniedException` "Access is denied" before the IAM decision, on
+//! `StreamArn` and on a tag operation's `ResourceArn`
+//! ([`check_stream_arn_account`]). On `StreamArn` the shape step runs first, so
+//! a value that is not a stream ARN, or has an empty partition, service or
+//! account field, is `Invalid StreamArn` in any account. Its partition, service
+//! and region must be this deployment's: on `StreamArn` (DescribeStream,
+//! GetShardIterator) another spelling is refused before the IAM decision with
+//! `Invalid StreamArn` ([`check_stream_arn_scope`]); on a tag operation's
+//! `ResourceArn` the IAM decision runs first, on the ARN as sent, and the
+//! engine refuses the spelling afterwards with the same message. A `GetRecords`
+//! iterator whose stream ARN is out of scope in any field, the account
+//! included, is refused before the IAM decision with ValidationException
+//! `Invalid ShardIterator` ([`getrecords_resource`]); GetShardIterator issues
+//! no such iterator. So a respelled stream ARN is never served as the
+//! caller's own stream, and a stream ARN that reaches the engine spells the
+//! stream the engine reads.
 //!
 //! Fail-closed: a missing or unparseable resource field never falls back to
 //! `table/*`. It yields the field value verbatim (often empty), which an
@@ -50,6 +76,10 @@ use serde_json::Value;
 const PARTITION: &str = "aws";
 const SERVICE: &str = "dynamodb";
 
+/// Scope check for stream ARNs (issues #450 and #451).
+mod stream_scope;
+use stream_scope::{check_stream_arn_account, check_stream_arn_scope, stream_arn_in_scope};
+
 /// The item actions the service requires on the target table of a restore, in
 /// addition to the restore action itself (measured live). A Deny on any of
 /// these for the target table name blocks a restore into it.
@@ -147,34 +177,27 @@ fn backup_arn_field(input: &Value) -> String {
 /// The resource ARN a tag operation authorizes against.
 ///
 /// `validate_tag_resource_arn_shape` has already run: the value is a table ARN,
-/// a stream ARN, or a foreign-service ARN (an index, a backup, or a non-table
-/// `DynamoDB` ARN is rejected before authorization). A table or stream ARN is
-/// canonicalized to this deployment; a foreign-service ARN is left verbatim so
-/// authorization denies it, matching the service (measured, arn-1 table).
+/// a stream ARN in the caller's account, or a foreign-service ARN (an index, a
+/// backup, a non-table `DynamoDB` ARN, or a stream ARN in another account is
+/// rejected before authorization). A table ARN is canonicalized to this
+/// deployment. A stream ARN is authorized as sent: the service evaluates the
+/// IAM decision on the spelling the caller gave (a no-permissions principal's
+/// denial names `arn:aws-cn:...` as sent, and a principal allowed the exact
+/// canonical ARN is denied the `aws-cn` spelling; measured 2026-10-23, module
+/// docstring of `tests/test_stream_arn_scope.py`), and the engine refuses
+/// another partition, service or region after authorization with
+/// `Invalid StreamArn`. A foreign-service ARN is left
+/// verbatim so authorization denies it, matching the service (measured, arn-1
+/// table).
 fn tag_resource_arn(arn: &str, region: &str, account_id: &str) -> String {
     match parse_resource_identity(arn) {
-        Some(rid @ ResourceId { sub: None, .. })
-        | Some(
-            rid @ ResourceId {
-                sub: Some(Sub::Stream(_)),
-                ..
-            },
-        ) => canonical(region, account_id, &rid),
-        // A foreign-service ARN (or, defensively, an index/backup the shape
-        // check already rejected): authorization denies it.
+        Some(rid @ ResourceId { sub: None, .. }) => canonical(region, account_id, &rid),
+        // A stream ARN (authorized as sent), a foreign-service ARN, or,
+        // defensively, an index/backup the shape check already rejected.
         _ => arn.to_owned(),
     }
 }
 
-/// The resource kind a strictly validated ARN-bearing request field must name.
-/// Only stream and backup fields are validated before authorization; an export
-/// `TableArn` is checked only for parseability (see `validate_operation_arns`).
-#[derive(Clone, Copy)]
-enum ArnKind {
-    Stream,
-    Backup,
-}
-
 /// Messages DynamoDB returns, before the IAM decision, for a request field
 /// whose ARN is not a valid resource ARN of the kind the field requires
 /// (measured live, `live-r2/arn-kinds-live.json`).
@@ -183,11 +206,11 @@ const INVALID_BACKUP_ARN_MESSAGE: &str = "Invalid Backup ARN";
 const INVALID_EXPORT_TABLE_ARN_MESSAGE: &str =
     "One or more parameter values were invalid: tableArn is not a valid ARN";
 
-/// True when `arn` is a well-formed DynamoDB resource ARN of `kind`: it parses
-/// to that kind and its partition, service, region, and account are all
-/// non-empty. The field checks catch an empty-partition ARN
-/// (`arn::dynamodb:...`), which the service rejects with `Invalid StreamArn`.
-fn arn_is_kind(arn: &str, kind: ArnKind) -> bool {
+/// True when `arn` is a well-formed DynamoDB backup ARN: it parses to a backup
+/// and its partition, service, region, and account are all non-empty. A
+/// `StreamArn` takes [`check_stream_arn_scope`] instead, which runs the account
+/// check between its field checks in the order the service does.
+fn arn_is_backup(arn: &str) -> bool {
     let Some(f) = extenddb_core::arn::fields(arn) else {
         return false;
     };
@@ -196,25 +219,18 @@ fn arn_is_kind(arn: &str, kind: ArnKind) -> bool {
         return false;
     }
     matches!(
-        (parse_resource_identity(arn), kind),
-        (
-            Some(ResourceId {
-                sub: Some(Sub::Stream(_)),
-                ..
-            }),
-            ArnKind::Stream
-        ) | (
-            Some(ResourceId {
-                sub: Some(Sub::Backup(_)),
-                ..
-            }),
-            ArnKind::Backup
-        )
+        parse_resource_identity(arn),
+        Some(ResourceId {
+            sub: Some(Sub::Backup(_)),
+            ..
+        })
     )
 }
 
 /// Validate ARN-bearing fields before the IAM decision, matching the service
-/// (measured live, `live-r2/arn-kinds-live.json`).
+/// (measured live, `live-r2/arn-kinds-live.json`; for `StreamArn`, the tables
+/// in the module docstring of `tests/test_stream_arn_scope.py`, measured
+/// 2026-10-23 and 2026-10-28).
 ///
 /// The two field families differ, as the service does:
 ///
@@ -222,7 +238,12 @@ fn arn_is_kind(arn: &str, kind: ArnKind) -> bool {
 ///   (DescribeBackup, DeleteBackup) are validated strictly: a value that is not
 ///   a well-formed stream (resp. backup) ARN of the correct kind, including a
 ///   table ARN, an `xrn:` ARN, or an empty partition, is refused before
-///   authorization with `Invalid StreamArn` (resp. `Invalid Backup ARN`).
+///   authorization with `Invalid StreamArn` (resp. `Invalid Backup ARN`). A
+///   stream ARN is checked by [`check_stream_arn_scope`] in the service's
+///   order: its shape, then the account (another account is
+///   `AccessDeniedException` whatever the region field holds), then the
+///   partition, service and region values (another value, or an empty region,
+///   is `Invalid StreamArn`).
 /// - An export `TableArn` is validated loosely: only an UNPARSEABLE value (a
 ///   repeated `table/` segment such as `.../table/a/stream/table/b`, a bare
 ///   `table/x`, or an `xrn:` ARN) is refused before authorization, with
@@ -237,7 +258,12 @@ fn arn_is_kind(arn: &str, kind: ArnKind) -> bool {
 /// principal is denied) and the engine refuses it afterwards. Tag operations
 /// keep their own measured shape check (`validate_tag_resource_arn_shape`);
 /// `GetRecords` is handled through its iterator by [`getrecords_resource`].
-pub(crate) fn validate_operation_arns(operation: &str, input: &Value) -> Result<(), DynamoDbError> {
+pub(crate) fn validate_operation_arns(
+    operation: &str,
+    input: &Value,
+    region: &str,
+    account_id: &str,
+) -> Result<(), DynamoDbError> {
     let present = |key: &str| {
         input
             .get(key)
@@ -246,20 +272,13 @@ pub(crate) fn validate_operation_arns(operation: &str, input: &Value) -> Result<
     };
     match operation {
         "DescribeStream" | "GetShardIterator" => {
-            if let Some(arn) = present("StreamArn")
-                && !arn_is_kind(arn, ArnKind::Stream)
-            {
-                return Err(DynamoDbError::ValidationException(
-                    INVALID_STREAM_ARN_MESSAGE.to_owned(),
-                ));
+            if let Some(arn) = present("StreamArn") {
+                check_stream_arn_scope(arn, region, account_id)?;
             }
         }
         "DescribeBackup" | "DeleteBackup" => {
             if let Some(arn) = present("BackupArn")
-                && !arn_is_kind(
-                    extenddb_core::backup_format::trim_backup_arn(arn),
-                    ArnKind::Backup,
-                )
+                && !arn_is_backup(extenddb_core::backup_format::trim_backup_arn(arn))
             {
                 return Err(DynamoDbError::ValidationException(
                     INVALID_BACKUP_ARN_MESSAGE.to_owned(),
@@ -337,9 +356,16 @@ pub(crate) enum GetRecordsResource {
 ///
 /// The iterator is decoded once by the engine
 /// ([`extenddb_engine::classify_shard_iterator`]). A current token's stream ARN
-/// is canonicalized through the shared parser; a decoded stream ARN that does
-/// not parse to a stream is refused as Invalid rather than authorized verbatim,
-/// so no respelled or forged stream ARN is served.
+/// must parse to a stream and name this deployment's partition, service and
+/// region and the caller's account, or the token is refused as Invalid rather
+/// than canonicalized and served as the caller's own stream. GetShardIterator
+/// encodes the `StreamArn` it is sent and refuses one outside that scope, so
+/// every iterator issued from the canonical stream ARN passes. An iterator an
+/// earlier build issued from a respelled `StreamArn` (`aws-cn`, `US-EAST-1`,
+/// `DYNAMODB`), every iterator chained from it, and every iterator issued
+/// before a change of `server.region` are refused as Invalid. The service
+/// answers a token it did not issue with `Invalid ShardIterator` (measured
+/// 2026-10-22, `live-r2/iter-live.json`).
 pub(crate) fn getrecords_resource(
     input: &Value,
     region: &str,
@@ -356,7 +382,9 @@ pub(crate) fn getrecords_resource(
                     sub: Some(Sub::Stream(_)),
                     ..
                 },
-            ) => GetRecordsResource::Stream(canonical(region, account_id, &rid)),
+            ) if stream_arn_in_scope(&arn, region, account_id) => {
+                GetRecordsResource::Stream(canonical(region, account_id, &rid))
+            }
             _ => GetRecordsResource::Invalid,
         },
         extenddb_engine::ShardIteratorClass::Expired => GetRecordsResource::Expired,
@@ -401,23 +429,37 @@ fn backup_wildcard(region: &str, account_id: &str) -> String {
 /// as the service does (measured live). The service returns ValidationException
 /// to every principal, before the IAM decision, for an ARN that does not start
 /// with `arn:`, a DynamoDB ARN whose resource is not a table or stream, or an
-/// index ARN (tags are accepted only on a table or stream). A table or stream
-/// ARN passes; so does a foreign-service ARN (for example an S3 bucket), which
+/// index ARN (tags are accepted only on a table or stream). A table ARN passes;
+/// so does a foreign-service ARN (for example an S3 bucket), which
 /// authorization then denies. The partition, service, region, account, and
 /// existence of a table ARN are checked after authorization by the engine
 /// handler, so for those a no-permissions principal is denied first and only an
 /// authorized principal sees the shape error.
-pub(crate) fn validate_tag_resource_arn_shape(arn: &str) -> Result<(), DynamoDbError> {
+///
+/// A stream ARN's account is also checked here ([`check_stream_arn_account`]):
+/// the service answers `AccessDeniedException` "Access is denied" for a stream
+/// ARN in another account on ListTagsOfResource and TagResource, to every
+/// principal, before the IAM decision (measured 2026-10-23, module docstring
+/// of `tests/test_stream_arn_scope.py`). Its
+/// partition, service and region are not checked here: the service runs the
+/// IAM decision on the ARN as sent and refuses `aws-cn`, `DYNAMODB` or
+/// `us-west-2` afterwards with `Invalid StreamArn`, which the engine's
+/// `validate_resource_arn` answers for an authorized principal.
+pub(crate) fn validate_tag_resource_arn_shape(
+    arn: &str,
+    account_id: &str,
+) -> Result<(), DynamoDbError> {
     if !arn.starts_with("arn:") {
         return Err(DynamoDbError::ValidationException(format!(
             "One or more parameter values were invalid: ARNs must start with 'arn:': {arn}"
         )));
     }
     match parse_resource_identity(arn) {
+        Some(ResourceId { sub: None, .. }) => Ok(()),
         Some(ResourceId {
-            sub: None | Some(Sub::Stream(_)),
+            sub: Some(Sub::Stream(_)),
             ..
-        }) => Ok(()),
+        }) => check_stream_arn_account(arn, account_id),
         // A backup ARN is not a taggable resource kind (measured live, with a
         // real backup ARN and a short backup id).
         Some(ResourceId {
@@ -680,7 +722,9 @@ mod tests {
             format!("\t{arn}\r\n"),
         ] {
             for op in ["DescribeBackup", "DeleteBackup"] {
-                assert!(validate_operation_arns(op, &json!({ "BackupArn": spelled })).is_ok());
+                assert!(
+                    validate_operation_arns(op, &json!({ "BackupArn": spelled }), R, A).is_ok()
+                );
                 assert_eq!(
                     only(op, json!({ "BackupArn": spelled })),
                     arn,
@@ -761,26 +805,6 @@ mod tests {
         );
     }
 
-    // -- canonicalization closes the respelling bypass --
-
-    #[test]
-    fn respelled_partition_service_region_account_canonicalize_to_same_arn() {
-        let canonical =
-            "arn:aws:dynamodb:us-east-1:964157134968:table/T/stream/2026-10-20T16:17:40.188";
-        for respelled in [
-            "arn:aws-cn:dynamodb:us-east-1:964157134968:table/T/stream/2026-10-20T16:17:40.188",
-            "arn:aws:s3:us-east-1:964157134968:table/T/stream/2026-10-20T16:17:40.188",
-            "arn:aws:dynamodb:us-west-2:964157134968:table/T/stream/2026-10-20T16:17:40.188",
-            "arn:aws:dynamodb:us-east-1:111122223333:table/T/stream/2026-10-20T16:17:40.188",
-        ] {
-            assert_eq!(
-                only("DescribeStream", json!({ "StreamArn": respelled })),
-                canonical,
-                "respelled {respelled} must canonicalize"
-            );
-        }
-    }
-
     // -- per-operation resource derivation (matches live DynamoDB) --
 
     #[test]
@@ -968,7 +992,7 @@ mod tests {
         // Both measured live.
         let index = "arn:aws:dynamodb:us-east-1:964157134968:table/bb/index/gsi1";
         assert!(matches!(
-            validate_tag_resource_arn_shape(index),
+            validate_tag_resource_arn_shape(index, A),
             Err(DynamoDbError::ValidationException(ref m))
                 if m.contains("only table or stream is accepted") && m.contains(index)
         ));
@@ -978,7 +1002,7 @@ mod tests {
         ] {
             assert!(
                 matches!(
-                    validate_tag_resource_arn_shape(backup),
+                    validate_tag_resource_arn_shape(backup, A),
                     Err(DynamoDbError::ValidationException(ref m))
                         if m == &format!(
                             "One or more parameter values were invalid: Provided Arn is not a \
@@ -990,12 +1014,13 @@ mod tests {
         }
         // A table ARN and a stream ARN pass the shape check.
         assert!(
-            validate_tag_resource_arn_shape("arn:aws:dynamodb:us-east-1:964157134968:table/bb")
+            validate_tag_resource_arn_shape("arn:aws:dynamodb:us-east-1:964157134968:table/bb", A)
                 .is_ok()
         );
         assert!(
             validate_tag_resource_arn_shape(
-                "arn:aws:dynamodb:us-east-1:964157134968:table/bb/stream/2026-10-20T16:17:40.188"
+                "arn:aws:dynamodb:us-east-1:964157134968:table/bb/stream/2026-10-20T16:17:40.188",
+                A
             )
             .is_ok()
         );
@@ -1005,14 +1030,14 @@ mod tests {
     fn tag_resource_arn_shape_matches_service_ordering() {
         // Measured live. ValidationException before authorization, every
         // principal:
-        let not_arn = validate_tag_resource_arn_shape("banana");
+        let not_arn = validate_tag_resource_arn_shape("banana", A);
         assert!(matches!(
             not_arn,
             Err(DynamoDbError::ValidationException(ref m))
                 if m.contains("ARNs must start with 'arn:': banana")
         ));
         let bad_type =
-            validate_tag_resource_arn_shape("arn:aws:dynamodb:us-east-1:964157134968:banana/x");
+            validate_tag_resource_arn_shape("arn:aws:dynamodb:us-east-1:964157134968:banana/x", A);
         assert!(matches!(
             bad_type,
             Err(DynamoDbError::ValidationException(ref m))
@@ -1020,16 +1045,20 @@ mod tests {
         ));
         // A foreign-service ARN whose resource is not a table passes the shape
         // check and is left to authorization (which denies it).
-        assert!(validate_tag_resource_arn_shape("arn:aws:s3:::mybucket").is_ok());
+        assert!(validate_tag_resource_arn_shape("arn:aws:s3:::mybucket", A).is_ok());
         // A table ARN whose partition, service, or region is this deployment's
         // or not passes the shape check; the engine handler validates those
         // fields after authorization.
         assert!(
-            validate_tag_resource_arn_shape("arn:aws-cn:dynamodb:us-east-1:964157134968:table/bb")
-                .is_ok()
+            validate_tag_resource_arn_shape(
+                "arn:aws-cn:dynamodb:us-east-1:964157134968:table/bb",
+                A
+            )
+            .is_ok()
         );
         assert!(
-            validate_tag_resource_arn_shape("arn:aws:s3:us-east-1:964157134968:table/bb").is_ok()
+            validate_tag_resource_arn_shape("arn:aws:s3:us-east-1:964157134968:table/bb", A)
+                .is_ok()
         );
     }
 
@@ -1163,17 +1192,17 @@ mod tests {
         let a1 = "arn:aws:dynamodb:us-east-1:1:table/allowed/stream/table/secret";
         assert!(arn::parse(a1).is_none());
         assert!(matches!(
-            validate_operation_arns("ExportTableToPointInTime", &json!({ "TableArn": a1 })),
+            validate_operation_arns("ExportTableToPointInTime", &json!({ "TableArn": a1 }), R, A),
             Err(DynamoDbError::ValidationException(ref m)) if m == INVALID_EXPORT_TABLE_ARN_MESSAGE
         ));
     }
 
     #[test]
     fn validate_operation_arns_matches_measured_live_behaviour() {
-        let err_msg = |op: &str, input: Value| match validate_operation_arns(op, &input) {
+        let err_msg = |op: &str, input: Value| match validate_operation_arns(op, &input, R, A) {
             Err(DynamoDbError::ValidationException(m)) => Some(m),
             Ok(()) => None,
-            other => panic!("unexpected {other:?}"),
+            Err(_) => panic!("unexpected non-validation error for {op}"),
         };
 
         // Export: only an UNPARSEABLE TableArn is refused before authz, with the
@@ -1221,10 +1250,11 @@ mod tests {
                 );
             }
         }
+        // A stream ARN in this deployment and the caller's account passes.
         assert_eq!(
             err_msg(
                 "DescribeStream",
-                json!({ "StreamArn": "arn:aws:dynamodb:us-east-1:1:table/t/stream/2026-10-20T16:17:40.188" })
+                json!({ "StreamArn": format!("arn:aws:dynamodb:{R}:{A}:table/t/stream/2026-10-20T16:17:40.188") })
             ),
             None
         );
@@ -1288,8 +1318,12 @@ mod tests {
 
     #[test]
     fn getrecords_resource_classifies_and_canonicalizes() {
-        // A current token authorizes against its canonical stream ARN.
-        let stream = "arn:aws:dynamodb:us-east-1:1:table/T/stream/2026-10-20T16:17:40.188";
+        // A current token issued by this server authorizes against the stream
+        // ARN it carries (a token naming any other account, partition, service
+        // or region is Invalid; see
+        // `getrecords_iterator_with_respelled_stream_arn_is_invalid`).
+        let stream =
+            "arn:aws:dynamodb:us-east-1:964157134968:table/T/stream/2026-10-20T16:17:40.188";
         let tok = extenddb_engine::encode_shard_iterator(
             "shardId-1",
             "AFTER_SEQUENCE_NUMBER",
@@ -1298,10 +1332,7 @@ mod tests {
             stream,
         );
         match getrecords_resource(&json!({ "ShardIterator": tok }), R, A) {
-            GetRecordsResource::Stream(arn) => assert_eq!(
-                arn,
-                "arn:aws:dynamodb:us-east-1:964157134968:table/T/stream/2026-10-20T16:17:40.188"
-            ),
+            GetRecordsResource::Stream(arn) => assert_eq!(arn, stream),
             _ => panic!("expected Stream"),
         }
         // A token that carries no stream ARN but a parseable timestamp is the
diff --git a/crates/server/src/authz_resource/stream_scope.rs b/crates/server/src/authz_resource/stream_scope.rs
new file mode 100644
index 000000000..d2a108315
--- /dev/null
+++ b/crates/server/src/authz_resource/stream_scope.rs
@@ -0,0 +1,348 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Scope check for stream ARNs (issues #450 and #451): the account must be the
+//! caller's, and the partition, service and region this deployment's. See the
+//! parent module docs for where each check runs relative to the IAM decision.
+
+use extenddb_core::error::DynamoDbError;
+
+use super::{INVALID_STREAM_ARN_MESSAGE, ResourceId, Sub, parse_resource_identity};
+
+/// Refuse a stream ARN whose account is not the caller's with
+/// `AccessDeniedException` "Access is denied". The service answers this before
+/// the IAM decision, for every principal, on `StreamArn` (DescribeStream,
+/// GetShardIterator) and on a tag operation's `ResourceArn` alike, and before
+/// it compares the partition, region or label (measured live 2026-10-23, in
+/// the module docstring of `tests/test_stream_arn_scope.py`: another 12-digit
+/// account, `abc`, and another account combined with `aws-cn`, `us-west-2` or
+/// an unknown label). On `StreamArn` an empty partition is refused before this
+/// check ([`check_stream_arn_scope`]).
+pub(super) fn check_stream_arn_account(arn: &str, account_id: &str) -> Result<(), DynamoDbError> {
+    let Some(f) = extenddb_core::arn::fields(arn) else {
+        return Err(DynamoDbError::ValidationException(
+            INVALID_STREAM_ARN_MESSAGE.to_owned(),
+        ));
+    };
+    if f.account != account_id {
+        return Err(DynamoDbError::AccessDeniedException(
+            "Access is denied".to_owned(),
+        ));
+    }
+    Ok(())
+}
+
+/// Refuse a `StreamArn` that is not a stream ARN in this deployment's scope,
+/// before the IAM decision, as the service does for DescribeStream and
+/// GetShardIterator (measured live 2026-10-23, identical for an admin, a
+/// no-permissions user, a user allowed the exact stream ARN and a user with a
+/// Deny on it). The checks run in the service's order, measured as an admin on
+/// DescribeStream and GetShardIterator on 2026-10-28; both measurements are
+/// tabled in the module docstring of `tests/test_stream_arn_scope.py`:
+///
+/// 1. not a stream ARN (a table ARN, an extra or missing field, a trailing
+///    slash), or an empty partition, service or account field:
+///    `ValidationException` `Invalid StreamArn`, for any account;
+/// 2. an account other than the caller's: `AccessDeniedException`
+///    ([`check_stream_arn_account`]), whatever the region field holds;
+/// 3. a partition other than `aws` (`aws-cn`, `aws-us-gov`), a service other
+///    than `dynamodb` (`DynamoDB`, `DYNAMODB`), or a region other than this
+///    server's (empty, `us-west-2`, `US-EAST-1`): `ValidationException`
+///    `Invalid StreamArn`. This is the rule the engine applies to a tag
+///    operation's `ResourceArn`, through the one predicate
+///    [`extenddb_core::arn::ArnFields::in_deployment`].
+///
+/// The order was measured on DescribeStream from 20:47 to 20:48 UTC and on
+/// GetShardIterator at 21:51 UTC. An empty service field, measured on both
+/// operations at 21:51 UTC, is step 1's refusal in another account too.
+///
+/// The stream engine resolves a stream by table name and label within the
+/// caller's account, so a stream ARN that passes here names exactly the stream
+/// the engine reads, and authorization evaluates that same ARN. Before this
+/// check existed, a respelled ARN was canonicalized for authorization and then
+/// served as the caller's own stream (issues #450 and #451).
+///
+/// A tag operation's `ResourceArn` does not take this check: the service runs
+/// the IAM decision on the ARN as sent and refuses the partition, service and
+/// region after it (`validate_resource_arn` in the engine answers
+/// `Invalid StreamArn` there).
+pub(super) fn check_stream_arn_scope(
+    arn: &str,
+    region: &str,
+    account_id: &str,
+) -> Result<(), DynamoDbError> {
+    let invalid = || {
+        Err(DynamoDbError::ValidationException(
+            INVALID_STREAM_ARN_MESSAGE.to_owned(),
+        ))
+    };
+    let is_stream = matches!(
+        parse_resource_identity(arn),
+        Some(ResourceId {
+            sub: Some(Sub::Stream(_)),
+            ..
+        })
+    );
+    let Some(f) = extenddb_core::arn::fields(arn).filter(|_| is_stream) else {
+        return invalid();
+    };
+    if f.partition.is_empty() || f.service.is_empty() || f.account.is_empty() {
+        return invalid();
+    }
+    check_stream_arn_account(arn, account_id)?;
+    if !f.in_deployment(region) {
+        return invalid();
+    }
+    Ok(())
+}
+
+/// True when every field of this stream ARN is this deployment's and the
+/// caller's. GetShardIterator refuses a `StreamArn` for which this is false, so
+/// a shard iterator carrying one was hand-built, or issued by an earlier build
+/// or under another `server.region` ([`super::getrecords_resource`]).
+pub(super) fn stream_arn_in_scope(arn: &str, region: &str, account_id: &str) -> bool {
+    check_stream_arn_scope(arn, region, account_id).is_ok()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::{
+        GetRecordsResource, authz_targets, getrecords_resource, validate_operation_arns,
+        validate_tag_resource_arn_shape,
+    };
+    use super::*;
+    use serde_json::{Value, json};
+
+    const R: &str = "us-east-1";
+    const A: &str = "964157134968";
+
+    const STREAM: &str =
+        "arn:aws:dynamodb:us-east-1:964157134968:table/T/stream/2026-10-20T16:17:40.188";
+
+    fn only(op: &str, input: Value) -> String {
+        let t = authz_targets(op, &input, R, A);
+        assert_eq!(t.len(), 1, "expected one target for {op}: {t:?}");
+        t[0].1.clone()
+    }
+
+    fn stream_arn_check(op: &str, arn: &str) -> Result<(), DynamoDbError> {
+        validate_operation_arns(op, &json!({ "StreamArn": arn }), R, A)
+    }
+
+    /// Names the result's variant for a failure message.
+    fn variant(r: &Result<(), DynamoDbError>) -> &'static str {
+        match r {
+            Ok(()) => "Ok",
+            Err(DynamoDbError::ValidationException(_)) => "ValidationException",
+            Err(DynamoDbError::AccessDeniedException(_)) => "AccessDeniedException",
+            Err(_) => "another error",
+        }
+    }
+
+    fn assert_invalid_stream_arn(r: Result<(), DynamoDbError>, what: &str, arn: &str) {
+        match r {
+            Err(DynamoDbError::ValidationException(m)) => {
+                assert_eq!(m, "Invalid StreamArn", "{what} {arn}");
+            }
+            other => panic!(
+                "{what} {arn}: expected Invalid StreamArn, got {}",
+                variant(&other)
+            ),
+        }
+    }
+
+    fn assert_access_denied(r: Result<(), DynamoDbError>, what: &str, arn: &str) {
+        match r {
+            Err(DynamoDbError::AccessDeniedException(m)) => {
+                assert_eq!(m, "Access is denied", "{what} {arn}");
+            }
+            other => panic!(
+                "{what} {arn}: expected Access is denied, got {}",
+                variant(&other)
+            ),
+        }
+    }
+
+    fn respellings() -> [String; 7] {
+        [
+            STREAM.replace("arn:aws:", "arn:aws-cn:"),
+            STREAM.replace("arn:aws:", "arn:aws-us-gov:"),
+            STREAM.replace(":dynamodb:", ":DYNAMODB:"),
+            STREAM.replace(":dynamodb:", ":DynamoDB:"),
+            STREAM.replace(":dynamodb:", ":s3:"),
+            STREAM.replace("us-east-1", "us-west-2"),
+            STREAM.replace("us-east-1", "US-EAST-1"),
+        ]
+    }
+
+    type Check = fn(Result<(), DynamoDbError>, &str, &str);
+
+    /// The nine StreamArn cells measured live on 2026-10-28 (the order table in
+    /// the module docstring of `tests/test_stream_arn_scope.py`), each with the
+    /// answer it expects: an empty partition or service is "Invalid StreamArn"
+    /// even in another account, another account is "Access is denied" whatever
+    /// the region field holds, and the caller's account with an empty or
+    /// another region is "Invalid StreamArn".
+    fn measured_order() -> [(String, Check); 9] {
+        let other = |arn: String| arn.replace(A, "111111111111");
+        let no_partition = STREAM.replace("arn:aws:", "arn::");
+        let no_service = STREAM.replace(":dynamodb:", "::");
+        let no_region = STREAM.replace("us-east-1", "");
+        let west = STREAM.replace("us-east-1", "us-west-2");
+        [
+            (other(no_partition.clone()), assert_invalid_stream_arn),
+            (no_partition, assert_invalid_stream_arn),
+            (other(no_service.clone()), assert_invalid_stream_arn),
+            (no_service, assert_invalid_stream_arn),
+            (other(no_region.clone()), assert_access_denied),
+            (other(west.clone()), assert_access_denied),
+            (other(STREAM.to_owned()), assert_access_denied),
+            (no_region, assert_invalid_stream_arn),
+            (west, assert_invalid_stream_arn),
+        ]
+    }
+
+    /// The StreamArn checks run in the measured order on DescribeStream and
+    /// GetShardIterator: an empty partition or service field, then the
+    /// account, then the region (measured on both operations on 2026-10-28,
+    /// GetShardIterator and the empty service at 21:51 UTC).
+    #[test]
+    fn stream_arn_checks_follow_the_measured_order() {
+        for (arn, check) in measured_order() {
+            for op in ["DescribeStream", "GetShardIterator"] {
+                check(stream_arn_check(op, &arn), op, &arn);
+            }
+        }
+    }
+
+    /// The service answers a stream ARN in another account with
+    /// AccessDeniedException "Access is denied" before the IAM decision, on
+    /// StreamArn and on a tag ResourceArn (measured 2026-10-23).
+    #[test]
+    fn stream_arn_in_another_account_is_denied_before_authz() {
+        for respelled in [
+            STREAM.replace(A, "111111111111"),
+            STREAM.replace(A, "111122223333"),
+            STREAM.replace(A, "abc"),
+        ] {
+            for op in ["DescribeStream", "GetShardIterator"] {
+                assert_access_denied(stream_arn_check(op, &respelled), op, &respelled);
+            }
+            assert_access_denied(
+                validate_tag_resource_arn_shape(&respelled, A),
+                "tag",
+                &respelled,
+            );
+        }
+        // An empty account is "Invalid StreamArn" on StreamArn (the kind check
+        // requires every field) and "Access is denied" on a tag ResourceArn.
+        let empty = STREAM.replace(A, "");
+        for op in ["DescribeStream", "GetShardIterator"] {
+            assert_invalid_stream_arn(stream_arn_check(op, &empty), op, &empty);
+        }
+        assert_access_denied(validate_tag_resource_arn_shape(&empty, A), "tag", &empty);
+    }
+
+    /// The service answers another partition, region or service spelling of a
+    /// StreamArn with ValidationException "Invalid StreamArn" before the IAM
+    /// decision (measured 2026-10-23 for aws-cn, aws-us-gov, DynamoDB,
+    /// DYNAMODB, us-west-2 and US-EAST-1). On a tag ResourceArn the same
+    /// spellings pass the pre-authorization check and are authorized as sent;
+    /// the engine refuses them after the IAM decision.
+    #[test]
+    fn stream_arn_with_other_partition_region_or_service_is_invalid() {
+        for respelled in &respellings() {
+            for op in ["DescribeStream", "GetShardIterator"] {
+                assert_invalid_stream_arn(stream_arn_check(op, respelled), op, respelled);
+            }
+            assert!(
+                validate_tag_resource_arn_shape(respelled, A).is_ok(),
+                "tag {respelled}"
+            );
+            assert_eq!(
+                only("TagResource", json!({ "ResourceArn": respelled })),
+                *respelled,
+                "a tag ResourceArn is authorized as sent"
+            );
+        }
+        // The canonical spelling passes both checks and authorizes verbatim.
+        assert!(stream_arn_check("DescribeStream", STREAM).is_ok());
+        assert!(validate_tag_resource_arn_shape(STREAM, A).is_ok());
+        assert_eq!(
+            only("DescribeStream", json!({ "StreamArn": STREAM })),
+            STREAM
+        );
+        assert_eq!(
+            only("TagResource", json!({ "ResourceArn": STREAM })),
+            STREAM
+        );
+    }
+
+    /// In the caller's account, the StreamArn check refuses exactly the stream
+    /// ARNs the shared predicate puts outside this deployment, the predicate the
+    /// engine tag handler applies to a tag ResourceArn
+    /// (`deployment_scope_error` in `crates/engine/src/tagging.rs`).
+    #[test]
+    fn stream_arn_scope_uses_the_shared_predicate() {
+        let mut arns = respellings().to_vec();
+        arns.push(STREAM.replace("us-east-1", ""));
+        arns.push(STREAM.to_owned());
+        for arn in &arns {
+            let f = extenddb_core::arn::fields(arn).unwrap();
+            assert_eq!(
+                check_stream_arn_scope(arn, R, A).is_ok(),
+                f.in_deployment(R),
+                "{arn}"
+            );
+        }
+    }
+
+    /// Account is checked before the partition, service and region (measured
+    /// 2026-10-23: another account combined with aws-cn or us-west-2 is
+    /// "Access is denied" on every operation).
+    #[test]
+    fn stream_arn_account_is_checked_before_partition() {
+        for both in [
+            STREAM
+                .replace("arn:aws:", "arn:aws-cn:")
+                .replace(A, "111111111111"),
+            STREAM
+                .replace("us-east-1", "us-west-2")
+                .replace(A, "111111111111"),
+        ] {
+            for op in ["DescribeStream", "GetShardIterator"] {
+                assert_access_denied(stream_arn_check(op, &both), op, &both);
+            }
+            assert_access_denied(validate_tag_resource_arn_shape(&both, A), "tag", &both);
+        }
+    }
+
+    /// A GetRecords iterator carrying a stream ARN outside this deployment's
+    /// scope is Invalid, never canonicalized. GetShardIterator issues no such
+    /// iterator.
+    #[test]
+    fn getrecords_iterator_with_respelled_stream_arn_is_invalid() {
+        let token = |arn: &str| {
+            json!({ "ShardIterator": extenddb_engine::encode_shard_iterator(
+                "shardId-1", "AFTER_SEQUENCE_NUMBER", "", 1_000, arn) })
+        };
+        for respelled in [
+            STREAM.replace(A, "111111111111"),
+            STREAM.replace("arn:aws:", "arn:aws-cn:"),
+            STREAM.replace(":dynamodb:", ":DYNAMODB:"),
+            STREAM.replace("us-east-1", "us-west-2"),
+        ] {
+            assert!(
+                matches!(
+                    getrecords_resource(&token(&respelled), R, A),
+                    GetRecordsResource::Invalid
+                ),
+                "{respelled}"
+            );
+        }
+        match getrecords_resource(&token(STREAM), R, A) {
+            GetRecordsResource::Stream(arn) => assert_eq!(arn, STREAM),
+            _ => panic!("canonical iterator must authorize its stream"),
+        }
+    }
+}
diff --git a/crates/server/src/request_helpers.rs b/crates/server/src/request_helpers.rs
index 9ca4d7dce..188c466f4 100755
--- a/crates/server/src/request_helpers.rs
+++ b/crates/server/src/request_helpers.rs
@@ -137,6 +137,73 @@ pub(crate) async fn authorize_request(
         }
     }
 
+    // The checks from here to the dev-mode return are input validation, not the
+    // IAM decision. The service applies them to every principal, so they run in
+    // dev mode too, and a dev-mode server refuses the same malformed,
+    // out-of-scope and unissued resources as any other build (#450, #451).
+    //
+    // Tag operations validate the ResourceArn shape before the IAM decision, as
+    // the service does: an index ARN (tags are accepted only on a table or
+    // stream) is rejected with ValidationException for every principal, before
+    // authorization. A table ARN passes; its account, region, and existence
+    // are checked after authorization by the engine tag handler. A stream ARN
+    // must name the caller's account here, before authorization, as the
+    // service does (AccessDeniedException "Access is denied" for every
+    // principal, #450); its partition, service and region are checked by the
+    // engine after authorization, which runs on the ARN as sent (#451).
+    if matches!(
+        operation,
+        "TagResource" | "UntagResource" | "ListTagsOfResource"
+    ) && let Some(arn) = input.get("ResourceArn").and_then(Value::as_str)
+        && !arn.is_empty()
+    {
+        authz_resource::validate_tag_resource_arn_shape(arn, account_id)?;
+    }
+
+    // A StreamArn (DescribeStream, GetShardIterator) or BackupArn
+    // (DescribeBackup, DeleteBackup) that is not a well-formed ARN of that kind,
+    // and an export TableArn that does not parse, are refused with
+    // ValidationException before the IAM decision. Without this, an ARN the
+    // engine resolves one way and authorization reads another lets a Deny on
+    // the real resource be bypassed. A StreamArn must also be in this
+    // deployment's scope (partition, service, region) and the caller's account,
+    // so the stream authorized is the stream the engine reads and a respelled
+    // ARN is never served (#450, #451). A restore's source ARN, and an export
+    // TableArn of another kind, are refused by the engine after the IAM
+    // decision.
+    authz_resource::validate_operation_arns(operation, input, &state.region, account_id)?;
+
+    // GetRecords carries its resource only inside the opaque shard iterator, not
+    // in a request body field, so the iterator is decoded here for the stream
+    // ARN the IAM decision below authorizes against.
+    //
+    // A token that does not decode, or decodes to neither a current token nor
+    // the pre-upgrade shape, is rejected with ValidationException "Invalid
+    // ShardIterator" before authorization. A pre-upgrade token (the four
+    // pre-stream-ARN fields with a parseable timestamp) is answered with
+    // ExpiredIteratorException before authorization, so a consumer reading
+    // across the upgrade renews its iterator instead of failing on the
+    // non-retryable "Invalid ShardIterator". The service answers an iterator it
+    // did not issue the same way whatever the caller's policy (measured
+    // 2026-10-20).
+    let getrecords_stream = if operation == "GetRecords" {
+        match authz_resource::getrecords_resource(input, &state.region, account_id) {
+            authz_resource::GetRecordsResource::Stream(arn) => Some(arn),
+            authz_resource::GetRecordsResource::Expired => {
+                return Err(DynamoDbError::ExpiredIteratorException(
+                    extenddb_engine::SHARD_ITERATOR_EXPIRED_MESSAGE.to_owned(),
+                ));
+            }
+            authz_resource::GetRecordsResource::Invalid => {
+                return Err(DynamoDbError::ValidationException(
+                    "Invalid ShardIterator".to_owned(),
+                ));
+            }
+        }
+    } else {
+        None
+    };
+
     // Developer mode opens authorization for the authenticated caller. SigV4
     // verification already ran upstream, so the request is authenticated; only
     // the IAM policy decision is skipped. The key info is still returned so the
@@ -186,58 +253,11 @@ pub(crate) async fn authorize_request(
         return Ok(authorized_key_info);
     }
 
-    // Tag operations validate the ResourceArn shape before the IAM decision, as
-    // the service does: an index ARN (tags are accepted only on a table or
-    // stream) is rejected with ValidationException for every principal, before
-    // authorization. A table or stream ARN passes; its account, region, and
-    // existence are checked after authorization by the engine tag handler.
-    if matches!(
-        operation,
-        "TagResource" | "UntagResource" | "ListTagsOfResource"
-    ) && let Some(arn) = input.get("ResourceArn").and_then(Value::as_str)
-        && !arn.is_empty()
-    {
-        authz_resource::validate_tag_resource_arn_shape(arn)?;
-    }
-
-    // Every other ARN-bearing field (StreamArn, BackupArn, the export TableArn,
-    // a restore SourceTableArn) must parse to the resource kind the operation
-    // takes, or the request is refused with ValidationException before the IAM
-    // decision. Without this, an ARN the engine resolves one way and
-    // authorization reads another lets a Deny on the real resource be bypassed.
-    authz_resource::validate_operation_arns(operation, input)?;
-
-    // GetRecords carries its resource only inside the opaque shard iterator, not
-    // in a request body field. Decode the iterator for the stream ARN and
-    // authorize against that specific stream, as the service does. The storage
-    // read resolves the shard against that stream as part of the read, so a
-    // token that names an allowed stream but a different stream's shard reads
-    // nothing.
-    //
-    // A token that does not decode, or decodes to neither a current token nor
-    // the pre-upgrade shape, is rejected with ValidationException "Invalid
-    // ShardIterator" before authorization. A pre-upgrade token (the four
-    // pre-stream-ARN fields with a parseable timestamp) is answered with
-    // ExpiredIteratorException before authorization, so a consumer reading
-    // across the upgrade renews its iterator instead of failing on the
-    // non-retryable "Invalid ShardIterator". The service answers an iterator it
-    // did not issue the same way whatever the caller's policy (measured
-    // 2026-10-20).
-    if operation == "GetRecords" {
-        let resource_arn =
-            match authz_resource::getrecords_resource(input, &state.region, account_id) {
-                authz_resource::GetRecordsResource::Stream(arn) => arn,
-                authz_resource::GetRecordsResource::Expired => {
-                    return Err(DynamoDbError::ExpiredIteratorException(
-                        extenddb_engine::SHARD_ITERATOR_EXPIRED_MESSAGE.to_owned(),
-                    ));
-                }
-                authz_resource::GetRecordsResource::Invalid => {
-                    return Err(DynamoDbError::ValidationException(
-                        "Invalid ShardIterator".to_owned(),
-                    ));
-                }
-            };
+    // GetRecords is authorized against the specific stream its iterator names,
+    // as the service does. The storage read resolves the shard against that
+    // stream as part of the read, so a token that names an allowed stream but a
+    // different stream's shard reads nothing.
+    if let Some(resource_arn) = getrecords_stream {
         authorization::check_authorization(
             state.authz_cache.as_ref(),
             identity,
diff --git a/docs/manuals/07-upgrade-manual.md b/docs/manuals/07-upgrade-manual.md
index f4451a0f5..93141a232 100755
--- a/docs/manuals/07-upgrade-manual.md
+++ b/docs/manuals/07-upgrade-manual.md
@@ -667,6 +667,12 @@ No catalog migration is required. Before upgrading a deployment that restricts t
 
 ---
 
+### Streams: a `StreamArn` with another partition is refused
+
+DescribeStream and GetShardIterator now refuse a client-built `StreamArn` with the `arn:aws-cn:` or `arn:aws-us-gov:` partition with `ValidationException` "Invalid StreamArn", also on a deployment whose `server.region` is a China or GovCloud region. The server issues every stream ARN as `arn:aws:`, so a `StreamArn` taken from DescribeTable or ListStreams is not affected. No catalog migration is required.
+
+---
+
 ## License
 
 Copyright 2026 ExtendDB contributors. Licensed under the Apache License, Version 2.0.
diff --git a/tests/python/test_streams.py b/tests/python/test_streams.py
index 4ce98acf1..7f055eeda 100755
--- a/tests/python/test_streams.py
+++ b/tests/python/test_streams.py
@@ -165,14 +165,15 @@ def test_describe_stream_basic(self, dynamodb_client, streams_client):
             wait_for_deleted(dynamodb_client, name)
 
     def test_describe_stream_invalid_arn(self, streams_client):
-        """DescribeStream with invalid ARN fails."""
+        """DescribeStream with a StreamArn in another account (000000000000) is
+        refused before the IAM decision with AccessDeniedException "Access is
+        denied", as the service answers (measured 2026-10-23, issue #450). A
+        missing stream in the caller's own account is covered in
+        tests/test_stream_arn_scope.py."""
         with pytest.raises(ClientError) as exc:
             streams_client.describe_stream(StreamArn="arn:aws:dynamodb:us-east-1:000000000000:table/nonexistent/stream/2026-01-01T00:00:00.000")
-        # Could be ResourceNotFoundException or ValidationException
-        assert exc.value.response["Error"]["Code"] in (
-            "ResourceNotFoundException",
-            "ValidationException",
-        )
+        assert exc.value.response["Error"]["Code"] == "AccessDeniedException"
+        assert exc.value.response["Error"]["Message"] == "Access is denied"
 
 
 def _arn_with_another_label(stream_arn: str) -> str:
diff --git a/tests/test_authz_resource_arn.py b/tests/test_authz_resource_arn.py
index 3d4accb65..b50ab1b3c 100644
--- a/tests/test_authz_resource_arn.py
+++ b/tests/test_authz_resource_arn.py
@@ -570,13 +570,18 @@ def test_import_deny_new_table_blocks(self, table):
     # -- fail closed: a respelled ARN cannot evade a Deny on the canonical ARN --
 
     def test_respelled_stream_arn_cannot_bypass_deny(self, table):
-        """A Deny on the canonical stream ARN must also block a request that
-        respells the partition. ExtendDB resolves the stream in this account
-        regardless of the ARN's partition, so authorizing the respelled string
-        verbatim would bypass the Deny."""
+        """A principal with a Deny on the canonical stream ARN gets no stream
+        back for a request that respells the partition. The service refuses the
+        respelled ARN before the IAM decision with ValidationException "Invalid
+        StreamArn" (measured 2026-10-22), so ExtendDB does the same instead of
+        canonicalizing it; the fuller matrix is in test_stream_arn_scope.py."""
         respelled = table["stream_arn"].replace("arn:aws:", "arn:aws-cn:")
         s = self._ddb(_allow_all_deny("DescribeStream", table["stream_arn"]), "dynamodbstreams")
-        self._assert_denied(lambda: s.describe_stream(StreamArn=respelled))
+        with pytest.raises(ClientError) as exc:
+            s.describe_stream(StreamArn=respelled)
+        err = exc.value.response["Error"]
+        assert err["Code"] == "ValidationException", err
+        assert err.get("Message") == "Invalid StreamArn", err
 
     def test_table_arn_in_tablename_deny_blocks(self, table):
         """A Deny on the table ARN blocks GetItem whether TableName is the bare
diff --git a/tests/test_cross_account_isolation.py b/tests/test_cross_account_isolation.py
index cb88bc3d6..6b5d98058 100755
--- a/tests/test_cross_account_isolation.py
+++ b/tests/test_cross_account_isolation.py
@@ -519,16 +519,18 @@ def old_format(shard_id: str) -> str:
         assert get_records_error(old_format(missing_shard)) == err
 
         # A token in the current format that names account A's stream ARN.
-        # Authorization pins the stream ARN to account B, and the engine looks
-        # the stream up in account B, where it does not exist, so the request
-        # fails the same way for a real and a made-up shard.
+        # GetShardIterator refuses a StreamArn in another account, so account B
+        # is never issued a token naming A's stream. This one is refused as an
+        # invalid iterator before authorization, the same way for a real and a
+        # made-up shard, and the stream ARN it carries is never rewritten to
+        # account B.
         def current_format(shard_id: str) -> str:
             return base64.b64encode(
                 f"{shard_id}|AFTER_SEQUENCE_NUMBER||{int(time.time())}|{arn}".encode()
             ).decode()
 
         err = get_records_error(current_format(target_shard))
-        assert err["Code"] == "ResourceNotFoundException", err
+        assert err == {"Code": "ValidationException", "Message": "Invalid ShardIterator"}, err
         assert get_records_error(current_format(missing_shard)) == err
 
         # An iterator that does not decode gets DynamoDB's verbatim answer.
diff --git a/tests/test_dev_mode_authorization.py b/tests/test_dev_mode_authorization.py
index 354fffff8..734ae0082 100644
--- a/tests/test_dev_mode_authorization.py
+++ b/tests/test_dev_mode_authorization.py
@@ -15,6 +15,11 @@
 `BatchGetItem`, `BatchWriteItem`, `TransactGetItems` and `TransactWriteItems`
 failed while every single-item operation succeeded.
 
+Dev mode skips only the IAM decision. The input checks the service applies to
+every principal before that decision still run, so the stream ARN scope
+refusals of `tests/test_stream_arn_scope.py` (#450, #451) are asserted here
+against the dev-mode server too.
+
 These require a server built and started in dev mode, so they are gated on
 EXTENDDB_TEST_DEV_MODE and excluded from the backend-agnostic suite, which runs
 against a production-mode server where these operations are correctly denied.
@@ -22,9 +27,13 @@
 
 from __future__ import annotations
 
+import base64
 import os
+import time
 
+import boto3
 import pytest
+from botocore.config import Config as BotoConfig
 from botocore.exceptions import ClientError
 
 from conftest import wait_for_active
@@ -159,3 +168,115 @@ def test_transact_get_items_is_not_denied(self, dynamodb_client, dev_table):
             ),
         )
         assert resp["Responses"][0]["Item"]["pk"]["S"] == "tg1"
+
+
+OTHER_ACCOUNT = "111111111111"
+
+
+def _error(call, *args) -> dict:
+    with pytest.raises(ClientError) as exc:
+        call(*args)
+    return exc.value.response["Error"]
+
+
+@pytest.fixture()
+def stream_table(dynamodb_client, endpoint_url, unique_table_name):
+    """A table with a stream and one record. Returns the issued stream ARN, a
+    shard id and a streams client with retries off."""
+    dynamodb_client.create_table(
+        TableName=unique_table_name,
+        KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}],
+        BillingMode="PAY_PER_REQUEST",
+        StreamSpecification={"StreamEnabled": True, "StreamViewType": "NEW_IMAGE"},
+    )
+    wait_for_active(dynamodb_client, unique_table_name)
+    dynamodb_client.put_item(TableName=unique_table_name, Item={"pk": {"S": "s1"}})
+    streams = boto3.client(
+        "dynamodbstreams",
+        endpoint_url=endpoint_url,
+        region_name=os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
+        config=BotoConfig(retries={"max_attempts": 0}),
+    )
+    arn = dynamodb_client.describe_table(TableName=unique_table_name)["Table"]["LatestStreamArn"]
+    shard_id = streams.describe_stream(StreamArn=arn)["StreamDescription"]["Shards"][0]["ShardId"]
+    yield arn, shard_id, streams
+    try:
+        dynamodb_client.delete_table(TableName=unique_table_name)
+    except ClientError:
+        pass
+
+
+class TestDevModeKeepsStreamArnScope:
+    """The stream ARN refusals run before the dev-mode return (#450, #451).
+    The expected answers are the service's, measured 2026-10-23 and tabled in
+    the module docstring of tests/test_stream_arn_scope.py, except "Shard
+    iterator has expired" for a pre-upgrade token, which is this server's
+    upgrade answer."""
+
+    def test_stream_arn_outside_scope_is_refused(self, stream_table):
+        arn, shard_id, streams = stream_table
+        account = arn.split(":")[4]
+        cases = (
+            (arn.replace("arn:aws:", "arn:aws-cn:", 1),
+             "ValidationException", "Invalid StreamArn"),
+            (arn.replace(f":{account}:", f":{OTHER_ACCOUNT}:", 1),
+             "AccessDeniedException", "Access is denied"),
+        )
+        calls = {
+            "DescribeStream": lambda a: streams.describe_stream(StreamArn=a),
+            "GetShardIterator": lambda a: streams.get_shard_iterator(
+                StreamArn=a, ShardId=shard_id, ShardIteratorType="TRIM_HORIZON"),
+        }
+        for sent, code, message in cases:
+            for op, call in calls.items():
+                err = _error(call, sent)
+                assert (err["Code"], err.get("Message")) == (code, message), (op, sent, err)
+
+    def test_tag_resource_arn_with_trailing_slash_is_refused(self, dynamodb_client, stream_table):
+        """The issued stream ARN with a trailing slash, as a tag operation's
+        ResourceArn, is not a DynamoDB resource ARN: the service answers this
+        for every principal, so the shape check runs before the dev-mode
+        return. Without it TagResource writes a tag under that ARN."""
+        sent = stream_table[0] + "/"
+        expected = {
+            "Code": "ValidationException",
+            "Message": (
+                "One or more parameter values were invalid: "
+                f"Provided Arn is not a DynamoDB resource arn: {sent}"
+            ),
+        }
+        calls = {
+            "TagResource": lambda: dynamodb_client.tag_resource(
+                ResourceArn=sent, Tags=[{"Key": "k", "Value": "v"}]),
+            "ListTagsOfResource": lambda: dynamodb_client.list_tags_of_resource(ResourceArn=sent),
+        }
+        for op, call in calls.items():
+            err = _error(call)
+            assert err == expected, (op, err)
+
+    def test_getrecords_token_not_issued_here_is_refused(self, stream_table):
+        """A token is base64 of `shard|type|seq|ts|stream_arn`. One naming
+        another account is "Invalid ShardIterator"; GetShardIterator issues no
+        such token. A token in the pre-upgrade shape (four fields, no stream
+        ARN) is answered "Shard iterator has expired", as in every other build."""
+        arn, shard_id, streams = stream_table
+        account = arn.split(":")[4]
+        now = int(time.time())
+
+        def token(*fields: str) -> str:
+            return base64.standard_b64encode("|".join(fields).encode()).decode()
+
+        def read(t: str):
+            return streams.get_records(ShardIterator=t, Limit=10)
+
+        # Control: the same hand-built shape with the issued ARN is served, so
+        # the refusals below come from the ARN and the shape, not the encoding.
+        read(token(shard_id, "AFTER_SEQUENCE_NUMBER", "", str(now), arn))
+        foreign = arn.replace(f":{account}:", f":{OTHER_ACCOUNT}:", 1)
+        err = _error(read, token(shard_id, "AFTER_SEQUENCE_NUMBER", "", str(now), foreign))
+        assert err == {"Code": "ValidationException", "Message": "Invalid ShardIterator"}, err
+        err = _error(read, token(shard_id, "AFTER_SEQUENCE_NUMBER", "", str(now)))
+        assert err == {
+            "Code": "ExpiredIteratorException", "Message": "Shard iterator has expired"
+        }, err
diff --git a/tests/test_stream_arn_scope.py b/tests/test_stream_arn_scope.py
new file mode 100644
index 000000000..e2c19213c
--- /dev/null
+++ b/tests/test_stream_arn_scope.py
@@ -0,0 +1,549 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Stream ARN scope: account, partition, region and service (issues #450, #451).
+
+A StreamArn (DescribeStream, GetShardIterator), a stream ARN in a tag
+operation's ResourceArn, and the stream ARN inside a GetRecords iterator must
+name the caller's account and this deployment's partition, service and region.
+Before this fix ExtendDB resolved a stream by table name and label only, so a
+StreamArn in another account, partition or region, or with an upper-cased
+service or region, was served as the caller's own stream.
+
+Every expected answer in the StreamArn and ResourceArn tables below was
+measured against live DynamoDB on 2026-10-23 (12:22 to 12:27 UTC, and
+UntagResource from 14:03 to 14:09 UTC with a recheck of GetShardIterator on an
+unknown label in that second run) for four principals: an admin (Allow
+dynamodb:* on *), a user with no stream permissions, a user allowed exactly
+the canonical stream ARN, and a user with Allow * plus a Deny on the canonical
+stream ARN. The order table between them was measured on 2026-10-28 as the
+admin only.
+
+StreamArn (DescribeStream, GetShardIterator), the same for every principal, so
+before the IAM decision:
+
+  | spelling                                          | answer                              |
+  |---------------------------------------------------|-------------------------------------|
+  | another account (111111111111, abc), alone or with | AccessDeniedException               |
+  |   aws-cn, us-west-2 or an unknown label            |   "Access is denied"                |
+  | empty account; aws-cn; aws-us-gov; DynamoDB;       | ValidationException                 |
+  |   DYNAMODB; us-west-2; US-EAST-1; an extra field;  |   "Invalid StreamArn"               |
+  |   a missing region field; a trailing slash         |                                     |
+  | a label the table never had                        | IAM decision first (noperm, exact   |
+  |                                                    |   denied); then ResourceNotFound    |
+
+The order of those StreamArn checks, measured on 2026-10-28 in us-east-1 as an
+admin, two samples per row: the shape first (an empty partition or service
+field), then the account, then the region. The empty-partition, account and
+region rows were measured on DescribeStream from 20:47 to 20:48 UTC and on
+GetShardIterator at 21:51 UTC; the empty-service rows on both operations at
+21:51 UTC.
+
+  | StreamArn                                          | answer                              |
+  |----------------------------------------------------|-------------------------------------|
+  | empty partition, in another account or the         | ValidationException                 |
+  |   caller's                                         |   "Invalid StreamArn"               |
+  | empty service, in another account (21:51 UTC,      | ValidationException                 |
+  |   DescribeStream and GetShardIterator)             |   "Invalid StreamArn"               |
+  | empty service, in the caller's account (21:51      | ValidationException                 |
+  |   UTC, DescribeStream and GetShardIterator)        |   "Invalid StreamArn"               |
+  | another account with an empty region, with         | AccessDeniedException               |
+  |   us-west-2, or with every other field the         |   "Access is denied"                |
+  |   caller's                                         |                                     |
+  | the caller's account with an empty region or       | ValidationException                 |
+  |   us-west-2                                        |   "Invalid StreamArn"               |
+
+ResourceArn (ListTagsOfResource, TagResource, UntagResource):
+
+  | spelling                                          | answer                              |
+  |---------------------------------------------------|-------------------------------------|
+  | another account, abc, empty account               | AccessDeniedException               |
+  |                                                    |   "Access is denied", every principal |
+  | an extra field; a trailing slash                   | ValidationException "One or more    |
+  |                                                    |   parameter values were invalid:    |
+  |                                                    |   Provided Arn is not a DynamoDB    |
+  |                                                    |   resource arn: <arn>", every       |
+  |                                                    |   principal                         |
+  | aws-cn; aws-us-gov; DynamoDB; DYNAMODB; us-west-2; | IAM decision first, on the ARN as   |
+  |   US-EAST-1                                        |   sent (noperm and exact denied,    |
+  |                                                    |   naming that ARN); then            |
+  |                                                    |   ValidationException               |
+  |                                                    |   "Invalid StreamArn"               |
+  | a label the table never had                        | IAM decision first; then            |
+  |                                                    |   ResourceNotFoundException         |
+  |                                                    |   "Requested resource not found:    |
+  |                                                    |   ResourceArn: <arn> not found"     |
+
+A GetRecords token the service did not issue is ValidationException "Invalid
+ShardIterator" whatever stream ARN it carries.
+
+Prerequisites:
+  - extenddb running with `auth.provider = "builtin"` on EXTENDDB_TEST_ENDPOINT
+  - Admin credentials in EXTENDDB_ADMIN_USER / EXTENDDB_ADMIN_PASSWORD
+"""
+
+from __future__ import annotations
+
+import base64
+import os
+import time
+import uuid
+from typing import Any, Callable
+
+import boto3
+import pytest
+from botocore.config import Config as BotoConfig
+from botocore.exceptions import ClientError
+
+from conftest import wait_for_active
+from management_helpers import ManagementClient
+
+INVALID_STREAM_ARN = "Invalid StreamArn"
+ACCESS_IS_DENIED = "Access is denied"
+OTHER_ACCOUNT = "111111111111"
+
+STREAM_OPS = ("DescribeStream", "GetShardIterator")
+TAG_OPS = ("ListTagsOfResource", "TagResource", "UntagResource")
+
+
+def _require_auth_env() -> tuple[str, str, str]:
+    endpoint = os.environ.get("EXTENDDB_TEST_ENDPOINT", "").strip()
+    admin_user = os.environ.get("EXTENDDB_ADMIN_USER", "").strip()
+    admin_pass = os.environ.get("EXTENDDB_ADMIN_PASSWORD", "").strip()
+    if not endpoint or not admin_user or not admin_pass:
+        pytest.fail(
+            "MISCONFIGURED: stream ARN scope tests require EXTENDDB_TEST_ENDPOINT, "
+            "EXTENDDB_ADMIN_USER, and EXTENDDB_ADMIN_PASSWORD."
+        )
+    return endpoint, admin_user, admin_pass
+
+
+def _client(endpoint: str, ak: str, sk: str, region: str, service: str) -> Any:
+    kwargs: dict = dict(
+        service_name=service,
+        endpoint_url=endpoint,
+        aws_access_key_id=ak,
+        aws_secret_access_key=sk,
+        region_name=region,
+        config=BotoConfig(retries={"max_attempts": 0}),
+    )
+    if endpoint.startswith("https://"):
+        kwargs["verify"] = False
+    return boto3.client(**kwargs)
+
+
+ALLOW_ALL = {
+    "Version": "2012-10-17",
+    "Statement": [{"Effect": "Allow", "Action": "dynamodb:*", "Resource": "*"}],
+}
+NO_PERMISSIONS = {
+    "Version": "2012-10-17",
+    "Statement": [{"Effect": "Allow", "Action": "dynamodb:ListTables", "Resource": "*"}],
+}
+
+
+def _allow_only(resource: str) -> dict:
+    return {
+        "Version": "2012-10-17",
+        "Statement": [{"Effect": "Allow", "Action": "dynamodb:*", "Resource": resource}],
+    }
+
+
+def _allow_all_deny(action: str, resource: str) -> dict:
+    return {
+        "Version": "2012-10-17",
+        "Statement": [
+            {"Effect": "Allow", "Action": "dynamodb:*", "Resource": "*"},
+            {"Effect": "Deny", "Action": f"dynamodb:{action}", "Resource": resource},
+        ],
+    }
+
+
+@pytest.fixture(scope="module")
+def auth_env():
+    return _require_auth_env()
+
+
+@pytest.fixture(scope="module")
+def mgmt(auth_env) -> ManagementClient:
+    endpoint, admin_user, admin_pass = auth_env
+    return ManagementClient(endpoint, admin_user, admin_pass)
+
+
+@pytest.fixture(scope="module")
+def region() -> str:
+    return os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
+
+
+@pytest.fixture(scope="module")
+def account_id(mgmt) -> str:
+    acct = f"{uuid.uuid4().int % 10**12:012d}"
+    resp = mgmt.create_account(acct, f"streamscope-{acct}")
+    assert resp.status_code == 201, resp.text
+    yield acct
+    mgmt.delete_account(acct)
+
+
+def _user_with_policy(mgmt: ManagementClient, account_id: str, policy: dict) -> tuple[str, str, str]:
+    user = f"u{uuid.uuid4().hex[:10]}"
+    assert mgmt.create_user(account_id, user, "TestPass123!").status_code == 201
+    assert mgmt.put_user_policy(account_id, user, "p", policy).status_code == 204
+    resp = mgmt.create_access_key(account_id, user)
+    assert resp.status_code == 201, resp.text
+    creds = resp.json()
+    return user, creds["access_key_id"], creds["secret_access_key"]
+
+
+@pytest.fixture(scope="module")
+def table(auth_env, mgmt, account_id, region):
+    """A table with a stream and one item. Returns its ARNs and a shard id."""
+    endpoint = auth_env[0]
+    _, ak, sk = _user_with_policy(mgmt, account_id, ALLOW_ALL)
+    ddb = _client(endpoint, ak, sk, region, "dynamodb")
+    streams = _client(endpoint, ak, sk, region, "dynamodbstreams")
+    name = f"s{uuid.uuid4().hex[:8]}"
+    ddb.create_table(
+        TableName=name,
+        AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}],
+        KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}],
+        BillingMode="PAY_PER_REQUEST",
+        StreamSpecification={"StreamEnabled": True, "StreamViewType": "NEW_AND_OLD_IMAGES"},
+    )
+    wait_for_active(ddb, name)
+    ddb.put_item(TableName=name, Item={"pk": {"S": "p1"}})
+    desc = ddb.describe_table(TableName=name)["Table"]
+    stream_arn = desc["LatestStreamArn"]
+    shard_id = streams.describe_stream(StreamArn=stream_arn)["StreamDescription"]["Shards"][0]["ShardId"]
+    return {
+        "name": name,
+        "table_arn": desc["TableArn"],
+        "stream_arn": stream_arn,
+        "shard_id": shard_id,
+    }
+
+
+def _variants(stream_arn: str, region: str, account_id: str) -> dict[str, str]:
+    """The spellings measured on 2026-10-23 and 2026-10-28 (module docstring),
+    keyed by their names."""
+    other_region = "us-west-2" if region != "us-west-2" else "us-east-1"
+    table_arn = stream_arn.split("/stream/")[0]
+    v = {
+        "account_other": stream_arn.replace(f":{account_id}:", f":{OTHER_ACCOUNT}:", 1),
+        "account_empty": stream_arn.replace(f":{account_id}:", "::", 1),
+        "account_nonnumeric": stream_arn.replace(f":{account_id}:", ":abc:", 1),
+        "partition_aws_cn": stream_arn.replace("arn:aws:", "arn:aws-cn:", 1),
+        "partition_aws_us_gov": stream_arn.replace("arn:aws:", "arn:aws-us-gov:", 1),
+        "service_DynamoDB": stream_arn.replace(":dynamodb:", ":DynamoDB:", 1),
+        "service_DYNAMODB": stream_arn.replace(":dynamodb:", ":DYNAMODB:", 1),
+        "region_other": stream_arn.replace(f":{region}:", f":{other_region}:", 1),
+        "region_upper": stream_arn.replace(f":{region}:", f":{region.upper()}:", 1),
+        "field_extra": stream_arn.replace(f":{account_id}:", f":{account_id}:x:", 1),
+        "field_missing_region": stream_arn.replace(f":{region}:", ":", 1),
+        "trailing_slash": stream_arn + "/",
+        "label_unknown": table_arn + "/stream/2000-01-01T00:00:00.000",
+    }
+    v["account_other_and_aws_cn"] = v["account_other"].replace("arn:aws:", "arn:aws-cn:", 1)
+    v["account_other_and_label_unknown"] = v["label_unknown"].replace(
+        f":{account_id}:", f":{OTHER_ACCOUNT}:", 1)
+    v["account_other_and_region_other"] = v["account_other"].replace(
+        f":{region}:", f":{other_region}:", 1)
+    v["partition_empty"] = stream_arn.replace("arn:aws:", "arn::", 1)
+    v["region_empty"] = stream_arn.replace(f":{region}:", "::", 1)
+    v["account_other_and_partition_empty"] = v["account_other"].replace("arn:aws:", "arn::", 1)
+    v["account_other_and_region_empty"] = v["account_other"].replace(f":{region}:", "::", 1)
+    v["service_empty"] = stream_arn.replace(":dynamodb:", "::", 1)
+    v["account_other_and_service_empty"] = v["account_other"].replace(":dynamodb:", "::", 1)
+    return v
+
+
+ACCOUNT_VARIANTS = (
+    "account_other", "account_nonnumeric", "account_other_and_aws_cn",
+    "account_other_and_label_unknown", "account_other_and_region_other",
+)
+RESPELLINGS = (
+    "partition_aws_cn", "partition_aws_us_gov", "service_DynamoDB",
+    "service_DYNAMODB", "region_other", "region_upper",
+)
+MALFORMED = ("field_extra", "trailing_slash")
+
+
+def _not_dynamodb_resource(arn: str) -> str:
+    return (
+        "One or more parameter values were invalid: "
+        f"Provided Arn is not a DynamoDB resource arn: {arn}"
+    )
+
+
+def _iam_denied(account_id: str, user: str, op: str, arn: str) -> str:
+    return (
+        f"User: arn:aws:iam::{account_id}:user/{user} is not authorized to perform: "
+        f"dynamodb:{op} on resource: {arn} because no identity-based policy allows "
+        f"the dynamodb:{op} action"
+    )
+
+
+class _Principal:
+    def __init__(self, endpoint: str, region: str, account_id: str, user: str, ak: str, sk: str):
+        self.user = user
+        self.account_id = account_id
+        self.ddb = _client(endpoint, ak, sk, region, "dynamodb")
+        self.streams = _client(endpoint, ak, sk, region, "dynamodbstreams")
+
+    def call(self, op: str, arn: str, shard_id: str) -> Callable[[], Any]:
+        return {
+            "DescribeStream": lambda: self.streams.describe_stream(StreamArn=arn),
+            "GetShardIterator": lambda: self.streams.get_shard_iterator(
+                StreamArn=arn, ShardId=shard_id, ShardIteratorType="TRIM_HORIZON"),
+            "ListTagsOfResource": lambda: self.ddb.list_tags_of_resource(ResourceArn=arn),
+            "TagResource": lambda: self.ddb.tag_resource(
+                ResourceArn=arn, Tags=[{"Key": "k", "Value": "v"}]),
+            "UntagResource": lambda: self.ddb.untag_resource(ResourceArn=arn, TagKeys=["k"]),
+        }[op]
+
+    def error(self, op: str, arn: str, shard_id: str) -> dict:
+        with pytest.raises(ClientError) as exc:
+            self.call(op, arn, shard_id)()
+        return exc.value.response["Error"]
+
+    def iam_denied(self, op: str, arn: str) -> str:
+        return _iam_denied(self.account_id, self.user, op, arn)
+
+
+class TestStreamArnScope:
+    @pytest.fixture(autouse=True)
+    def setup(self, auth_env, mgmt, account_id, region, table):
+        self.endpoint = auth_env[0]
+        self.mgmt = mgmt
+        self.account_id = account_id
+        self.region = region
+        self.table = table
+        self.shard_id = table["shard_id"]
+        self.variants = _variants(table["stream_arn"], region, account_id)
+
+    def _principal(self, policy: dict) -> _Principal:
+        user, ak, sk = _user_with_policy(self.mgmt, self.account_id, policy)
+        return _Principal(self.endpoint, self.region, self.account_id, user, ak, sk)
+
+    def _principals(self) -> dict[str, _Principal]:
+        """The four principals every case was measured as."""
+        canonical = self.table["stream_arn"]
+        return {
+            "admin": self._principal(ALLOW_ALL),
+            "noperm": self._principal(NO_PERMISSIONS),
+            "exact": self._principal(_allow_only(canonical)),
+            "deny": self._principal(_allow_all_deny("*", canonical)),
+        }
+
+    @staticmethod
+    def _assert_err(err: dict, code: str, message: str | None, ctx: tuple) -> None:
+        assert err["Code"] == code, (*ctx, err)
+        if message is not None:
+            assert err.get("Message") == message, (*ctx, err)
+
+    # -- #450: another account is AccessDeniedException ----------------------
+
+    def test_stream_arn_in_another_account_is_denied(self):
+        """Another 12-digit account, `abc`, and another account combined with
+        aws-cn, us-west-2 or an unknown label: AccessDeniedException "Access is
+        denied" on all five operations, for every principal, so before the IAM
+        decision and before the partition, region or label is looked at."""
+        for who, p in self._principals().items():
+            for label in ACCOUNT_VARIANTS:
+                arn = self.variants[label]
+                for op in STREAM_OPS + TAG_OPS:
+                    err = p.error(op, arn, self.shard_id)
+                    self._assert_err(err, "AccessDeniedException", ACCESS_IS_DENIED, (who, label, op))
+
+    def test_empty_account_and_missing_region(self):
+        """An empty account field is "Invalid StreamArn" on StreamArn and
+        "Access is denied" on a tag ResourceArn, for every principal. A missing
+        region field is "Invalid StreamArn" on StreamArn for every principal.
+        (On a tag ResourceArn the service answers a missing region field with
+        "Access is denied"; ExtendDB reads the shifted fields as a malformed
+        resource first. That ordering is pre-existing and outside this fix, so
+        it is not asserted here.)"""
+        principals = self._principals()
+        for label in ("account_empty", "field_missing_region"):
+            arn = self.variants[label]
+            for who, p in principals.items():
+                for op in STREAM_OPS:
+                    err = p.error(op, arn, self.shard_id)
+                    self._assert_err(err, "ValidationException", INVALID_STREAM_ARN, (who, label, op))
+        arn = self.variants["account_empty"]
+        for who, p in principals.items():
+            for op in TAG_OPS:
+                err = p.error(op, arn, self.shard_id)
+                self._assert_err(err, "AccessDeniedException", ACCESS_IS_DENIED, (who, "account_empty", op))
+
+    def test_partition_then_account_then_region(self):
+        """The order cells measured on 2026-10-28 (the order table above), on
+        DescribeStream and GetShardIterator: an empty partition or service is
+        "Invalid StreamArn" even in another account; another account is
+        "Access is denied" whatever the region field holds; the caller's
+        account with an empty or another region is "Invalid StreamArn".
+        Measured as an admin; GetShardIterator and the empty service at 21:51
+        UTC."""
+        admin = self._principal(ALLOW_ALL)
+        cells = (
+            ("account_other_and_partition_empty", "ValidationException", INVALID_STREAM_ARN),
+            ("partition_empty", "ValidationException", INVALID_STREAM_ARN),
+            ("account_other_and_service_empty", "ValidationException", INVALID_STREAM_ARN),
+            ("service_empty", "ValidationException", INVALID_STREAM_ARN),
+            ("account_other_and_region_empty", "AccessDeniedException", ACCESS_IS_DENIED),
+            ("account_other_and_region_other", "AccessDeniedException", ACCESS_IS_DENIED),
+            ("account_other", "AccessDeniedException", ACCESS_IS_DENIED),
+            ("region_empty", "ValidationException", INVALID_STREAM_ARN),
+            ("region_other", "ValidationException", INVALID_STREAM_ARN),
+        )
+        for label, code, message in cells:
+            for op in STREAM_OPS:
+                err = admin.error(op, self.variants[label], self.shard_id)
+                self._assert_err(err, code, message, ("admin", label, op))
+
+    # -- #451: another partition, region or service ---------------------------
+
+    def test_respelled_stream_arn_on_stream_operations_is_invalid(self):
+        """aws-cn, aws-us-gov, DynamoDB, DYNAMODB, us-west-2 and US-EAST-1 in
+        StreamArn: ValidationException "Invalid StreamArn" for every principal,
+        so before the IAM decision."""
+        for who, p in self._principals().items():
+            for label in RESPELLINGS:
+                arn = self.variants[label]
+                for op in STREAM_OPS:
+                    err = p.error(op, arn, self.shard_id)
+                    self._assert_err(err, "ValidationException", INVALID_STREAM_ARN, (who, label, op))
+
+    def test_respelled_stream_arn_on_tag_operations(self):
+        """The same six spellings in a tag ResourceArn: the IAM decision runs
+        first, on the ARN as sent. A no-permissions principal and a principal
+        allowed only the canonical ARN are denied with a message naming the
+        spelling they sent; the admin, and a principal with Allow * plus a Deny
+        on the canonical ARN (which the respelled ARN does not match), get
+        ValidationException "Invalid StreamArn"."""
+        principals = self._principals()
+        for label in RESPELLINGS:
+            arn = self.variants[label]
+            for op in TAG_OPS:
+                for who in ("admin", "deny"):
+                    err = principals[who].error(op, arn, self.shard_id)
+                    self._assert_err(err, "ValidationException", INVALID_STREAM_ARN, (who, label, op))
+                for who in ("noperm", "exact"):
+                    p = principals[who]
+                    err = p.error(op, arn, self.shard_id)
+                    self._assert_err(err, "AccessDeniedException", p.iam_denied(op, arn), (who, label, op))
+
+    def test_malformed_stream_arn(self):
+        """An extra field and a trailing slash: "Invalid StreamArn" on
+        StreamArn; "Provided Arn is not a DynamoDB resource arn: <arn>" on a tag
+        ResourceArn. Every principal, so before the IAM decision."""
+        for who, p in self._principals().items():
+            for label in MALFORMED:
+                arn = self.variants[label]
+                for op in STREAM_OPS:
+                    err = p.error(op, arn, self.shard_id)
+                    self._assert_err(err, "ValidationException", INVALID_STREAM_ARN, (who, label, op))
+                for op in TAG_OPS:
+                    err = p.error(op, arn, self.shard_id)
+                    self._assert_err(
+                        err, "ValidationException", _not_dynamodb_resource(arn), (who, label, op))
+
+    # -- positive control and the Deny on the canonical ARN -------------------
+
+    def test_canonical_stream_arn(self):
+        """The ARN the server issued: served to the admin and to the principal
+        allowed exactly that ARN on all five operations; the no-permissions
+        principal gets the IAM denial naming it; the Deny principal gets the
+        explicit-deny message."""
+        p = self._principals()
+        arn = self.table["stream_arn"]
+        for who in ("admin", "exact"):
+            for op in STREAM_OPS + TAG_OPS:
+                p[who].call(op, arn, self.shard_id)()
+        for op in STREAM_OPS + TAG_OPS:
+            err = p["noperm"].error(op, arn, self.shard_id)
+            self._assert_err(err, "AccessDeniedException", p["noperm"].iam_denied(op, arn), ("noperm", op))
+            err = p["deny"].error(op, arn, self.shard_id)
+            expected = (
+                f"User: arn:aws:iam::{self.account_id}:user/{p['deny'].user} is not authorized "
+                f"to perform: dynamodb:{op} on resource: {arn} with an explicit deny in an "
+                "identity-based policy"
+            )
+            self._assert_err(err, "AccessDeniedException", expected, ("deny", op))
+
+    def test_deny_on_real_stream_arn_is_not_evaded_by_respelling(self):
+        """A principal with Allow * and a per-operation Deny on the canonical
+        stream ARN gets no stream back for any spelling: another account is
+        "Access is denied", the six respellings are "Invalid StreamArn", on
+        every operation."""
+        canonical = self.table["stream_arn"]
+        for op in STREAM_OPS + TAG_OPS:
+            p = self._principal(_allow_all_deny(op, canonical))
+            err = p.error(op, canonical, self.shard_id)
+            self._assert_err(err, "AccessDeniedException", None, ("canonical", op))
+            for label in ACCOUNT_VARIANTS:
+                err = p.error(op, self.variants[label], self.shard_id)
+                self._assert_err(err, "AccessDeniedException", ACCESS_IS_DENIED, (label, op))
+            for label in RESPELLINGS:
+                err = p.error(op, self.variants[label], self.shard_id)
+                self._assert_err(err, "ValidationException", INVALID_STREAM_ARN, (label, op))
+
+    # -- GetRecords: a token the server did not issue -------------------------
+
+    def test_getrecords_iterator_with_respelled_stream_arn_is_invalid(self):
+        """A shard iterator is base64 of `shard|type|seq|ts|stream_arn`. A token
+        carrying a stream ARN the server did not issue (another account, aws-cn,
+        us-west-2, DYNAMODB) was canonicalized and served before; it is now
+        ValidationException "Invalid ShardIterator", the service's answer to a
+        token it did not issue, for an admin and for a no-permissions principal
+        alike."""
+        now = int(time.time())
+
+        def token(arn: str) -> str:
+            return base64.standard_b64encode(
+                f"{self.shard_id}|AFTER_SEQUENCE_NUMBER||{now}|{arn}".encode()
+            ).decode()
+
+        admin = self._principal(ALLOW_ALL)
+        noperm = self._principal(NO_PERMISSIONS)
+        # Control for this test only: the same hand-built token with the issued
+        # stream ARN reads the stream, so the refusals below come from the
+        # respelled ARN and not from the token shape. This is not parity: the
+        # service answers "Invalid ShardIterator" to every token it did not
+        # issue, including this one (measured 2026-10-23), while ExtendDB serves
+        # it because its iterators carry no signature. That gap exists on main.
+        admin.streams.get_records(ShardIterator=token(self.table["stream_arn"]), Limit=10)
+        for label in ("account_other", "partition_aws_cn", "region_other", "service_DYNAMODB"):
+            for who, p in (("admin", admin), ("noperm", noperm)):
+                with pytest.raises(ClientError) as exc:
+                    p.streams.get_records(ShardIterator=token(self.variants[label]), Limit=10)
+                err = exc.value.response["Error"]
+                assert err == {"Code": "ValidationException", "Message": "Invalid ShardIterator"}, (
+                    who, label, err)
+
+    # -- a label the table never had ------------------------------------------
+
+    def test_unknown_label(self):
+        """IAM decision first (noperm and exact are denied naming the ARN) on all
+        five operations, then ResourceNotFoundException for the admin and the
+        Deny principal (whose Deny names another label). DescribeStream:
+        "Requested resource not found: Stream: <arn> not found" with no trailing
+        period (#451). GetShardIterator: "Requested resource not found: Stream
+        does not exist". Both sentences are the service's (measured 2026-10-23).
+        (The tag operations answer "Requested resource not found: ResourceArn:
+        <arn> not found" on the service; ExtendDB checks only that the table
+        exists, which is pre-existing and outside this fix, so the authorized
+        principals are not asserted on them here.)"""
+        p = self._principals()
+        arn = self.variants["label_unknown"]
+        for op in STREAM_OPS + TAG_OPS:
+            for who in ("noperm", "exact"):
+                err = p[who].error(op, arn, self.shard_id)
+                self._assert_err(err, "AccessDeniedException", p[who].iam_denied(op, arn), (who, op))
+        for op in STREAM_OPS:
+            for who in ("admin", "deny"):
+                err = p[who].error(op, arn, self.shard_id)
+                expected = (
+                    f"Requested resource not found: Stream: {arn} not found"
+                    if op == "DescribeStream"
+                    else "Requested resource not found: Stream does not exist"
+                )
+                self._assert_err(err, "ResourceNotFoundException", expected, (who, op))