fix(ec2): DescribeVolumes MaxResults guard + change_state TOCTOU + empty-filter-value (bug-hunt)

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

@@ -820,6 +820,41 @@ fn transition_allowed(current: i64, new_code: i64) -> bool {
     true
 }
 
+/// Validate a single instance's state transition + stop/termination protection,
+/// returning the exact error AWS returns when it is illegal. Shared by the
+/// pre-flight read check and the re-check under the write lock so both critical
+/// sections enforce identical rules (TOCTOU-safe: a concurrent Terminate/Stop
+/// landing between the read and the write must fail this call, not be silently
+/// clobbered — bug-hunt 2026-07 finding 4.1).
+fn check_transition(inst: &Instance, id: &str, new_code: i64) -> Result<(), AwsServiceError> {
+    if !transition_allowed(inst.state_code, new_code) {
+        return Err(crate::service_helpers::incorrect_instance_state(
+            id,
+            &inst.state_name,
+        ));
+    }
+    // Termination / stop protection.
+    if new_code == 48 && inst.disable_api_termination {
+        return Err(AwsServiceError::aws_error(
+            http::StatusCode::BAD_REQUEST,
+            "OperationNotPermitted",
+            format!(
+                "The instance '{id}' may not be terminated. Modify its 'disableApiTermination' instance attribute and try again."
+            ),
+        ));
+    }
+    if new_code == 80 && inst.disable_api_stop {
+        return Err(AwsServiceError::aws_error(
+            http::StatusCode::BAD_REQUEST,
+            "OperationNotPermitted",
+            format!(
+                "The instance '{id}' may not be stopped. Modify its 'disableApiStop' instance attribute and try again."
+            ),
+        ));
+    }
+    Ok(())
+}
+
 async fn change_state(
     svc: &Ec2Service,
     req: &AwsRequest,
@@ -841,31 +876,7 @@ async fn change_state(
                 .instances
                 .get(id)
                 .ok_or_else(|| crate::service_helpers::instance_not_found(id))?;
-            if !transition_allowed(inst.state_code, new_code) {
-                return Err(crate::service_helpers::incorrect_instance_state(
-                    id,
-                    &inst.state_name,
-                ));
-            }
-            // Termination / stop protection.
-            if new_code == 48 && inst.disable_api_termination {
-                return Err(AwsServiceError::aws_error(
-                    http::StatusCode::BAD_REQUEST,
-                    "OperationNotPermitted",
-                    format!(
-                        "The instance '{id}' may not be terminated. Modify its 'disableApiTermination' instance attribute and try again."
-                    ),
-                ));
-            }
-            if new_code == 80 && inst.disable_api_stop {
-                return Err(AwsServiceError::aws_error(
-                    http::StatusCode::BAD_REQUEST,
-                    "OperationNotPermitted",
-                    format!(
-                        "The instance '{id}' may not be stopped. Modify its 'disableApiStop' instance attribute and try again."
-                    ),
-                ));
-            }
+            check_transition(inst, id, new_code)?;
         }
     }
 
