Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ELBv2 improvements (#4808)

Browse Source
This commit is contained in:
Bert Blommers 2022-01-29 11:04:14 -01:00 committed by GitHub
parent e1dbec1dff
commit ed86df6bae
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 129 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
moto/elbv2/models.py
View File

@ -62,6 +62,7 @@ class FakeTargetGroup(CloudFormationModel):
vpc_id,
protocol,
port,
protocol_version=None,
healthcheck_protocol=None,
healthcheck_port=None,
healthcheck_path=None,
@ -79,6 +80,7 @@ class FakeTargetGroup(CloudFormationModel):
self.arn = arn
self.vpc_id = vpc_id
self.protocol = protocol
self.protocol_version = protocol_version or "HTTP1"
self.port = port
self.healthcheck_protocol = healthcheck_protocol or self.protocol
self.healthcheck_port = healthcheck_port
@ -912,7 +914,7 @@ Member must satisfy regular expression pattern: {}".format(
if target_group.name == name:
raise DuplicateTargetGroupName()
valid_protocols = ["HTTPS", "HTTP", "TCP"]
valid_protocols = ["HTTPS", "HTTP", "TCP", "TLS", "UDP", "TCP_UDP", "GENEVE"]
if (
kwargs.get("healthcheck_protocol")
and kwargs["healthcheck_protocol"] not in valid_protocols
@ -948,6 +950,13 @@ Member must satisfy regular expression pattern: {}".format(
self.target_groups[target_group.arn] = target_group
return target_group
def modify_target_group_attributes(self, target_group_arn, attributes):
target_group = self.target_groups.get(target_group_arn)
if not target_group:
raise TargetGroupNotFoundError()
target_group.attributes.update(attributes)
def convert_and_validate_certificates(self, certificates):
# transform default certificate to conform with the rest of the code and XML templates
@ -1355,7 +1364,7 @@ Member must satisfy regular expression pattern: {}".format(
"HttpCode must be like 200 | 200-399 | 200,201 ...",
)
if http_codes is not None:
if http_codes is not None and target_group.protocol in ["HTTP", "HTTPS"]:
target_group.matcher["HttpCode"] = http_codes
if health_check_interval is not None:
target_group.healthcheck_interval_seconds = health_check_interval
@ -1397,15 +1406,15 @@ Member must satisfy regular expression pattern: {}".format(
if port is not None:
listener.port = port
if protocol is not None:
if protocol not in ("HTTP", "HTTPS", "TCP"):
if protocol not in (None, "HTTP", "HTTPS", "TCP"):
raise RESTError(
"UnsupportedProtocol",
"Protocol {0} is not supported".format(protocol),
"UnsupportedProtocol", "Protocol {0} is not supported".format(protocol),
)
# HTTPS checks
if protocol == "HTTPS":
protocol_becomes_https = protocol == "HTTPS"
protocol_stays_https = protocol is None and listener.protocol == "HTTPS"
if protocol_becomes_https or protocol_stays_https:
# Check certificates exist
if certificates:
default_cert = certificates[0]
@ -1425,6 +1434,7 @@ Member must satisfy regular expression pattern: {}".format(
"You must provide a list containing exactly one certificate if the listener protocol is HTTPS.",
)
if protocol is not None:
listener.protocol = protocol
if ssl_policy is not None:

18
moto/elbv2/responses.py
View File

