fix(auth): authorize resource-ARN operations against the resource they act on

crates/auth/src/policy/evaluator.rs

@@ -77,6 +77,35 @@ pub fn evaluate_policies_arc(
     )
 }
 
+/// Returns true if an explicit Deny statement in any identity policy,
+/// permissions boundary, or session policy matches the action, resource and
+/// context. The server uses this only to choose the AccessDenied message form;
+/// it does not change the authorization decision, which `evaluate_policies_arc`
+/// already made.
+#[must_use]
+pub fn explicit_deny_matches_arc(
+    identity_policies: &[std::sync::Arc<PolicyDocument>],
+    permissions_boundary: Option<&PolicyDocument>,
+    session_policy: Option<&PolicyDocument>,
+    action: &str,
+    resource_arn: &str,
+    context: &impl ConditionContext,
+) -> bool {
+    identity_policies
+        .iter()
+        .map(std::sync::Arc::as_ref)
+        .chain(permissions_boundary)
+        .chain(session_policy)
+        .any(|policy| {
+            policy.statements.iter().any(|stmt| {
+                stmt.effect == Effect::Deny
+                    && action_matches(stmt, action)
+                    && resource_matches(stmt, resource_arn)
+                    && deny_conditions_apply(stmt, context)
+            })
+        })
+}
+
 fn evaluate_with<'a, I>(
     identity_policies: I,
     permissions_boundary: Option<&'a PolicyDocument>,

crates/auth/src/policy/matcher.rs

@@ -4,8 +4,10 @@
 //! Wildcard and ARN matching for IAM policy evaluation.
 //!
 //! `wildcard_match` supports `*` (zero or more characters) and `?` (exactly one
-//! character). `arn_match` splits on `:` and matches each segment independently,
-//! with `*` in a segment matching any value for that segment.
+//! character). `arn_match` splits an ARN into its first five fields plus the
+//! resource field (everything after the fifth `:`) as one field, and matches
+//! each field with `wildcard_match`, so `*` and `?` in the resource field span
+//! `:` and `/`.
 
 /// Match a pattern against a value. Supports `*` and `?` wildcards.
 /// Comparison is case-sensitive.
@@ -87,11 +89,16 @@ fn wildcard_match_impl(p: &[u8], v: &[u8], ignore_case: bool) -> bool {
     pi == plen
 }
 
-/// Match an ARN pattern against an ARN value.
+/// Match an ARN pattern against an ARN value, the way IAM does.
 ///
-/// ARN matching is segment-aware: each colon-separated segment is matched
-/// independently using `wildcard_match`. Both pattern and value must have
-/// the same number of segments (6 for standard ARNs).
+/// The ARN is split into its first five fields (`arn`, partition, service,
+/// region, account) plus the resource field (everything after the fifth `:`)
+/// as a single field. Each field is matched with `wildcard_match`, so `*` and
+/// `?` in the resource field span `:` and `/`. This matters for stream ARNs,
+/// whose label contains `:` (for example
+/// `.../stream/2026-04-08T08:40:22.123`), and for sub-resources separated by
+/// `/` such as `.../index/<name>` and `.../stream/<label>`. Both sides must
+/// have the same number of fields.
 ///
 /// # Examples
 ///
