Add exact Number, exact String.Array and attribute key matching to SNS subscription filter policy and validate filter policy

moto/sns/exceptions.py

@@ -40,3 +40,11 @@ class InvalidParameterValue(RESTError):
     def __init__(self, message):
         super(InvalidParameterValue, self).__init__(
             "InvalidParameterValue", message)
+
+
+class InternalError(RESTError):
+    code = 500
+
+    def __init__(self, message):
+        super(InternalError, self).__init__(
+            "InternalFailure", message)

moto/sns/models.py

@@ -18,7 +18,7 @@ from moto.awslambda import lambda_backends
 
 from .exceptions import (
     SNSNotFoundError, DuplicateSnsEndpointError, SnsEndpointDisabled, SNSInvalidParameter,
-    InvalidParameterValue
+    InvalidParameterValue, InternalError
 )
 from .utils import make_arn_for_topic, make_arn_for_subscription
 
@@ -131,13 +131,47 @@ class Subscription(BaseModel):
             message_attributes = {}
 
         def _field_match(field, rules, message_attributes):
-            if field not in message_attributes:
-                return False
             for rule in rules:
+                #  TODO: boolean value matching is not supported, SNS behavior unknown
                 if isinstance(rule, six.string_types):
-                    # only string value matching is supported
+                    if field not in message_attributes:
+                        return False
                     if message_attributes[field]['Value'] == rule:
                         return True
+                    try:
+                        json_data = json.loads(message_attributes[field]['Value'])
+                        if rule in json_data:
+                            return True
+                    except (ValueError, TypeError):
+                        pass
+                if isinstance(rule, (six.integer_types, float)):
+                    if field not in message_attributes:
+                        return False
+                    if message_attributes[field]['Type'] == 'Number':
+                        attribute_values = [message_attributes[field]['Value']]
+                    elif message_attributes[field]['Type'] == 'String.Array':
+                        try:
+                            attribute_values = json.loads(message_attributes[field]['Value'])
+                            if not isinstance(attribute_values, list):
+                                attribute_values = [attribute_values]
+                        except (ValueError, TypeError):
+                            return False
+                    else:
+                        return False
+
+                    for attribute_values in attribute_values:
+                        # Even the offical documentation states a 5 digits of accuracy after the decimal point for numerics, in reality it is 6
+                        # https://docs.aws.amazon.com/sns/latest/dg/sns-subscription-filter-policies.html#subscription-filter-policy-constraints
+                        if int(attribute_values * 1000000) == int(rule * 1000000):
+                            return True
+                if isinstance(rule, dict):
+                    keyword = list(rule.keys())[0]
+                    attributes = list(rule.values())[0]
+                    if keyword == 'exists':
+                        if attributes and field in message_attributes:
+                            return True
+                        elif not attributes and field not in message_attributes:
+                            return True
             return False
 
         return all(_field_match(field, rules, message_attributes)
@@ -421,7 +455,49 @@ class SNSBackend(BaseBackend):
         subscription.attributes[name] = value
 
         if name == 'FilterPolicy':
-            subscription._filter_policy = json.loads(value)
+            filter_policy = json.loads(value)
+            self._validate_filter_policy(filter_policy)
+            subscription._filter_policy = filter_policy
+
+    def _validate_filter_policy(self, value):
+        # TODO: extend validation checks
+        combinations = 1
+        for rules in six.itervalues(value):
+            combinations *= len(rules)
+        # Even the offical documentation states the total combination of values must not exceed 100, in reality it is 150
+        # https://docs.aws.amazon.com/sns/latest/dg/sns-subscription-filter-policies.html#subscription-filter-policy-constraints
+        if combinations > 150:
+            raise SNSInvalidParameter("Invalid parameter: FilterPolicy: Filter policy is too complex")
+
+        for field, rules in six.iteritems(value):
+            for rule in rules:
+                if rule is None:
+                    continue
+                if isinstance(rule, six.string_types):
+                    continue
+                if isinstance(rule, bool):
+                    continue
+                if isinstance(rule, (six.integer_types, float)):
+                    if rule <= -1000000000 or rule >= 1000000000:
+                        raise InternalError("Unknown")
+                    continue
+                if isinstance(rule, dict):
+                    keyword = list(rule.keys())[0]
+                    attributes = list(rule.values())[0]
+                    if keyword == 'anything-but':
+                        continue
+                    elif keyword == 'exists':
+                        if not isinstance(attributes, bool):
+                            raise SNSInvalidParameter("Invalid parameter: FilterPolicy: exists match pattern must be either true or false.")
+                        continue
+                    elif keyword == 'numeric':
+                        continue
+                    elif keyword == 'prefix':
+                        continue
+                    else:
+                        raise SNSInvalidParameter("Invalid parameter: FilterPolicy: Unrecognized match type {type}".format(type=keyword))
+
+                raise SNSInvalidParameter("Invalid parameter: FilterPolicy: Match value must be String, number, true, false, or null")
 
 
 sns_backends = {}

moto/sns/responses.py

@@ -57,7 +57,16 @@ class SNSResponse(BaseResponse):
 
             transform_value = None
             if 'StringValue' in value:
-                transform_value = value['StringValue']
+                if data_type == 'Number':
+                    try:
+                        transform_value = float(value['StringValue'])
+                    except ValueError:
+                        raise InvalidParameterValue(
+                            "An error occurred (ParameterValueInvalid) "
+                            "when calling the Publish operation: "
+                            "Could not cast message attribute '{0}' value to number.".format(name))
+                else:
+                    transform_value = value['StringValue']
             elif 'BinaryValue' in value:
                 transform_value = value['BinaryValue']
             if not transform_value:

tests/test_sns/test_publishing_boto3.py

@@ -109,6 +109,17 @@ def test_publish_to_sqs_bad():
             }})
     except ClientError as err:
         err.response['Error']['Code'].should.equal('InvalidParameterValue')
