Merge pull request #2418 from nadlerjessie/elbv2-fixed-response-action

Elbv2 fixed response action

moto/elbv2/exceptions.py

@@ -131,7 +131,7 @@ class InvalidActionTypeError(ELBClientError):
     def __init__(self, invalid_name, index):
         super(InvalidActionTypeError, self).__init__(
             "ValidationError",
-            "1 validation error detected: Value '%s' at 'actions.%s.member.type' failed to satisfy constraint: Member must satisfy enum value set: [forward, redirect]" % (invalid_name, index)
+            "1 validation error detected: Value '%s' at 'actions.%s.member.type' failed to satisfy constraint: Member must satisfy enum value set: [forward, redirect, fixed-response]" % (invalid_name, index)
         )
 
 
@@ -190,3 +190,18 @@ class InvalidModifyRuleArgumentsError(ELBClientError):
             "ValidationError",
             "Either conditions or actions must be specified"
         )
+
+
+class InvalidStatusCodeActionTypeError(ELBClientError):
+    def __init__(self, msg):
+        super(InvalidStatusCodeActionTypeError, self).__init__(
+            "ValidationError", msg
+        )
+
+
+class InvalidLoadBalancerActionException(ELBClientError):
+
+    def __init__(self, msg):
+        super(InvalidLoadBalancerActionException, self).__init__(
+            "InvalidLoadBalancerAction", msg
+        )

moto/elbv2/models.py

@@ -3,10 +3,11 @@ from __future__ import unicode_literals
 import datetime
 import re
 from jinja2 import Template
+from botocore.exceptions import ParamValidationError
 from moto.compat import OrderedDict
 from moto.core.exceptions import RESTError
 from moto.core import BaseBackend, BaseModel
-from moto.core.utils import camelcase_to_underscores
+from moto.core.utils import camelcase_to_underscores, underscores_to_camelcase
 from moto.ec2.models import ec2_backends
 from moto.acm.models import acm_backends
 from .utils import make_arn_for_target_group
@@ -31,8 +32,8 @@ from .exceptions import (
     RuleNotFoundError,
     DuplicatePriorityError,
     InvalidTargetGroupNameError,
-    InvalidModifyRuleArgumentsError
+    InvalidModifyRuleArgumentsError,
-)
+    InvalidStatusCodeActionTypeError, InvalidLoadBalancerActionException)
 
 
 class FakeHealthStatus(BaseModel):
@@ -220,9 +221,9 @@ class FakeListener(BaseModel):
                 action_type = action['Type']
                 if action_type == 'forward':
                     default_actions.append({'type': action_type, 'target_group_arn': action['TargetGroupArn']})
-                elif action_type in ['redirect', 'authenticate-cognito']:
+                elif action_type in ['redirect', 'authenticate-cognito', 'fixed-response']:
                     redirect_action = {'type': action_type}
-                    key = 'RedirectConfig' if action_type == 'redirect' else 'AuthenticateCognitoConfig'
+                    key = underscores_to_camelcase(action_type.capitalize().replace('-', '_')) + 'Config'
                     for redirect_config_key, redirect_config_value in action[key].items():
                         # need to match the output of _get_list_prefix
                         redirect_action[camelcase_to_underscores(key) + '._' + camelcase_to_underscores(redirect_config_key)] = redirect_config_value
@@ -258,6 +259,12 @@ class FakeAction(BaseModel):
                 <UserPoolClientId>{{ action.data["authenticate_cognito_config._user_pool_client_id"] }}</UserPoolClientId>
                 <UserPoolDomain>{{ action.data["authenticate_cognito_config._user_pool_domain"] }}</UserPoolDomain>
             </AuthenticateCognitoConfig>