@@ -102,8 +109,8 @@ fn wildcard_match_impl(p: &[u8], v: &[u8], ignore_case: bool) -> bool {
 ///     "arn:aws:dynamodb:us-east-1:123456789012:table/Users"
 /// ));
 /// assert!(arn_match(
-///     "arn:aws:dynamodb:*:*:table/User*",
-///     "arn:aws:dynamodb:us-east-1:123456789012:table/Users"
+///     "arn:aws:dynamodb:*:*:table/T/stream/*",
+///     "arn:aws:dynamodb:us-east-1:123456789012:table/T/stream/2026-04-08T08:40:22.123"
 /// ));
 /// ```
 #[must_use]
@@ -113,16 +120,18 @@ pub fn arn_match(pattern: &str, value: &str) -> bool {
         return true;
     }
 
-    let p_segments: Vec<&str> = pattern.split(':').collect();
-    let v_segments: Vec<&str> = value.split(':').collect();
+    // Split into at most six fields so the resource field (index 5) keeps its
+    // own `:` and `/`; wildcards in it then span those characters.
+    let p_fields: Vec<&str> = pattern.splitn(6, ':').collect();
+    let v_fields: Vec<&str> = value.splitn(6, ':').collect();
 
-    if p_segments.len() != v_segments.len() {
+    if p_fields.len() != v_fields.len() {
         return false;
     }
 
-    p_segments
+    p_fields
         .iter()
-        .zip(v_segments.iter())
+        .zip(v_fields.iter())
         .all(|(p, v)| wildcard_match(p, v))
 }
 
@@ -403,6 +412,57 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn arn_wildcard_stream_label_spans_colons() {
+        // A stream label contains ':'. A trailing '*' on the label must match
+        // across those colons.
+        let value = "arn:aws:dynamodb:us-east-1:123:table/T/stream/2026-10-20T16:17:40.188";
+        assert!(arn_match(
+            "arn:aws:dynamodb:us-east-1:123:table/T/stream/*",
+            value
+        ));
+        assert!(arn_match("arn:aws:dynamodb:*:*:table/T/stream/*", value));
+    }
+
+    #[test]
+    fn arn_exact_stream_label_matches() {
+        let value = "arn:aws:dynamodb:us-east-1:123:table/T/stream/2026-10-20T16:17:40.188";
+        assert!(arn_match(value, value));
+    }
+
+    #[test]
+    fn arn_resource_wildcard_spans_slash() {
+        // 'table/T/*' matches any sub-resource of T (index or stream).
+        assert!(arn_match(
+            "arn:aws:dynamodb:*:*:table/T/*",
+            "arn:aws:dynamodb:us-east-1:123:table/T/index/gsi1"
+        ));
+        assert!(arn_match(
+            "arn:aws:dynamodb:*:*:table/T/*",
+            "arn:aws:dynamodb:us-east-1:123:table/T/stream/2026-10-20T16:17:40.188"
+        ));
+    }
+
+    #[test]
+    fn arn_stream_wildcard_does_not_match_other_table() {
+        assert!(!arn_match(
+            "arn:aws:dynamodb:*:*:table/T/stream/*",
+            "arn:aws:dynamodb:us-east-1:123:table/Other/stream/2026-10-20T16:17:40.188"
+        ));
+    }
+
+    #[test]
+    fn arn_resource_field_count_is_fixed_at_six() {
+        // A value with extra ':' only in the resource still has six fields and
+        // matches an exact pattern field for field.
+        let value = "arn:aws:dynamodb:us-east-1:123:table/T/stream/a:b:c";
+        assert!(arn_match(value, value));
+        assert!(arn_match(
+            "arn:aws:dynamodb:us-east-1:123:table/T/stream/*",
+            value
+        ));
+    }
+
     // --- ip_in_cidr ---
 
     #[test]

crates/core/src/arn.rs

@@ -0,0 +1,220 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Strict parsing of `DynamoDB` resource ARNs.
+//!
+//! One parser serves request validation, authorization resource derivation, and
+//! the engine resolvers, so they never disagree about the resource an ARN names.
+//! When authorization reads a different table, stream, or backup from a field
+//! than the engine resolves from the same field, a Deny or a narrow Allow
+//! written against the real resource is bypassed. Every entry point that turns
+//! an ARN into a resource identity routes through [`parse`] (or [`fields`] for
+//! the raw five-field split), so a respelling that one layer accepts and another
+//! rejects cannot exist.
+//!
+//! [`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
+//! its own deployment) reads only the table and sub-resource.
+
+/// A table sub-resource: a secondary index, a stream, or a backup.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SubResource {
+    Index(String),
+    Stream(String),
+    Backup(String),
+}
+
+/// The five colon-separated fields of an ARN, with the resource kept whole.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ArnFields<'a> {
+    pub partition: &'a str,
+    pub service: &'a str,
+    pub region: &'a str,
+    pub account: &'a str,
+    /// Everything after the fifth `:`; keeps any `:` and `/` it contains.
+    pub resource: &'a str,
+}
+
+/// A parsed `DynamoDB` table resource ARN: its ARN fields, the table name, and
+/// an optional sub-resource.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct TableResource {
+    pub partition: String,
+    pub service: String,
+    pub region: String,
+    pub account: String,
+    pub table: String,
+    pub sub: Option<SubResource>,
+}
+
+/// 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
+/// label (`.../stream/2026-10-20T16:17:40.188`) survives intact. Returns `None`
+/// unless the string begins with `arn:` and has five colon-separated fields.
+#[must_use]
+pub fn fields(arn: &str) -> Option<ArnFields<'_>> {
+    let rest = arn.strip_prefix("arn:")?;
+    let mut it = rest.splitn(5, ':');
+    Some(ArnFields {
+        partition: it.next()?,
+        service: it.next()?,
+        region: it.next()?,
+        account: it.next()?,
+        resource: it.next()?,
+    })
+}
+
+/// Parse an ARN that names a `DynamoDB` table resource.
+///
+/// Returns `Some` only for
+/// `arn:<partition>:<service>:<region>:<account>:table/<name>` or
+/// `arn:<partition>:<service>:<region>:<account>:table/<name>/<kind>/<id>`
+/// with a non-empty table name and, when a sub-resource is present, a known
+/// kind (`index`, `stream`, or `backup`). The resource must split into exactly
+/// one segment (bare table) or three (table, kind, id); any other shape is
+/// `None`. Returns `None` for a missing `arn:` prefix, fewer than five fields,
+/// a resource that is not `table/...`, an empty table name, an unknown
+/// sub-resource kind, a trailing slash (`table/T/`), an empty sub-resource id
+/// (`table/T/stream/`), a repeated `table/` segment
+/// (`table/a/stream/table/b`), or an id that itself contains `/`
+/// (`table/T/stream/a/b`). Real stream labels, index names, and backup ids
+/// carry no `/`, so this is the service's notion of a well-formed resource ARN.
+#[must_use]
+pub fn parse(arn: &str) -> Option<TableResource> {
+    let f = fields(arn)?;
+    let tail = f.resource.strip_prefix("table/")?;
+    // `tail` is `<table>` or `<table>/<kind>/<id>`. Table names, kinds, and ids
+    // contain no `/`, so the resource splits into exactly 1 segment (a bare
+    // table) or 3 (table, kind, id). Any other count, including a repeated
+    // `table/` segment or an id that itself contains `/`, is not a well-formed
+    // DynamoDB resource ARN and is rejected. Real stream labels are timestamps,
+    // index names and backup ids carry no `/`, so nothing legitimate is lost,
+    // and the engine stream resolver (built on this) rejects exactly what the
+    // service treats as malformed.
+    let segs: Vec<&str> = tail.split('/').collect();
+    match segs.as_slice() {
+        [table] if !table.is_empty() => Some(TableResource {
+            partition: f.partition.to_owned(),
+            service: f.service.to_owned(),
+            region: f.region.to_owned(),
+            account: f.account.to_owned(),
+            table: (*table).to_owned(),
+            sub: None,
+        }),
+        [table, kind, id] if !table.is_empty() && !id.is_empty() => {
+            let sub = match *kind {
+                "index" => SubResource::Index((*id).to_owned()),
+                "stream" => SubResource::Stream((*id).to_owned()),
+                "backup" => SubResource::Backup((*id).to_owned()),
+                // An unrecognized sub-resource kind is not a nameable resource.
+                _ => return None,
+            };
+            Some(TableResource {
+                partition: f.partition.to_owned(),
+                service: f.service.to_owned(),
+                region: f.region.to_owned(),
+                account: f.account.to_owned(),
+                table: (*table).to_owned(),
+                sub: Some(sub),
+            })
+        }
+        _ => None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn fields_requires_arn_prefix_and_five_fields() {
+        assert_eq!(fields("banana"), None);
+        // A leading field other than `arn` has no `arn:` prefix.
+        assert_eq!(
+            fields("xrn:aws:dynamodb:us-east-1:1:table/T/stream/x"),
+            None
+        );
+        // An empty leading field (`:aws:...`) has no `arn:` prefix either.
+        assert_eq!(fields(":aws:dynamodb:us-east-1:1:table/T"), None);
+        assert_eq!(fields("arn:aws:dynamodb:us-east-1"), None);
+        let f = fields("arn:aws:dynamodb:us-east-1:1:table/T").unwrap();
+        assert_eq!(f.partition, "aws");
+        assert_eq!(f.service, "dynamodb");
+        assert_eq!(f.region, "us-east-1");
+        assert_eq!(f.account, "1");
+        assert_eq!(f.resource, "table/T");
+    }
+
+    #[test]
+    fn parses_table_index_stream_backup() {
+        assert_eq!(
+            parse("arn:aws:dynamodb:us-east-1:1:table/T").unwrap().sub,
+            None
+        );
+        assert_eq!(
+            parse("arn:aws:dynamodb:us-east-1:1:table/T/index/gsi1")
+                .unwrap()
+                .sub,
+            Some(SubResource::Index("gsi1".into()))
+        );
+        assert_eq!(
+            parse("arn:aws:dynamodb:us-east-1:1:table/T/stream/2026-10-20T16:17:40.188")
+                .unwrap()
+                .sub,
+            Some(SubResource::Stream("2026-10-20T16:17:40.188".into()))
+        );
+        assert_eq!(
+            parse("arn:aws:dynamodb:us-east-1:1:table/T/backup/01-ab")
+                .unwrap()
+                .sub,
+            Some(SubResource::Backup("01-ab".into()))
+        );
+    }
+
+    #[test]
+    fn captures_all_fields() {
+        let r = parse("arn:aws-cn:dynamodb:eu-west-1:111122223333:table/T/index/g").unwrap();
+        assert_eq!(r.partition, "aws-cn");
+        assert_eq!(r.service, "dynamodb");
+        assert_eq!(r.region, "eu-west-1");
+        assert_eq!(r.account, "111122223333");
+        assert_eq!(r.table, "T");
+        assert_eq!(r.sub, Some(SubResource::Index("g".into())));
+    }
+
+    #[test]
+    fn trailing_slash_empty_id_and_inner_slashes_are_rejected() {
+        // A trailing slash, an empty sub-resource id, a repeated table/
+        // segment, and an id containing '/' are all malformed: the resource
+        // must split into exactly one or three non-empty-table segments.
+        assert_eq!(parse("arn:aws:dynamodb:us-east-1:1:table/T/"), None);
+        assert_eq!(parse("arn:aws:dynamodb:us-east-1:1:table/T/stream/"), None);
+        assert_eq!(
+            parse("arn:aws:dynamodb:us-east-1:1:table/T/stream/a/b"),
+            None
+        );
+        assert_eq!(
+            parse("arn:aws:dynamodb:us-east-1:1:table/a/stream/table/b"),
+            None
+        );
+        assert_eq!(
+            parse("arn:aws:dynamodb:us-east-1:1:table/a/index/table/b"),
+            None
+        );
+    }
+
+    #[test]
+    fn rejects_non_table_empty_table_and_unknown_kind() {
+        assert_eq!(parse("banana"), None);
+        assert_eq!(parse("arn:aws:dynamodb:us-east-1:1:banana/x"), None);
+        assert_eq!(parse("arn:aws:dynamodb:us-east-1:1:table/"), None);
+        assert_eq!(parse("arn:aws:dynamodb:us-east-1:1:table//index/g"), None);
+        assert_eq!(parse("arn:aws:dynamodb:us-east-1:1:table/T/weird/x"), None);
+        // No `arn:` prefix: the A2 bypass input is rejected here, so every
+        // caller built on this parser rejects it.
+        assert_eq!(parse("xrn:aws:dynamodb:us-east-1:1:table/T/stream/x"), None);
+    }
+}

crates/core/src/lib.rs

@@ -7,6 +7,7 @@
 //! It defines the Virtual `DynamoDB` type system, request/response envelopes,
 //! validation logic, and error types shared across all other crates.
 
+pub mod arn;
 pub mod error;
 pub mod expression;
 pub mod limits;