@@ -882,6 +893,22 @@ async fn change_state(
     {
         let mut accounts = svc.state.write();
         let state = accounts.get_or_create(&req.account_id);
+        // Re-run existence + transition/protection checks against the freshly
+        // re-read state INSIDE the write lock before mutating anything. The
+        // read-phase check above ran under a different (dropped) lock, so a
+        // concurrent Terminate/Stop could have landed in between; without this
+        // re-check a StartInstances that passed the read check would overwrite
+        // a terminated instance's state code, resurrecting it past the boot
+        // task's terminal guard (bug-hunt 2026-07 finding 4.1). AWS applies the
+        // whole call atomically, so any id now failing fails the entire call
+        // with nothing mutated (the guard is still held, so no partial writes).
+        for id in &ids {
+            let inst = state
+                .instances
+                .get(id)
+                .ok_or_else(|| crate::service_helpers::instance_not_found(id))?;
+            check_transition(inst, id, new_code)?;
+        }
         for id in &ids {
             let (prev_code, prev_name) = state
                 .instances
@@ -2328,4 +2355,111 @@ mod modify_tests {
         assert_eq!(inst.state_code, 16);
         assert_eq!(inst.state_name, "running");
     }
+
+    fn state_of(svc: &Ec2Service, id: &str) -> (i64, Option<String>) {
+        let accounts = svc.state.read();
+        let inst = &accounts.get("000000000000").unwrap().instances[id];
+        (inst.state_code, inst.container_id.clone())
+    }
+
+    #[test]
+    fn check_transition_enforces_terminal_and_protection() {
+        // Direct unit test of the shared re-check helper the write-lock TOCTOU
+        // guard relies on (bug-hunt finding 4.1).
+        let svc = Ec2Service::new();
+        seed_instance(&svc, "i-1");
+        let mut accounts = svc.state.write();
+        let state = accounts.get_or_create("000000000000");
+        let inst = state.instances.get_mut("i-1").unwrap();
+
+        // Running -> start/stop/terminate all legal.
+        assert!(check_transition(inst, "i-1", 16).is_ok());
+        assert!(check_transition(inst, "i-1", 80).is_ok());
+        assert!(check_transition(inst, "i-1", 48).is_ok());
+
+        // Terminated is terminal: no Start (16) allowed, only re-terminate.
+        inst.state_code = 48;
+        inst.state_name = "terminated".into();
+        assert_eq!(
+            check_transition(inst, "i-1", 16).unwrap_err().code(),
+            "IncorrectInstanceState"
+        );
+        assert!(check_transition(inst, "i-1", 48).is_ok());
+
+        // Protection flags map to OperationNotPermitted.
+        inst.state_code = 16;
+        inst.state_name = "running".into();
+        inst.disable_api_termination = true;
+        assert_eq!(
+            check_transition(inst, "i-1", 48).unwrap_err().code(),
+            "OperationNotPermitted"
+        );
+        inst.disable_api_termination = false;
+        inst.disable_api_stop = true;
+        assert_eq!(
+            check_transition(inst, "i-1", 80).unwrap_err().code(),
+            "OperationNotPermitted"
+        );
+    }
+
+    #[tokio::test]
+    async fn start_after_terminate_does_not_resurrect() {
+        // End-to-end: once terminated, a StartInstances must be rejected rather
+        // than overwriting code 48 with pending (the resurrection the write-lock
+        // re-check closes). Metadata-only mode has no runtime, so the terminal
+        // state is fully determined by the control-plane path.
+        let svc = Ec2Service::new();
+        seed_instance(&svc, "i-1");
+
+        terminate_instances(&svc, &req("TerminateInstances", &[("InstanceId.1", "i-1")]))
+            .await
+            .unwrap();
+        let (code, container) = state_of(&svc, "i-1");
+        assert_eq!(code, 48, "terminate must set code 48");
+        assert_eq!(container, None, "terminate must drop the container handle");
+
+        let err = start_instances(&svc, &req("StartInstances", &[("InstanceId.1", "i-1")]))
+            .await
+            .err()
+            .expect("starting a terminated instance must fail");
+        assert_eq!(err.code(), "IncorrectInstanceState");
+
+        // State is untouched: still terminated, no partial resurrection.
+        let (code, _) = state_of(&svc, "i-1");
+        assert_eq!(code, 48, "instance must stay terminated");
+    }
+
+    #[tokio::test]
+    async fn change_state_is_atomic_on_mixed_ids() {
+        // AWS applies the whole call or none: a batch where one id is illegal
+        // (terminated) must fail entirely and leave the other id untouched.
+        let svc = Ec2Service::new();
+        seed_instance(&svc, "i-ok");
+        seed_instance(&svc, "i-dead");
+        terminate_instances(
+            &svc,
+            &req("TerminateInstances", &[("InstanceId.1", "i-dead")]),
+        )
+        .await
+        .unwrap();
+
+        let err = start_instances(
+            &svc,
+            &req(
+                "StartInstances",
+                &[("InstanceId.1", "i-ok"), ("InstanceId.2", "i-dead")],
+            ),
+        )
+        .await
+        .err()
+        .expect("batch with a terminated id must fail");
+        assert_eq!(err.code(), "IncorrectInstanceState");
+
+        // i-ok must NOT have been flipped to pending by a partial write.
+        let (code, _) = state_of(&svc, "i-ok");
+        assert_eq!(
+            code, 16,
+            "healthy instance must be untouched on atomic fail"
+        );
+    }
 }

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

@@ -7,7 +7,7 @@ use fakecloud_core::service::{AwsRequest, AwsResponse, AwsServiceError};
 use crate::service::Ec2Service;
 use crate::service_helpers::{
     filter_value_matches, gen_id, indexed_list, not_found, paginate, parse_filters, require,
-    validate_enum, Filter,
+    validate_enum, validate_max_results, Filter,
 };
 use crate::state::{Ec2State, Tag, Volume, VolumeAttachment};
 
@@ -159,6 +159,7 @@ pub(crate) fn describe_volumes(
     svc: &Ec2Service,
     req: &AwsRequest,
 ) -> Result<AwsResponse, AwsServiceError> {
+    validate_max_results(&req.query_params, 5, 1000)?;
     let filters = parse_filters(&req.query_params);
     let wanted = indexed_list(&req.query_params, "VolumeId");
     let accounts = svc.state.read();
@@ -798,15 +799,43 @@ mod tests {
     #[test]
     fn describe_volumes_paginates() {
         let svc = Ec2Service::new();
-        for i in 0..3 {
+        // MaxResults must be within [5, 1000]; seed enough volumes to force a
+        // second page at the minimum page size.
+        for i in 0..6 {
             seed_volume(&svc, &format!("vol-{i}"), "available", false);
         }
         let body = body_of(
-            describe_volumes(&svc, &req("DescribeVolumes", &[("MaxResults", "2")])).unwrap(),
+            describe_volumes(&svc, &req("DescribeVolumes", &[("MaxResults", "5")])).unwrap(),
         );
         assert!(body.contains("<nextToken>"), "expected a NextToken: {body}");
     }
 
+    #[test]
+    fn describe_volumes_rejects_max_results_zero() {
+        // MaxResults=0 with >=1 volume previously produced a self-referential
+        // NextToken=0 that looped forever (bug-hunt finding 1.1). Like its
+        // sibling paginators, DescribeVolumes must reject out-of-range
+        // MaxResults with InvalidParameterValue instead.
+        let svc = Ec2Service::new();
+        seed_volume(&svc, "vol-1", "available", false);
+        let err = err_of(describe_volumes(
+            &svc,
+            &req("DescribeVolumes", &[("MaxResults", "0")]),
+        ));
+        assert_eq!(err.code(), "InvalidParameterValue");
+    }
+
+    #[test]
+    fn describe_volumes_rejects_max_results_above_max() {
+        let svc = Ec2Service::new();
+        seed_volume(&svc, "vol-1", "available", false);
+        let err = err_of(describe_volumes(
+            &svc,
+            &req("DescribeVolumes", &[("MaxResults", "1001")]),
+        ));
+        assert_eq!(err.code(), "InvalidParameterValue");
+    }
+
     #[test]
     fn attach_volume_rejects_nonexistent() {
         let svc = Ec2Service::new();

crates/fakecloud-ec2/src/service_helpers.rs

@@ -255,8 +255,11 @@ pub fn validate_length(
 /// Collect a 1-based indexed list, e.g. `ResourceId.1`, `ResourceId.2`, ….
 ///
 /// EC2 list members are contiguous from index 1; collection stops at the first
-/// missing index. Empty values terminate the list too (matching how the SDKs
-/// never emit a gap).
+/// missing index. A present-but-empty value terminates the list too — for
+/// id-lists (`InstanceId.N`, `GroupId.N`, `VolumeId.N`, …) an empty member is
+/// meaningless, and treating it as a terminator matches how the SDKs never emit
+/// a gap. Filter *values* differ (an empty value is a legitimate member); use
+/// [`indexed_list_keep_empty`] there.
 pub fn indexed_list(params: &HashMap<String, String>, prefix: &str) -> Vec<String> {
     let mut out = Vec::new();
     let mut i = 1usize;
@@ -271,6 +274,25 @@ pub fn indexed_list(params: &HashMap<String, String>, prefix: &str) -> Vec<Strin
     out
 }
 
+/// Like [`indexed_list`], but a present-but-empty value (`Filter.1.Value.1=`)
+/// is preserved as a legitimate empty-string member rather than terminating the
+/// list — the terminator is "next index absent", not "value empty". Used for
+/// filter values, where filtering for an absent/empty tag value is valid and an
+/// empty value must NOT truncate (or self-referentially loop) the list.
+pub fn indexed_list_keep_empty(params: &HashMap<String, String>, prefix: &str) -> Vec<String> {
+    let mut out = Vec::new();
+    let mut i = 1usize;
+    loop {
+        let key = format!("{prefix}.{i}");
+        match params.get(&key) {
+            Some(v) => out.push(v.clone()),
+            None => break,
+        }
+        i += 1;
+    }
+    out
+}
+
 /// Parse `Filter.N.Name` + `Filter.N.Value.M` into [`Filter`] entries.
 pub fn parse_filters(params: &HashMap<String, String>) -> Vec<Filter> {
     let mut out = Vec::new();
@@ -280,7 +302,7 @@ pub fn parse_filters(params: &HashMap<String, String>) -> Vec<Filter> {
         let Some(name) = params.get(&name_key).filter(|v| !v.is_empty()) else {
             break;
         };
-        let values = indexed_list(params, &format!("Filter.{i}.Value"));
+        let values = indexed_list_keep_empty(params, &format!("Filter.{i}.Value"));
         out.push(Filter {
             name: name.clone(),
             values,
@@ -339,6 +361,46 @@ mod tests {
         assert_eq!(indexed_list(&params, "ResourceId"), vec!["vpc-1"]);
     }
 
+    #[test]
+    fn indexed_list_empty_value_terminates_for_id_lists() {
+        // Plain id-lists keep the original semantics: a present-but-empty member
+        // is meaningless and terminates (SDKs never emit an empty id member).
+        let params = p(&[("InstanceId.1", ""), ("InstanceId.2", "i-2")]);
+        assert_eq!(indexed_list(&params, "InstanceId"), Vec::<String>::new());
+    }
+
+    #[test]
+    fn indexed_list_keep_empty_preserves_present_but_empty_value() {
+        // `Filter.1.Value.1=` (explicit empty string) is a legitimate member,
+        // not a terminator: it must be kept, and a following contiguous index
+        // must still be collected rather than truncated at the empty one.
+        let params = p(&[("Value.1", ""), ("Value.2", "x")]);
+        assert_eq!(indexed_list_keep_empty(&params, "Value"), vec!["", "x"]);
+    }
+
+    #[test]
+    fn indexed_list_keep_empty_then_absent_stops() {
+        // A trailing empty value followed by an absent index still terminates
+        // (no infinite loop on a genuinely-absent index).
+        let params = p(&[("Value.1", "")]);
+        assert_eq!(indexed_list_keep_empty(&params, "Value"), vec![""]);
+    }
+
+    #[test]
+    fn parse_filters_keeps_empty_filter_value() {
+        // Filtering for an empty/absent tag value: `Filter.1.Value.1=` must be
+        // preserved as a single empty-string value, not dropped.
+        let params = p(&[("Filter.1.Name", "tag:env"), ("Filter.1.Value.1", "")]);
+        let filters = parse_filters(&params);
+        assert_eq!(
+            filters,
+            vec![Filter {
+                name: "tag:env".into(),
+                values: vec!["".into()]
+            }]
+        );
+    }
+
     #[test]
     fn parse_filters_groups_name_and_values() {
         let params = p(&[