8000 Add EC2 support for GetSecurityGroupsForVpc API operation (#12602) by iamramtin · Pull Request #12615 · localstack/localstack · GitHub
[go: up one dir, main page]

Skip to content

Add EC2 support for GetSecurityGroupsForVpc API operation (#12602) #12615

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

27 changes: 27 additions & 0 deletions localstack-core/localstack/services/ec2/provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,8 @@
DnsOptionsSpecification,
DnsRecordIpType,
Ec2Api,
GetSecurityGroupsForVpcRequest,
GetSecurityGroupsForVpcResult,
InstanceType,
IpAddressType,
LaunchTemplate,
Expand All @@ -70,6 +72,7 @@
RevokeSecurityGroupEgressRequest,
RevokeSecurityGroupEgressResult,
RIProductDescription,
SecurityGroupForVpc,
String,
SubnetConfigurationsList,
Tenancy,
Expand Down Expand Up @@ -539,6 +542,30 @@ def create_flow_logs(

return response

@handler("GetSecurityGroupsForVpc", expand=False)
def get_security_groups_for_vpc(
self,
context: RequestContext,
get_security_groups_for_vpc_request: GetSecurityGroupsForVpcRequest,
) -> GetSecurityGroupsForVpcResult:
vpc_id = get_security_groups_for_vpc_request.get("VpcId")
backend = get_ec2_backend(context.account_id, context.region)
filters = {"vpc-id": [vpc_id]}
filtered_sgs = backend.describe_security_groups(filters=filters)

sgs = [
SecurityGroupForVpc(
Description=sg.description,
GroupId=sg.id,
GroupName=sg.name,
OwnerId=context.account_id,
PrimaryVpcId=sg.vpc_id,
Tags=[{"Key": tag.get("key"), "Value": tag.get("value")} for tag in sg.get_tags()],
)
for sg in filtered_sgs
]
return GetSecurityGroupsForVpcResult(SecurityGroupForVpcs=sgs, NextToken=None)


@patch(SubnetBackend.modify_subnet_attribute)
def modify_subnet_attribute(fn, self, subnet_id, attr_name, attr_value):
Expand Down
5 changes: 5 additions & 0 deletions localstack-core/localstack/testing/pytest/fixtures.py
Original file line number Diff line number Diff line change
Expand Up @@ -2009,11 +2009,16 @@ def factory(ports=None, ip_protocol: str = "tcp", **kwargs):
:param ip_protocol: the ip protocol for the permissions (tcp by default)
"""
if "GroupName" not in kwargs:
# FIXME: This will fail against AWS since the sg prefix is not valid for GroupName
# > "Group names may not be in the format sg-*".
kwargs["GroupName"] = f"sg-{short_uid()}"
# Making sure the call to CreateSecurityGroup gets the right arguments
_args = select_from_typed_dict(CreateSecurityGroupRequest, kwargs)
security_group = aws_client.ec2.create_security_group(**_args)
security_group_id = security_group["GroupId"]

# FIXME: If 'ports' is None or an empty list, authorize_security_group_ingress will fail due to missing IpPermissions.
# Must ensure ports are explicitly provided or skip authorization entirely if not required.
permissions = [
{
"FromPort": port,
Expand Down
66 changes: 66 additions & 0 deletions tests/aws/services/ec2/test_ec2.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

import pytest
from botocore.exceptions import ClientError
from localstack_snapshot.snapshots.transformer import SortingTransformer
from moto.ec2 import ec2_backends
from moto.ec2.utils import (
random_security_group_id,
Expand Down Expand Up @@ -694,6 +695,71 @@ def _create_security_group() -> dict:
assert e.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
assert e.value.response["Error"]["Code"] == "InvalidSecurityGroupId.DuplicateCustomId"

@markers.snapshot.skip_snapshot_verify(
paths=[
"$..Tags", # Tags can differ between environments
"$..Vpc.IsDefault", # TODO: CreateVPC should return an IsDefault param
"$..Vpc.DhcpOptionsId", # FIXME: DhcpOptionsId uses different reference formats in AWS vs LocalStack
]
)
@markers.aws.validated
def test_get_security_groups_for_vpc(
self, snapshot, aws_client, create_vpc, ec2_create_security_group
):
group_name = f"test-security-group-{short_uid()}"
group_description = f"Description for {group_name}"

# Returned security groups appear to be sorted by the randomly generated GroupId field,
# so we should sort snapshots by this value to mitigate flakiness for runs against AWS.
snapshot.add_transformer(
SortingTransformer("SecurityGroupForVpcs", lambda x: x["GroupName"])
)
snapshot.add_transformer(snapshot.transform.key_value("GroupId"))
snapshot.add_transformer(snapshot.transform.key_value("GroupName"))
snapshot.add_transformer(snapshot.transform.key_value("VpcId"))
snapshot.add_transformer(snapshot.transform.key_value("AssociationId"))
snapshot.add_transformer(snapshot.transform.key_value("DhcpOptionsId"))

# Create VPC for testing
vpc: dict = create_vpc(
cidr_block="10.0.0.0/16",
tag_specifications=[
{
"ResourceType": "vpc",
"Tags": [
{"Key": "test-key", "Value": "test-value"},
],
}
],
)
vpc_id: str = vpc["Vpc"]["VpcId"]
snapshot.match("create_vpc_response", vpc)

# Wait to ensure VPC is available
waiter = aws_client.ec2.get_waiter("vpc_available")
waiter.wait(VpcIds=[vpc_id])

# Get all security groups in the VPC
get_security_groups_for_vpc = aws_client.ec2.get_security_groups_for_vpc(VpcId=vpc_id)
snapshot.match("get_security_groups_for_vpc", get_security_groups_for_vpc)

# Create new security group in the VPC
create_security_group = ec2_create_security_group(
GroupName=group_name,
Description=group_description,
VpcId=vpc_id,
ports=[22], # TODO: Handle port issues in the fixture
)
snapshot.match("create_security_group", create_security_group)

# Ensure new security group is in the VPC
get_security_groups_for_vpc_after_addition = aws_client.ec2.get_security_groups_for_vpc(
VpcId=vpc_id
)
snapshot.match(
"get_security_groups_for_vpc_after_addition", get_security_groups_for_vpc_after_addition
)


@markers.snapshot.skip_snapshot_verify(
# Moto and LS do not return the ClientToken
Expand Down
84 changes: 84 additions & 0 deletions tests/aws/services/ec2/test_ec2.snapshot.json
A3D4
Original file line number Diff line number Diff line change
Expand Up @@ -335,5 +335,89 @@
}
}
}
},
"tests/aws/services/ec2/test_ec2.py::TestEc2Integrations::test_get_security_groups_for_vpc": {
"recorded-date": "19-05-2025, 13:53:56",
"recorded-content": {
"create_vpc_response": {
"Vpc": {
"CidrBlock": "10.0.0.0/16",
"CidrBlockAssociationSet": [
{
"AssociationId": "<association-id:1>",
"CidrBlock": "10.0.0.0/16",
"CidrBlockState": {
"State": "associated"
}
}
],
"DhcpOptionsId": "<dhcp-options-id:1>",
"InstanceTenancy": "<group-name:1>",
"Ipv6CidrBlockAssociationSet": [],
"IsDefault": false,
"OwnerId": "111111111111",
"State": "pending",
"Tags": [
{
"Key": "test-key",
"Value": "test-value"
}
],
"VpcId": "<vpc-id:1>"
},
"ResponseMetadata": {
"HTTPHeaders": {},
"HTTPStatusCode": 200
}
},
"get_security_groups_for_vpc": {
"SecurityGroupForVpcs": [
{
"Description": "<group-name:1> VPC security group",
"GroupId": "<group-id:1>",
"GroupName": "<group-name:1>",
"OwnerId": "111111111111",
"PrimaryVpcId": "<vpc-id:1>",
"Tags": []
}
],
"ResponseMetadata": {
"HTTPHeaders": {},
"HTTPStatusCode": 200
}
},
"create_security_group": {
"GroupId": "<group-id:2>",
"SecurityGroupArn": "arn:<partition>:ec2:<region>:111111111111:security-group/<group-id:2>",
"ResponseMetadata": {
"HTTPHeaders": {},
"HTTPStatusCode": 200
}
},
"get_security_groups_for_vpc_after_addition": {
"SecurityGroupForVpcs": [
{
"Description": "<group-name:1> VPC security group",
"GroupId": "<group-id:1>",
"GroupName": "<group-name:1>",
"OwnerId": "111111111111",
"PrimaryVpcId": "<vpc-id:1>",
"Tags": []
},
{
"Description": "Description for <group-name:2>",
"GroupId": "<group-id:2>",
"GroupName": "<group-name:2>",
"OwnerId": "111111111111",
"PrimaryVpcId": "<vpc-id:1>",
"Tags": []
}
],
"ResponseMetadata": {
"HTTPHeaders": {},
"HTTPStatusCode": 200
}
}
}
}
}
3 changes: 3 additions & 0 deletions tests/aws/services/ec2/test_ec2.validation.json
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,9 @@
"tests/aws/services/ec2/test_ec2.py::TestEc2Integrations::test_describe_vpn_gateways_filter_by_vpc": {
"last_validated_date": "2024-06-07T01:11:12+00:00"
},
"tests/aws/services/ec2/test_ec2.py::TestEc2Integrations::test_get_security_groups_for_vpc": {
"last_validated_date": "2025-05-19T13:54:09+00:00"
},
"tests/aws/services/ec2/test_ec2.py::TestEc2Integrations::test_vcp_peering_difference_regions": {
"last_validated_date": "2024-06-07T21:28:25+00:00"
},
Expand Down
0