We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Govcloud SNS prefix incorrect
Hello,

It seems that cloud custodian is putting the SNS arn prefix incorrectly for govcloud. This is the error I'm getting:

botocore.errorfactory.InvalidParameterException: An error occurred (InvalidParameter) when calling the Publish operation: Invalid parameter: TopicArn Reason: A us-gov-west-1 ARN must begin with arn:aws-us-gov, not arn:aws:sns:us-gov-west-1:xxxxxxxxxxx:snsname


policies:
  - name: custodian-tags
    resource: ec2
    filters:
      - or:
        - "tag:CustodianTest": present
        - "tag:CustodianTest2": present
    actions:
      - type: notify
        to:
          - my@email.com
        template: default
        transport:
          type: sns
          region: us-gov-west-1
          topic: snsname


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `c7n/actions/notify.py`
Content:
```
1 # Copyright 2017-2018 Capital One Services, LLC
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 
16 import base64
17 import copy
18 import zlib
19 
20 from .core import EventAction
21 from c7n import utils
22 from c7n.exceptions import PolicyValidationError
23 from c7n.resolver import ValuesFrom
24 
25 
26 class BaseNotify(EventAction):
27 
28     batch_size = 250
29 
30     def expand_variables(self, message):
31         """expand any variables in the action to_from/cc_from fields.
32         """
33         p = copy.deepcopy(self.data)
34         if 'to_from' in self.data:
35             to_from = self.data['to_from'].copy()
36             to_from['url'] = to_from['url'].format(**message)
37             if 'expr' in to_from:
38                 to_from['expr'] = to_from['expr'].format(**message)
39             p.setdefault('to', []).extend(ValuesFrom(to_from, self.manager).get_values())
40         if 'cc_from' in self.data:
41             cc_from = self.data['cc_from'].copy()
42             cc_from['url'] = cc_from['url'].format(**message)
43             if 'expr' in cc_from:
44                 cc_from['expr'] = cc_from['expr'].format(**message)
45             p.setdefault('cc', []).extend(ValuesFrom(cc_from, self.manager).get_values())
46         return p
47 
48     def pack(self, message):
49         dumped = utils.dumps(message)
50         compressed = zlib.compress(dumped.encode('utf8'))
51         b64encoded = base64.b64encode(compressed)
52         return b64encoded.decode('ascii')
53 
54 
55 class Notify(BaseNotify):
56     """
57     Flexible notifications require quite a bit of implementation support
58     on pluggable transports, templates, address resolution, variable
59     extraction, batch periods, etc.
60 
61     For expedience and flexibility then, we instead send the data to
62     an sqs queue, for processing. ie. actual communications can be enabled
63     with the c7n-mailer tool, found under tools/c7n_mailer.
64 
65     Attaching additional string message attributes are supported on the SNS
66     transport, with the exception of the ``mtype`` attribute, which is a
67     reserved attribute used by Cloud Custodian.
68 
69     Example::
70 
71       policies:
72         - name: ec2-bad-instance-kill
73           resource: ec2
74           filters:
75            - Name: bad-instance
76           actions:
77            - terminate
78            - type: notify
79              to:
80               - event-user
81               - resource-creator
82               - email@address
83              owner_absent_contact:
84               - other_email@address
85              # which template for the email should we use
86              template: policy-template
87              transport:
88                type: sqs
89                region: us-east-1
90                queue: xyz
91 
92         - name: ec2-notify-with-attributes
93           resource: ec2
94           filters:
95            - Name: bad-instance
96           actions:
97            - type: notify
98              to:
99               - event-user
100               - resource-creator
101               - email@address
102              owner_absent_contact:
103               - other_email@address
104              # which template for the email should we use
105              template: policy-template
106              transport:
107                type: sns
108                region: us-east-1
109                topic: your-notify-topic
110                attributes:
111                  - attribute_key: attribute_value
112                  - attribute_key_2: attribute_value_2
113     """
114 
115     C7N_DATA_MESSAGE = "maidmsg/1.0"
116 
117     schema_alias = True
118     schema = {
119         'type': 'object',
120         'anyOf': [
121             {'required': ['type', 'transport', 'to']},
122             {'required': ['type', 'transport', 'to_from']}],
123         'properties': {
124             'type': {'enum': ['notify']},
125             'to': {'type': 'array', 'items': {'type': 'string'}},
126             'owner_absent_contact': {'type': 'array', 'items': {'type': 'string'}},
127             'to_from': ValuesFrom.schema,
128             'cc': {'type': 'array', 'items': {'type': 'string'}},
129             'cc_from': ValuesFrom.schema,
130             'cc_manager': {'type': 'boolean'},
131             'from': {'type': 'string'},
132             'subject': {'type': 'string'},
133             'template': {'type': 'string'},
134             'transport': {
135                 'oneOf': [
136                     {'type': 'object',
137                      'required': ['type', 'queue'],
138                      'properties': {
139                          'queue': {'type': 'string'},
140                          'type': {'enum': ['sqs']}}},
141                     {'type': 'object',
142                      'required': ['type', 'topic'],
143                      'properties': {
144                          'topic': {'type': 'string'},
145                          'type': {'enum': ['sns']},
146                          'attributes': {'type': 'object'},
147                      }}]
148             },
149             'assume_role': {'type': 'boolean'}
150         }
151     }
152 
153     def __init__(self, data=None, manager=None, log_dir=None):
154         super(Notify, self).__init__(data, manager, log_dir)
155         self.assume_role = data.get('assume_role', True)
156 
157     def validate(self):
158         if self.data.get('transport', {}).get('type') == 'sns' and \
159                 self.data.get('transport').get('attributes') and \
160                 'mtype' in self.data.get('transport').get('attributes').keys():
161                     raise PolicyValidationError(
162                         "attribute: mtype is a reserved attribute for sns transport")
163         return self
164 
165     def get_permissions(self):
166         if self.data.get('transport', {}).get('type') == 'sns':
167             return ('sns:Publish',)
168         if self.data.get('transport', {'type': 'sqs'}).get('type') == 'sqs':
169             return ('sqs:SendMessage',)
170         return ()
171 
172     def process(self, resources, event=None):
173         alias = utils.get_account_alias_from_sts(
174             utils.local_session(self.manager.session_factory))
175         message = {
176             'event': event,
177             'account_id': self.manager.config.account_id,
178             'account': alias,
179             'region': self.manager.config.region,
180             'policy': self.manager.data}
181         message['action'] = self.expand_variables(message)
182 
183         for batch in utils.chunks(resources, self.batch_size):
184             message['resources'] = self.prepare_resources(batch)
185             receipt = self.send_data_message(message)
186             self.log.info("sent message:%s policy:%s template:%s count:%s" % (
187                 receipt, self.manager.data['name'],
188                 self.data.get('template', 'default'), len(batch)))
189 
190     def prepare_resources(self, resources):
191         """Resources preparation for transport.
192 
193         If we have sensitive or overly large resource metadata we want to
194         remove or additional serialization we need to perform, this
195         provides a mechanism.
196 
197         TODO: consider alternative implementations, at min look at adding
198         provider as additional discriminator to resource type. One alternative
199         would be dynamically adjusting buffer size based on underlying
200         transport.
201         """
202         handler = getattr(self, "prepare_%s" % (
203             self.manager.type.replace('-', '_')),
204             None)
205         if handler is None:
206             return resources
207         return handler(resources)
208 
209     def prepare_launch_config(self, resources):
210         for r in resources:
211             r.pop('UserData', None)
212         return resources
213 
214     def prepare_asg(self, resources):
215         for r in resources:
216             if 'c7n:user-data' in r:
217                 r.pop('c7n:user-data', None)
218         return resources
219 
220     def prepare_ec2(self, resources):
221         for r in resources:
222             if 'c7n:user-data' in r:
223                 r.pop('c7n:user-data')
224         return resources
225 
226     def send_data_message(self, message):
227         if self.data['transport']['type'] == 'sqs':
228             return self.send_sqs(message)
229         elif self.data['transport']['type'] == 'sns':
230             return self.send_sns(message)
231 
232     def send_sns(self, message):
233         topic = self.data['transport']['topic'].format(**message)
234         user_attributes = self.data['transport'].get('attributes')
235         if topic.startswith('arn:aws:sns'):
236             region = region = topic.split(':', 5)[3]
237             topic_arn = topic
238         else:
239             region = message['region']
240             topic_arn = "arn:aws:sns:%s:%s:%s" % (
241                 message['region'], message['account_id'], topic)
242         client = self.manager.session_factory(
243             region=region, assume=self.assume_role).client('sns')
244         attrs = {
245             'mtype': {
246                 'DataType': 'String',
247                 'StringValue': self.C7N_DATA_MESSAGE,
248             },
249         }
250         if user_attributes:
251             for k, v in user_attributes.items():
252                 if k != 'mtype':
253                     attrs[k] = {'DataType': 'String', 'StringValue': v}
254         client.publish(
255             TopicArn=topic_arn,
256             Message=self.pack(message),
257             MessageAttributes=attrs
258         )
259 
260     def send_sqs(self, message):
261         queue = self.data['transport']['queue'].format(**message)
262         if queue.startswith('https://queue.amazonaws.com'):
263             region = 'us-east-1'
264             queue_url = queue
265         elif 'queue.amazonaws.com' in queue:
266             region = queue[len('https://'):].split('.', 1)[0]
267             queue_url = queue
268         elif queue.startswith('https://sqs.'):
269             region = queue.split('.', 2)[1]
270             queue_url = queue
271         elif queue.startswith('arn:aws:sqs'):
272             queue_arn_split = queue.split(':', 5)
273             region = queue_arn_split[3]
274             owner_id = queue_arn_split[4]
275             queue_name = queue_arn_split[5]
276             queue_url = "https://sqs.%s.amazonaws.com/%s/%s" % (
277                 region, owner_id, queue_name)
278         else:
279             region = self.manager.config.region
280             owner_id = self.manager.config.account_id
281             queue_name = queue
282             queue_url = "https://sqs.%s.amazonaws.com/%s/%s" % (
283                 region, owner_id, queue_name)
284         client = self.manager.session_factory(
285             region=region, assume=self.assume_role).client('sqs')
286         attrs = {
287             'mtype': {
288                 'DataType': 'String',
289                 'StringValue': self.C7N_DATA_MESSAGE,
290             },
291         }
292         result = client.send_message(
293             QueueUrl=queue_url,
294             MessageBody=self.pack(message),
295             MessageAttributes=attrs)
296         return result['MessageId']
297
```
--- 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/c7n/actions/notify.py b/c7n/actions/notify.py
--- a/c7n/actions/notify.py
+++ b/c7n/actions/notify.py
@@ -232,7 +232,7 @@
     def send_sns(self, message):
         topic = self.data['transport']['topic'].format(**message)
         user_attributes = self.data['transport'].get('attributes')
