Repository: getmoto/moto
Pull request: EC2: SG rule tagging
Closes https://github.com/getmoto/moto/issues/7408

Problem statement:
EC2: security group rule bad behaviours
First Issue:

I have create an sg rule like that:

```
response = ec2_client.authorize_security_group_ingress(
             GroupId=group_id,
            IpPermissions=[
                  {
                      'IpProtocol': rule['ip_protocol'],
                      'FromPort': rule['from_port'],
                      'ToPort': rule['to_port'],
                      'IpRanges': [{'CidrIp': rule['cidr_ipv4'], 'Description': 'Ip test'} if 'cidr_ipv4' in rule else {'CidrIpv6': rule['cidr_ipv6'],'Description': 'Ip test'}]
                }
            ],
            TagSpecifications=[{
                'ResourceType': 'security-group-rule',
                'Tags': [
                    {
                        'Key': 'Automation',
                        'Value': 'Lambda'
                    },
                    {
                        'Key': 'Partner',
                        'Value': 'test'
                    },
                ]
            
            }]
        )
```

It's return a response with no tags:
```
{'SecurityGroupRules': [{'SecurityGroupRuleId': 'sgr-f1cb3d13e054dcc7a', 'GroupId': 'sg-b54b06d08c2abe95b', 'GroupOwnerId': '123456789012', 'IsEgress': True, 'IpProtocol': '-1', 'CidrIpv4': '0.0.0.0/0', 'Tags': []}, {'SecurityGroupRuleId': 'sgr-8537473175c427978', 'GroupId': 'sg-0a0ae1fd2ff8bd88a', 'GroupOwnerId': '123456789012', 'IsEgress': False, 'IpProtocol': 'tcp', 'FromPort': 80, 'ToPort': 80, 'CidrIpv4': '1.1.1.1/32', 'Tags': []}], ...}

```

The workaround is to create tags in a second step with ec2.create_tags...

Second issue:

```
response = ec2_client.describe_security_group_rules(Filters=[{'Name': 'tag:Partner', 'Values': ['test]}])
```
Response:
```
botocore.exceptions.ClientError: An error occurred (InvalidGroup.NotFound) when calling the DescribeSecurityGroupRules operation: The security group 'No security groups found matching the filters provided.' does not exist

```
Without using Filter argument:

```
{'SecurityGroupRules': [{'SecurityGroupRuleId': 'sgr-440c08b2647bb596a', 'GroupId': 'sg-47513810a50b65e5c', 'GroupOwnerId': '123456789012', 'IsEgress': True, 'IpProtocol': '-1', 'CidrIpv4': '0.0.0.0/0', 'Tags': []}, {'SecurityGroupRuleId': 'sgr-5d0d181c0dbca42b8', 'GroupId': 'sg-e7fb2d342af75c7cd', 'GroupOwnerId': '123456789012', 'IsEgress': False, 'IpProtocol': 'tcp', 'FromPort': 80, 'ToPort': 80, 'CidrIpv4': '1.1.1.1/32', 'Tags': [{'Key': 'Automation', 'Value': 'Lambda'}, {'Key': 'Partner', 'Value': 'test'}]}], ...}}

```

Hints from issue discussion:
@bblommers I can take a look

Commit message: chore: run ci