@ -178,6 +178,7 @@ class ELBV2Response(BaseResponse):
name = self._get_param("Name")
vpc_id = self._get_param("VpcId")
protocol = self._get_param("Protocol")
protocol_version = self._get_param("ProtocolVersion", "HTTP1")
port = self._get_param("Port")
healthcheck_protocol = self._get_param("HealthCheckProtocol")
healthcheck_port = self._get_param("HealthCheckPort")
@ -194,6 +195,7 @@ class ELBV2Response(BaseResponse):
name,
vpc_id=vpc_id,
protocol=protocol,
protocol_version=protocol_version,
port=port,
healthcheck_protocol=healthcheck_protocol,
healthcheck_port=healthcheck_port,
@ -366,14 +368,10 @@ class ELBV2Response(BaseResponse):
@amzn_request_id
def modify_target_group_attributes(self):
target_group_arn = self._get_param("TargetGroupArn")
target_group = self.elbv2_backend.target_groups.get(target_group_arn)
attributes = {
attr["key"]: attr["value"]
for attr in self._get_list_prefix("Attributes.member")
}
target_group.attributes.update(attributes)
if not target_group:
raise TargetGroupNotFoundError()
attributes = self._get_list_prefix("Attributes.member")
attributes = {attr["key"]: attr["value"] for attr in attributes}
self.elbv2_backend.modify_target_group_attributes(target_group_arn, attributes)
template = self.response_template(MODIFY_TARGET_GROUP_ATTRIBUTES_TEMPLATE)
return template.render(attributes=attributes)
@ -1085,6 +1083,7 @@ DESCRIBE_TARGET_GROUPS_TEMPLATE = """<DescribeTargetGroupsResponse xmlns="http:/
<TargetGroupName>{{ target_group.name }}</TargetGroupName>
{% if target_group.protocol %}
<Protocol>{{ target_group.protocol }}</Protocol>
<ProtocolVersion>{{ target_group.protocol_version }}</ProtocolVersion>
{% endif %}
{% if target_group.port %}
<Port>{{ target_group.port }}</Port>
@ -1608,6 +1607,7 @@ DESCRIBE_LOADBALANCER_ATTRS_TEMPLATE = """<DescribeLoadBalancerAttributesRespons
</ResponseMetadata>
</DescribeLoadBalancerAttributesResponse>"""
MODIFY_TARGET_GROUP_TEMPLATE = """<ModifyTargetGroupResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<ModifyTargetGroupResult>
<TargetGroups>
@ -1650,7 +1650,7 @@ MODIFY_LISTENER_TEMPLATE = """<ModifyListenerResponse xmlns="http://elasticloadb
<Certificates>
{% for cert in listener.certificates %}
<member>
<CertificateArn>{{ cert }}</CertificateArn>
<CertificateArn>{{ cert["certificate_arn"] }}</CertificateArn>
</member>
{% endfor %}
</Certificates>

81
tests/test_elbv2/test_elbv2.py
View File

@ -1354,7 +1354,7 @@ def test_modify_listener_http_to_https():
)
# Bad cert
with pytest.raises(ClientError):
with pytest.raises(ClientError) as exc:
client.modify_listener(
ListenerArn=listener_arn,
Port=443,
@ -1363,6 +1363,85 @@ def test_modify_listener_http_to_https():
Certificates=[{"CertificateArn": "lalala", "IsDefault": True}],
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
)
err = exc.value.response["Error"]
err["Message"].should.equal("Certificate lalala not found")
# Unknown protocol
with pytest.raises(ClientError) as exc:
client.modify_listener(
ListenerArn=listener_arn,
Port=443,
Protocol="HTP",
SslPolicy="ELBSecurityPolicy-TLS-1-2-2017-01",
Certificates=[{"CertificateArn": yahoo_arn, "IsDefault": True}],
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
)
err = exc.value.response["Error"]
err["Message"].should.equal("Protocol HTP is not supported")
@mock_acm
@mock_ec2
@mock_elbv2
def test_modify_listener_of_https_target_group():
# Verify we can add a listener for a TargetGroup that is already HTTPS
client = boto3.client("elbv2", region_name="eu-central-1")
acm = boto3.client("acm", region_name="eu-central-1")
ec2 = boto3.resource("ec2", region_name="eu-central-1")
security_group = ec2.create_security_group(
GroupName="a-security-group", Description="First One"
)
vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
subnet1 = ec2.create_subnet(
VpcId=vpc.id, CidrBlock="172.28.7.192/26", AvailabilityZone="eu-central-1a"
)
response = client.create_load_balancer(
Name="my-lb",
Subnets=[subnet1.id],
SecurityGroups=[security_group.id],
Scheme="internal",
Tags=[{"Key": "key_name", "Value": "a_value"}],
)
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
response = client.create_target_group(
Name="a-target", Protocol="HTTPS", Port=8443, VpcId=vpc.id,
)
target_group = response.get("TargetGroups")[0]
target_group_arn = target_group["TargetGroupArn"]
# HTTPS listener
response = acm.request_certificate(
DomainName="google.com", SubjectAlternativeNames=["google.com"],
)
google_arn = response["CertificateArn"]
response = client.create_listener(
LoadBalancerArn=load_balancer_arn,
Protocol="HTTPS",
Port=443,
Certificates=[{"CertificateArn": google_arn}],
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
)
listener_arn = response["Listeners"][0]["ListenerArn"]
# Now modify the HTTPS listener with a different certificate
response = acm.request_certificate(
DomainName="yahoo.com", SubjectAlternativeNames=["yahoo.com"],
)
yahoo_arn = response["CertificateArn"]
listener = client.modify_listener(
ListenerArn=listener_arn,
Certificates=[{"CertificateArn": yahoo_arn,},],
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
)["Listeners"][0]
listener["Certificates"].should.equal([{"CertificateArn": yahoo_arn}])
listener = client.describe_listeners(ListenerArns=[listener_arn])["Listeners"][0]
listener["Certificates"].should.equal([{"CertificateArn": yahoo_arn}])
@mock_elbv2

4
tests/test_elbv2/test_elbv2_target_groups.py
View File

@ -32,7 +32,7 @@ def test_create_target_group_with_invalid_healthcheck_protocol():
err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError")
err["Message"].should.equal(
"Value /HTTP at 'healthCheckProtocol' failed to satisfy constraint: Member must satisfy enum value set: ['HTTPS', 'HTTP', 'TCP']"
"Value /HTTP at 'healthCheckProtocol' failed to satisfy constraint: Member must satisfy enum value set: ['HTTPS', 'HTTP', 'TCP', 'TLS', 'UDP', 'TCP_UDP', 'GENEVE']"
)
@ -499,6 +499,8 @@ def test_modify_target_group():
response["TargetGroups"][0]["HealthCheckProtocol"].should.equal("HTTPS")
response["TargetGroups"][0]["HealthCheckTimeoutSeconds"].should.equal(10)
response["TargetGroups"][0]["HealthyThresholdCount"].should.equal(10)
response["TargetGroups"][0].should.have.key("Protocol").equals("HTTP")
response["TargetGroups"][0].should.have.key("ProtocolVersion").equals("HTTP1")
response["TargetGroups"][0]["UnhealthyThresholdCount"].should.equal(4)