-        if topic.startswith('arn:aws:sns'):
+        if topic.startswith('arn:'):
             region = region = topic.split(':', 5)[3]
             topic_arn = topic
         else:

{"golden_diff": "diff --git a/c7n/actions/notify.py b/c7n/actions/notify.py\n--- a/c7n/actions/notify.py\n+++ b/c7n/actions/notify.py\n@@ -232,7 +232,7 @@\n     def send_sns(self, message):\n         topic = self.data['transport']['topic'].format(**message)\n         user_attributes = self.data['transport'].get('attributes')\n-        if topic.startswith('arn:aws:sns'):\n+        if topic.startswith('arn:'):\n             region = region = topic.split(':', 5)[3]\n             topic_arn = topic\n         else:\n", "issue": "Govcloud SNS prefix incorrect\nHello,\r\n\r\nIt seems that cloud custodian is putting the SNS arn prefix incorrectly for govcloud. This is the error I'm getting:\r\n\r\nbotocore.errorfactory.InvalidParameterException: An error occurred (InvalidParameter) when calling the Publish operation: Invalid parameter: TopicArn Reason: A us-gov-west-1 ARN must begin with arn:aws-us-gov, not arn:aws:sns:us-gov-west-1:xxxxxxxxxxx:snsname\r\n\r\n\r\npolicies:\r\n  - name: custodian-tags\r\n    resource: ec2\r\n    filters:\r\n      - or:\r\n        - \"tag:CustodianTest\": present\r\n        - \"tag:CustodianTest2\": present\r\n    actions:\r\n      - type: notify\r\n        to:\r\n          - my@email.com\r\n        template: default\r\n        transport:\r\n          type: sns\r\n          region: us-gov-west-1\r\n          topic: snsname\r\n\n", "before_files": [{"content": "# Copyright 2017-2018 Capital One Services, LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\nimport base64\nimport copy\nimport zlib\n\nfrom .core import EventAction\nfrom c7n import utils\nfrom c7n.exceptions import PolicyValidationError\nfrom c7n.resolver import ValuesFrom\n\n\nclass BaseNotify(EventAction):\n\n    batch_size = 250\n\n    def expand_variables(self, message):\n        \"\"\"expand any variables in the action to_from/cc_from fields.\n        \"\"\"\n        p = copy.deepcopy(self.data)\n        if 'to_from' in self.data:\n            to_from = self.data['to_from'].copy()\n            to_from['url'] = to_from['url'].format(**message)\n            if 'expr' in to_from:\n                to_from['expr'] = to_from['expr'].format(**message)\n            p.setdefault('to', []).extend(ValuesFrom(to_from, self.manager).get_values())\n        if 'cc_from' in self.data:\n            cc_from = self.data['cc_from'].copy()\n            cc_from['url'] = cc_from['url'].format(**message)\n            if 'expr' in cc_from:\n                cc_from['expr'] = cc_from['expr'].format(**message)\n            p.setdefault('cc', []).extend(ValuesFrom(cc_from, self.manager).get_values())\n        return p\n\n    def pack(self, message):\n        dumped = utils.dumps(message)\n        compressed = zlib.compress(dumped.encode('utf8'))\n        b64encoded = base64.b64encode(compressed)\n        return b64encoded.decode('ascii')\n\n\nclass Notify(BaseNotify):\n    \"\"\"\n    Flexible notifications require quite a bit of implementation support\n    on pluggable transports, templates, address resolution, variable\n    extraction, batch periods, etc.\n\n    For expedience and flexibility then, we instead send the data to\n    an sqs queue, for processing. ie. actual communications can be enabled\n    with the c7n-mailer tool, found under tools/c7n_mailer.\n\n    Attaching additional string message attributes are supported on the SNS\n    transport, with the exception of the ``mtype`` attribute, which is a\n    reserved attribute used by Cloud Custodian.\n\n    Example::\n\n      policies:\n        - name: ec2-bad-instance-kill\n          resource: ec2\n          filters:\n           - Name: bad-instance\n          actions:\n           - terminate\n           - type: notify\n             to:\n              - event-user\n              - resource-creator\n              - email@address\n             owner_absent_contact:\n              - other_email@address\n             # which template for the email should we use\n             template: policy-template\n             transport:\n               type: sqs\n               region: us-east-1\n               queue: xyz\n\n        - name: ec2-notify-with-attributes\n          resource: ec2\n          filters:\n           - Name: bad-instance\n          actions:\n           - type: notify\n             to:\n              - event-user\n              - resource-creator\n              - email@address\n             owner_absent_contact:\n              - other_email@address\n             # which template for the email should we use\n             template: policy-template\n             transport:\n               type: sns\n               region: us-east-1\n               topic: your-notify-topic\n               attributes:\n                 - attribute_key: attribute_value\n                 - attribute_key_2: attribute_value_2\n    \"\"\"\n\n    C7N_DATA_MESSAGE = \"maidmsg/1.0\"\n\n    schema_alias = True\n    schema = {\n        'type': 'object',\n        'anyOf': [\n            {'required': ['type', 'transport', 'to']},\n            {'required': ['type', 'transport', 'to_from']}],\n        'properties': {\n            'type': {'enum': ['notify']},\n            'to': {'type': 'array', 'items': {'type': 'string'}},\n            'owner_absent_contact': {'type': 'array', 'items': {'type': 'string'}},\n            'to_from': ValuesFrom.schema,\n            'cc': {'type': 'array', 'items': {'type': 'string'}},\n            'cc_from': ValuesFrom.schema,\n            'cc_manager': {'type': 'boolean'},\n            'from': {'type': 'string'},\n            'subject': {'type': 'string'},\n            'template': {'type': 'string'},\n            'transport': {\n                'oneOf': [\n                    {'type': 'object',\n                     'required': ['type', 'queue'],\n                     'properties': {\n                         'queue': {'type': 'string'},\n                         'type': {'enum': ['sqs']}}},\n                    {'type': 'object',\n                     'required': ['type', 'topic'],\n                     'properties': {\n                         'topic': {'type': 'string'},\n                         'type': {'enum': ['sns']},\n                         'attributes': {'type': 'object'},\n                     }}]\n            },\n            'assume_role': {'type': 'boolean'}\n        }\n    }\n\n    def __init__(self, data=None, manager=None, log_dir=None):\n        super(Notify, self).__init__(data, manager, log_dir)\n        self.assume_role = data.get('assume_role', True)\n\n    def validate(self):\n        if self.data.get('transport', {}).get('type') == 'sns' and \\\n                self.data.get('transport').get('attributes') and \\\n                'mtype' in self.data.get('transport').get('attributes').keys():\n                    raise PolicyValidationError(\n                        \"attribute: mtype is a reserved attribute for sns transport\")\n        return self\n\n    def get_permissions(self):\n        if self.data.get('transport', {}).get('type') == 'sns':\n            return ('sns:Publish',)\n        if self.data.get('transport', {'type': 'sqs'}).get('type') == 'sqs':\n            return ('sqs:SendMessage',)\n        return ()\n\n    def process(self, resources, event=None):\n        alias = utils.get_account_alias_from_sts(\n            utils.local_session(self.manager.session_factory))\n        message = {\n            'event': event,\n            'account_id': self.manager.config.account_id,\n            'account': alias,\n            'region': self.manager.config.region,\n            'policy': self.manager.data}\n        message['action'] = self.expand_variables(message)\n\n        for batch in utils.chunks(resources, self.batch_size):\n            message['resources'] = self.prepare_resources(batch)\n            receipt = self.send_data_message(message)\n            self.log.info(\"sent message:%s policy:%s template:%s count:%s\" % (\n                receipt, self.manager.data['name'],\n                self.data.get('template', 'default'), len(batch)))\n\n    def prepare_resources(self, resources):\n        \"\"\"Resources preparation for transport.\n\n        If we have sensitive or overly large resource metadata we want to\n        remove or additional serialization we need to perform, this\n        provides a mechanism.\n\n        TODO: consider alternative implementations, at min look at adding\n        provider as additional discriminator to resource type. One alternative\n        would be dynamically adjusting buffer size based on underlying\n        transport.\n        \"\"\"\n        handler = getattr(self, \"prepare_%s\" % (\n            self.manager.type.replace('-', '_')),\n            None)\n        if handler is None:\n            return resources\n        return handler(resources)\n\n    def prepare_launch_config(self, resources):\n        for r in resources:\n            r.pop('UserData', None)\n        return resources\n\n    def prepare_asg(self, resources):\n        for r in resources:\n            if 'c7n:user-data' in r:\n                r.pop('c7n:user-data', None)\n        return resources\n\n    def prepare_ec2(self, resources):\n        for r in resources:\n            if 'c7n:user-data' in r:\n                r.pop('c7n:user-data')\n        return resources\n\n    def send_data_message(self, message):\n        if self.data['transport']['type'] == 'sqs':\n            return self.send_sqs(message)\n        elif self.data['transport']['type'] == 'sns':\n            return self.send_sns(message)\n\n    def send_sns(self, message):\n        topic = self.data['transport']['topic'].format(**message)\n        user_attributes = self.data['transport'].get('attributes')\n        if topic.startswith('arn:aws:sns'):\n            region = region = topic.split(':', 5)[3]\n            topic_arn = topic\n        else:\n            region = message['region']\n            topic_arn = \"arn:aws:sns:%s:%s:%s\" % (\n                message['region'], message['account_id'], topic)\n        client = self.manager.session_factory(\n            region=region, assume=self.assume_role).client('sns')\n        attrs = {\n            'mtype': {\n                'DataType': 'String',\n                'StringValue': self.C7N_DATA_MESSAGE,\n            },\n        }\n        if user_attributes:\n            for k, v in user_attributes.items():\n                if k != 'mtype':\n                    attrs[k] = {'DataType': 'String', 'StringValue': v}\n        client.publish(\n            TopicArn=topic_arn,\n            Message=self.pack(message),\n            MessageAttributes=attrs\n        )\n\n    def send_sqs(self, message):\n        queue = self.data['transport']['queue'].format(**message)\n        if queue.startswith('https://queue.amazonaws.com'):\n            region = 'us-east-1'\n            queue_url = queue\n        elif 'queue.amazonaws.com' in queue:\n            region = queue[len('https://'):].split('.', 1)[0]\n            queue_url = queue\n        elif queue.startswith('https://sqs.'):\n            region = queue.split('.', 2)[1]\n            queue_url = queue\n        elif queue.startswith('arn:aws:sqs'):\n            queue_arn_split = queue.split(':', 5)\n            region = queue_arn_split[3]\n            owner_id = queue_arn_split[4]\n            queue_name = queue_arn_split[5]\n            queue_url = \"https://sqs.%s.amazonaws.com/%s/%s\" % (\n                region, owner_id, queue_name)\n        else:\n            region = self.manager.config.region\n            owner_id = self.manager.config.account_id\n            queue_name = queue\n            queue_url = \"https://sqs.%s.amazonaws.com/%s/%s\" % (\n                region, owner_id, queue_name)\n        client = self.manager.session_factory(\n            region=region, assume=self.assume_role).client('sqs')\n        attrs = {\n            'mtype': {\n                'DataType': 'String',\n                'StringValue': self.C7N_DATA_MESSAGE,\n            },\n        }\n        result = client.send_message(\n            QueueUrl=queue_url,\n            MessageBody=self.pack(message),\n            MessageAttributes=attrs)\n        return result['MessageId']\n", "path": "c7n/actions/notify.py"}], "after_files": [{"content": "# Copyright 2017-2018 Capital One Services, LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\nimport base64\nimport copy\nimport zlib\n\nfrom .core import EventAction\nfrom c7n import utils\nfrom c7n.exceptions import PolicyValidationError\nfrom c7n.resolver import ValuesFrom\n\n\nclass BaseNotify(EventAction):\n\n    batch_size = 250\n\n    def expand_variables(self, message):\n        \"\"\"expand any variables in the action to_from/cc_from fields.\n        \"\"\"\n        p = copy.deepcopy(self.data)\n        if 'to_from' in self.data:\n            to_from = self.data['to_from'].copy()\n            to_from['url'] = to_from['url'].format(**message)\n            if 'expr' in to_from:\n                to_from['expr'] = to_from['expr'].format(**message)\n            p.setdefault('to', []).extend(ValuesFrom(to_from, self.manager).get_values())\n        if 'cc_from' in self.data:\n            cc_from = self.data['cc_from'].copy()\n            cc_from['url'] = cc_from['url'].format(**message)\n            if 'expr' in cc_from:\n                cc_from['expr'] = cc_from['expr'].format(**message)\n            p.setdefault('cc', []).extend(ValuesFrom(cc_from, self.manager).get_values())\n        return p\n\n    def pack(self, message):\n        dumped = utils.dumps(message)\n        compressed = zlib.compress(dumped.encode('utf8'))\n        b64encoded = base64.b64encode(compressed)\n        return b64encoded.decode('ascii')\n\n\nclass Notify(BaseNotify):\n    \"\"\"\n    Flexible notifications require quite a bit of implementation support\n    on pluggable transports, templates, address resolution, variable\n    extraction, batch periods, etc.\n\n    For expedience and flexibility then, we instead send the data to\n    an sqs queue, for processing. ie. actual communications can be enabled\n    with the c7n-mailer tool, found under tools/c7n_mailer.\n\n    Attaching additional string message attributes are supported on the SNS\n    transport, with the exception of the ``mtype`` attribute, which is a\n    reserved attribute used by Cloud Custodian.\n\n    Example::\n\n      policies:\n        - name: ec2-bad-instance-kill\n          resource: ec2\n          filters:\n           - Name: bad-instance\n          actions:\n           - terminate\n           - type: notify\n             to:\n              - event-user\n              - resource-creator\n              - email@address\n             owner_absent_contact:\n              - other_email@address\n             # which template for the email should we use\n             template: policy-template\n             transport:\n               type: sqs\n               region: us-east-1\n               queue: xyz\n\n        - name: ec2-notify-with-attributes\n          resource: ec2\n          filters:\n           - Name: bad-instance\n          actions:\n           - type: notify\n             to:\n              - event-user\n              - resource-creator\n              - email@address\n             owner_absent_contact:\n              - other_email@address\n             # which template for the email should we use\n             template: policy-template\n             transport:\n               type: sns\n               region: us-east-1\n               topic: your-notify-topic\n               attributes:\n                 - attribute_key: attribute_value\n                 - attribute_key_2: attribute_value_2\n    \"\"\"\n\n    C7N_DATA_MESSAGE = \"maidmsg/1.0\"\n\n    schema_alias = True\n    schema = {\n        'type': 'object',\n        'anyOf': [\n            {'required': ['type', 'transport', 'to']},\n            {'required': ['type', 'transport', 'to_from']}],\n        'properties': {\n            'type': {'enum': ['notify']},\n            'to': {'type': 'array', 'items': {'type': 'string'}},\n            'owner_absent_contact': {'type': 'array', 'items': {'type': 'string'}},\n            'to_from': ValuesFrom.schema,\n            'cc': {'type': 'array', 'items': {'type': 'string'}},\n            'cc_from': ValuesFrom.schema,\n            'cc_manager': {'type': 'boolean'},\n            'from': {'type': 'string'},\n            'subject': {'type': 'string'},\n            'template': {'type': 'string'},\n            'transport': {\n                'oneOf': [\n                    {'type': 'object',\n                     'required': ['type', 'queue'],\n                     'properties': {\n                         'queue': {'type': 'string'},\n                         'type': {'enum': ['sqs']}}},\n                    {'type': 'object',\n                     'required': ['type', 'topic'],\n                     'properties': {\n                         'topic': {'type': 'string'},\n                         'type': {'enum': ['sns']},\n                         'attributes': {'type': 'object'},\n                     }}]\n            },\n            'assume_role': {'type': 'boolean'}\n        }\n    }\n\n    def __init__(self, data=None, manager=None, log_dir=None):\n        super(Notify, self).__init__(data, manager, log_dir)\n        self.assume_role = data.get('assume_role', True)\n\n    def validate(self):\n        if self.data.get('transport', {}).get('type') == 'sns' and \\\n                self.data.get('transport').get('attributes') and \\\n                'mtype' in self.data.get('transport').get('attributes').keys():\n                    raise PolicyValidationError(\n                        \"attribute: mtype is a reserved attribute for sns transport\")\n        return self\n\n    def get_permissions(self):\n        if self.data.get('transport', {}).get('type') == 'sns':\n            return ('sns:Publish',)\n        if self.data.get('transport', {'type': 'sqs'}).get('type') == 'sqs':\n            return ('sqs:SendMessage',)\n        return ()\n\n    def process(self, resources, event=None):\n        alias = utils.get_account_alias_from_sts(\n            utils.local_session(self.manager.session_factory))\n        message = {\n            'event': event,\n            'account_id': self.manager.config.account_id,\n            'account': alias,\n            'region': self.manager.config.region,\n            'policy': self.manager.data}\n        message['action'] = self.expand_variables(message)\n\n        for batch in utils.chunks(resources, self.batch_size):\n            message['resources'] = self.prepare_resources(batch)\n            receipt = self.send_data_message(message)\n            self.log.info(\"sent message:%s policy:%s template:%s count:%s\" % (\n                receipt, self.manager.data['name'],\n                self.data.get('template', 'default'), len(batch)))\n\n    def prepare_resources(self, resources):\n        \"\"\"Resources preparation for transport.\n\n        If we have sensitive or overly large resource metadata we want to\n        remove or additional serialization we need to perform, this\n        provides a mechanism.\n\n        TODO: consider alternative implementations, at min look at adding\n        provider as additional discriminator to resource type. One alternative\n        would be dynamically adjusting buffer size based on underlying\n        transport.\n        \"\"\"\n        handler = getattr(self, \"prepare_%s\" % (\n            self.manager.type.replace('-', '_')),\n            None)\n        if handler is None:\n            return resources\n        return handler(resources)\n\n    def prepare_launch_config(self, resources):\n        for r in resources:\n            r.pop('UserData', None)\n        return resources\n\n    def prepare_asg(self, resources):\n        for r in resources:\n            if 'c7n:user-data' in r:\n                r.pop('c7n:user-data', None)\n        return resources\n\n    def prepare_ec2(self, resources):\n        for r in resources:\n            if 'c7n:user-data' in r:\n                r.pop('c7n:user-data')\n        return resources\n\n    def send_data_message(self, message):\n        if self.data['transport']['type'] == 'sqs':\n            return self.send_sqs(message)\n        elif self.data['transport']['type'] == 'sns':\n            return self.send_sns(message)\n\n    def send_sns(self, message):\n        topic = self.data['transport']['topic'].format(**message)\n        user_attributes = self.data['transport'].get('attributes')\n        if topic.startswith('arn:'):\n            region = region = topic.split(':', 5)[3]\n            topic_arn = topic\n        else:\n            region = message['region']\n            topic_arn = \"arn:aws:sns:%s:%s:%s\" % (\n                message['region'], message['account_id'], topic)\n        client = self.manager.session_factory(\n            region=region, assume=self.assume_role).client('sns')\n        attrs = {\n            'mtype': {\n                'DataType': 'String',\n                'StringValue': self.C7N_DATA_MESSAGE,\n            },\n        }\n        if user_attributes:\n            for k, v in user_attributes.items():\n                if k != 'mtype':\n                    attrs[k] = {'DataType': 'String', 'StringValue': v}\n        client.publish(\n            TopicArn=topic_arn,\n            Message=self.pack(message),\n            MessageAttributes=attrs\n        )\n\n    def send_sqs(self, message):\n        queue = self.data['transport']['queue'].format(**message)\n        if queue.startswith('https://queue.amazonaws.com'):\n            region = 'us-east-1'\n            queue_url = queue\n        elif 'queue.amazonaws.com' in queue:\n            region = queue[len('https://'):].split('.', 1)[0]\n            queue_url = queue\n        elif queue.startswith('https://sqs.'):\n            region = queue.split('.', 2)[1]\n            queue_url = queue\n        elif queue.startswith('arn:aws:sqs'):\n            queue_arn_split = queue.split(':', 5)\n            region = queue_arn_split[3]\n            owner_id = queue_arn_split[4]\n            queue_name = queue_arn_split[5]\n            queue_url = \"https://sqs.%s.amazonaws.com/%s/%s\" % (\n                region, owner_id, queue_name)\n        else:\n            region = self.manager.config.region\n            owner_id = self.manager.config.account_id\n            queue_name = queue\n            queue_url = \"https://sqs.%s.amazonaws.com/%s/%s\" % (\n                region, owner_id, queue_name)\n        client = self.manager.session_factory(\n            region=region, assume=self.assume_role).client('sqs')\n        attrs = {\n            'mtype': {\n                'DataType': 'String',\n                'StringValue': self.C7N_DATA_MESSAGE,\n            },\n        }\n        result = client.send_message(\n            QueueUrl=queue_url,\n            MessageBody=self.pack(message),\n            MessageAttributes=attrs)\n        return result['MessageId']\n", "path": "c7n/actions/notify.py"}]}