Repository: getmoto/moto
Pull request #8917 (2025-05-22 11:00:29)

EC2: Implement ModifySecurityGroupRules
This PR implements the EC2 [ModifySecurityGroupRules](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_ModifySecurityGroupRules.html) operation. Only IPv4 CIDR ranges are supported.

There is an accompanying AWS validated test.


Patch:
diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index 81fdfb9b9..43f97c9a5 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -3185,7 +3185,7 @@
 - [ ] modify_public_ip_dns_name_options
 - [ ] modify_reserved_instances
 - [ ] modify_route_server
-- [ ] modify_security_group_rules
+- [X] modify_security_group_rules
 - [ ] modify_snapshot_attribute
 - [ ] modify_snapshot_tier
 - [X] modify_spot_fleet_request
@@ -10365,4 +10365,4 @@
 - workmailmessageflow
 - workspaces-thin-client
 - xray
-</details>
\ No newline at end of file
+</details>
diff --git a/docs/docs/services/ec2.rst b/docs/docs/services/ec2.rst
index 90dfe366d..d1cefa531 100644
--- a/docs/docs/services/ec2.rst
+++ b/docs/docs/services/ec2.rst
@@ -628,7 +628,7 @@ ec2
 - [ ] modify_public_ip_dns_name_options
 - [ ] modify_reserved_instances
 - [ ] modify_route_server
-- [ ] modify_security_group_rules
+- [X] modify_security_group_rules
 - [ ] modify_snapshot_attribute
 - [ ] modify_snapshot_tier
 - [X] modify_spot_fleet_request
diff --git a/moto/batch/models.py b/moto/batch/models.py
index 7eeac5fb6..ffa34d066 100644
--- a/moto/batch/models.py
+++ b/moto/batch/models.py
@@ -1219,6 +1219,7 @@ class BatchBackend(BaseBackend):
                 "type": environment.env_type,
                 "status": "VALID",
                 "statusReason": "Compute environment is available",
+                "tags": self.list_tags_for_resource(arn),
             }
             if environment.env_type == "MANAGED":
                 json_part["computeResources"] = environment.compute_resources
@@ -1234,6 +1235,7 @@ class BatchBackend(BaseBackend):
         state: str,
         compute_resources: Dict[str, Any],
         service_role: str,
+        tags: Optional[Dict[str, str]] = None,
     ) -> ComputeEnvironment:
         # Validate
         if COMPUTE_ENVIRONMENT_NAME_REGEX.match(compute_environment_name) is None:
@@ -1289,6 +1291,9 @@ class BatchBackend(BaseBackend):
         )
         self._compute_environments[new_comp_env.arn] = new_comp_env
 
+        if tags:
+            self.tag_resource(new_comp_env.arn, tags)
+
         # Ok by this point, everything is legit, so if its Managed then start some instances
         if _type == "MANAGED" and "FARGATE" not in compute_resources["type"]:
             cpus = int(
diff --git a/moto/batch/responses.py b/moto/batch/responses.py
index 6d1c1eb93..db59fed0d 100644
--- a/moto/batch/responses.py
+++ b/moto/batch/responses.py
@@ -30,6 +30,7 @@ class BatchResponse(BaseResponse):
         service_role = self._get_param("serviceRole")
         state = self._get_param("state")
         _type = self._get_param("type")
+        tags = self._get_param("tags")
 
         env = self.batch_backend.create_compute_environment(
             compute_environment_name=compute_env_name,
@@ -37,6 +38,7 @@ class BatchResponse(BaseResponse):
             state=state,
             compute_resources=compute_resource,
             service_role=service_role,
+            tags=tags,
         )
 
         result = {"computeEnvironmentArn": env.arn, "computeEnvironmentName": env.name}
diff --git a/moto/ec2/exceptions.py b/moto/ec2/exceptions.py
index 4ae7eaaf7..5d3d848be 100644
--- a/moto/ec2/exceptions.py
+++ b/moto/ec2/exceptions.py
@@ -828,6 +828,22 @@ class InvalidParameter(EC2ClientError):
         )
 
 