Patch to review:
diff --git a/moto/ec2/models/core.py b/moto/ec2/models/core.py
index 0ecc94dcf930..581e44dd3fcd 100644
--- a/moto/ec2/models/core.py
+++ b/moto/ec2/models/core.py
@@ -41,3 +41,10 @@ def get_filter_value(
             return value
 
         raise FilterNotImplementedError(filter_name, method_name)
+
+    def match_tags(self, filters: Dict[str, str]) -> bool:
+        for tag_name in filters.keys():
+            tag_value = self.get_filter_value(tag_name)
+            if tag_value == filters[tag_name][0]:
+                return True
+        return False
diff --git a/moto/ec2/models/security_groups.py b/moto/ec2/models/security_groups.py
index b432cca3920a..0a776d76f06b 100644
--- a/moto/ec2/models/security_groups.py
+++ b/moto/ec2/models/security_groups.py
@@ -33,12 +33,15 @@ def __init__(
         self,
         ec2_backend: Any,
         ip_protocol: str,
+        group_id: str,
         from_port: Optional[str],
         to_port: Optional[str],
         ip_ranges: Optional[List[Any]],
         source_groups: List[Dict[str, Any]],
         prefix_list_ids: Optional[List[Dict[str, str]]] = None,
         is_egress: bool = True,
+        tags: Dict[str, str] = {},
+        description: str = "",
     ):
         self.ec2_backend = ec2_backend
         self.id = random_security_group_rule_id()
@@ -48,6 +51,8 @@ def __init__(
         self.prefix_list_ids = prefix_list_ids or []
         self.from_port = self.to_port = None
         self.is_egress = is_egress
+        self.description = description
+        self.group_id = group_id
 
         if self.ip_protocol and self.ip_protocol != "-1":
             self.from_port = int(from_port)  # type: ignore[arg-type]
@@ -73,6 +78,7 @@ def __init__(
             else None
         )
         self.ip_protocol = proto if proto else self.ip_protocol
+        self.add_tags(tags)
 
     @property
     def owner_id(self) -> str:
@@ -156,6 +162,7 @@ def __init__(
                     SecurityRule(
                         self.ec2_backend,
                         "-1",
+                        self.id,
                         None,
                         None,
                         [{"CidrIp": "0.0.0.0/0"}],
@@ -167,6 +174,7 @@ def __init__(
                     SecurityRule(
                         self.ec2_backend,
                         "-1",
+                        self.id,
                         None,
                         None,
                         [{"CidrIpv6": "::/0"}],
@@ -571,22 +579,60 @@ def describe_security_groups(
         return matches
 
     def describe_security_group_rules(
-        self, group_ids: Optional[List[str]] = None, filters: Any = None
-    ) -> Dict[str, List[SecurityRule]]:
-        matches = self.describe_security_groups(group_ids=group_ids, filters=filters)
-        if not matches:
-            raise InvalidSecurityGroupNotFoundError(
-                "No security groups found matching the filters provided."
+        self,
+        group_ids: Optional[List[str]] = None,
+        sg_rule_ids: List[str] = [],
+        filters: Any = None,
+    ) -> List[SecurityRule]:
+        results = []
+
+        if sg_rule_ids:
+            # go thru all the rules in the backend to find a match
+            for sg_rule_id in sg_rule_ids:
+                for sg in self.sg_old_ingress_ruls:
+                    for rule in self.sg_old_ingress_ruls[sg]:
+                        if rule.id == sg_rule_id:
+                            results.append(rule)
+
+            return results
+
+        if group_ids:
+            all_sgs = self.describe_security_groups(group_ids=group_ids)
+            for group in all_sgs:
+                results.extend(group.ingress_rules)
+                results.extend(group.egress_rules)
+
+            return results
+
+        if filters and "group-id" in filters:
+            matches = self.describe_security_groups(
+                group_ids=group_ids, filters=filters
             )
-        rules = {}
-        for group in matches:
-            group_rules = []
-            group_rules.extend(group.ingress_rules)
-            group_rules.extend(group.egress_rules)
-            if group_rules:
-                rules[group.group_id] = group_rules
+            if not matches:
+                raise InvalidSecurityGroupNotFoundError(
+                    "No security groups found matching the filters provided."
+                )
+            for group in matches:
+                results.extend(group.ingress_rules)
+                results.extend(group.egress_rules)
+
+            return results
+
+        all_sgs = self.describe_security_groups()
 
-        return rules
+        for group in all_sgs:
+            results.extend(self._match_sg_rules(group.ingress_rules, filters))
+            results.extend(self._match_sg_rules(group.egress_rules, filters))
+
+        return results
+
+    @staticmethod
+    def _match_sg_rules(rules_list: List[SecurityRule], filters: Any) -> List[SecurityRule]:  # type: ignore[misc]
+        results = []
+        for rule in rules_list:
+            if rule.match_tags(filters):
+                results.append(rule)
+        return results
 
     def _delete_security_group(self, vpc_id: Optional[str], group_id: str) -> None:
         vpc_id = vpc_id or self.default_vpc.id  # type: ignore[attr-defined]
@@ -657,6 +703,7 @@ def authorize_security_group_ingress(
         from_port: str,
         to_port: str,
         ip_ranges: List[Any],
+        sgrule_tags: Dict[str, str] = {},
         source_groups: Optional[List[Dict[str, str]]] = None,
         prefix_list_ids: Optional[List[Dict[str, str]]] = None,
         security_rule_ids: Optional[List[str]] = None,  # pylint:disable=unused-argument
@@ -695,12 +742,14 @@ def authorize_security_group_ingress(
         security_rule = SecurityRule(
             self,
             ip_protocol,
+            group.group_id,
             from_port,
             to_port,
             ip_ranges,
             _source_groups,
             prefix_list_ids,
             is_egress=False,
+            tags=sgrule_tags,
         )
 
         if security_rule in group.ingress_rules:
@@ -738,6 +787,7 @@ def authorize_security_group_ingress(
                 break
         else:
             group.add_ingress_rule(security_rule)
+
         return security_rule, group
 
     def revoke_security_group_ingress(
@@ -765,6 +815,7 @@ def revoke_security_group_ingress(
         security_rule = SecurityRule(
             self,
             ip_protocol,
+            group.group_id,
             from_port,
             to_port,
             ip_ranges,
@@ -819,6 +870,7 @@ def authorize_security_group_egress(
         from_port: str,
         to_port: str,
         ip_ranges: List[Any],
+        sgrule_tags: Dict[str, str] = {},
         source_groups: Optional[List[Dict[str, Any]]] = None,
         prefix_list_ids: Optional[List[Dict[str, str]]] = None,
         security_rule_ids: Optional[List[str]] = None,  # pylint:disable=unused-argument
@@ -860,11 +912,13 @@ def authorize_security_group_egress(
         security_rule = SecurityRule(
             self,
             ip_protocol,
+            group.group_id,
             from_port,
             to_port,
             ip_ranges,
             _source_groups,
             prefix_list_ids,
+            tags=sgrule_tags,
         )
 
         if security_rule in group.egress_rules:
@@ -944,6 +998,7 @@ def revoke_security_group_egress(
         security_rule = SecurityRule(
             self,
             ip_protocol,
+            group.group_id,
             from_port,
             to_port,
             ip_ranges,
@@ -1030,6 +1085,7 @@ def update_security_group_rule_descriptions_ingress(
         security_rule = SecurityRule(
             self,
             ip_protocol,
+            group.group_id,
             from_port,
             to_port,
             ip_ranges,
@@ -1085,6 +1141,7 @@ def update_security_group_rule_descriptions_egress(
         security_rule = SecurityRule(
             self,
             ip_protocol,
+            group.group_id,
             from_port,
             to_port,
             ip_ranges,
diff --git a/moto/ec2/responses/security_groups.py b/moto/ec2/responses/security_groups.py
index b65949291432..5ef75b6d35cc 100644
--- a/moto/ec2/responses/security_groups.py
+++ b/moto/ec2/responses/security_groups.py
@@ -86,6 +86,15 @@ def _process_rules_from_querystring(self) -> Any:
                 d = d[subkey]
             d[key_splitted[-1]] = value
 
+        sg_rule_tags = {}
+        qs_tags = querytree.get("TagSpecification", {})
+        if qs_tags and 1 in qs_tags and "Tag" in qs_tags[1]:
+            temp_tags = {}
+            for tag_item in qs_tags[1]["Tag"].items():
+                temp_tags[tag_item[1]["Key"][0]] = tag_item[1]["Value"][0]
+            if qs_tags[1]["ResourceType"][0] == "security-group-rule":
+                sg_rule_tags = temp_tags
+
         if "IpPermissions" not in querytree:
             # Handle single rule syntax
             (
@@ -97,16 +106,17 @@ def _process_rules_from_querystring(self) -> Any:
                 prefix_list_ids,
             ) = parse_sg_attributes_from_dict(querytree)
 
-            yield (
-                group_name_or_id,
-                ip_protocol,
-                from_port,
-                to_port,
-                ip_ranges,
-                source_groups,
-                prefix_list_ids,
-                security_rule_ids,
-            )
+            yield {
+                "group_name_or_id": group_name_or_id,
+                "ip_protocol": ip_protocol,
+                "from_port": from_port,
+                "to_port": to_port,
+                "ip_ranges": ip_ranges,
+                "sgrule_tags": sg_rule_tags,
+                "source_groups": source_groups,
+                "prefix_list_ids": prefix_list_ids,
+                "security_rule_ids": security_rule_ids,
+            }
 
         ip_permissions = querytree.get("IpPermissions") or {}
         for ip_permission_idx in sorted(ip_permissions.keys()):
@@ -121,22 +131,23 @@ def _process_rules_from_querystring(self) -> Any:
                 prefix_list_ids,
             ) = parse_sg_attributes_from_dict(ip_permission)
 
-            yield (
-                group_name_or_id,
-                ip_protocol,
-                from_port,
-                to_port,
-                ip_ranges,
-                source_groups,
-                prefix_list_ids,
-                security_rule_ids,
-            )
+            yield {
+                "group_name_or_id": group_name_or_id,
+                "ip_protocol": ip_protocol,
+                "from_port": from_port,
+                "to_port": to_port,
+                "ip_ranges": ip_ranges,
+                "sgrule_tags": sg_rule_tags,
+                "source_groups": source_groups,
+                "prefix_list_ids": prefix_list_ids,
+                "security_rule_ids": security_rule_ids,
+            }
 
     def authorize_security_group_egress(self) -> str:
         self.error_on_dryrun()
 
         for args in self._process_rules_from_querystring():
-            rule, group = self.ec2_backend.authorize_security_group_egress(*args)
+            rule, group = self.ec2_backend.authorize_security_group_egress(**args)
         self.ec2_backend.sg_old_egress_ruls[group.id] = group.egress_rules.copy()
         template = self.response_template(AUTHORIZE_SECURITY_GROUP_EGRESS_RESPONSE)
         return template.render(rule=rule, group=group)
@@ -145,7 +156,7 @@ def authorize_security_group_ingress(self) -> str:
         self.error_on_dryrun()
 
         for args in self._process_rules_from_querystring():
-            rule, group = self.ec2_backend.authorize_security_group_ingress(*args)
+            rule, group = self.ec2_backend.authorize_security_group_ingress(**args)
         self.ec2_backend.sg_old_ingress_ruls[group.id] = group.ingress_rules.copy()
         template = self.response_template(AUTHORIZE_SECURITY_GROUP_INGRESS_RESPONSE)
         return template.render(rule=rule, group=group)
@@ -198,11 +209,19 @@ def describe_security_groups(self) -> str:
 
     def describe_security_group_rules(self) -> str:
         group_id = self._get_param("GroupId")
+        sg_rule_ids = self._get_param("SecurityGroupRuleId.1")
         filters = self._filters_from_querystring()
 
         self.error_on_dryrun()
 
-        rules = self.ec2_backend.describe_security_group_rules(group_id, filters)
+        # if sg rule ids are not None then wrap in a list
+        # as expected by ec2_backend.describe_security_group_rules
+        if sg_rule_ids:
+            sg_rule_ids = [sg_rule_ids]
+
+        rules = self.ec2_backend.describe_security_group_rules(
+            group_id, sg_rule_ids, filters
+        )
         template = self.response_template(DESCRIBE_SECURITY_GROUP_RULES_RESPONSE)
         return template.render(rules=rules)
 
@@ -210,28 +229,36 @@ def revoke_security_group_egress(self) -> str:
         self.error_on_dryrun()
 
         for args in self._process_rules_from_querystring():
-            self.ec2_backend.revoke_security_group_egress(*args)
+            # we don't need this parameter to revoke
+            del args["sgrule_tags"]
+            self.ec2_backend.revoke_security_group_egress(**args)
         return REVOKE_SECURITY_GROUP_EGRESS_RESPONSE
 
     def revoke_security_group_ingress(self) -> str:
         self.error_on_dryrun()
 
         for args in self._process_rules_from_querystring():
-            self.ec2_backend.revoke_security_group_ingress(*args)
+            # we don't need this parameter to revoke
+            del args["sgrule_tags"]
+            self.ec2_backend.revoke_security_group_ingress(**args)
         return REVOKE_SECURITY_GROUP_INGRESS_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
+            del args["sgrule_tags"]
             group = self.ec2_backend.update_security_group_rule_descriptions_ingress(
-                *args
+                **args
             )
         self.ec2_backend.sg_old_ingress_ruls[group.id] = group.ingress_rules.copy()
         return UPDATE_SECURITY_GROUP_RULE_DESCRIPTIONS_INGRESS
 
     def update_security_group_rule_descriptions_egress(self) -> str:
         for args in self._process_rules_from_querystring():
+            # we don't need this parameter to revoke
+            del args["sgrule_tags"]
             group = self.ec2_backend.update_security_group_rule_descriptions_egress(
-                *args
+                **args
             )
         self.ec2_backend.sg_old_egress_ruls[group.id] = group.egress_rules.copy()
         return UPDATE_SECURITY_GROUP_RULE_DESCRIPTIONS_EGRESS
@@ -255,8 +282,7 @@ def update_security_group_rule_descriptions_egress(self) -> str:
 <DescribeSecurityGroupRulesResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">
   <requestId>{{ request_id }}</requestId>
   <securityGroupRuleSet>
-    {% for group, rule_list in rules.items() %}
-        {% for rule in rule_list %}
+        {% for rule in rules %}
             <item>
                 {% if rule.from_port is not none %}
                 <fromPort>{{ rule.from_port }}</fromPort>
@@ -268,7 +294,7 @@ def update_security_group_rule_descriptions_egress(self) -> str:
                   <cidrIpv4>{{ rule.ip_ranges[0]['CidrIp'] }}</cidrIpv4>
                 {% endif %}
                 <ipProtocol>{{ rule.ip_protocol }}</ipProtocol>
-                <groupId>{{ group }}</groupId>
+                <groupId>{{ rule.group_id }}</groupId>
                 <groupOwnerId>{{ rule.owner_id }}</groupOwnerId>
                 <isEgress>{{ 'true' if rule.is_egress else 'false' }}</isEgress>
                 <securityGroupRuleId>{{ rule.id }}</securityGroupRuleId>
@@ -282,7 +308,6 @@ def update_security_group_rule_descriptions_egress(self) -> str:
                 </tagSet> 
             </item>
         {% endfor %}
-    {% endfor %}
   </securityGroupRuleSet>
 </DescribeSecurityGroupRulesResponse>"""
 
@@ -470,6 +495,14 @@ def update_security_group_rule_descriptions_egress(self) -> str:
             {% if rule.to_port is not none %}
             <toPort>{{ rule.to_port }}</toPort>
             {% endif %}
+            <tagSet>
+                {% for tag in rule.get_tags() %}
+                    <item>
+                      <key>{{ tag.key }}</key>
+                      <value>{{ tag.value }}</value>
+                    </item>
+                {% endfor %}
+            </tagSet> 
         </item>
     {% endfor %}
     {% for item in rule.prefix_list_ids %}
@@ -550,6 +583,14 @@ def update_security_group_rule_descriptions_egress(self) -> str:
             {% if rule.to_port is not none %}
             <toPort>{{ rule.to_port }}</toPort>
             {% endif %}
+            <tagSet>
+                {% for tag in rule.get_tags() %}
+                    <item>
+                      <key>{{ tag.key }}</key>
+                      <value>{{ tag.value }}</value>
+                    </item>
+                {% endfor %}
+            </tagSet> 
         </item>
     {% endfor %}
     {% for item in rule.prefix_list_ids %}
diff --git a/tests/test_ec2/test_security_groups.py b/tests/test_ec2/test_security_groups.py
index d826c1578d6a..3486c395c15a 100644
--- a/tests/test_ec2/test_security_groups.py
+++ b/tests/test_ec2/test_security_groups.py
@@ -13,11 +13,13 @@
 from moto.core import DEFAULT_ACCOUNT_ID
 from moto.ec2 import ec2_backends
 
+REGION = "us-east-1"
+
 
 @mock_aws
 def test_create_and_describe_security_group():
-    ec2 = boto3.resource("ec2", "us-west-1")
-    client = boto3.client("ec2", "us-west-1")
+    ec2 = boto3.resource("ec2", REGION)
+    client = boto3.client("ec2", REGION)
 
     with pytest.raises(ClientError) as ex:
         client.create_security_group(GroupName="test", Description="test", DryRun=True)
@@ -51,7 +53,7 @@ def test_create_and_describe_security_group():
 
 @mock_aws
 def test_create_security_group_without_description_raises_error():
-    ec2 = boto3.resource("ec2", "us-west-1")
+    ec2 = boto3.resource("ec2", REGION)
 
     with pytest.raises(ClientError) as ex:
         ec2.create_security_group(GroupName="test security group", Description="")
@@ -62,15 +64,15 @@ def test_create_security_group_without_description_raises_error():
 
 @mock_aws
 def test_default_security_group():
-    client = boto3.client("ec2", "us-west-1")
+    client = boto3.client("ec2", REGION)
     groups = retrieve_all_sgs(client)
     assert "default" in [g["GroupName"] for g in groups]
 
 
 @mock_aws
 def test_create_and_describe_vpc_security_group():
-    ec2 = boto3.resource("ec2", "us-west-1")
-    client = boto3.client("ec2", "us-west-1")
+    ec2 = boto3.resource("ec2", REGION)
+    client = boto3.client("ec2", REGION)
 
     name = str(uuid4())
     vpc_id = f"vpc-{str(uuid4())[0:6]}"
@@ -111,8 +113,8 @@ def test_create_and_describe_vpc_security_group():
 
 @mock_aws
 def test_create_two_security_groups_with_same_name_in_different_vpc():
-    ec2 = boto3.resource("ec2", "us-east-1")
-    client = boto3.client("ec2", "us-east-1")
+    ec2 = boto3.resource("ec2", REGION)
+    client = boto3.client("ec2", REGION)
 
     name = str(uuid4())
     vpc_id = "vpc-5300000c"
@@ -132,7 +134,7 @@ def test_create_two_security_groups_with_same_name_in_different_vpc():
 
 @mock_aws
 def test_create_two_security_groups_in_vpc_with_ipv6_enabled():
-    ec2 = boto3.resource("ec2", region_name="us-west-1")
+    ec2 = boto3.resource("ec2", region_name=REGION)
     vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16", AmazonProvidedIpv6CidrBlock=True)
 
     security_group = ec2.create_security_group(
@@ -145,8 +147,8 @@ def test_create_two_security_groups_in_vpc_with_ipv6_enabled():
 
 @mock_aws
 def test_deleting_security_groups():
-    ec2 = boto3.resource("ec2", "us-west-1")
-    client = boto3.client("ec2", "us-west-1")
+    ec2 = boto3.resource("ec2", REGION)
+    client = boto3.client("ec2", REGION)
     sg_name1 = str(uuid4())
     sg_name2 = str(uuid4())
     group1 = ec2.create_security_group(GroupName=sg_name1, Description="test desc 1")
@@ -188,8 +190,8 @@ def test_deleting_security_groups():
 
 @mock_aws
 def test_delete_security_group_in_vpc():
-    ec2 = boto3.resource("ec2", "us-west-1")
-    client = boto3.client("ec2", "us-west-1")
+    ec2 = boto3.resource("ec2", REGION)
+    client = boto3.client("ec2", REGION)
 
     group = ec2.create_security_group(
         GroupName=str(uuid4()), Description="test1", VpcId="vpc-12345"
@@ -207,8 +209,8 @@ def test_delete_security_group_in_vpc():
 
 @mock_aws
 def test_authorize_ip_range_and_revoke():
-    ec2 = boto3.resource("ec2", "us-west-1")
-    client = boto3.client("ec2", "us-west-1")
+    ec2 = boto3.resource("ec2", REGION)
+    client = boto3.client("ec2", REGION)
     security_group = ec2.create_security_group(
         GroupName=str(uuid4()), Description="test"
     )
@@ -332,8 +334,8 @@ def test_authorize_ip_range_and_revoke():
 
 @mock_aws
 def test_authorize_other_group_and_revoke():
-    ec2 = boto3.resource("ec2", "us-west-1")
-    client = boto3.client("ec2", "us-west-1")
+    ec2 = boto3.resource("ec2", REGION)
+    client = boto3.client("ec2", REGION)
     sg_name = str(uuid4())
     security_group = ec2.create_security_group(
         GroupName=sg_name, Description="test desc"
@@ -389,7 +391,7 @@ def test_authorize_other_group_and_revoke():
 
 @mock_aws
 def test_authorize_other_group_egress_and_revoke():
-    ec2 = boto3.resource("ec2", region_name="us-west-1")
+    ec2 = boto3.resource("ec2", region_name=REGION)
 
     vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
 
@@ -476,8 +478,8 @@ def test_authorize_group_in_vpc():
 
 @mock_aws
 def test_describe_security_groups():
-    ec2 = boto3.resource("ec2", "us-west-1")
-    client = boto3.client("ec2", "us-west-1")
+    ec2 = boto3.resource("ec2", REGION)
+    client = boto3.client("ec2", REGION)
     vpc_id = f"vpc-{str(uuid4())[0:6]}"
     name_1 = str(uuid4())
     desc_1 = str(uuid4())
@@ -514,7 +516,7 @@ def test_describe_security_groups():
 
 @mock_aws
 def test_authorize_bad_cidr_throws_invalid_parameter_value():
-    ec2 = boto3.resource("ec2", "us-west-1")
+    ec2 = boto3.resource("ec2", REGION)
     sec_group = ec2.create_security_group(GroupName=str(uuid4()), Description="test")
     with pytest.raises(ClientError) as ex:
         permissions = [
@@ -533,8 +535,8 @@ def test_authorize_bad_cidr_throws_invalid_parameter_value():
 
 @mock_aws
 def test_security_group_tag_filtering():
-    ec2 = boto3.resource("ec2", region_name="us-east-1")
-    client = boto3.client("ec2", region_name="us-east-1")
+    ec2 = boto3.resource("ec2", region_name=REGION)
+    client = boto3.client("ec2", region_name=REGION)
     sg = ec2.create_security_group(GroupName=str(uuid4()), Description="Test SG")
     tag_name = str(uuid4())[0:6]
     tag_val = str(uuid4())
@@ -553,8 +555,8 @@ def test_security_group_tag_filtering():
 
 @mock_aws
 def test_authorize_all_protocols_with_no_port_specification():
-    ec2 = boto3.resource("ec2", region_name="us-east-1")
-    client = boto3.client("ec2", region_name="us-east-1")
+    ec2 = boto3.resource("ec2", region_name=REGION)
+    client = boto3.client("ec2", region_name=REGION)
     sg_name = str(uuid4())
     sg = ec2.create_security_group(GroupName=sg_name, Description="test desc")
 
@@ -570,9 +572,9 @@ def test_authorize_all_protocols_with_no_port_specification():
 
 
 @mock_aws
-def test_security_group_rule_tagging():
-    ec2 = boto3.resource("ec2", "us-east-1")
-    client = boto3.client("ec2", "us-east-1")
+def test_security_group_rule_filtering_group_id():
+    ec2 = boto3.resource("ec2", REGION)
+    client = boto3.client("ec2", REGION)
     vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
 
     sg_name = str(uuid4())
@@ -598,10 +600,55 @@ def test_security_group_rule_tagging():
     assert response["SecurityGroupRules"][0]["Tags"][0]["Value"] == tag_val
 
 
+@mock_aws
+def test_security_group_rule_filtering_tags():
+    # Setup
+    ec2 = boto3.resource("ec2", REGION)
+    client = boto3.client("ec2", REGION)
+    vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
+    tags = [
+        {"Key": "Automation", "Value": "Lambda"},
+        {"Key": "Partner", "Value": "test"},
+    ]
+
+    sg_name = str(uuid4())
+    sg = client.create_security_group(
+        Description="Test SG",
+        GroupName=sg_name,
+        VpcId=vpc.id,
+    )
+
+    # Execute
+    response1 = client.authorize_security_group_ingress(
+        GroupId=sg["GroupId"],
+        IpPermissions=[
+            {
+                "IpProtocol": "tcp",
+                "FromPort": 80,
+                "ToPort": 80,
+                "IpRanges": [
+                    {"CidrIp": "1.2.3.4/32", "Description": "Test description"}
+                ],
+            }
+        ],
+        TagSpecifications=[{"ResourceType": "security-group-rule", "Tags": tags}],
+    )
+
+    response2 = client.describe_security_group_rules(
+        Filters=[{"Name": "tag:Partner", "Values": ["test"]}]
+    )
+
+    # Verify
+    assert response1["SecurityGroupRules"][0]["Tags"] == tags
+    assert "Tags" in response2["SecurityGroupRules"][0]
+    assert response2["SecurityGroupRules"][0]["Tags"][1]["Key"] == "Partner"
+    assert response2["SecurityGroupRules"][0]["Tags"][1]["Value"] == "test"
+
+
 @mock_aws
 def test_create_and_describe_security_grp_rule():
-    ec2 = boto3.resource("ec2", "us-east-1")
-    client = boto3.client("ec2", "us-east-1")
+    ec2 = boto3.resource("ec2", REGION)
+    client = boto3.client("ec2", REGION)
     vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
 
     sg_name = str(uuid4())
@@ -627,8 +674,8 @@ def test_create_and_describe_security_grp_rule():
 @mock_aws
 @pytest.mark.parametrize("use_vpc", [True, False], ids=["Use VPC", "Without VPC"])
 def test_sec_group_rule_limit(use_vpc):
-    ec2 = boto3.resource("ec2", region_name="us-west-1")
-    client = boto3.client("ec2", region_name="us-west-1")
+    ec2 = boto3.resource("ec2", region_name=REGION)
+    client = boto3.client("ec2", region_name=REGION)
 
     limit = 60
     if use_vpc:
@@ -728,7 +775,7 @@ def test_sec_group_rule_limit(use_vpc):
 
 @mock_aws
 def test_add_same_rule_twice_throws_error():
-    ec2 = boto3.resource("ec2", region_name="us-west-1")
+    ec2 = boto3.resource("ec2", region_name=REGION)
 
     vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
     sg = ec2.create_security_group(
@@ -768,8 +815,8 @@ def test_add_same_rule_twice_throws_error():
 
 @mock_aws
 def test_description_in_ip_permissions():
-    ec2 = boto3.resource("ec2", region_name="us-west-1")
-    conn = boto3.client("ec2", region_name="us-east-1")
+    ec2 = boto3.resource("ec2", region_name=REGION)
+    conn = boto3.client("ec2", region_name=REGION)
     vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
     sg = conn.create_security_group(
         GroupName="sg1", Description="Test security group sg1", VpcId=vpc.id
@@ -818,7 +865,7 @@ def test_description_in_ip_permissions():
 
 @mock_aws
 def test_security_group_tagging():
-    conn = boto3.client("ec2", region_name="us-east-1")
+    conn = boto3.client("ec2", region_name=REGION)
 
     sg = conn.create_security_group(GroupName=str(uuid4()), Description="Test SG")
 
@@ -849,7 +896,7 @@ def test_security_group_tagging():
 
 @mock_aws
 def test_security_group_wildcard_tag_filter():
-    conn = boto3.client("ec2", region_name="us-east-1")
+    conn = boto3.client("ec2", region_name=REGION)
     sg = conn.create_security_group(GroupName=str(uuid4()), Description="Test SG")
 
     rand_name = str(uuid4())[0:6]
@@ -868,10 +915,10 @@ def test_security_group_wildcard_tag_filter():
 
 @mock_aws
 def test_security_group_filter_ip_permission():
-    ec2 = boto3.resource("ec2", region_name="us-east-1")
+    ec2 = boto3.resource("ec2", region_name=REGION)
     vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
 
-    conn = boto3.client("ec2", region_name="us-east-1")
+    conn = boto3.client("ec2", region_name=REGION)
     sg_name = str(uuid4())[0:6]
     sg = ec2.create_security_group(
         GroupName=sg_name, Description="Test SG", VpcId=vpc.id
@@ -910,7 +957,7 @@ def retrieve_all_sgs(conn, filters=[]):  # pylint: disable=W0102
 
 @mock_aws
 def test_authorize_and_revoke_in_bulk():
-    ec2 = boto3.resource("ec2", region_name="us-west-1")
+    ec2 = boto3.resource("ec2", region_name=REGION)
 
     vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
 
@@ -1090,8 +1137,8 @@ def test_security_group_ingress_without_multirule_after_reload():
 
 @mock_aws
 def test_get_all_security_groups_filter_with_same_vpc_id():
-    ec2 = boto3.resource("ec2", region_name="us-east-1")
-    client = boto3.client("ec2", region_name="us-east-1")
+    ec2 = boto3.resource("ec2", region_name=REGION)
+    client = boto3.client("ec2", region_name=REGION)
     vpc_id = "vpc-5300000c"
     security_group = ec2.create_security_group(
         GroupName=str(uuid4()), Description="test1", VpcId=vpc_id
@@ -1117,7 +1164,7 @@ def test_get_all_security_groups_filter_with_same_vpc_id():
 
 @mock_aws
 def test_revoke_security_group_egress():
-    ec2 = boto3.resource("ec2", "us-east-1")
+    ec2 = boto3.resource("ec2", REGION)
     vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
     sg = ec2.create_security_group(
         Description="Test SG", GroupName=str(uuid4()), VpcId=vpc.id
@@ -1177,8 +1224,8 @@ def test_revoke_security_group_egress__without_ipprotocol():
 
 @mock_aws
 def test_update_security_group_rule_descriptions_egress():
-    ec2 = boto3.resource("ec2", "us-east-1")
-    client = boto3.client("ec2", "us-east-1")
+    ec2 = boto3.resource("ec2", REGION)
+    client = boto3.client("ec2", REGION)
     vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
     sg_name = str(uuid4())
     sg = ec2.create_security_group(
@@ -1214,8 +1261,8 @@ def test_update_security_group_rule_descriptions_egress():
 
 @mock_aws
 def test_update_security_group_rule_descriptions_ingress():
-    ec2 = boto3.resource("ec2", "us-east-1")
-    client = boto3.client("ec2", "us-east-1")
+    ec2 = boto3.resource("ec2", REGION)
+    client = boto3.client("ec2", REGION)
     vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
     sg_name = str(uuid4())
     sg = ec2.create_security_group(
@@ -1231,8 +1278,13 @@ def test_update_security_group_rule_descriptions_ingress():
             "IpRanges": [{"CidrIp": "1.2.3.4/32", "Description": "first desc"}],
         }
     ]
-    client.authorize_security_group_ingress(GroupId=sg_id, IpPermissions=ip_permissions)
-
+    client.authorize_security_group_ingress(
+        GroupId=sg_id,
+        IpPermissions=ip_permissions,
+    )
+    client.describe_security_group_rules(
+        Filters=[{"Name": "tag:Partner", "Values": ["test"]}]
+    )
     ip_ranges = client.describe_security_groups(GroupIds=[sg_id])["SecurityGroups"][0][
         "IpPermissions"
     ][0]["IpRanges"]
@@ -1260,7 +1312,7 @@ def test_update_security_group_rule_descriptions_ingress():
 
 @mock_aws
 def test_non_existent_security_group_raises_error_on_authorize():
-    client = boto3.client("ec2", "us-east-1")
+    client = boto3.client("ec2", REGION)
     non_existent_sg = "sg-123abc"
     expected_error = f"The security group '{non_existent_sg}' does not exist"
     authorize_funcs = [
@@ -1278,9 +1330,9 @@ def test_non_existent_security_group_raises_error_on_authorize():
 def test_security_group_rules_added_via_the_backend_can_be_revoked_via_the_api():
     if settings.TEST_SERVER_MODE:
         raise unittest.SkipTest("Can't test backend directly in server mode.")
-    ec2_backend = ec2_backends[DEFAULT_ACCOUNT_ID]["us-east-1"]
-    ec2_resource = boto3.resource("ec2", region_name="us-east-1")
-    ec2_client = boto3.client("ec2", region_name="us-east-1")
+    ec2_backend = ec2_backends[DEFAULT_ACCOUNT_ID][REGION]
+    ec2_resource = boto3.resource("ec2", region_name=REGION)
+    ec2_client = boto3.client("ec2", region_name=REGION)
     vpc = ec2_resource.create_vpc(CidrBlock="10.0.0.0/16")
     group_name = str(uuid4())
     sg = ec2_resource.create_security_group(
@@ -1329,7 +1381,7 @@ def test_security_group_rules_added_via_the_backend_can_be_revoked_via_the_api()
 
 @mock_aws
 def test_filter_description():
-    ec2r = boto3.resource("ec2", region_name="us-west-1")
+    ec2r = boto3.resource("ec2", region_name=REGION)
     vpc = ec2r.create_vpc(CidrBlock="10.250.0.0/16")
 
     unique = str(uuid4())
@@ -1360,7 +1412,7 @@ def test_filter_ip_permission__cidr():
         raise SkipTest(
             "CIDR's might already exist due to other tests creating IP ranges"
         )
-    ec2r = boto3.resource("ec2", region_name="us-west-1")
+    ec2r = boto3.resource("ec2", region_name=REGION)
     vpc = ec2r.create_vpc(CidrBlock="10.250.1.0/16")
 
     sg1 = vpc.create_security_group(
@@ -1407,7 +1459,7 @@ def test_filter_egress__ip_permission__cidr():
         raise SkipTest(
             "CIDR's might already exist due to other tests creating IP ranges"
         )
-    ec2r = boto3.resource("ec2", region_name="us-west-1")
+    ec2r = boto3.resource("ec2", region_name=REGION)
     vpc = ec2r.create_vpc(CidrBlock="10.250.1.0/16")
 
     sg1 = vpc.create_security_group(
@@ -1450,7 +1502,7 @@ def test_filter_egress__ip_permission__cidr():
 
 @mock_aws
 def test_filter_egress__ip_permission__from_port():
-    ec2r = boto3.resource("ec2", region_name="us-west-1")
+    ec2r = boto3.resource("ec2", region_name=REGION)
     vpc = ec2r.create_vpc(CidrBlock="10.250.1.0/16")
 
     sg1 = vpc.create_security_group(
@@ -1494,7 +1546,7 @@ def test_filter_egress__ip_permission__from_port():
 
 @mock_aws
 def test_filter_egress__ip_permission__group_id():
-    ec2r = boto3.resource("ec2", region_name="us-west-1")
+    ec2r = boto3.resource("ec2", region_name=REGION)
     vpc = ec2r.create_vpc(CidrBlock="10.250.1.0/16")
 
     sg1 = vpc.create_security_group(
@@ -1547,7 +1599,7 @@ def test_filter_egress__ip_permission__group_name_create_with_id_filter_by_name(
     """
     this fails to find the group in the AWS API, so we should also fail to find it
     """
-    ec2r = boto3.resource("ec2", region_name="us-west-1")
+    ec2r = boto3.resource("ec2", region_name=REGION)
     vpc = ec2r.create_vpc(CidrBlock="10.250.1.0/16")
 
     sg1 = vpc.create_security_group(
@@ -1599,7 +1651,7 @@ def test_filter_egress__ip_permission__group_name_create_with_id_filter_by_name(
 
 @mock_aws
 def test_filter_egress__ip_permission__group_name_create_with_id_filter_by_id():
-    ec2r = boto3.resource("ec2", region_name="us-west-1")
+    ec2r = boto3.resource("ec2", region_name=REGION)
     vpc = ec2r.create_vpc(CidrBlock="10.250.1.0/16")
 
     sg1 = vpc.create_security_group(
@@ -1652,7 +1704,7 @@ def test_filter_egress__ip_permission__group_name_create_with_id_filter_by_id():
 
 @mock_aws
 def test_filter_egress__ip_permission__protocol():
-    ec2r = boto3.resource("ec2", region_name="us-west-1")
+    ec2r = boto3.resource("ec2", region_name=REGION)
     vpc = ec2r.create_vpc(CidrBlock="10.250.1.0/16")
 
     sg1 = vpc.create_security_group(
@@ -1695,7 +1747,7 @@ def test_filter_egress__ip_permission__protocol():
 
 @mock_aws
 def test_filter_egress__ip_permission__to_port():
-    ec2r = boto3.resource("ec2", region_name="us-west-1")
+    ec2r = boto3.resource("ec2", region_name=REGION)
     vpc = ec2r.create_vpc(CidrBlock="10.250.1.0/16")
 
     sg1 = vpc.create_security_group(
@@ -1739,7 +1791,7 @@ def test_filter_egress__ip_permission__to_port():
 
 @mock_aws
 def test_get_groups_by_ippermissions_group_id_filter():
-    ec2r = boto3.resource("ec2", region_name="us-west-1")
+    ec2r = boto3.resource("ec2", region_name=REGION)
     vpc = ec2r.create_vpc(CidrBlock="10.250.0.0/16")
     sg1 = vpc.create_security_group(Description="test", GroupName="test-1")
     sg2 = vpc.create_security_group(Description="test", GroupName="test-2")
@@ -1772,7 +1824,7 @@ def test_get_groups_by_ippermissions_group_id_filter():
 def test_get_groups_by_ippermissions_group_id_filter_across_vpcs():
     # setup 2 VPCs, each with a single Security Group
     # where one security group authorizes the other sg (in another vpc) via GroupId
-    ec2r = boto3.resource("ec2", region_name="us-west-1")
+    ec2r = boto3.resource("ec2", region_name=REGION)
 
     vpc1 = ec2r.create_vpc(CidrBlock="10.250.0.0/16")
     vpc2 = ec2r.create_vpc(CidrBlock="10.251.0.0/16")
@@ -1809,7 +1861,7 @@ def test_filter_group_name():
     """
     this filter is an exact match, not a glob
     """
-    ec2r = boto3.resource("ec2", region_name="us-west-1")
+    ec2r = boto3.resource("ec2", region_name=REGION)
     vpc = ec2r.create_vpc(CidrBlock="10.250.1.0/16")
 
     uniq_sg_name_prefix = str(uuid4())[0:6]
@@ -1843,7 +1895,7 @@ def test_filter_group_name():
 
 @mock_aws
 def test_revoke_security_group_ingress():
-    ec2 = boto3.client("ec2", region_name="us-east-1")
+    ec2 = boto3.client("ec2", region_name=REGION)
 
     vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
 


Code review comments:
diff_hunk: @@ -121,22 +131,23 @@ def _process_rules_from_querystring(self) -> Any:
                 prefix_list_ids,
             ) = parse_sg_attributes_from_dict(ip_permission)
 
-            yield (
-                group_name_or_id,
-                ip_protocol,
-                from_port,
-                to_port,
-                ip_ranges,
-                source_groups,
-                prefix_list_ids,
-                security_rule_ids,
-            )
+            yield {
+                "group_name_or_id": group_name_or_id,
+                "ip_protocol": ip_protocol,
+                "from_port": from_port,
+                "to_port": to_port,
+                "ip_ranges": ip_ranges,
+                "sgrule_tags": sg_rule_tags,
+                "source_groups": source_groups,
+                "prefix_list_ids": prefix_list_ids,
+                "security_rule_ids": security_rule_ids,
+            }
 
     def authorize_security_group_egress(self) -> str:
         self.error_on_dryrun()
 
         for args in self._process_rules_from_querystring():
-            rule, group = self.ec2_backend.authorize_security_group_egress(*args)
+            rule, group = self.ec2_backend.authorize_security_group_egress(**args)
path: moto/ec2/responses/security_groups.py
text: @user1:
```suggestion
            rule, group = self.ec2_backend.authorize_security_group_egress(**kwargs)
```
I like the change to make them keyword-arguments! Makes it easier to debug. Just a minor nitpick that the name should reflect this.

diff_hunk: @@ -86,6 +86,15 @@ def _process_rules_from_querystring(self) -> Any:
                 d = d[subkey]
             d[key_splitted[-1]] = value
 
+        sg_rule_tags = {}
path: moto/ec2/responses/security_groups.py
text: @user1:
I think we should process the tag-retrieval inside the `authorize_..`-method itself. It is a toplevel attribute, and arguably unrelated to the other SecurityGroupRule-properties that we're parsing.

Ideally we can just use 
`tags = self._parse_tag_specification().get("security-group-rule", {})`,
similar to how it's done in `create_security_group`. That way we don't have to have any complicated parsing here