We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Send SMS via SNS fails with 502
Hi, we are trying to mock the SMS sending via SMS.
```
aws --endpoint-url=http://localhost:4575 --region=us-east-1 sns publish --message "lorem ipsum" --phone-number 12345
```
Checking the localstack image I can see a Python error in the logs:
```
2018-10-30T13:46:19:ERROR:localstack.services.generic_proxy: Error forwarding request: None Traceback (most recent call last):
  File "/opt/code/localstack/localstack/services/generic_proxy.py", line 181, in forward
    path=path, data=data, headers=forward_headers)
  File "/opt/code/localstack/localstack/services/sns/sns_listener.py", line 68, in forward_request
    for subscriber in SNS_SUBSCRIPTIONS[topic_arn]:
KeyError: None
```
Checking the source code I can see that topic_arn is None because it is not informed in the SNS API Call. If I inform topic_arn everything works fine:
```
aws --endpoint-url=http://localhost:4575 --region=us-east-1 sns publish --message "lorem ipsum" --phone-number 12345 --topic-arn abcd
{
    "MessageId": "245292e5"
}
```

Can you guys please check that?
Thanks

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `localstack/services/sns/sns_listener.py`
Content:
```
1 import ast
2 import json
3 import logging
4 import requests
5 import uuid
6 import xmltodict
7 from requests.models import Response
8 from six.moves.urllib import parse as urlparse
9 from localstack.utils.aws import aws_stack
10 from localstack.utils.common import short_uid, to_str
11 from localstack.services.awslambda import lambda_api
12 from localstack.services.generic_proxy import ProxyListener
13 
14 # mappings for SNS topic subscriptions
15 SNS_SUBSCRIPTIONS = {}
16 
17 # set up logger
18 LOGGER = logging.getLogger(__name__)
19 
20 
21 class ProxyListenerSNS(ProxyListener):
22 
23     def forward_request(self, method, path, data, headers):
24 
25         # check region
26         try:
27             aws_stack.check_valid_region(headers)
28         except Exception as e:
29             return make_error(message=str(e), code=400)
30 
31         if method == 'POST' and path == '/':
32             req_data = urlparse.parse_qs(to_str(data))
33             req_action = req_data['Action'][0]
34             topic_arn = req_data.get('TargetArn') or req_data.get('TopicArn')
35 
36             if topic_arn:
37                 topic_arn = topic_arn[0]
38                 do_create_topic(topic_arn)
39 
40             if req_action == 'SetSubscriptionAttributes':
41                 sub = get_subscription_by_arn(req_data['SubscriptionArn'][0])
42                 if not sub:
43                     return make_error(message='Unable to find subscription for given ARN', code=400)
44                 attr_name = req_data['AttributeName'][0]
45                 attr_value = req_data['AttributeValue'][0]
46                 sub[attr_name] = attr_value
47                 return make_response(req_action)
48             elif req_action == 'GetSubscriptionAttributes':
49                 sub = get_subscription_by_arn(req_data['SubscriptionArn'][0])
50                 if not sub:
51                     return make_error(message='Unable to find subscription for given ARN', code=400)
52                 content = '<Attributes>'
53                 for key, value in sub.items():
54                     content += '<entry><key>%s</key><value>%s</value></entry>\n' % (key, value)
55                 content += '</Attributes>'
56                 return make_response(req_action, content=content)
57             elif req_action == 'Subscribe':
58                 if 'Endpoint' not in req_data:
59                     return make_error(message='Endpoint not specified in subscription', code=400)
60             elif req_action == 'Unsubscribe':
61                 if 'SubscriptionArn' not in req_data:
62                     return make_error(message='SubscriptionArn not specified in unsubscribe request', code=400)
63                 do_unsubscribe(req_data.get('SubscriptionArn')[0])
64 
65             elif req_action == 'Publish':
66                 message = req_data['Message'][0]
67                 sqs_client = aws_stack.connect_to_service('sqs')
68                 for subscriber in SNS_SUBSCRIPTIONS[topic_arn]:
69                     filter_policy = json.loads(subscriber.get('FilterPolicy', '{}'))
70                     message_attributes = get_message_attributes(req_data)
71                     if check_filter_policy(filter_policy, message_attributes):
72                         if subscriber['Protocol'] == 'sqs':
73                             endpoint = subscriber['Endpoint']
74                             if 'sqs_queue_url' in subscriber:
75                                 queue_url = subscriber.get('sqs_queue_url')
76                             elif '://' in endpoint:
77                                 queue_url = endpoint
78                             else:
79                                 queue_name = endpoint.split(':')[5]
80                                 queue_url = aws_stack.get_sqs_queue_url(queue_name)
81                                 subscriber['sqs_queue_url'] = queue_url
82                             try:
83                                 sqs_client.send_message(
84                                     QueueUrl=queue_url,
85                                     MessageBody=create_sns_message_body(subscriber, req_data)
86                                 )
87                             except Exception as exc:
88                                 return make_error(message=str(exc), code=400)
89                         elif subscriber['Protocol'] == 'lambda':
90                             lambda_api.process_sns_notification(
91                                 subscriber['Endpoint'],
92                                 topic_arn, message, subject=req_data.get('Subject', [None])[0]
93                             )
94                         elif subscriber['Protocol'] in ['http', 'https']:
95                             try:
96                                 message_body = create_sns_message_body(subscriber, req_data)
97                             except Exception as exc:
98                                 return make_error(message=str(exc), code=400)
99                             requests.post(
100                                 subscriber['Endpoint'],
101                                 headers={
102                                     'Content-Type': 'text/plain',
103                                     'x-amz-sns-message-type': 'Notification'
104                                 },
105                                 data=message_body
106                             )
107                         else:
108                             LOGGER.warning('Unexpected protocol "%s" for SNS subscription' % subscriber['Protocol'])
109                 # return response here because we do not want the request to be forwarded to SNS
110                 return make_response(req_action)
111 
112         return True
113 
114     def return_response(self, method, path, data, headers, response):
115         # This method is executed by the proxy after we've already received a
116         # response from the backend, hence we can utilize the "response" variable here
117         if method == 'POST' and path == '/':
118             req_data = urlparse.parse_qs(to_str(data))
119             req_action = req_data['Action'][0]
120             if req_action == 'Subscribe' and response.status_code < 400:
121                 response_data = xmltodict.parse(response.content)
122                 topic_arn = (req_data.get('TargetArn') or req_data.get('TopicArn'))[0]
123                 sub_arn = response_data['SubscribeResponse']['SubscribeResult']['SubscriptionArn']
124                 do_subscribe(topic_arn, req_data['Endpoint'][0], req_data['Protocol'][0], sub_arn)
125 
126 
127 # instantiate listener
128 UPDATE_SNS = ProxyListenerSNS()
129 
130 
131 def do_create_topic(topic_arn):
132     if topic_arn not in SNS_SUBSCRIPTIONS:
133         SNS_SUBSCRIPTIONS[topic_arn] = []
134 
135 
136 def do_subscribe(topic_arn, endpoint, protocol, subscription_arn):
137     subscription = {
138         # http://docs.aws.amazon.com/cli/latest/reference/sns/get-subscription-attributes.html
139         'TopicArn': topic_arn,
140         'Endpoint': endpoint,
141         'Protocol': protocol,
142         'SubscriptionArn': subscription_arn,
143         'RawMessageDelivery': 'false'
144     }
145     SNS_SUBSCRIPTIONS[topic_arn].append(subscription)
146 
147 
148 def do_unsubscribe(subscription_arn):
149     for topic_arn in SNS_SUBSCRIPTIONS:
150         SNS_SUBSCRIPTIONS[topic_arn] = [
151             sub for sub in SNS_SUBSCRIPTIONS[topic_arn]
152             if sub['SubscriptionArn'] != subscription_arn
153         ]
154 
155 
156 # ---------------
157 # HELPER METHODS
158 # ---------------
159 
160 def get_topic_by_arn(topic_arn):
161     if topic_arn in SNS_SUBSCRIPTIONS:
162         return SNS_SUBSCRIPTIONS[topic_arn]
163     else:
164         return None
165 
166 
167 def get_subscription_by_arn(sub_arn):
168     # TODO maintain separate map instead of traversing all items
169     for key, subscriptions in SNS_SUBSCRIPTIONS.items():
170         for sub in subscriptions:
171             if sub['SubscriptionArn'] == sub_arn:
172                 return sub
173 
174 
175 def make_response(op_name, content=''):
176     response = Response()
177     if not content:
178         content = '<MessageId>%s</MessageId>' % short_uid()
179     response._content = """<{op_name}Response xmlns="http://sns.amazonaws.com/doc/2010-03-31/">
180         <{op_name}Result>
181             {content}
182         </{op_name}Result>
183         <ResponseMetadata><RequestId>{req_id}</RequestId></ResponseMetadata>
184         </{op_name}Response>""".format(op_name=op_name, content=content, req_id=short_uid())
185     response.status_code = 200
186     return response
187 
188 
189 def make_error(message, code=400, code_string='InvalidParameter'):
190     response = Response()
191     response._content = """<ErrorResponse xmlns="http://sns.amazonaws.com/doc/2010-03-31/"><Error>
192         <Type>Sender</Type>
193         <Code>{code_string}</Code>
194         <Message>{message}</Message>
195         </Error><RequestId>{req_id}</RequestId>
196         </ErrorResponse>""".format(message=message, code_string=code_string, req_id=short_uid())
197     response.status_code = code
198     return response
199 
200 
201 def create_sns_message_body(subscriber, req_data):
202     message = req_data['Message'][0]
203     subject = req_data.get('Subject', [None])[0]
204     protocol = subscriber['Protocol']
205 
206     if subscriber['RawMessageDelivery'] == 'true':
207         return message
208 
209     if req_data.get('MessageStructure') == ['json']:
210         message = json.loads(message)
211         try:
212             message = message.get(protocol, message['default'])
213         except KeyError:
214             raise Exception("Unable to find 'default' key in message payload")
215 
216     data = {}
217     data['MessageId'] = str(uuid.uuid4())
218     data['Type'] = 'Notification'
219     data['Message'] = message
220     data['TopicArn'] = subscriber['TopicArn']
221     if subject is not None:
222         data['Subject'] = subject
223     attributes = get_message_attributes(req_data)
224     if attributes:
225         data['MessageAttributes'] = attributes
226     return json.dumps(data)
227 
228 
229 def get_message_attributes(req_data):
230     attributes = {}
231     x = 1
232     while True:
233         name = req_data.get('MessageAttributes.entry.' + str(x) + '.Name', [None])[0]
234         if name is not None:
235             attribute = {}
236             attribute['Type'] = req_data.get('MessageAttributes.entry.' + str(x) + '.Value.DataType', [None])[0]
237             string_value = req_data.get('MessageAttributes.entry.' + str(x) + '.Value.StringValue', [None])[0]
238             binary_value = req_data.get('MessageAttributes.entry.' + str(x) + '.Value.BinaryValue', [None])[0]
239             if string_value is not None:
240                 attribute['Value'] = string_value
241             elif binary_value is not None:
242                 attribute['Value'] = binary_value
243 
244             attributes[name] = attribute
245             x += 1
246         else:
247             break
248 
249     return attributes
250 
251 
252 def is_number(x):
253     try:
254         float(x)
255         return True
256     except ValueError:
257         return False
258 
259 
260 def evaluate_numeric_condition(conditions, value):
261     if not is_number(value):
262         return False
263 
264     for i in range(0, len(conditions), 2):
265         operator = conditions[i]
266         operand = conditions[i + 1]
267 
268         if operator == '=':
269             if value != operand:
270                 return False
271         elif operator == '>':
272             if value <= operand:
273                 return False
274         elif operator == '<':
275             if value >= operand:
276                 return False
277         elif operator == '>=':
278             if value < operand:
279                 return False
280         elif operator == '<=':
281             if value > operand:
282                 return False
283 
284     return True
285 
286 
287 def evaluate_condition(value, condition):
288     if type(condition) is not dict:
289         return value == condition
290     elif condition.get('anything-but'):
291         return value not in condition.get('anything-but')
292     elif condition.get('prefix'):
293         prefix = condition.get('prefix')
294         return value.startswith(prefix)
295     elif condition.get('numeric'):
296         return evaluate_numeric_condition(condition.get('numeric'), value)
297 
298     return False
299 
300 
301 def evaluate_filter_policy_conditions(conditions, attribute):
302     if type(conditions) is not list:
303         conditions = [conditions]
304 
305     if attribute['Type'] == 'String.Array':
306         values = ast.literal_eval(attribute['Value'])
307         for value in values:
308             for condition in conditions:
309                 if evaluate_condition(value, condition):
310                     return True
311     else:
312         for condition in conditions:
313             if evaluate_condition(attribute['Value'], condition):
314                 return True
315 
316     return False
317 
318 
319 def check_filter_policy(filter_policy, message_attributes):
320     if not filter_policy:
321         return True
322 
323     for criteria in filter_policy:
324         conditions = filter_policy.get(criteria)
325         attribute = message_attributes.get(criteria)
326 
327         if attribute is None:
328             return False
329 
330         if evaluate_filter_policy_conditions(conditions, attribute) is False:
331             return False
332 
333     return True
334
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/localstack/services/sns/sns_listener.py b/localstack/services/sns/sns_listener.py
--- a/localstack/services/sns/sns_listener.py
+++ b/localstack/services/sns/sns_listener.py
@@ -65,7 +65,7 @@
             elif req_action == 'Publish':
                 message = req_data['Message'][0]
                 sqs_client = aws_stack.connect_to_service('sqs')
-                for subscriber in SNS_SUBSCRIPTIONS[topic_arn]:
+                for subscriber in SNS_SUBSCRIPTIONS.get(topic_arn, []):
                     filter_policy = json.loads(subscriber.get('FilterPolicy', '{}'))
                     message_attributes = get_message_attributes(req_data)
                     if check_filter_policy(filter_policy, message_attributes):

{"golden_diff": "diff --git a/localstack/services/sns/sns_listener.py b/localstack/services/sns/sns_listener.py\n--- a/localstack/services/sns/sns_listener.py\n+++ b/localstack/services/sns/sns_listener.py\n@@ -65,7 +65,7 @@\n             elif req_action == 'Publish':\n                 message = req_data['Message'][0]\n                 sqs_client = aws_stack.connect_to_service('sqs')\n-                for subscriber in SNS_SUBSCRIPTIONS[topic_arn]:\n+                for subscriber in SNS_SUBSCRIPTIONS.get(topic_arn, []):\n                     filter_policy = json.loads(subscriber.get('FilterPolicy', '{}'))\n                     message_attributes = get_message_attributes(req_data)\n                     if check_filter_policy(filter_policy, message_attributes):\n", "issue": "Send SMS via SNS fails with 502\nHi, we are trying to mock the SMS sending via SMS.\r\n```\r\naws --endpoint-url=http://localhost:4575 --region=us-east-1 sns publish --message \"lorem ipsum\" --phone-number 12345\r\n```\r\nChecking the localstack image I can see a Python error in the logs:\r\n```\r\n2018-10-30T13:46:19:ERROR:localstack.services.generic_proxy: Error forwarding request: None Traceback (most recent call last):\r\n  File \"/opt/code/localstack/localstack/services/generic_proxy.py\", line 181, in forward\r\n    path=path, data=data, headers=forward_headers)\r\n  File \"/opt/code/localstack/localstack/services/sns/sns_listener.py\", line 68, in forward_request\r\n    for subscriber in SNS_SUBSCRIPTIONS[topic_arn]:\r\nKeyError: None\r\n```\r\nChecking the source code I can see that topic_arn is None because it is not informed in the SNS API Call. If I inform topic_arn everything works fine:\r\n```\r\naws --endpoint-url=http://localhost:4575 --region=us-east-1 sns publish --message \"lorem ipsum\" --phone-number 12345 --topic-arn abcd\r\n{\r\n    \"MessageId\": \"245292e5\"\r\n}\r\n```\r\n\r\nCan you guys please check that?\r\nThanks\n", "before_files": [{"content": "import ast\nimport json\nimport logging\nimport requests\nimport uuid\nimport xmltodict\nfrom requests.models import Response\nfrom six.moves.urllib import parse as urlparse\nfrom localstack.utils.aws import aws_stack\nfrom localstack.utils.common import short_uid, to_str\nfrom localstack.services.awslambda import lambda_api\nfrom localstack.services.generic_proxy import ProxyListener\n\n# mappings for SNS topic subscriptions\nSNS_SUBSCRIPTIONS = {}\n\n# set up logger\nLOGGER = logging.getLogger(__name__)\n\n\nclass ProxyListenerSNS(ProxyListener):\n\n    def forward_request(self, method, path, data, headers):\n\n        # check region\n        try:\n            aws_stack.check_valid_region(headers)\n        except Exception as e:\n            return make_error(message=str(e), code=400)\n\n        if method == 'POST' and path == '/':\n            req_data = urlparse.parse_qs(to_str(data))\n            req_action = req_data['Action'][0]\n            topic_arn = req_data.get('TargetArn') or req_data.get('TopicArn')\n\n            if topic_arn:\n                topic_arn = topic_arn[0]\n                do_create_topic(topic_arn)\n\n            if req_action == 'SetSubscriptionAttributes':\n                sub = get_subscription_by_arn(req_data['SubscriptionArn'][0])\n                if not sub:\n                    return make_error(message='Unable to find subscription for given ARN', code=400)\n                attr_name = req_data['AttributeName'][0]\n                attr_value = req_data['AttributeValue'][0]\n                sub[attr_name] = attr_value\n                return make_response(req_action)\n            elif req_action == 'GetSubscriptionAttributes':\n                sub = get_subscription_by_arn(req_data['SubscriptionArn'][0])\n                if not sub:\n                    return make_error(message='Unable to find subscription for given ARN', code=400)\n                content = '<Attributes>'\n                for key, value in sub.items():\n                    content += '<entry><key>%s</key><value>%s</value></entry>\\n' % (key, value)\n                content += '</Attributes>'\n                return make_response(req_action, content=content)\n            elif req_action == 'Subscribe':\n                if 'Endpoint' not in req_data:\n                    return make_error(message='Endpoint not specified in subscription', code=400)\n            elif req_action == 'Unsubscribe':\n                if 'SubscriptionArn' not in req_data:\n                    return make_error(message='SubscriptionArn not specified in unsubscribe request', code=400)\n                do_unsubscribe(req_data.get('SubscriptionArn')[0])\n\n            elif req_action == 'Publish':\n                message = req_data['Message'][0]\n                sqs_client = aws_stack.connect_to_service('sqs')\n                for subscriber in SNS_SUBSCRIPTIONS[topic_arn]:\n                    filter_policy = json.loads(subscriber.get('FilterPolicy', '{}'))\n                    message_attributes = get_message_attributes(req_data)\n                    if check_filter_policy(filter_policy, message_attributes):\n                        if subscriber['Protocol'] == 'sqs':\n                            endpoint = subscriber['Endpoint']\n                            if 'sqs_queue_url' in subscriber:\n                                queue_url = subscriber.get('sqs_queue_url')\n                            elif '://' in endpoint:\n                                queue_url = endpoint\n                            else:\n                                queue_name = endpoint.split(':')[5]\n                                queue_url = aws_stack.get_sqs_queue_url(queue_name)\n                                subscriber['sqs_queue_url'] = queue_url\n                            try:\n                                sqs_client.send_message(\n                                    QueueUrl=queue_url,\n                                    MessageBody=create_sns_message_body(subscriber, req_data)\n                                )\n                            except Exception as exc:\n                                return make_error(message=str(exc), code=400)\n                        elif subscriber['Protocol'] == 'lambda':\n                            lambda_api.process_sns_notification(\n                                subscriber['Endpoint'],\n                                topic_arn, message, subject=req_data.get('Subject', [None])[0]\n                            )\n                        elif subscriber['Protocol'] in ['http', 'https']:\n                            try:\n                                message_body = create_sns_message_body(subscriber, req_data)\n                            except Exception as exc:\n                                return make_error(message=str(exc), code=400)\n                            requests.post(\n                                subscriber['Endpoint'],\n                                headers={\n                                    'Content-Type': 'text/plain',\n                                    'x-amz-sns-message-type': 'Notification'\n                                },\n                                data=message_body\n                            )\n                        else:\n                            LOGGER.warning('Unexpected protocol \"%s\" for SNS subscription' % subscriber['Protocol'])\n                # return response here because we do not want the request to be forwarded to SNS\n                return make_response(req_action)\n\n        return True\n\n    def return_response(self, method, path, data, headers, response):\n        # This method is executed by the proxy after we've already received a\n        # response from the backend, hence we can utilize the \"response\" variable here\n        if method == 'POST' and path == '/':\n            req_data = urlparse.parse_qs(to_str(data))\n            req_action = req_data['Action'][0]\n            if req_action == 'Subscribe' and response.status_code < 400:\n                response_data = xmltodict.parse(response.content)\n                topic_arn = (req_data.get('TargetArn') or req_data.get('TopicArn'))[0]\n                sub_arn = response_data['SubscribeResponse']['SubscribeResult']['SubscriptionArn']\n                do_subscribe(topic_arn, req_data['Endpoint'][0], req_data['Protocol'][0], sub_arn)\n\n\n# instantiate listener\nUPDATE_SNS = ProxyListenerSNS()\n\n\ndef do_create_topic(topic_arn):\n    if topic_arn not in SNS_SUBSCRIPTIONS:\n        SNS_SUBSCRIPTIONS[topic_arn] = []\n\n\ndef do_subscribe(topic_arn, endpoint, protocol, subscription_arn):\n    subscription = {\n        # http://docs.aws.amazon.com/cli/latest/reference/sns/get-subscription-attributes.html\n        'TopicArn': topic_arn,\n        'Endpoint': endpoint,\n        'Protocol': protocol,\n        'SubscriptionArn': subscription_arn,\n        'RawMessageDelivery': 'false'\n    }\n    SNS_SUBSCRIPTIONS[topic_arn].append(subscription)\n\n\ndef do_unsubscribe(subscription_arn):\n    for topic_arn in SNS_SUBSCRIPTIONS:\n        SNS_SUBSCRIPTIONS[topic_arn] = [\n            sub for sub in SNS_SUBSCRIPTIONS[topic_arn]\n            if sub['SubscriptionArn'] != subscription_arn\n        ]\n\n\n# ---------------\n# HELPER METHODS\n# ---------------\n\ndef get_topic_by_arn(topic_arn):\n    if topic_arn in SNS_SUBSCRIPTIONS:\n        return SNS_SUBSCRIPTIONS[topic_arn]\n    else:\n        return None\n\n\ndef get_subscription_by_arn(sub_arn):\n    # TODO maintain separate map instead of traversing all items\n    for key, subscriptions in SNS_SUBSCRIPTIONS.items():\n        for sub in subscriptions:\n            if sub['SubscriptionArn'] == sub_arn:\n                return sub\n\n\ndef make_response(op_name, content=''):\n    response = Response()\n    if not content:\n        content = '<MessageId>%s</MessageId>' % short_uid()\n    response._content = \"\"\"<{op_name}Response xmlns=\"http://sns.amazonaws.com/doc/2010-03-31/\">\n        <{op_name}Result>\n            {content}\n        </{op_name}Result>\n        <ResponseMetadata><RequestId>{req_id}</RequestId></ResponseMetadata>\n        </{op_name}Response>\"\"\".format(op_name=op_name, content=content, req_id=short_uid())\n    response.status_code = 200\n    return response\n\n\ndef make_error(message, code=400, code_string='InvalidParameter'):\n    response = Response()\n    response._content = \"\"\"<ErrorResponse xmlns=\"http://sns.amazonaws.com/doc/2010-03-31/\"><Error>\n        <Type>Sender</Type>\n        <Code>{code_string}</Code>\n        <Message>{message}</Message>\n        </Error><RequestId>{req_id}</RequestId>\n        </ErrorResponse>\"\"\".format(message=message, code_string=code_string, req_id=short_uid())\n    response.status_code = code\n    return response\n\n\ndef create_sns_message_body(subscriber, req_data):\n    message = req_data['Message'][0]\n    subject = req_data.get('Subject', [None])[0]\n    protocol = subscriber['Protocol']\n\n    if subscriber['RawMessageDelivery'] == 'true':\n        return message\n\n    if req_data.get('MessageStructure') == ['json']:\n        message = json.loads(message)\n        try:\n            message = message.get(protocol, message['default'])\n        except KeyError:\n            raise Exception(\"Unable to find 'default' key in message payload\")\n\n    data = {}\n    data['MessageId'] = str(uuid.uuid4())\n    data['Type'] = 'Notification'\n    data['Message'] = message\n    data['TopicArn'] = subscriber['TopicArn']\n    if subject is not None:\n        data['Subject'] = subject\n    attributes = get_message_attributes(req_data)\n    if attributes:\n        data['MessageAttributes'] = attributes\n    return json.dumps(data)\n\n\ndef get_message_attributes(req_data):\n    attributes = {}\n    x = 1\n    while True:\n        name = req_data.get('MessageAttributes.entry.' + str(x) + '.Name', [None])[0]\n        if name is not None:\n            attribute = {}\n            attribute['Type'] = req_data.get('MessageAttributes.entry.' + str(x) + '.Value.DataType', [None])[0]\n            string_value = req_data.get('MessageAttributes.entry.' + str(x) + '.Value.StringValue', [None])[0]\n            binary_value = req_data.get('MessageAttributes.entry.' + str(x) + '.Value.BinaryValue', [None])[0]\n            if string_value is not None:\n                attribute['Value'] = string_value\n            elif binary_value is not None:\n                attribute['Value'] = binary_value\n\n            attributes[name] = attribute\n            x += 1\n        else:\n            break\n\n    return attributes\n\n\ndef is_number(x):\n    try:\n        float(x)\n        return True\n    except ValueError:\n        return False\n\n\ndef evaluate_numeric_condition(conditions, value):\n    if not is_number(value):\n        return False\n\n    for i in range(0, len(conditions), 2):\n        operator = conditions[i]\n        operand = conditions[i + 1]\n\n        if operator == '=':\n            if value != operand:\n                return False\n        elif operator == '>':\n            if value <= operand:\n                return False\n        elif operator == '<':\n            if value >= operand:\n                return False\n        elif operator == '>=':\n            if value < operand:\n                return False\n        elif operator == '<=':\n            if value > operand:\n                return False\n\n    return True\n\n\ndef evaluate_condition(value, condition):\n    if type(condition) is not dict:\n        return value == condition\n    elif condition.get('anything-but'):\n        return value not in condition.get('anything-but')\n    elif condition.get('prefix'):\n        prefix = condition.get('prefix')\n        return value.startswith(prefix)\n    elif condition.get('numeric'):\n        return evaluate_numeric_condition(condition.get('numeric'), value)\n\n    return False\n\n\ndef evaluate_filter_policy_conditions(conditions, attribute):\n    if type(conditions) is not list:\n        conditions = [conditions]\n\n    if attribute['Type'] == 'String.Array':\n        values = ast.literal_eval(attribute['Value'])\n        for value in values:\n            for condition in conditions:\n                if evaluate_condition(value, condition):\n                    return True\n    else:\n        for condition in conditions:\n            if evaluate_condition(attribute['Value'], condition):\n                return True\n\n    return False\n\n\ndef check_filter_policy(filter_policy, message_attributes):\n    if not filter_policy:\n        return True\n\n    for criteria in filter_policy:\n        conditions = filter_policy.get(criteria)\n        attribute = message_attributes.get(criteria)\n\n        if attribute is None:\n            return False\n\n        if evaluate_filter_policy_conditions(conditions, attribute) is False:\n            return False\n\n    return True\n", "path": "localstack/services/sns/sns_listener.py"}], "after_files": [{"content": "import ast\nimport json\nimport logging\nimport requests\nimport uuid\nimport xmltodict\nfrom requests.models import Response\nfrom six.moves.urllib import parse as urlparse\nfrom localstack.utils.aws import aws_stack\nfrom localstack.utils.common import short_uid, to_str\nfrom localstack.services.awslambda import lambda_api\nfrom localstack.services.generic_proxy import ProxyListener\n\n# mappings for SNS topic subscriptions\nSNS_SUBSCRIPTIONS = {}\n\n# set up logger\nLOGGER = logging.getLogger(__name__)\n\n\nclass ProxyListenerSNS(ProxyListener):\n\n    def forward_request(self, method, path, data, headers):\n\n        # check region\n        try:\n            aws_stack.check_valid_region(headers)\n        except Exception as e:\n            return make_error(message=str(e), code=400)\n\n        if method == 'POST' and path == '/':\n            req_data = urlparse.parse_qs(to_str(data))\n            req_action = req_data['Action'][0]\n            topic_arn = req_data.get('TargetArn') or req_data.get('TopicArn')\n\n            if topic_arn:\n                topic_arn = topic_arn[0]\n                do_create_topic(topic_arn)\n\n            if req_action == 'SetSubscriptionAttributes':\n                sub = get_subscription_by_arn(req_data['SubscriptionArn'][0])\n                if not sub:\n                    return make_error(message='Unable to find subscription for given ARN', code=400)\n                attr_name = req_data['AttributeName'][0]\n                attr_value = req_data['AttributeValue'][0]\n                sub[attr_name] = attr_value\n                return make_response(req_action)\n            elif req_action == 'GetSubscriptionAttributes':\n                sub = get_subscription_by_arn(req_data['SubscriptionArn'][0])\n                if not sub:\n                    return make_error(message='Unable to find subscription for given ARN', code=400)\n                content = '<Attributes>'\n                for key, value in sub.items():\n                    content += '<entry><key>%s</key><value>%s</value></entry>\\n' % (key, value)\n                content += '</Attributes>'\n                return make_response(req_action, content=content)\n            elif req_action == 'Subscribe':\n                if 'Endpoint' not in req_data:\n                    return make_error(message='Endpoint not specified in subscription', code=400)\n            elif req_action == 'Unsubscribe':\n                if 'SubscriptionArn' not in req_data:\n                    return make_error(message='SubscriptionArn not specified in unsubscribe request', code=400)\n                do_unsubscribe(req_data.get('SubscriptionArn')[0])\n\n            elif req_action == 'Publish':\n                message = req_data['Message'][0]\n                sqs_client = aws_stack.connect_to_service('sqs')\n                for subscriber in SNS_SUBSCRIPTIONS.get(topic_arn, []):\n                    filter_policy = json.loads(subscriber.get('FilterPolicy', '{}'))\n                    message_attributes = get_message_attributes(req_data)\n                    if check_filter_policy(filter_policy, message_attributes):\n                        if subscriber['Protocol'] == 'sqs':\n                            endpoint = subscriber['Endpoint']\n                            if 'sqs_queue_url' in subscriber:\n                                queue_url = subscriber.get('sqs_queue_url')\n                            elif '://' in endpoint:\n                                queue_url = endpoint\n                            else:\n                                queue_name = endpoint.split(':')[5]\n                                queue_url = aws_stack.get_sqs_queue_url(queue_name)\n                                subscriber['sqs_queue_url'] = queue_url\n                            try:\n                                sqs_client.send_message(\n                                    QueueUrl=queue_url,\n                                    MessageBody=create_sns_message_body(subscriber, req_data)\n                                )\n                            except Exception as exc:\n                                return make_error(message=str(exc), code=400)\n                        elif subscriber['Protocol'] == 'lambda':\n                            lambda_api.process_sns_notification(\n                                subscriber['Endpoint'],\n                                topic_arn, message, subject=req_data.get('Subject', [None])[0]\n                            )\n                        elif subscriber['Protocol'] in ['http', 'https']:\n                            try:\n                                message_body = create_sns_message_body(subscriber, req_data)\n                            except Exception as exc:\n                                return make_error(message=str(exc), code=400)\n                            requests.post(\n                                subscriber['Endpoint'],\n                                headers={\n                                    'Content-Type': 'text/plain',\n                                    'x-amz-sns-message-type': 'Notification'\n                                },\n                                data=message_body\n                            )\n                        else:\n                            LOGGER.warning('Unexpected protocol \"%s\" for SNS subscription' % subscriber['Protocol'])\n                # return response here because we do not want the request to be forwarded to SNS\n                return make_response(req_action)\n\n        return True\n\n    def return_response(self, method, path, data, headers, response):\n        # This method is executed by the proxy after we've already received a\n        # response from the backend, hence we can utilize the \"response\" variable here\n        if method == 'POST' and path == '/':\n            req_data = urlparse.parse_qs(to_str(data))\n            req_action = req_data['Action'][0]\n            if req_action == 'Subscribe' and response.status_code < 400:\n                response_data = xmltodict.parse(response.content)\n                topic_arn = (req_data.get('TargetArn') or req_data.get('TopicArn'))[0]\n                sub_arn = response_data['SubscribeResponse']['SubscribeResult']['SubscriptionArn']\n                do_subscribe(topic_arn, req_data['Endpoint'][0], req_data['Protocol'][0], sub_arn)\n\n\n# instantiate listener\nUPDATE_SNS = ProxyListenerSNS()\n\n\ndef do_create_topic(topic_arn):\n    if topic_arn not in SNS_SUBSCRIPTIONS:\n        SNS_SUBSCRIPTIONS[topic_arn] = []\n\n\ndef do_subscribe(topic_arn, endpoint, protocol, subscription_arn):\n    subscription = {\n        # http://docs.aws.amazon.com/cli/latest/reference/sns/get-subscription-attributes.html\n        'TopicArn': topic_arn,\n        'Endpoint': endpoint,\n        'Protocol': protocol,\n        'SubscriptionArn': subscription_arn,\n        'RawMessageDelivery': 'false'\n    }\n    SNS_SUBSCRIPTIONS[topic_arn].append(subscription)\n\n\ndef do_unsubscribe(subscription_arn):\n    for topic_arn in SNS_SUBSCRIPTIONS:\n        SNS_SUBSCRIPTIONS[topic_arn] = [\n            sub for sub in SNS_SUBSCRIPTIONS[topic_arn]\n            if sub['SubscriptionArn'] != subscription_arn\n        ]\n\n\n# ---------------\n# HELPER METHODS\n# ---------------\n\ndef get_topic_by_arn(topic_arn):\n    if topic_arn in SNS_SUBSCRIPTIONS:\n        return SNS_SUBSCRIPTIONS[topic_arn]\n    else:\n        return None\n\n\ndef get_subscription_by_arn(sub_arn):\n    # TODO maintain separate map instead of traversing all items\n    for key, subscriptions in SNS_SUBSCRIPTIONS.items():\n        for sub in subscriptions:\n            if sub['SubscriptionArn'] == sub_arn:\n                return sub\n\n\ndef make_response(op_name, content=''):\n    response = Response()\n    if not content:\n        content = '<MessageId>%s</MessageId>' % short_uid()\n    response._content = \"\"\"<{op_name}Response xmlns=\"http://sns.amazonaws.com/doc/2010-03-31/\">\n        <{op_name}Result>\n            {content}\n        </{op_name}Result>\n        <ResponseMetadata><RequestId>{req_id}</RequestId></ResponseMetadata>\n        </{op_name}Response>\"\"\".format(op_name=op_name, content=content, req_id=short_uid())\n    response.status_code = 200\n    return response\n\n\ndef make_error(message, code=400, code_string='InvalidParameter'):\n    response = Response()\n    response._content = \"\"\"<ErrorResponse xmlns=\"http://sns.amazonaws.com/doc/2010-03-31/\"><Error>\n        <Type>Sender</Type>\n        <Code>{code_string}</Code>\n        <Message>{message}</Message>\n        </Error><RequestId>{req_id}</RequestId>\n        </ErrorResponse>\"\"\".format(message=message, code_string=code_string, req_id=short_uid())\n    response.status_code = code\n    return response\n\n\ndef create_sns_message_body(subscriber, req_data):\n    message = req_data['Message'][0]\n    subject = req_data.get('Subject', [None])[0]\n    protocol = subscriber['Protocol']\n\n    if subscriber['RawMessageDelivery'] == 'true':\n        return message\n\n    if req_data.get('MessageStructure') == ['json']:\n        message = json.loads(message)\n        try:\n            message = message.get(protocol, message['default'])\n        except KeyError:\n            raise Exception(\"Unable to find 'default' key in message payload\")\n\n    data = {}\n    data['MessageId'] = str(uuid.uuid4())\n    data['Type'] = 'Notification'\n    data['Message'] = message\n    data['TopicArn'] = subscriber['TopicArn']\n    if subject is not None:\n        data['Subject'] = subject\n    attributes = get_message_attributes(req_data)\n    if attributes:\n        data['MessageAttributes'] = attributes\n    return json.dumps(data)\n\n\ndef get_message_attributes(req_data):\n    attributes = {}\n    x = 1\n    while True:\n        name = req_data.get('MessageAttributes.entry.' + str(x) + '.Name', [None])[0]\n        if name is not None:\n            attribute = {}\n            attribute['Type'] = req_data.get('MessageAttributes.entry.' + str(x) + '.Value.DataType', [None])[0]\n            string_value = req_data.get('MessageAttributes.entry.' + str(x) + '.Value.StringValue', [None])[0]\n            binary_value = req_data.get('MessageAttributes.entry.' + str(x) + '.Value.BinaryValue', [None])[0]\n            if string_value is not None:\n                attribute['Value'] = string_value\n            elif binary_value is not None:\n                attribute['Value'] = binary_value\n\n            attributes[name] = attribute\n            x += 1\n        else:\n            break\n\n    return attributes\n\n\ndef is_number(x):\n    try:\n        float(x)\n        return True\n    except ValueError:\n        return False\n\n\ndef evaluate_numeric_condition(conditions, value):\n    if not is_number(value):\n        return False\n\n    for i in range(0, len(conditions), 2):\n        operator = conditions[i]\n        operand = conditions[i + 1]\n\n        if operator == '=':\n            if value != operand:\n                return False\n        elif operator == '>':\n            if value <= operand:\n                return False\n        elif operator == '<':\n            if value >= operand:\n                return False\n        elif operator == '>=':\n            if value < operand:\n                return False\n        elif operator == '<=':\n            if value > operand:\n                return False\n\n    return True\n\n\ndef evaluate_condition(value, condition):\n    if type(condition) is not dict:\n        return value == condition\n    elif condition.get('anything-but'):\n        return value not in condition.get('anything-but')\n    elif condition.get('prefix'):\n        prefix = condition.get('prefix')\n        return value.startswith(prefix)\n    elif condition.get('numeric'):\n        return evaluate_numeric_condition(condition.get('numeric'), value)\n\n    return False\n\n\ndef evaluate_filter_policy_conditions(conditions, attribute):\n    if type(conditions) is not list:\n        conditions = [conditions]\n\n    if attribute['Type'] == 'String.Array':\n        values = ast.literal_eval(attribute['Value'])\n        for value in values:\n            for condition in conditions:\n                if evaluate_condition(value, condition):\n                    return True\n    else:\n        for condition in conditions:\n            if evaluate_condition(attribute['Value'], condition):\n                return True\n\n    return False\n\n\ndef check_filter_policy(filter_policy, message_attributes):\n    if not filter_policy:\n        return True\n\n    for criteria in filter_policy:\n        conditions = filter_policy.get(criteria)\n        attribute = message_attributes.get(criteria)\n\n        if attribute is None:\n            return False\n\n        if evaluate_filter_policy_conditions(conditions, attribute) is False:\n            return False\n\n    return True\n", "path": "localstack/services/sns/sns_listener.py"}]}