+class InvalidParameterValue(EC2ClientError):
+    def __init__(self, message: str):
+        super().__init__(
+            "InvalidParameterValue",
+            message,
+        )
+
+
+class MissingParameter(EC2ClientError):
+    def __init__(self, message: str):
+        super().__init__(
+            "MissingParameter",
+            message,
+        )
+
+
 class InvalidSubnetCidrBlockAssociationID(EC2ClientError):
     def __init__(self, association_id: str):
         super().__init__(
diff --git a/moto/ec2/models/security_groups.py b/moto/ec2/models/security_groups.py
index 5bb11d39f..cf3f02d3a 100644
--- a/moto/ec2/models/security_groups.py
+++ b/moto/ec2/models/security_groups.py
@@ -10,11 +10,14 @@ from moto.core.utils import aws_api_matches
 from ..exceptions import (
     InvalidCIDRSubnetError,
     InvalidGroupIdMalformedError,
+    InvalidParameterCombination,
+    InvalidParameterValue,
     InvalidPermissionDuplicateError,
     InvalidPermissionNotFoundError,
     InvalidSecurityGroupDuplicateError,
     InvalidSecurityGroupNotFoundError,
     InvalidSecurityGroupRuleIdNotFoundError,
+    MissingParameter,
     MissingParameterError,
     MotoNotImplementedError,
     RulesPerSecurityGroupLimitExceededError,
@@ -524,6 +527,13 @@ class SecurityGroup(TaggedEC2Resource, CloudFormationModel):
             return self.id
         raise UnformattedGetAttTemplateException()
 
+    def get_rule(self, rule_id: str) -> Optional[SecurityRule]:
+        """Retrieve a security group rule by its ID."""
+        for rule in list(itertools.chain(self.egress_rules, self.ingress_rules)):
+            if rule.id == rule_id:
+                return rule
+        return None
+
     def add_ingress_rule(self, rule: SecurityRule) -> None:
         if rule in self.ingress_rules:
             raise InvalidPermissionDuplicateError()
@@ -785,6 +795,69 @@ class SecurityGroupBackend:
                 tags=tags,
             )
 
+    def modify_security_group_rules(
+        self,
+        group_id: str,
+        rules: dict[str, dict[str, Any]],
+    ) -> None:
+        group = self.get_security_group_by_name_or_id(group_id)
+        if group is None:
+            raise InvalidSecurityGroupNotFoundError(group_id)
+
+        for rule_id, new_rule in rules.items():
+            cidr_ipv4 = new_rule.get("CidrIpv4")
+            cidr_ipv6 = new_rule.get("CidrIpv6")
+            prefix_list_id = new_rule.get("PrefixListId")
+            reference_group_id = new_rule.get("ReferenceGroupId")
+
+            set_param_count = sum(
+                item is not None
+                for item in [cidr_ipv4, cidr_ipv6, prefix_list_id, reference_group_id]
+            )
+            if set_param_count > 1:
+                raise InvalidParameterCombination(
+                    "Only one of cidrIp, cidrIpv6, prefixListId, or referencedGroupId can be specified"
+                )
+            if set_param_count < 1:
+                raise MissingParameter(
+                    "The request must contain exactly one of: cidrIp, cidrIpv6, prefixListId, or referencedGroupId"
+                )
+
+            existing_rule = group.get_rule(rule_id)
+            if existing_rule is None:
+                raise InvalidSecurityGroupRuleIdNotFoundError(rule_id)
+
+            ip_protocol = new_rule.get("IpProtocol")
+            if ip_protocol is None:
+                raise InvalidParameterValue(
+                    "Invalid value 'null' for protocol. VPC security group rules must specify protocols explicitly."
+                )
+
+            from_port = new_rule.get("FromPort")
+            to_port = new_rule.get("ToPort")
+            if from_port is None or to_port is None:
+                raise InvalidParameterValue(
+                    "Invalid value for portRange. Must specify both from and to ports with TCP/UDP."
+                )
+
+            if cidr_ipv4 and "CidrIpv6" in existing_rule.ip_range:
+                raise InvalidParameterValue(
+                    f"Invalid rule type for security group rule '{rule_id}'. You may not specify CidrIpv6 for an existing IPv4 CIDR rule."
+                )
+
+            existing_rule.ip_protocol = ip_protocol
+            existing_rule.from_port = from_port
+            existing_rule.to_port = to_port
+
+            # TODO: Handle cidr_ipv6, prefix_list_id and reference_group_id
+
+            description = new_rule.get("Description")
+            if description is not None:
+                existing_rule.ip_range["Description"] = description
+
+            if cidr_ipv4 is not None:
+                existing_rule.ip_range["CidrIp"] = cidr_ipv4
+
     def authorize_security_group_ingress(
         self,
         group_name_or_id: str,
diff --git a/moto/ec2/responses/security_groups.py b/moto/ec2/responses/security_groups.py
index 1b05adc6a..717603525 100644
--- a/moto/ec2/responses/security_groups.py
+++ b/moto/ec2/responses/security_groups.py
@@ -221,6 +221,19 @@ class SecurityGroups(EC2BaseResponse):
             self.ec2_backend.revoke_security_group_ingress(**args)
         return REVOKE_SECURITY_GROUP_INGRESS_RESPONSE
 
+    def modify_security_group_rules(self) -> str:
+        self.error_on_dryrun()
+
+        rules = {}
+        security_group_rules_param = self._get_params()["SecurityGroupRule"]
+        for idx, sgr in security_group_rules_param.items():
+            rules[sgr["SecurityGroupRuleId"]] = sgr["SecurityGroupRule"]
+
+        group_id = self._get_param("GroupId")
+        self.ec2_backend.modify_security_group_rules(group_id, rules)
+
+        return MODIFY_SECURITY_GROUP_RULES_RESPONSE
+
     def update_security_group_rule_descriptions_ingress(self) -> str:
         for args in self._process_rules_from_querystring():
             # we don't need this parameter to revoke
@@ -577,6 +590,13 @@ REVOKE_SECURITY_GROUP_EGRESS_RESPONSE = """<RevokeSecurityGroupEgressResponse xm
   <return>true</return>
 </RevokeSecurityGroupEgressResponse>"""
 
+MODIFY_SECURITY_GROUP_RULES_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
+<ModifySecurityGroupRulesResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">
+  <requestId>04d8f970-9213-41fa-81d2-50825EXAMPLE</requestId>
+  <return>true</return>
+</ModifySecurityGroupRulesResponse>
+"""
+
 UPDATE_SECURITY_GROUP_RULE_DESCRIPTIONS_INGRESS = """<UpdateSecurityGroupRuleDescriptionsIngressResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
   <requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
   <return>true</return>


Test patch:
diff --git a/.github/workflows/tests_sdk_ruby.yml b/.github/workflows/tests_sdk_ruby.yml
index 26d85d51f..bc5e97bfe 100644
--- a/.github/workflows/tests_sdk_ruby.yml
+++ b/.github/workflows/tests_sdk_ruby.yml
@@ -13,7 +13,7 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - name: Set up Ruby ${{ matrix.ruby-version }}
-        uses: ruby/setup-ruby@c95ae3725f6ebdd095f2bd19caed7ebc14435ba5
+        uses: ruby/setup-ruby@13e7a03dc3ac6c3798f4570bfead2aed4d96abfb
         with:
           ruby-version: ${{ matrix.ruby-version }}
       - name: Set up Python 3.12
diff --git a/tests/test_batch/test_batch_compute_envs.py b/tests/test_batch/test_batch_compute_envs.py
index 37b6f43b9..455d9fb3a 100644
--- a/tests/test_batch/test_batch_compute_envs.py
+++ b/tests/test_batch/test_batch_compute_envs.py
@@ -60,6 +60,62 @@ def test_create_managed_compute_environment(use_arn_for_instance_role):
     assert our_env["ecsClusterArn"] in all_clusters
 
 
+@mock_aws
+@pytest.mark.parametrize("use_arn_for_instance_role", [True, False])
+def test_create_managed_compute_environment_with_tags(use_arn_for_instance_role):
+    ec2_client, iam_client, ecs_client, _, batch_client = _get_clients()
+    _, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)
+
+    instance_role = (
+        iam_arn.replace("role", "instance-profile")
+        if use_arn_for_instance_role
+        else iam_arn.split("/")[-1]
+    )
+    compute_name = str(uuid4())
+    resp = batch_client.create_compute_environment(
+        computeEnvironmentName=compute_name,
+        type="MANAGED",
+        state="ENABLED",
+        computeResources={
+            "type": "EC2",
+            "minvCpus": 5,
+            "maxvCpus": 10,
+            "desiredvCpus": 5,
+            "instanceTypes": ["t2.small", "t2.medium"],
+            "imageId": "some_image_id",
+            "subnets": [subnet_id],
+            "securityGroupIds": [sg_id],
+            "ec2KeyPair": "string",
+            "instanceRole": instance_role,
+            "tags": {"string": "string"},
+            "bidPercentage": 123,
+            "spotIamFleetRole": "string",
+        },
+        serviceRole=iam_arn,
+        tags={"name": "compute-environment-1"},
+    )
+    assert "computeEnvironmentArn" in resp
+    assert resp["computeEnvironmentName"] == compute_name
+
+    our_env = batch_client.describe_compute_environments(
+        computeEnvironments=[compute_name]
+    )["computeEnvironments"][0]
+
+    # Given a t2.medium is 2 vcpu and t2.small is 1, therefore 2 mediums and 1 small should be created
+    if not settings.TEST_SERVER_MODE:
+        # Can't verify this in ServerMode, as other tests may have created instances
+        resp = ec2_client.describe_instances()
+        assert "Reservations" in resp
+        assert len(resp["Reservations"]) == 3
+
+    # Should have created 1 ECS cluster
+    all_clusters = ecs_client.list_clusters()["clusterArns"]
+    assert our_env["ecsClusterArn"] in all_clusters
+
+    assert "tags" in our_env
+    assert len(our_env["tags"]) == 1
+
+
 @mock_aws
 def test_create_managed_compute_environment_with_instance_family():
     """
diff --git a/tests/test_ec2/test_security_groups.py b/tests/test_ec2/test_security_groups.py
index 63bc9b74d..9a979c062 100644
--- a/tests/test_ec2/test_security_groups.py
+++ b/tests/test_ec2/test_security_groups.py
@@ -768,6 +768,256 @@ def test_create_and_describe_security_grp_rule():
     _verify_egress_rule(rules[0])
 
 
+@ec2_aws_verified(create_vpc=True, create_sg=True)
+@pytest.mark.aws_verified
+def test_modify_security_group_rules(ec2_client=None, vpc_id=None, sg_id=None):
+    client = ec2_client
+    group_id = sg_id
+
+    ip_permissions = [
+        {
+            "IpProtocol": "tcp",
+            "FromPort": 30900,
+            "ToPort": 31000,
+            "IpRanges": [
+                {"CidrIp": "1.2.3.4/32", "Description": "original rule"},
+            ],
+        }
+    ]
+    rule_id = client.authorize_security_group_ingress(
+        GroupId=group_id, IpPermissions=ip_permissions
+    )["SecurityGroupRules"][0]["SecurityGroupRuleId"]
+
+    # Ensure bad group ID raises
+    with pytest.raises(ClientError) as ex:
+        client.modify_security_group_rules(
+            GroupId="sg-00000000",
+            SecurityGroupRules=[
+                {
+                    "SecurityGroupRuleId": rule_id,
+                    "SecurityGroupRule": {
+                        "IpProtocol": "udp",
+                        "FromPort": 27010,
+                        "ToPort": 27011,
+                        "CidrIpv4": "2.3.4.5/32",
+                        "Description": "modification that should fail",
+                    },
+                }
+            ],
+        )
+    assert ex.value.response["Error"]["Code"] == "InvalidGroup.NotFound"
+    assert (
+        ex.value.response["Error"]["Message"]
+        == "The security group 'sg-00000000' does not exist"
+    )
+
+    # Ensure bad rule ID raises
+    with pytest.raises(ClientError) as ex:
+        client.modify_security_group_rules(
+            GroupId=group_id,
+            SecurityGroupRules=[
+                {
+                    "SecurityGroupRuleId": "sgr-00000000",
+                    "SecurityGroupRule": {
+                        "IpProtocol": "udp",
+                        "FromPort": 27010,
+                        "ToPort": 27011,
+                        "CidrIpv4": "2.3.4.5/32",
+                        "Description": "modification that should fail",
+                    },
+                }
+            ],
+        )
+    assert ex.value.response["Error"]["Code"] == "InvalidSecurityGroupRuleId.NotFound"
+    assert (
+        ex.value.response["Error"]["Message"]
+        == "The security group rule ID 'sgr-00000000' does not exist"
+    )
+
+    # Ensure missing protocol raises
+    with pytest.raises(ClientError) as ex:
+        client.modify_security_group_rules(
+            GroupId=group_id,
+            SecurityGroupRules=[
+                {
+                    "SecurityGroupRuleId": rule_id,
+                    "SecurityGroupRule": {
+                        "FromPort": 27010,
+                        "ToPort": 27011,
+                        "CidrIpv4": "2.3.4.5/32",
+                        "Description": "modification that should fail",
+                    },
+                }
+            ],
+        )
+    assert ex.value.response["Error"]["Code"] == "InvalidParameterValue"
+    assert (
+        ex.value.response["Error"]["Message"]
+        == "Invalid value 'null' for protocol. VPC security group rules must specify protocols explicitly."
+    )
+
+    # Ensure missing port range raises
+    with pytest.raises(ClientError) as ex:
+        client.modify_security_group_rules(
+            GroupId=group_id,
+            SecurityGroupRules=[
+                {
+                    "SecurityGroupRuleId": rule_id,
+                    "SecurityGroupRule": {
+                        "IpProtocol": "udp",
+                        "ToPort": 27011,
+                        "CidrIpv4": "2.3.4.5/32",
+                        "Description": "modification that should fail",
+                    },
+                }
+            ],
+        )
+    assert ex.value.response["Error"]["Code"] == "InvalidParameterValue"
+    assert (
+        ex.value.response["Error"]["Message"]
+        == "Invalid value for portRange. Must specify both from and to ports with TCP/UDP."
+    )
+
+    # Ensure missing cidr raises
+    with pytest.raises(ClientError) as ex:
+        client.modify_security_group_rules(
+            GroupId=group_id,
+            SecurityGroupRules=[
+                {
+                    "SecurityGroupRuleId": rule_id,
+                    "SecurityGroupRule": {
+                        "IpProtocol": "udp",
+                        "FromPort": 27010,
+                        "ToPort": 27011,
+                        "Description": "modification that should fail",
+                    },
+                }
+            ],
+        )
+    assert ex.value.response["Error"]["Code"] == "MissingParameter"
+    assert (
+        ex.value.response["Error"]["Message"]
+        == "The request must contain exactly one of: cidrIp, cidrIpv6, prefixListId, or referencedGroupId"
+    )
+
+    # Ensure multiple cidr raises
+    with pytest.raises(ClientError) as ex:
+        client.modify_security_group_rules(
+            GroupId=group_id,
+            SecurityGroupRules=[
+                {
+                    "SecurityGroupRuleId": rule_id,
+                    "SecurityGroupRule": {
+                        "IpProtocol": "udp",
+                        "FromPort": 27010,
+                        "ToPort": 27011,
+                        "CidrIpv4": "2.3.4.5/32",
+                        "CidrIpv6": "2001:db8::/32",
+                        "Description": "modification that should fail ",
+                    },
+                }
+            ],
+        )
+    assert ex.value.response["Error"]["Code"] == "InvalidParameterCombination"
+    assert (
+        ex.value.response["Error"]["Message"]
+        == "Only one of cidrIp, cidrIpv6, prefixListId, or referencedGroupId can be specified"
+    )
+
+    # Happy path: Ensure all attributes can be modified
+    response = client.modify_security_group_rules(
+        GroupId=group_id,
+        SecurityGroupRules=[
+            {
+                "SecurityGroupRuleId": rule_id,
+                "SecurityGroupRule": {
+                    "IpProtocol": "udp",
+                    "FromPort": 26000,
+                    "ToPort": 26500,
+                    "CidrIpv4": "2.3.4.5/32",
+                    "Description": "first modification",
+                },
+            }
+        ],
+    )
+    assert response["Return"] is True
+
+    response = client.describe_security_group_rules(
+        Filters=[
+            {"Name": "group-id", "Values": [group_id]},
+            {"Name": "security-group-rule-id", "Values": [rule_id]},
+        ]
+    )
+    assert len(response["SecurityGroupRules"]) == 1
+    assert response["SecurityGroupRules"][0]["IpProtocol"] == "udp"
+    assert response["SecurityGroupRules"][0]["FromPort"] == 26000
+    assert response["SecurityGroupRules"][0]["ToPort"] == 26500
+    assert response["SecurityGroupRules"][0]["CidrIpv4"] == "2.3.4.5/32"
+    assert response["SecurityGroupRules"][0]["Description"] == "first modification"
+    assert "CidrIpv6" not in response["SecurityGroupRules"][0]
+
+    # Add another rule
+    rule_id_2 = client.authorize_security_group_egress(
+        GroupId=group_id, IpPermissions=ip_permissions
+    )["SecurityGroupRules"][0]["SecurityGroupRuleId"]
+
+    # Ensure multiple sg rules can be modified together
+    response = client.modify_security_group_rules(
+        GroupId=group_id,
+        SecurityGroupRules=[
+            {
+                "SecurityGroupRuleId": rule_id,
+                "SecurityGroupRule": {
+                    "IpProtocol": "tcp",
+                    "FromPort": 10,
+                    "ToPort": 11,
+                    "CidrIpv4": "1.1.1.1/32",
+                    "Description": "second modification",
+                },
+            },
+            {
+                "SecurityGroupRuleId": rule_id_2,
+                "SecurityGroupRule": {
+                    "IpProtocol": "udp",
+                    "FromPort": 20,
+                    "ToPort": 21,
+                    "CidrIpv4": "2.2.2.2/32",
+                    "Description": "first modification",
+                },
+            },
+        ],
+    )
+    assert response["Return"] is True
+
+    # Ensure first rule
+    response = client.describe_security_group_rules(
+        Filters=[
+            {"Name": "group-id", "Values": [group_id]},
+            {"Name": "security-group-rule-id", "Values": [rule_id]},
+        ]
+    )
+    assert len(response["SecurityGroupRules"]) == 1
+    assert response["SecurityGroupRules"][0]["IpProtocol"] == "tcp"
+    assert response["SecurityGroupRules"][0]["FromPort"] == 10
+    assert response["SecurityGroupRules"][0]["ToPort"] == 11
+    assert response["SecurityGroupRules"][0]["CidrIpv4"] == "1.1.1.1/32"
+    assert response["SecurityGroupRules"][0]["Description"] == "second modification"
+
+    # Ensure second rule
+    response = client.describe_security_group_rules(
+        Filters=[
+            {"Name": "group-id", "Values": [group_id]},
+            {"Name": "security-group-rule-id", "Values": [rule_id_2]},
+        ]
+    )
+    assert len(response["SecurityGroupRules"]) == 1
+    assert response["SecurityGroupRules"][0]["IpProtocol"] == "udp"
+    assert response["SecurityGroupRules"][0]["FromPort"] == 20
+    assert response["SecurityGroupRules"][0]["ToPort"] == 21
+    assert response["SecurityGroupRules"][0]["CidrIpv4"] == "2.2.2.2/32"
+    assert response["SecurityGroupRules"][0]["Description"] == "first modification"
+
+
 @mock_aws
 @pytest.mark.parametrize("use_vpc", [True, False], ids=["Use VPC", "Without VPC"])
 def test_sec_group_rule_limit(use_vpc):