+    try:
+        # Test Number DataType, with a non numeric value
+        conn.publish(
+            TopicArn=topic_arn, Message=message,
+            MessageAttributes={'price': {
+                'DataType': 'Number',
+                'StringValue': 'error'
+            }})
+    except ClientError as err:
+        err.response['Error']['Code'].should.equal('InvalidParameterValue')
+        err.response['Error']['Message'].should.equal("An error occurred (ParameterValueInvalid) when calling the Publish operation: Could not cast message attribute 'price' value to number.")
 
 
 @mock_sqs
@@ -487,3 +498,380 @@ def test_filtering_exact_string_no_attributes_no_match():
     message_attributes = [
         json.loads(m.body)['MessageAttributes'] for m in messages]
     message_attributes.should.equal([])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_exact_number_int():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'price': [100]})
+
+    topic.publish(
+        Message='match',
+        MessageAttributes={'price': {'DataType': 'Number',
+                                     'StringValue': '100'}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal(['match'])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal(
+        [{'price': {'Type': 'Number', 'Value': 100}}])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_exact_number_float():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'price': [100.1]})
+
+    topic.publish(
+        Message='match',
+        MessageAttributes={'price': {'DataType': 'Number',
+                                     'StringValue': '100.1'}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal(['match'])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal(
+        [{'price': {'Type': 'Number', 'Value': 100.1}}])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_exact_number_float_accuracy():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'price': [100.123456789]})
+
+    topic.publish(
+        Message='match',
+        MessageAttributes={'price': {'DataType': 'Number',
+                                     'StringValue': '100.1234561'}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal(['match'])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal(
+        [{'price': {'Type': 'Number', 'Value': 100.1234561}}])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_exact_number_no_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'price': [100]})
+
+    topic.publish(
+        Message='no match',
+        MessageAttributes={'price': {'DataType': 'Number',
+                                     'StringValue': '101'}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal([])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal([])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_exact_number_with_string_no_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'price': [100]})
+
+    topic.publish(
+        Message='no match',
+        MessageAttributes={'price': {'DataType': 'String',
+                                     'StringValue': '100'}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal([])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal([])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_string_array_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'customer_interests': ['basketball', 'baseball']})
+
+    topic.publish(
+        Message='match',
+        MessageAttributes={'customer_interests': {'DataType': 'String.Array',
+                                     'StringValue': json.dumps(['basketball', 'rugby'])}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal(['match'])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal(
+        [{'customer_interests': {'Type': 'String.Array', 'Value': json.dumps(['basketball', 'rugby'])}}])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_string_array_no_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'customer_interests': ['baseball']})
+
+    topic.publish(
+        Message='no_match',
+        MessageAttributes={'customer_interests': {'DataType': 'String.Array',
+                                     'StringValue': json.dumps(['basketball', 'rugby'])}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal([])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal([])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_string_array_with_number_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'price': [100, 500]})
+
+    topic.publish(
+        Message='match',
+        MessageAttributes={'price': {'DataType': 'String.Array',
+                                     'StringValue': json.dumps([100, 50])}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal(['match'])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal(
+        [{'price': {'Type': 'String.Array', 'Value': json.dumps([100, 50])}}])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_string_array_with_number_float_accuracy_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'price': [100.123456789, 500]})
+
+    topic.publish(
+        Message='match',
+        MessageAttributes={'price': {'DataType': 'String.Array',
+                                     'StringValue': json.dumps([100.1234561, 50])}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal(['match'])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal(
+        [{'price': {'Type': 'String.Array', 'Value': json.dumps([100.1234561, 50])}}])
+
+
+@mock_sqs
+@mock_sns
+# this is the correct behavior from SNS
+def test_filtering_string_array_with_number_no_array_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'price': [100, 500]})
+
+    topic.publish(
+        Message='match',
+        MessageAttributes={'price': {'DataType': 'String.Array',
+                                     'StringValue': '100'}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal(['match'])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal(
+        [{'price': {'Type': 'String.Array', 'Value': '100'}}])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_string_array_with_number_no_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'price': [500]})
+
+    topic.publish(
+        Message='no_match',
+        MessageAttributes={'price': {'DataType': 'String.Array',
+                                     'StringValue': json.dumps([100, 50])}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal([])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal([])
+
+
+@mock_sqs
+@mock_sns
+# this is the correct behavior from SNS
+def test_filtering_string_array_with_string_no_array_no_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'price': [100]})
+
+    topic.publish(
+        Message='no_match',
+        MessageAttributes={'price': {'DataType': 'String.Array',
+                                     'StringValue': 'one hundread'}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal([])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal([])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_attribute_key_exists_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'store': [{'exists': True}]})
+
+    topic.publish(
+        Message='match',
+        MessageAttributes={'store': {'DataType': 'String',
+                                     'StringValue': 'example_corp'}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal(['match'])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal(
+        [{'store': {'Type': 'String', 'Value': 'example_corp'}}])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_attribute_key_exists_no_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'store': [{'exists': True}]})
+
+    topic.publish(
+        Message='no match',
+        MessageAttributes={'event': {'DataType': 'String',
+                                     'StringValue': 'order_cancelled'}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal([])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal([])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_attribute_key_not_exists_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'store': [{'exists': False}]})
+
+    topic.publish(
+        Message='match',
+        MessageAttributes={'event': {'DataType': 'String',
+                                     'StringValue': 'order_cancelled'}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal(['match'])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal(
+        [{'event': {'Type': 'String', 'Value': 'order_cancelled'}}])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_attribute_key_not_exists_no_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'store': [{'exists': False}]})
+
+    topic.publish(
+        Message='no match',
+        MessageAttributes={'store': {'DataType': 'String',
+                                     'StringValue': 'example_corp'}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal([])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal([])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_all_AND_matching_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'store': [{'exists': True}],
+         'event': ['order_cancelled'],
+         'customer_interests': ['basketball', 'baseball'],
+         'price': [100]})
+
+    topic.publish(
+        Message='match',
+        MessageAttributes={'store': {'DataType': 'String',
+                                     'StringValue': 'example_corp'},
+                           'event': {'DataType': 'String',
+                                     'StringValue': 'order_cancelled'},
+                           'customer_interests': {'DataType': 'String.Array',
+                                                  'StringValue': json.dumps(['basketball', 'rugby'])},
+                           'price': {'DataType': 'Number',
+                                     'StringValue': '100'}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal(
+        ['match'])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal([{
+        'store': {'Type': 'String', 'Value': 'example_corp'},
+        'event': {'Type': 'String', 'Value': 'order_cancelled'},
+        'customer_interests': {'Type': 'String.Array', 'Value': json.dumps(['basketball', 'rugby'])},
+        'price': {'Type': 'Number', 'Value': 100}}])
+
+
+@mock_sqs
+@mock_sns
+def test_filtering_all_AND_matching_no_match():
+    topic, subscription, queue = _setup_filter_policy_test(
+        {'store': [{'exists': True}],
+         'event': ['order_cancelled'],
+         'customer_interests': ['basketball', 'baseball'],
+         'price': [100],
+         "encrypted": [False]})
+
+    topic.publish(
+        Message='no match',
+        MessageAttributes={'store': {'DataType': 'String',
+                                     'StringValue': 'example_corp'},
+                           'event': {'DataType': 'String',
+                                     'StringValue': 'order_cancelled'},
+                           'customer_interests': {'DataType': 'String.Array',
+                                                  'StringValue': json.dumps(['basketball', 'rugby'])},
+                           'price': {'DataType': 'Number',
+                                     'StringValue': '100'}})
+
+    messages = queue.receive_messages(MaxNumberOfMessages=5)
+    message_bodies = [json.loads(m.body)['Message'] for m in messages]
+    message_bodies.should.equal([])
+    message_attributes = [
+        json.loads(m.body)['MessageAttributes'] for m in messages]
+    message_attributes.should.equal([])

tests/test_sns/test_subscriptions_boto3.py

@@ -201,7 +201,9 @@ def test_creating_subscription_with_attributes():
         "store": ["example_corp"],
         "event": ["order_cancelled"],
         "encrypted": [False],
-        "customer_interests": ["basketball", "baseball"]
+        "customer_interests": ["basketball", "baseball"],
+        "price": [100, 100.12],
+        "error": [None]
     })
 
     conn.subscribe(TopicArn=topic_arn,
@@ -294,7 +296,9 @@ def test_set_subscription_attributes():
         "store": ["example_corp"],
         "event": ["order_cancelled"],
         "encrypted": [False],
-        "customer_interests": ["basketball", "baseball"]
+        "customer_interests": ["basketball", "baseball"],
+        "price": [100, 100.12],
+        "error": [None]
     })
     conn.set_subscription_attributes(
         SubscriptionArn=subscription_arn,
@@ -332,6 +336,77 @@ def test_set_subscription_attributes():
         )
 
 