+            {% elif action.type == "fixed-response" %}
+             <FixedResponseConfig>
+                <ContentType>{{ action.data["fixed_response_config._content_type"] }}</ContentType>
+                <MessageBody>{{ action.data["fixed_response_config._message_body"] }}</MessageBody>
+                <StatusCode>{{ action.data["fixed_response_config._status_code"] }}</StatusCode>
+            </FixedResponseConfig>
             {% endif %}
             """)
         return template.render(action=self)
@@ -482,11 +489,30 @@ class ELBv2Backend(BaseBackend):
                 action_target_group_arn = action.data['target_group_arn']
                 if action_target_group_arn not in target_group_arns:
                     raise ActionTargetGroupNotFoundError(action_target_group_arn)
+            elif action_type == 'fixed-response':
+                self._validate_fixed_response_action(action, i, index)
             elif action_type in ['redirect', 'authenticate-cognito']:
                 pass
             else:
                 raise InvalidActionTypeError(action_type, index)
 
+    def _validate_fixed_response_action(self, action, i, index):
+        status_code = action.data.get('fixed_response_config._status_code')
+        if status_code is None:
+            raise ParamValidationError(
+                report='Missing required parameter in Actions[%s].FixedResponseConfig: "StatusCode"' % i)
+        if not re.match(r'^(2|4|5)\d\d$', status_code):
+            raise InvalidStatusCodeActionTypeError(
+                "1 validation error detected: Value '%s' at 'actions.%s.member.fixedResponseConfig.statusCode' failed to satisfy constraint: \
+Member must satisfy regular expression pattern: ^(2|4|5)\d\d$" % (status_code, index)
+            )
+        content_type = action.data['fixed_response_config._content_type']
+        if content_type and content_type not in ['text/plain', 'text/css', 'text/html', 'application/javascript',
+                                                 'application/json']:
+            raise InvalidLoadBalancerActionException(
+                "The ContentType must be one of:'text/html', 'application/json', 'application/javascript', 'text/css', 'text/plain'"
+            )
+
     def create_target_group(self, name, **kwargs):
         if len(name) > 32:
             raise InvalidTargetGroupNameError(

tests/test_elbv2/test_elbv2.py

@@ -4,7 +4,7 @@ import json
 import os
 import boto3
 import botocore
-from botocore.exceptions import ClientError
+from botocore.exceptions import ClientError, ParamValidationError
 from nose.tools import assert_raises
 import sure  # noqa
 
@@ -2017,3 +2017,279 @@ def test_cognito_action_listener_rule_cloudformation():
             'UserPoolDomain': 'testpool',
         }
     },])
+
+
+@mock_elbv2
+@mock_ec2
+def test_fixed_response_action_listener_rule():
+    conn = boto3.client('elbv2', region_name='us-east-1')
+    ec2 = boto3.resource('ec2', region_name='us-east-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='us-east-1a')
+    subnet2 = ec2.create_subnet(
+        VpcId=vpc.id,
+        CidrBlock='172.28.7.128/26',
+        AvailabilityZone='us-east-1b')
+
+    response = conn.create_load_balancer(
+        Name='my-lb',
+        Subnets=[subnet1.id, subnet2.id],
+        SecurityGroups=[security_group.id],
+        Scheme='internal',
+        Tags=[{'Key': 'key_name', 'Value': 'a_value'}])
+    load_balancer_arn = response.get('LoadBalancers')[0].get('LoadBalancerArn')
+
+    action = {
+        'Type': 'fixed-response',
+        'FixedResponseConfig': {
+             'ContentType': 'text/plain',
+             'MessageBody': 'This page does not exist',
+             'StatusCode': '404',
+         }
+    }
+    response = conn.create_listener(LoadBalancerArn=load_balancer_arn,
+                                    Protocol='HTTP',
+                                    Port=80,
+                                    DefaultActions=[action])
+
+    listener = response.get('Listeners')[0]
+    listener.get('DefaultActions')[0].should.equal(action)
+    listener_arn = listener.get('ListenerArn')
+
+    describe_rules_response = conn.describe_rules(ListenerArn=listener_arn)
+    describe_rules_response['Rules'][0]['Actions'][0].should.equal(action)
+
+    describe_listener_response = conn.describe_listeners(ListenerArns=[listener_arn, ])
+    describe_listener_actions = describe_listener_response['Listeners'][0]['DefaultActions'][0]
+    describe_listener_actions.should.equal(action)
+
+
+@mock_elbv2
+@mock_cloudformation
+def test_fixed_response_action_listener_rule_cloudformation():
+    cnf_conn = boto3.client('cloudformation', region_name='us-east-1')
+    elbv2_client = boto3.client('elbv2', region_name='us-east-1')
+
+    template = {
+        "AWSTemplateFormatVersion": "2010-09-09",
+        "Description": "ECS Cluster Test CloudFormation",
+        "Resources": {
+            "testVPC": {
+                "Type": "AWS::EC2::VPC",
+                "Properties": {
+                    "CidrBlock": "10.0.0.0/16",
+                },
+            },
+            "subnet1": {
+                "Type": "AWS::EC2::Subnet",
+                "Properties": {
+                    "CidrBlock": "10.0.0.0/24",
+                    "VpcId": {"Ref": "testVPC"},
+                    "AvalabilityZone": "us-east-1b",
+                },
+            },
+            "subnet2": {
+                "Type": "AWS::EC2::Subnet",
+                "Properties": {
+                    "CidrBlock": "10.0.1.0/24",
+                    "VpcId": {"Ref": "testVPC"},
+                    "AvalabilityZone": "us-east-1b",
+                },
+            },
+            "testLb": {
+                "Type": "AWS::ElasticLoadBalancingV2::LoadBalancer",
+                "Properties": {
+                    "Name": "my-lb",
+                    "Subnets": [{"Ref": "subnet1"}, {"Ref": "subnet2"}],
+                    "Type": "application",
+                    "SecurityGroups": [],
+                }
+            },
+            "testListener": {
+                "Type": "AWS::ElasticLoadBalancingV2::Listener",
+                "Properties": {
+                    "LoadBalancerArn": {"Ref": "testLb"},
+                    "Port": 80,
+                    "Protocol": "HTTP",
+                    "DefaultActions": [{
+                        "Type": "fixed-response",
+                        "FixedResponseConfig": {
+                            'ContentType': 'text/plain',
+                            'MessageBody': 'This page does not exist',
+                            'StatusCode': '404',
+                        }
+                    }]
+                }
+
+            }
+        }
+    }
+    template_json = json.dumps(template)
+    cnf_conn.create_stack(StackName="test-stack", TemplateBody=template_json)
+
+    describe_load_balancers_response = elbv2_client.describe_load_balancers(Names=['my-lb',])
+    load_balancer_arn = describe_load_balancers_response['LoadBalancers'][0]['LoadBalancerArn']
+    describe_listeners_response = elbv2_client.describe_listeners(LoadBalancerArn=load_balancer_arn)
+
+    describe_listeners_response['Listeners'].should.have.length_of(1)
+    describe_listeners_response['Listeners'][0]['DefaultActions'].should.equal([{
+        'Type': 'fixed-response',
+        "FixedResponseConfig": {
+            'ContentType': 'text/plain',
+            'MessageBody': 'This page does not exist',
+            'StatusCode': '404',
+        }
+    },])
+
+
+@mock_elbv2
+@mock_ec2
+def test_fixed_response_action_listener_rule_validates_status_code():
+    conn = boto3.client('elbv2', region_name='us-east-1')
+    ec2 = boto3.resource('ec2', region_name='us-east-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='us-east-1a')
+    subnet2 = ec2.create_subnet(
+        VpcId=vpc.id,
+        CidrBlock='172.28.7.128/26',
+        AvailabilityZone='us-east-1b')
+
+    response = conn.create_load_balancer(
+        Name='my-lb',
+        Subnets=[subnet1.id, subnet2.id],
+        SecurityGroups=[security_group.id],
+        Scheme='internal',
+        Tags=[{'Key': 'key_name', 'Value': 'a_value'}])
+    load_balancer_arn = response.get('LoadBalancers')[0].get('LoadBalancerArn')
+
+    missing_status_code_action = {
+        'Type': 'fixed-response',
+        'FixedResponseConfig': {
+             'ContentType': 'text/plain',
+             'MessageBody': 'This page does not exist',
+         }
+    }
+    with assert_raises(ParamValidationError):
+        conn.create_listener(LoadBalancerArn=load_balancer_arn,
+                                    Protocol='HTTP',
+                                    Port=80,
+                                    DefaultActions=[missing_status_code_action])
+
+    invalid_status_code_action = {
+        'Type': 'fixed-response',
+        'FixedResponseConfig': {
+            'ContentType': 'text/plain',
+            'MessageBody': 'This page does not exist',
+            'StatusCode': '100'
+        }
+    }
+
+    @mock_elbv2
+    @mock_ec2
+    def test_fixed_response_action_listener_rule_validates_status_code():
+        conn = boto3.client('elbv2', region_name='us-east-1')
+        ec2 = boto3.resource('ec2', region_name='us-east-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='us-east-1a')
+        subnet2 = ec2.create_subnet(
+            VpcId=vpc.id,
+            CidrBlock='172.28.7.128/26',
+            AvailabilityZone='us-east-1b')
+
+        response = conn.create_load_balancer(
+            Name='my-lb',
+            Subnets=[subnet1.id, subnet2.id],
+            SecurityGroups=[security_group.id],
+            Scheme='internal',
+            Tags=[{'Key': 'key_name', 'Value': 'a_value'}])
+        load_balancer_arn = response.get('LoadBalancers')[0].get('LoadBalancerArn')
+
+        missing_status_code_action = {
+            'Type': 'fixed-response',
+            'FixedResponseConfig': {
+                'ContentType': 'text/plain',
+                'MessageBody': 'This page does not exist',
+            }
+        }
+        with assert_raises(ParamValidationError):
+            conn.create_listener(LoadBalancerArn=load_balancer_arn,
+                                 Protocol='HTTP',
+                                 Port=80,
+                                 DefaultActions=[missing_status_code_action])
+
+        invalid_status_code_action = {
+            'Type': 'fixed-response',
+            'FixedResponseConfig': {
+                'ContentType': 'text/plain',
+                'MessageBody': 'This page does not exist',
+                'StatusCode': '100'
+            }
+        }
+
+        with assert_raises(ClientError) as invalid_status_code_exception:
+            conn.create_listener(LoadBalancerArn=load_balancer_arn,
+                                 Protocol='HTTP',
+                                 Port=80,
+                                 DefaultActions=[invalid_status_code_action])
+
+        invalid_status_code_exception.exception.response['Error']['Code'].should.equal('ValidationError')
+
+
+@mock_elbv2
+@mock_ec2
+def test_fixed_response_action_listener_rule_validates_content_type():
+    conn = boto3.client('elbv2', region_name='us-east-1')
+    ec2 = boto3.resource('ec2', region_name='us-east-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='us-east-1a')
+    subnet2 = ec2.create_subnet(
+        VpcId=vpc.id,
+        CidrBlock='172.28.7.128/26',
+        AvailabilityZone='us-east-1b')
+
+    response = conn.create_load_balancer(
+        Name='my-lb',
+        Subnets=[subnet1.id, subnet2.id],
+        SecurityGroups=[security_group.id],
+        Scheme='internal',
+        Tags=[{'Key': 'key_name', 'Value': 'a_value'}])
+    load_balancer_arn = response.get('LoadBalancers')[0].get('LoadBalancerArn')
+
+    invalid_content_type_action = {
+        'Type': 'fixed-response',
+        'FixedResponseConfig': {
+            'ContentType': 'Fake content type',
+            'MessageBody': 'This page does not exist',
+            'StatusCode': '200'
+        }
+    }
+    with assert_raises(ClientError) as invalid_content_type_exception:
+        conn.create_listener(LoadBalancerArn=load_balancer_arn,
+                             Protocol='HTTP',
+                             Port=80,
+                             DefaultActions=[invalid_content_type_action])
+    invalid_content_type_exception.exception.response['Error']['Code'].should.equal('InvalidLoadBalancerAction')