fix(kinesis,ec2,sns,iam,ecs,secretsmanager,bedrock): validation + pagination correctness (bug-hunt) (#2286)

Author: vieiralucas

crates/fakecloud-bedrock/src/inference_profiles.rs

@@ -123,23 +123,22 @@ pub(crate) fn list_inference_profiles(
         .and_then(|v| v.parse::<usize>().ok())
         .unwrap_or(100)
         .max(1);
-    let next_token = req.query_params.get("nextToken");
+    // The nextToken is an opaque numeric offset into the combined result list.
+    // A prior version emitted the last item's Id but resumed by comparing it to
+    // the ARN, and computed the offset against the APPLICATION-only list before
+    // prepending SYSTEM_DEFINED profiles — so page 2 skipped or duplicated rows.
+    let start = req
+        .query_params
+        .get("nextToken")
+        .and_then(|t| t.parse::<usize>().ok())
+        .unwrap_or(0);
 
     let accts = state.read();
     let empty = crate::state::BedrockState::new(&req.account_id, &req.region);
     let s = accts.get(&req.account_id).unwrap_or(&empty);
     let mut items: Vec<&InferenceProfile> = s.inference_profiles.values().collect();
     items.sort_by(|a, b| a.inference_profile_arn.cmp(&b.inference_profile_arn));
 
-    let start = if let Some(token) = next_token {
-        items
-            .iter()
-            .position(|p| p.inference_profile_arn.as_str() > token.as_str())
-            .unwrap_or(items.len())
-    } else {
-        0
-    };
-
     // SYSTEM_DEFINED (AWS-managed cross-region) profiles exist out of the box;
     // list them alongside any user-created APPLICATION profiles. A `type`
     // filter, when present, narrows to one kind.
@@ -180,14 +179,8 @@ pub(crate) fn list_inference_profiles(
     let mut resp = json!({ "inferenceProfileSummaries": page });
     let end = start.saturating_add(max_results);
     if end < total {
-        // Page on the last item's id so the caller can continue.
-        if let Some(last) = resp["inferenceProfileSummaries"]
-            .as_array()
-            .and_then(|a| a.last())
-            .and_then(|v| v["inferenceProfileId"].as_str())
-        {
-            resp["nextToken"] = json!(last);
-        }
+        // Resume at the next offset into the combined list.
+        resp["nextToken"] = json!(end.to_string());
     }
 
     Ok(AwsResponse::ok_json(resp))

crates/fakecloud-e2e/tests/bedrock.rs

@@ -2353,3 +2353,53 @@ async fn bedrock_inference_profiles_catalogue_and_application_arn() {
         .inference_profile_arn()
         .contains("application-inference-profile/"));
 }
+
+#[tokio::test]
+async fn bedrock_list_inference_profiles_paginates_without_skip_or_dup() {
+    let server = TestServer::start().await;
+    let client = server.bedrock_client().await;
+
+    // Full list in one shot as the source of truth.
+    let all = client.list_inference_profiles().send().await.unwrap();
+    let expected: Vec<String> = all
+        .inference_profile_summaries()
+        .iter()
+        .map(|s| s.inference_profile_id().to_string())
+        .collect();
+    assert!(
+        expected.len() >= 2,
+        "need at least 2 profiles to exercise pagination"
+    );
+
+    // Page through with maxResults=1; the cursor must neither skip nor
+    // duplicate an entry (regression: token was an Id but resume compared ARNs
+    // and offset indexed the wrong list).
+    let mut collected: Vec<String> = Vec::new();
+    let mut token: Option<String> = None;
+    loop {
+        let mut req = client.list_inference_profiles().max_results(1);
+        if let Some(t) = &token {
+            req = req.next_token(t.clone());
+        }
+        let page = req.send().await.unwrap();
+        for s in page.inference_profile_summaries() {
+            collected.push(s.inference_profile_id().to_string());
+        }
+        match page.next_token() {
+            Some(t) => token = Some(t.to_string()),
+            None => break,
+        }
+    }
+
+    let mut got = collected.clone();
+    got.sort();
+    got.dedup();
+    assert_eq!(
+        got.len(),
+        collected.len(),
+        "pagination duplicated an entry: {collected:?}"
+    );
+    let mut exp_sorted = expected.clone();
+    exp_sorted.sort();
+    assert_eq!(got, exp_sorted, "pagination skipped or added entries");
+}

crates/fakecloud-e2e/tests/ec2_vpc_control_plane.rs

@@ -415,3 +415,43 @@ async fn replace_route_table_association_moves_main_to_new_table() {
         Some(new_rt_id.as_str())
     );
 }
+
+#[tokio::test]
+async fn associate_address_rejects_unknown_allocation_id() {
+    let server = TestServer::start().await;
+    let c = server.ec2_client().await;
+
+    // A real allocation associates fine.
+    let alloc = c
+        .allocate_address()
+        .domain(aws_sdk_ec2::types::DomainType::Vpc)
+        .send()
+        .await
+        .expect("allocate_address");
+    let alloc_id = alloc.allocation_id().expect("allocation_id").to_string();
+    let ok = c
+        .associate_address()
+        .allocation_id(&alloc_id)
+        .network_interface_id("eni-00000000000000000")
+        .send()
+        .await
+        .expect("associate real allocation");
+    assert!(ok.association_id().is_some());
+
+    // An unknown AllocationId must error, not fabricate a phantom association.
+    let err = c
+        .associate_address()
+        .allocation_id("eipalloc-deadbeefdeadbeef0")
+        .network_interface_id("eni-00000000000000000")
+        .send()
+        .await
+        .expect_err("unknown allocation must be rejected");
+    // EC2's query-protocol error envelope isn't parsed into meta().code() by
+    // aws-sdk-ec2 (a separate, pre-existing format gap), so assert on the raw
+    // error text which carries the code.
+    let msg = format!("{}", aws_sdk_ec2::error::DisplayErrorContext(&err));
+    assert!(
+        msg.contains("InvalidAllocationID.NotFound"),
+        "expected InvalidAllocationID.NotFound, got: {msg}"
+    );
+}

crates/fakecloud-e2e/tests/secretsmanager.rs

@@ -315,6 +315,54 @@ async fn secretsmanager_batch_get_secret_value() {
     assert!(resp.errors().is_empty());
 }
 
+#[tokio::test]
+async fn secretsmanager_batch_get_secret_value_paginates_with_filters() {
+    use aws_sdk_secretsmanager::types::{Filter, FilterNameStringType};
+    let server = TestServer::start().await;
+    let client = server.secretsmanager_client().await;
+
+    for name in ["pag-a", "pag-b", "pag-c"] {
+        client
+            .create_secret()
+            .name(name)
+            .secret_string("v")
+            .send()
+            .await
+            .unwrap();
+    }
+
+    let name_filter = Filter::builder()
+        .key(FilterNameStringType::Name)
+        .values("pag-")
+        .build();
+
+    // Page 1: MaxResults=2 must return 2 and a NextToken.
+    let page1 = client
+        .batch_get_secret_value()
+        .filters(name_filter.clone())
+        .max_results(2)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(page1.secret_values().len(), 2);
+    let token = page1
+        .next_token()
+        .expect("NextToken when more results remain")
+        .to_string();
+
+    // Page 2: resumes past the first two, returning the last one, no token.
+    let page2 = client
+        .batch_get_secret_value()
+        .filters(name_filter)
+        .max_results(2)
+        .next_token(token)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(page2.secret_values().len(), 1);
+    assert!(page2.next_token().is_none());
+}
+
 #[tokio::test]
 async fn secretsmanager_get_random_password() {
     let server = TestServer::start().await;

crates/fakecloud-e2e/tests/sns.rs

@@ -264,6 +264,45 @@ async fn sns_publish_delivers_to_sqs_subscriber() {
     assert_eq!(envelope["TopicArn"], topic_arn);
 }
 
+/// Subscribe with an unknown protocol must be rejected with InvalidParameter
+/// instead of creating a phantom subscription that can never deliver.
+#[tokio::test]
+async fn sns_subscribe_rejects_unknown_protocol() {
+    let server = TestServer::start().await;
+    let sns = server.sns_client().await;
+    let topic_arn = sns
+        .create_topic()
+        .name("proto-topic")
+        .send()
+        .await
+        .unwrap()
+        .topic_arn()
+        .unwrap()
+        .to_string();
+
+    let err = sns
+        .subscribe()
+        .topic_arn(&topic_arn)
+        .protocol("carrier-pigeon")
+        .endpoint("nowhere")
+        .send()
+        .await
+        .expect_err("unknown protocol must be rejected");
+    assert_eq!(
+        err.into_service_error().meta().code(),
+        Some("InvalidParameter")
+    );
+
+    // No phantom subscription was created.
+    let subs = sns
+        .list_subscriptions_by_topic()
+        .topic_arn(&topic_arn)
+        .send()
+        .await
+        .unwrap();
+    assert!(subs.subscriptions().is_empty());
+}
+
 /// An SQS subscriber whose target queue is gone must route the notification to
 /// the subscription's RedrivePolicy DLQ instead of silently dropping it — the
 /// DLQ was previously honored for HTTP subscribers only.

crates/fakecloud-ec2/src/service/eip.rs

@@ -184,14 +184,43 @@ pub(crate) fn associate_address(
     req: &AwsRequest,
 ) -> Result<AwsResponse, AwsServiceError> {
     let assoc_id = gen_id("eipassoc");
-    if let Some(alloc) = req.query_params.get("AllocationId") {
-        let mut accounts = svc.state.write();
-        let state = accounts.get_or_create(&req.account_id);
-        if let Some(e) = state.elastic_ips.get_mut(alloc) {
-            e.association_id = Some(assoc_id.clone());
-            e.instance_id = req.query_params.get("InstanceId").cloned();
-            e.network_interface_id = req.query_params.get("NetworkInterfaceId").cloned();
+    let mut accounts = svc.state.write();
+    let state = accounts.get_or_create(&req.account_id);
+    // Resolve the target address by AllocationId (VPC) or PublicIp (Classic).
+    // An unknown identifier must error rather than fabricate a phantom
+    // association that no Describe would ever reflect.
+    let target_alloc = if let Some(alloc) = req.query_params.get("AllocationId") {
+        if !state.elastic_ips.contains_key(alloc) {
+            return Err(not_found("InvalidAllocationID.NotFound", alloc));
+        }
+        alloc.clone()
+    } else if let Some(ip) = req.query_params.get("PublicIp") {
+        match state
+            .elastic_ips
+            .values()
+            .find(|e| &e.public_ip == ip)
+            .map(|e| e.allocation_id.clone())
+        {
+            Some(id) => id,
+            None => {
+                return Err(AwsServiceError::aws_error(
+                    http::StatusCode::BAD_REQUEST,
+                    "InvalidAddress.NotFound",
+                    format!("Address '{ip}' not found."),
+                ));
+            }
         }
+    } else {
+        return Err(AwsServiceError::aws_error(
+            http::StatusCode::BAD_REQUEST,
+            "MissingParameter",
+            "The request must contain either AllocationId or PublicIp.".to_string(),
+        ));
+    };
+    if let Some(e) = state.elastic_ips.get_mut(&target_alloc) {
+        e.association_id = Some(assoc_id.clone());
+        e.instance_id = req.query_params.get("InstanceId").cloned();
+        e.network_interface_id = req.query_params.get("NetworkInterfaceId").cloned();
     }
     Ok(Ec2Service::respond(
         "AssociateAddress",

crates/fakecloud-ecs/src/helpers.rs

@@ -87,13 +87,29 @@ pub(crate) fn opt_str<'a>(body: &'a Value, field: &str) -> Option<&'a str> {
 /// an empty page — restarting at the beginning would silently re-list page 1
 /// and risk an infinite pagination loop for the client.
 pub(crate) fn paginate_arns(
+    arns: Vec<String>,
+    next_token: &str,
+    max_results: usize,
+) -> (Vec<String>, Option<String>) {
+    paginate_arns_dir(arns, next_token, max_results, false)
+}
+
+/// Like [`paginate_arns`] but honors a sort direction. `descending` reverses the
+/// canonical (ascending, deduped) order and adjusts the resume cursor
+/// comparison so DESC pages advance correctly instead of being silently
+/// re-sorted back to ASC.
+pub(crate) fn paginate_arns_dir(
     mut arns: Vec<String>,
     next_token: &str,
     max_results: usize,
+    descending: bool,
 ) -> (Vec<String>, Option<String>) {
     use base64::Engine;
     arns.sort();
     arns.dedup();
+    if descending {
+        arns.reverse();
+    }
     let cursor = if next_token.is_empty() {
         None
     } else {
@@ -106,9 +122,17 @@ pub(crate) fn paginate_arns(
         0
     } else {
         match &cursor {
+            // Resume just past the cursor. For ASC the next element is the first
+            // greater than the cursor; for DESC it's the first less than it.
             Some(c) => arns
                 .iter()
-                .position(|a| a.as_str() > c.as_str())
+                .position(|a| {
+                    if descending {
+                        a.as_str() < c.as_str()
+                    } else {
+                        a.as_str() > c.as_str()
+                    }
+                })
                 .unwrap_or(arns.len()),
             // Non-empty but undecodable token: a cursor we never emitted.
             // Treat as past the end (empty page) rather than restarting at 0.
@@ -1571,7 +1595,21 @@ pub(crate) fn task_protection_json(task: &Task) -> Value {
 
 #[cfg(test)]
 mod pagination_tests {
-    use super::paginate_arns;
+    use super::{paginate_arns, paginate_arns_dir};
+
+    #[test]
+    fn paginate_arns_dir_descending_pages_in_reverse() {
+        let all: Vec<String> = (0..5).map(|i| format!("arn:{i}")).collect();
+        // DESC page 1: highest first.
+        let (p1, next) = paginate_arns_dir(all.clone(), "", 2, true);
+        assert_eq!(p1, vec!["arn:4", "arn:3"]);
+        let token = next.expect("nextToken after DESC page 1");
+        // DESC page 2 continues downward, not silently re-sorted to ASC.
+        let (p2, next2) = paginate_arns_dir(all.clone(), &token, 2, true);
+        assert_eq!(p2, vec!["arn:2", "arn:1"]);
+        let (p3, _) = paginate_arns_dir(all, &next2.unwrap(), 2, true);
+        assert_eq!(p3, vec!["arn:0"]);
+    }
 
     #[test]
     fn paginate_arns_key_cursor_is_stable_across_delete() {

crates/fakecloud-ecs/src/service_task_definitions.rs

@@ -337,13 +337,7 @@ impl EcsService {
                 }
             }
         }
-        if sort == "DESC" {
-            arns.sort();
-            arns.reverse();
-        } else {
-            arns.sort();
-        }
-        let (page, next) = paginate_arns(arns, next_token, max_results);
+        let (page, next) = paginate_arns_dir(arns, next_token, max_results, sort == "DESC");
         let mut out = json!({"taskDefinitionArns": page});
         if let Some(n) = next {
             out.as_object_mut()