+@mock_sns
+def test_subscribe_invalid_filter_policy():
+    conn = boto3.client('sns', region_name = 'us-east-1')
+    conn.create_topic(Name = 'some-topic')
+    response = conn.list_topics()
+    topic_arn = response['Topics'][0]['TopicArn']
+
+    try:
+        conn.subscribe(TopicArn = topic_arn,
+                       Protocol = 'http',
+                       Endpoint = 'http://example.com/',
+                       Attributes = {
+                           'FilterPolicy': json.dumps({
+                               'store': [str(i) for i in range(151)]
+                           })
+                       })
+    except ClientError as err:
+        err.response['Error']['Code'].should.equal('InvalidParameter')
+        err.response['Error']['Message'].should.equal('Invalid parameter: FilterPolicy: Filter policy is too complex')
+
+    try:
+        conn.subscribe(TopicArn = topic_arn,
+                       Protocol = 'http',
+                       Endpoint = 'http://example.com/',
+                       Attributes = {
+                           'FilterPolicy': json.dumps({
+                               'store': [['example_corp']]
+                           })
+                       })
+    except ClientError as err:
+        err.response['Error']['Code'].should.equal('InvalidParameter')
+        err.response['Error']['Message'].should.equal('Invalid parameter: FilterPolicy: Match value must be String, number, true, false, or null')
+
+    try:
+        conn.subscribe(TopicArn = topic_arn,
+                       Protocol = 'http',
+                       Endpoint = 'http://example.com/',
+                       Attributes = {
+                           'FilterPolicy': json.dumps({
+                               'store': [{'exists': None}]
+                           })
+                       })
+    except ClientError as err:
+        err.response['Error']['Code'].should.equal('InvalidParameter')
+        err.response['Error']['Message'].should.equal('Invalid parameter: FilterPolicy: exists match pattern must be either true or false.')
+
+    try:
+        conn.subscribe(TopicArn = topic_arn,
+                       Protocol = 'http',
+                       Endpoint = 'http://example.com/',
+                       Attributes = {
+                           'FilterPolicy': json.dumps({
+                                'store': [{'error': True}]
+                            })
+                       })
+    except ClientError as err:
+        err.response['Error']['Code'].should.equal('InvalidParameter')
+        err.response['Error']['Message'].should.equal('Invalid parameter: FilterPolicy: Unrecognized match type error')
+
+    try:
+        conn.subscribe(TopicArn = topic_arn,
+                       Protocol = 'http',
+                       Endpoint = 'http://example.com/',
+                       Attributes = {
+                           'FilterPolicy': json.dumps({
+                               'store': [1000000001]
+                           })
+                       })
+    except ClientError as err:
+        err.response['Error']['Code'].should.equal('InternalFailure')
+
 @mock_sns
 def test_check_not_opted_out():
     conn = boto3.client('sns', region_name='us-east-1')