We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
aws.secrets-manager:tag action tries to retag secret with existing tags as well
### Describe the bug

I have a policy that checks all AWS Secret Manager secrets for the existence of two particular tags. If the secret is missing either of these tags, it tries to makr (i.e. tag) the secret for later action. However, Custodian fails to apply its marking tag with the following error:

2023-10-18 14:25:20,524 - custodian.policy - INFO - policy:secrets-manager-secret-mark-untagged resource:secrets-manager region:us-west-2 count:2 time:0.03
2023-10-18 14:25:20,525 - custodian.actions - INFO - Tagging 2 resources for notify on 2023/10/19
2023-10-18 14:25:21,519 - custodian.actions - ERROR - Exception with tags: [{'Key': 'c7n-secrets-manager-secret-untagged', 'Value': 'Resource does not meet policy: notify@2023/10/19'}]  An error occurred (InvalidParameterException) when calling the TagResource operation: You can't modify system tags. Tag key: 'aws:cloudformation:stack-name'.

It appears Custodian is trying to retag with all existing tags in addition to the new marking tag. AWS doesn't like you messing with the system tags?

### What did you expect to happen?

I expected Custodian to only add the marking tag rather than try to apply all the existing tags as well.

Is this possible? Is it a bug in Custodian or in the AWS Secrets Manager API?

### Cloud Provider

Amazon Web Services (AWS)

### Cloud Custodian version and dependency information

```shell
Custodian:   0.9.32
Python:      3.8.16 (default, Mar 10 2023, 03:07:48) 
             [GCC 7.3.1 20180712 (Red Hat 7.3.1-15)]
Platform:    posix.uname_result(sysname='Linux', nodename='custodian', release='4.14.320-243.544.amzn2.x86_64', version='#1 SMP Tue Aug 1 21:03:08 UTC 2023', machine='x86_64')
Using venv:  False
Docker: False
Installed: 

argcomplete==3.1.2
attrs==23.1.0
boto3==1.28.58
botocore==1.31.58
docutils==0.18.1
importlib-metadata==5.2.0
importlib-resources==6.1.0
jmespath==1.0.1
jsonschema==4.19.1
jsonschema-specifications==2023.7.1
pkgutil-resolve-name==1.3.10
python-dateutil==2.8.2
pyyaml==6.0.1
referencing==0.30.2
rpds-py==0.10.3
s3transfer==0.7.0
six==1.16.0
tabulate==0.9.0
urllib3==1.26.17
zipp==3.17.0
```


### Policy

```shell
Here is the action bit:

actions:
      - type: mark-for-op
        tag: c7n-secrets-manager-secret-untagged
        op: notify
        days: 1
```


### Relevant log/traceback output

```shell
2023-10-18 14:25:20,524 - custodian.policy - INFO - policy:secrets-manager-secret-mark-untagged resource:secrets-manager region:us-west-2 count:2 time:0.03
2023-10-18 14:25:20,525 - custodian.actions - INFO - Tagging 2 resources for notify on 2023/10/19
2023-10-18 14:25:21,519 - custodian.actions - ERROR - Exception with tags: [{'Key': 'c7n-secrets-manager-secret-untagged', 'Value': 'Resource does not meet policy: notify@2023/10/19'}]  An error occurred (InvalidParameterException) when calling the TagResource operation: You can't modify system tags. Tag key: 'aws:cloudformation:stack-name'.
```


### Extra information or context

_No response_

--- 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/resources/secretsmanager.py`
Content:
```
1 # Copyright The Cloud Custodian Authors.
2 # SPDX-License-Identifier: Apache-2.0
3 import json
4 from c7n.manager import resources
5 from c7n.actions import BaseAction, RemovePolicyBase
6 from c7n.exceptions import PolicyValidationError
7 from c7n.filters import iamaccess
8 from c7n.query import QueryResourceManager, TypeInfo
9 from c7n.filters.kms import KmsRelatedFilter
10 from c7n.tags import RemoveTag, Tag, TagActionFilter, TagDelayedAction, Action
11 from c7n.utils import local_session, type_schema, jmespath_search
12 from c7n.filters.policystatement import HasStatementFilter
13 
14 
15 @resources.register('secrets-manager')
16 class SecretsManager(QueryResourceManager):
17 
18     permissions = ('secretsmanager:ListSecretVersionIds',)
19 
20     class resource_type(TypeInfo):
21         service = 'secretsmanager'
22         enum_spec = ('list_secrets', 'SecretList', None)
23         detail_spec = ('describe_secret', 'SecretId', 'Name', None)
24         config_type = cfn_type = 'AWS::SecretsManager::Secret'
25         name = id = 'Name'
26         arn = 'ARN'
27 
28 
29 SecretsManager.filter_registry.register('marked-for-op', TagActionFilter)
30 
31 
32 @SecretsManager.filter_registry.register('cross-account')
33 class CrossAccountAccessFilter(iamaccess.CrossAccountAccessFilter):
34 
35     policy_annotation = "c7n:AccessPolicy"
36     permissions = ("secretsmanager:GetResourcePolicy",)
37 
38     def process(self, resources, event=None):
39         self.client = local_session(self.manager.session_factory).client('secretsmanager')
40         return super(CrossAccountAccessFilter, self).process(resources)
41 
42     def get_resource_policy(self, r):
43         if self.policy_annotation in r:
44             return r[self.policy_annotation]
45         r[self.policy_annotation] = p = self.client.get_resource_policy(
46             SecretId=r['Name']).get('ResourcePolicy', None)
47         return p
48 
49 
50 @SecretsManager.filter_registry.register('kms-key')
51 class KmsFilter(KmsRelatedFilter):
52     RelatedIdsExpression = 'KmsKeyId'
53 
54 
55 @SecretsManager.filter_registry.register('has-statement')
56 class HasStatementFilter(HasStatementFilter):
57 
58     def get_std_format_args(self, secret):
59         return {
60             'secret_arn': secret['ARN'],
61             'account_id': self.manager.config.account_id,
62             'region': self.manager.config.region
63         }
64 
65     def process(self, resources, event=None):
66         self.client = local_session(self.manager.session_factory).client('secretsmanager')
67         for r in resources:
68             try:
69                 policy = self.client.get_resource_policy(SecretId=r['Name'])
70                 if policy.get('ResourcePolicy'):
71                     r['Policy'] = policy['ResourcePolicy']
72             except self.client.exceptions.ResourceNotFoundException:
73                 continue
74 
75         return list(filter(None, map(self.process_resource, resources)))
76 
77 
78 @SecretsManager.action_registry.register('tag')
79 class TagSecretsManagerResource(Tag):
80     """Action to create tag(s) on a Secret resource
81 
82     :example:
83 
84     .. code-block:: yaml
85 
86         policies:
87             - name: tag-secret
88               resource: secrets-manager
89               actions:
90                 - type: tag
91                   key: tag-key
92                   value: tag-value
93     """
94 
95     permissions = ('secretsmanager:TagResource',)
96 
97     def process_resource_set(self, client, resources, new_tags):
98         for r in resources:
99             tags = {t['Key']: t['Value'] for t in r.get('Tags', ())}
100             for t in new_tags:
101                 tags[t['Key']] = t['Value']
102             formatted_tags = [{'Key': k, 'Value': v} for k, v in tags.items()]
103             client.tag_resource(SecretId=r['ARN'], Tags=formatted_tags)
104 
105 
106 @SecretsManager.action_registry.register('remove-tag')
107 class RemoveTagSecretsManagerResource(RemoveTag):
108     """Action to remove tag(s) on a Secret resource
109 
110     :example:
111 
112     .. code-block:: yaml
113 
114         policies:
115             - name: untag-secret
116               resource: secrets-manager
117               actions:
118                 - type: remove-tag
119                   tags: ['tag-to-be-removed']
120     """
121 
122     permissions = ('secretsmanager:UntagResource',)
123 
124     def process_resource_set(self, client, resources, keys):
125         for r in resources:
126             client.untag_resource(SecretId=r['ARN'], TagKeys=keys)
127 
128 
129 @SecretsManager.action_registry.register('mark-for-op')
130 class MarkSecretForOp(TagDelayedAction):
131     """Action to mark a Secret resource for deferred action :example:
132 
133     .. code-block:: yaml
134 
135         policies:
136             - name: mark-secret-for-delete
137               resource: secrets-manager
138               actions:
139                 - type: mark-for-op
140                   op: tag
141                   days: 1
142     """
143 
144 
145 @SecretsManager.action_registry.register('delete')
146 class DeleteSecretsManager(BaseAction):
147     """Delete a secret and all of its versions.
148     The recovery window is the number of days from 7 to 30 that
149     Secrets Manager waits before permanently deleting the secret
150     with default as 30
151 
152     :example:
153 
154     .. code-block:: yaml
155 
156             policies:
157               - name: delete-cross-account-secrets
158                 resource: aws.secrets-manager
159                 filters:
160                   - type: cross-account
161                 actions:
162                   - type: delete
163                     recovery_window: 10
164     """
165 
166     schema = type_schema('delete', recovery_window={'type': 'integer'})
167     permissions = ('secretsmanager:DeleteSecret',)
168 
169     def process(self, resources):
170         client = local_session(
171             self.manager.session_factory).client('secretsmanager')
172 
173         for r in resources:
174             if 'ReplicationStatus' in r:
175                 rep_regions = jmespath_search('ReplicationStatus[*].Region', r)
176                 self.manager.retry(client.remove_regions_from_replication,
177                   SecretId=r['ARN'], RemoveReplicaRegions=rep_regions)
178             self.manager.retry(client.delete_secret,
179               SecretId=r['ARN'], RecoveryWindowInDays=self.data.get('recovery_window', 30))
180 
181 
182 @SecretsManager.action_registry.register('remove-statements')
183 class SecretsManagerRemovePolicyStatement(RemovePolicyBase):
184     """
185     Action to remove resource based policy statements from secrets manager
186 
187     :example:
188 
189     .. code-block:: yaml
190 
191         policies:
192           - name: secrets-manager-cross-account
193             resource: aws.secrets-manager
194             filters:
195               - type: cross-account
196             actions:
197               - type: remove-statements
198                 statement_ids: matched
199     """
200 
201     permissions = ("secretsmanager:DeleteResourcePolicy", "secretsmanager:PutResourcePolicy",)
202 
203     def validate(self):
204         for f in self.manager.iter_filters():
205             if isinstance(f, CrossAccountAccessFilter):
206                 return self
207         raise PolicyValidationError(
208             '`remove-statements` may only be used in '
209             'conjunction with `cross-account` filter on %s' % (self.manager.data,))
210 
211     def process(self, resources):
212         client = local_session(self.manager.session_factory).client('secretsmanager')
213         for r in resources:
214             try:
215                 self.process_resource(client, r)
216             except Exception:
217                 self.log.exception("Error processing secretsmanager:%s", r['ARN'])
218 
219     def process_resource(self, client, resource):
220         p = json.loads(resource.get('c7n:AccessPolicy'))
221         if p is None:
222             return
223 
224         statements, found = self.process_policy(
225             p, resource, CrossAccountAccessFilter.annotation_key)
226 
227         if not found:
228             return
229         if statements:
230             client.put_resource_policy(
231                 SecretId=resource['ARN'],
232                 ResourcePolicy=json.dumps(p)
233             )
234         else:
235             client.delete_resource_policy(SecretId=resource['ARN'])
236 
237 
238 @SecretsManager.action_registry.register('set-encryption')
239 class SetEncryptionAction(Action):
240     """
241     Set kms encryption key for secrets, key supports ARN, ID, or alias
242 
243     :example:
244 
245     .. code-block:: yaml
246 
247         policies:
248             - name: set-secret-encryption
249               resource: aws.secrets-manager
250               actions:
251                 - type: set-encryption
252                   key: alias/foo/bar
253     """
254 
255     schema = type_schema('set-encryption', key={'type': 'string'}, required=['key'])
256     permissions = ('secretsmanager:UpdateSecret', )
257 
258     def process(self, resources):
259         key = self.data['key']
260         client = local_session(self.manager.session_factory).client('secretsmanager')
261         for r in resources:
262             client.update_secret(
263                 SecretId=r['Name'],
264                 KmsKeyId=key
265             )
266
```
--- 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/resources/secretsmanager.py b/c7n/resources/secretsmanager.py
--- a/c7n/resources/secretsmanager.py
+++ b/c7n/resources/secretsmanager.py
@@ -96,7 +96,8 @@
 
     def process_resource_set(self, client, resources, new_tags):
         for r in resources:
-            tags = {t['Key']: t['Value'] for t in r.get('Tags', ())}
+            tags = {t['Key']: t['Value'] for t in r.get('Tags', ())
+                    if not t['Key'].startswith('aws:')}
             for t in new_tags:
                 tags[t['Key']] = t['Value']
             formatted_tags = [{'Key': k, 'Value': v} for k, v in tags.items()]

{"golden_diff": "diff --git a/c7n/resources/secretsmanager.py b/c7n/resources/secretsmanager.py\n--- a/c7n/resources/secretsmanager.py\n+++ b/c7n/resources/secretsmanager.py\n@@ -96,7 +96,8 @@\n \n     def process_resource_set(self, client, resources, new_tags):\n         for r in resources:\n-            tags = {t['Key']: t['Value'] for t in r.get('Tags', ())}\n+            tags = {t['Key']: t['Value'] for t in r.get('Tags', ())\n+                    if not t['Key'].startswith('aws:')}\n             for t in new_tags:\n                 tags[t['Key']] = t['Value']\n             formatted_tags = [{'Key': k, 'Value': v} for k, v in tags.items()]\n", "issue": "aws.secrets-manager:tag action tries to retag secret with existing tags as well\n### Describe the bug\n\nI have a policy that checks all AWS Secret Manager secrets for the existence of two particular tags. If the secret is missing either of these tags, it tries to makr (i.e. tag) the secret for later action. However, Custodian fails to apply its marking tag with the following error:\r\n\r\n2023-10-18 14:25:20,524 - custodian.policy - INFO - policy:secrets-manager-secret-mark-untagged resource:secrets-manager region:us-west-2 count:2 time:0.03\r\n2023-10-18 14:25:20,525 - custodian.actions - INFO - Tagging 2 resources for notify on 2023/10/19\r\n2023-10-18 14:25:21,519 - custodian.actions - ERROR - Exception with tags: [{'Key': 'c7n-secrets-manager-secret-untagged', 'Value': 'Resource does not meet policy: notify@2023/10/19'}]  An error occurred (InvalidParameterException) when calling the TagResource operation: You can't modify system tags. Tag key: 'aws:cloudformation:stack-name'.\r\n\r\nIt appears Custodian is trying to retag with all existing tags in addition to the new marking tag. AWS doesn't like you messing with the system tags?\n\n### What did you expect to happen?\n\nI expected Custodian to only add the marking tag rather than try to apply all the existing tags as well.\r\n\r\nIs this possible? Is it a bug in Custodian or in the AWS Secrets Manager API?\n\n### Cloud Provider\n\nAmazon Web Services (AWS)\n\n### Cloud Custodian version and dependency information\n\n```shell\nCustodian:   0.9.32\r\nPython:      3.8.16 (default, Mar 10 2023, 03:07:48) \r\n             [GCC 7.3.1 20180712 (Red Hat 7.3.1-15)]\r\nPlatform:    posix.uname_result(sysname='Linux', nodename='custodian', release='4.14.320-243.544.amzn2.x86_64', version='#1 SMP Tue Aug 1 21:03:08 UTC 2023', machine='x86_64')\r\nUsing venv:  False\r\nDocker: False\r\nInstalled: \r\n\r\nargcomplete==3.1.2\r\nattrs==23.1.0\r\nboto3==1.28.58\r\nbotocore==1.31.58\r\ndocutils==0.18.1\r\nimportlib-metadata==5.2.0\r\nimportlib-resources==6.1.0\r\njmespath==1.0.1\r\njsonschema==4.19.1\r\njsonschema-specifications==2023.7.1\r\npkgutil-resolve-name==1.3.10\r\npython-dateutil==2.8.2\r\npyyaml==6.0.1\r\nreferencing==0.30.2\r\nrpds-py==0.10.3\r\ns3transfer==0.7.0\r\nsix==1.16.0\r\ntabulate==0.9.0\r\nurllib3==1.26.17\r\nzipp==3.17.0\n```\n\n\n### Policy\n\n```shell\nHere is the action bit:\r\n\r\nactions:\r\n      - type: mark-for-op\r\n        tag: c7n-secrets-manager-secret-untagged\r\n        op: notify\r\n        days: 1\n```\n\n\n### Relevant log/traceback output\n\n```shell\n2023-10-18 14:25:20,524 - custodian.policy - INFO - policy:secrets-manager-secret-mark-untagged resource:secrets-manager region:us-west-2 count:2 time:0.03\r\n2023-10-18 14:25:20,525 - custodian.actions - INFO - Tagging 2 resources for notify on 2023/10/19\r\n2023-10-18 14:25:21,519 - custodian.actions - ERROR - Exception with tags: [{'Key': 'c7n-secrets-manager-secret-untagged', 'Value': 'Resource does not meet policy: notify@2023/10/19'}]  An error occurred (InvalidParameterException) when calling the TagResource operation: You can't modify system tags. Tag key: 'aws:cloudformation:stack-name'.\n```\n\n\n### Extra information or context\n\n_No response_\n", "before_files": [{"content": "# Copyright The Cloud Custodian Authors.\n# SPDX-License-Identifier: Apache-2.0\nimport json\nfrom c7n.manager import resources\nfrom c7n.actions import BaseAction, RemovePolicyBase\nfrom c7n.exceptions import PolicyValidationError\nfrom c7n.filters import iamaccess\nfrom c7n.query import QueryResourceManager, TypeInfo\nfrom c7n.filters.kms import KmsRelatedFilter\nfrom c7n.tags import RemoveTag, Tag, TagActionFilter, TagDelayedAction, Action\nfrom c7n.utils import local_session, type_schema, jmespath_search\nfrom c7n.filters.policystatement import HasStatementFilter\n\n\n@resources.register('secrets-manager')\nclass SecretsManager(QueryResourceManager):\n\n    permissions = ('secretsmanager:ListSecretVersionIds',)\n\n    class resource_type(TypeInfo):\n        service = 'secretsmanager'\n        enum_spec = ('list_secrets', 'SecretList', None)\n        detail_spec = ('describe_secret', 'SecretId', 'Name', None)\n        config_type = cfn_type = 'AWS::SecretsManager::Secret'\n        name = id = 'Name'\n        arn = 'ARN'\n\n\nSecretsManager.filter_registry.register('marked-for-op', TagActionFilter)\n\n\n@SecretsManager.filter_registry.register('cross-account')\nclass CrossAccountAccessFilter(iamaccess.CrossAccountAccessFilter):\n\n    policy_annotation = \"c7n:AccessPolicy\"\n    permissions = (\"secretsmanager:GetResourcePolicy\",)\n\n    def process(self, resources, event=None):\n        self.client = local_session(self.manager.session_factory).client('secretsmanager')\n        return super(CrossAccountAccessFilter, self).process(resources)\n\n    def get_resource_policy(self, r):\n        if self.policy_annotation in r:\n            return r[self.policy_annotation]\n        r[self.policy_annotation] = p = self.client.get_resource_policy(\n            SecretId=r['Name']).get('ResourcePolicy', None)\n        return p\n\n\n@SecretsManager.filter_registry.register('kms-key')\nclass KmsFilter(KmsRelatedFilter):\n    RelatedIdsExpression = 'KmsKeyId'\n\n\n@SecretsManager.filter_registry.register('has-statement')\nclass HasStatementFilter(HasStatementFilter):\n\n    def get_std_format_args(self, secret):\n        return {\n            'secret_arn': secret['ARN'],\n            'account_id': self.manager.config.account_id,\n            'region': self.manager.config.region\n        }\n\n    def process(self, resources, event=None):\n        self.client = local_session(self.manager.session_factory).client('secretsmanager')\n        for r in resources:\n            try:\n                policy = self.client.get_resource_policy(SecretId=r['Name'])\n                if policy.get('ResourcePolicy'):\n                    r['Policy'] = policy['ResourcePolicy']\n            except self.client.exceptions.ResourceNotFoundException:\n                continue\n\n        return list(filter(None, map(self.process_resource, resources)))\n\n\n@SecretsManager.action_registry.register('tag')\nclass TagSecretsManagerResource(Tag):\n    \"\"\"Action to create tag(s) on a Secret resource\n\n    :example:\n\n    .. code-block:: yaml\n\n        policies:\n            - name: tag-secret\n              resource: secrets-manager\n              actions:\n                - type: tag\n                  key: tag-key\n                  value: tag-value\n    \"\"\"\n\n    permissions = ('secretsmanager:TagResource',)\n\n    def process_resource_set(self, client, resources, new_tags):\n        for r in resources:\n            tags = {t['Key']: t['Value'] for t in r.get('Tags', ())}\n            for t in new_tags:\n                tags[t['Key']] = t['Value']\n            formatted_tags = [{'Key': k, 'Value': v} for k, v in tags.items()]\n            client.tag_resource(SecretId=r['ARN'], Tags=formatted_tags)\n\n\n@SecretsManager.action_registry.register('remove-tag')\nclass RemoveTagSecretsManagerResource(RemoveTag):\n    \"\"\"Action to remove tag(s) on a Secret resource\n\n    :example:\n\n    .. code-block:: yaml\n\n        policies:\n            - name: untag-secret\n              resource: secrets-manager\n              actions:\n                - type: remove-tag\n                  tags: ['tag-to-be-removed']\n    \"\"\"\n\n    permissions = ('secretsmanager:UntagResource',)\n\n    def process_resource_set(self, client, resources, keys):\n        for r in resources:\n            client.untag_resource(SecretId=r['ARN'], TagKeys=keys)\n\n\n@SecretsManager.action_registry.register('mark-for-op')\nclass MarkSecretForOp(TagDelayedAction):\n    \"\"\"Action to mark a Secret resource for deferred action :example:\n\n    .. code-block:: yaml\n\n        policies:\n            - name: mark-secret-for-delete\n              resource: secrets-manager\n              actions:\n                - type: mark-for-op\n                  op: tag\n                  days: 1\n    \"\"\"\n\n\n@SecretsManager.action_registry.register('delete')\nclass DeleteSecretsManager(BaseAction):\n    \"\"\"Delete a secret and all of its versions.\n    The recovery window is the number of days from 7 to 30 that\n    Secrets Manager waits before permanently deleting the secret\n    with default as 30\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: delete-cross-account-secrets\n                resource: aws.secrets-manager\n                filters:\n                  - type: cross-account\n                actions:\n                  - type: delete\n                    recovery_window: 10\n    \"\"\"\n\n    schema = type_schema('delete', recovery_window={'type': 'integer'})\n    permissions = ('secretsmanager:DeleteSecret',)\n\n    def process(self, resources):\n        client = local_session(\n            self.manager.session_factory).client('secretsmanager')\n\n        for r in resources:\n            if 'ReplicationStatus' in r:\n                rep_regions = jmespath_search('ReplicationStatus[*].Region', r)\n                self.manager.retry(client.remove_regions_from_replication,\n                  SecretId=r['ARN'], RemoveReplicaRegions=rep_regions)\n            self.manager.retry(client.delete_secret,\n              SecretId=r['ARN'], RecoveryWindowInDays=self.data.get('recovery_window', 30))\n\n\n@SecretsManager.action_registry.register('remove-statements')\nclass SecretsManagerRemovePolicyStatement(RemovePolicyBase):\n    \"\"\"\n    Action to remove resource based policy statements from secrets manager\n\n    :example:\n\n    .. code-block:: yaml\n\n        policies:\n          - name: secrets-manager-cross-account\n            resource: aws.secrets-manager\n            filters:\n              - type: cross-account\n            actions:\n              - type: remove-statements\n                statement_ids: matched\n    \"\"\"\n\n    permissions = (\"secretsmanager:DeleteResourcePolicy\", \"secretsmanager:PutResourcePolicy\",)\n\n    def validate(self):\n        for f in self.manager.iter_filters():\n            if isinstance(f, CrossAccountAccessFilter):\n                return self\n        raise PolicyValidationError(\n            '`remove-statements` may only be used in '\n            'conjunction with `cross-account` filter on %s' % (self.manager.data,))\n\n    def process(self, resources):\n        client = local_session(self.manager.session_factory).client('secretsmanager')\n        for r in resources:\n            try:\n                self.process_resource(client, r)\n            except Exception:\n                self.log.exception(\"Error processing secretsmanager:%s\", r['ARN'])\n\n    def process_resource(self, client, resource):\n        p = json.loads(resource.get('c7n:AccessPolicy'))\n        if p is None:\n            return\n\n        statements, found = self.process_policy(\n            p, resource, CrossAccountAccessFilter.annotation_key)\n\n        if not found:\n            return\n        if statements:\n            client.put_resource_policy(\n                SecretId=resource['ARN'],\n                ResourcePolicy=json.dumps(p)\n            )\n        else:\n            client.delete_resource_policy(SecretId=resource['ARN'])\n\n\n@SecretsManager.action_registry.register('set-encryption')\nclass SetEncryptionAction(Action):\n    \"\"\"\n    Set kms encryption key for secrets, key supports ARN, ID, or alias\n\n    :example:\n\n    .. code-block:: yaml\n\n        policies:\n            - name: set-secret-encryption\n              resource: aws.secrets-manager\n              actions:\n                - type: set-encryption\n                  key: alias/foo/bar\n    \"\"\"\n\n    schema = type_schema('set-encryption', key={'type': 'string'}, required=['key'])\n    permissions = ('secretsmanager:UpdateSecret', )\n\n    def process(self, resources):\n        key = self.data['key']\n        client = local_session(self.manager.session_factory).client('secretsmanager')\n        for r in resources:\n            client.update_secret(\n                SecretId=r['Name'],\n                KmsKeyId=key\n            )\n", "path": "c7n/resources/secretsmanager.py"}], "after_files": [{"content": "# Copyright The Cloud Custodian Authors.\n# SPDX-License-Identifier: Apache-2.0\nimport json\nfrom c7n.manager import resources\nfrom c7n.actions import BaseAction, RemovePolicyBase\nfrom c7n.exceptions import PolicyValidationError\nfrom c7n.filters import iamaccess\nfrom c7n.query import QueryResourceManager, TypeInfo\nfrom c7n.filters.kms import KmsRelatedFilter\nfrom c7n.tags import RemoveTag, Tag, TagActionFilter, TagDelayedAction, Action\nfrom c7n.utils import local_session, type_schema, jmespath_search\nfrom c7n.filters.policystatement import HasStatementFilter\n\n\n@resources.register('secrets-manager')\nclass SecretsManager(QueryResourceManager):\n\n    permissions = ('secretsmanager:ListSecretVersionIds',)\n\n    class resource_type(TypeInfo):\n        service = 'secretsmanager'\n        enum_spec = ('list_secrets', 'SecretList', None)\n        detail_spec = ('describe_secret', 'SecretId', 'Name', None)\n        config_type = cfn_type = 'AWS::SecretsManager::Secret'\n        name = id = 'Name'\n        arn = 'ARN'\n\n\nSecretsManager.filter_registry.register('marked-for-op', TagActionFilter)\n\n\n@SecretsManager.filter_registry.register('cross-account')\nclass CrossAccountAccessFilter(iamaccess.CrossAccountAccessFilter):\n\n    policy_annotation = \"c7n:AccessPolicy\"\n    permissions = (\"secretsmanager:GetResourcePolicy\",)\n\n    def process(self, resources, event=None):\n        self.client = local_session(self.manager.session_factory).client('secretsmanager')\n        return super(CrossAccountAccessFilter, self).process(resources)\n\n    def get_resource_policy(self, r):\n        if self.policy_annotation in r:\n            return r[self.policy_annotation]\n        r[self.policy_annotation] = p = self.client.get_resource_policy(\n            SecretId=r['Name']).get('ResourcePolicy', None)\n        return p\n\n\n@SecretsManager.filter_registry.register('kms-key')\nclass KmsFilter(KmsRelatedFilter):\n    RelatedIdsExpression = 'KmsKeyId'\n\n\n@SecretsManager.filter_registry.register('has-statement')\nclass HasStatementFilter(HasStatementFilter):\n\n    def get_std_format_args(self, secret):\n        return {\n            'secret_arn': secret['ARN'],\n            'account_id': self.manager.config.account_id,\n            'region': self.manager.config.region\n        }\n\n    def process(self, resources, event=None):\n        self.client = local_session(self.manager.session_factory).client('secretsmanager')\n        for r in resources:\n            try:\n                policy = self.client.get_resource_policy(SecretId=r['Name'])\n                if policy.get('ResourcePolicy'):\n                    r['Policy'] = policy['ResourcePolicy']\n            except self.client.exceptions.ResourceNotFoundException:\n                continue\n\n        return list(filter(None, map(self.process_resource, resources)))\n\n\n@SecretsManager.action_registry.register('tag')\nclass TagSecretsManagerResource(Tag):\n    \"\"\"Action to create tag(s) on a Secret resource\n\n    :example:\n\n    .. code-block:: yaml\n\n        policies:\n            - name: tag-secret\n              resource: secrets-manager\n              actions:\n                - type: tag\n                  key: tag-key\n                  value: tag-value\n    \"\"\"\n\n    permissions = ('secretsmanager:TagResource',)\n\n    def process_resource_set(self, client, resources, new_tags):\n        for r in resources:\n            tags = {t['Key']: t['Value'] for t in r.get('Tags', ())\n                    if not t['Key'].startswith('aws:')}\n            for t in new_tags:\n                tags[t['Key']] = t['Value']\n            formatted_tags = [{'Key': k, 'Value': v} for k, v in tags.items()]\n            client.tag_resource(SecretId=r['ARN'], Tags=formatted_tags)\n\n\n@SecretsManager.action_registry.register('remove-tag')\nclass RemoveTagSecretsManagerResource(RemoveTag):\n    \"\"\"Action to remove tag(s) on a Secret resource\n\n    :example:\n\n    .. code-block:: yaml\n\n        policies:\n            - name: untag-secret\n              resource: secrets-manager\n              actions:\n                - type: remove-tag\n                  tags: ['tag-to-be-removed']\n    \"\"\"\n\n    permissions = ('secretsmanager:UntagResource',)\n\n    def process_resource_set(self, client, resources, keys):\n        for r in resources:\n            client.untag_resource(SecretId=r['ARN'], TagKeys=keys)\n\n\n@SecretsManager.action_registry.register('mark-for-op')\nclass MarkSecretForOp(TagDelayedAction):\n    \"\"\"Action to mark a Secret resource for deferred action :example:\n\n    .. code-block:: yaml\n\n        policies:\n            - name: mark-secret-for-delete\n              resource: secrets-manager\n              actions:\n                - type: mark-for-op\n                  op: tag\n                  days: 1\n    \"\"\"\n\n\n@SecretsManager.action_registry.register('delete')\nclass DeleteSecretsManager(BaseAction):\n    \"\"\"Delete a secret and all of its versions.\n    The recovery window is the number of days from 7 to 30 that\n    Secrets Manager waits before permanently deleting the secret\n    with default as 30\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: delete-cross-account-secrets\n                resource: aws.secrets-manager\n                filters:\n                  - type: cross-account\n                actions:\n                  - type: delete\n                    recovery_window: 10\n    \"\"\"\n\n    schema = type_schema('delete', recovery_window={'type': 'integer'})\n    permissions = ('secretsmanager:DeleteSecret',)\n\n    def process(self, resources):\n        client = local_session(\n            self.manager.session_factory).client('secretsmanager')\n\n        for r in resources:\n            if 'ReplicationStatus' in r:\n                rep_regions = jmespath_search('ReplicationStatus[*].Region', r)\n                self.manager.retry(client.remove_regions_from_replication,\n                  SecretId=r['ARN'], RemoveReplicaRegions=rep_regions)\n            self.manager.retry(client.delete_secret,\n              SecretId=r['ARN'], RecoveryWindowInDays=self.data.get('recovery_window', 30))\n\n\n@SecretsManager.action_registry.register('remove-statements')\nclass SecretsManagerRemovePolicyStatement(RemovePolicyBase):\n    \"\"\"\n    Action to remove resource based policy statements from secrets manager\n\n    :example:\n\n    .. code-block:: yaml\n\n        policies:\n          - name: secrets-manager-cross-account\n            resource: aws.secrets-manager\n            filters:\n              - type: cross-account\n            actions:\n              - type: remove-statements\n                statement_ids: matched\n    \"\"\"\n\n    permissions = (\"secretsmanager:DeleteResourcePolicy\", \"secretsmanager:PutResourcePolicy\",)\n\n    def validate(self):\n        for f in self.manager.iter_filters():\n            if isinstance(f, CrossAccountAccessFilter):\n                return self\n        raise PolicyValidationError(\n            '`remove-statements` may only be used in '\n            'conjunction with `cross-account` filter on %s' % (self.manager.data,))\n\n    def process(self, resources):\n        client = local_session(self.manager.session_factory).client('secretsmanager')\n        for r in resources:\n            try:\n                self.process_resource(client, r)\n            except Exception:\n                self.log.exception(\"Error processing secretsmanager:%s\", r['ARN'])\n\n    def process_resource(self, client, resource):\n        p = json.loads(resource.get('c7n:AccessPolicy'))\n        if p is None:\n            return\n\n        statements, found = self.process_policy(\n            p, resource, CrossAccountAccessFilter.annotation_key)\n\n        if not found:\n            return\n        if statements:\n            client.put_resource_policy(\n                SecretId=resource['ARN'],\n                ResourcePolicy=json.dumps(p)\n            )\n        else:\n            client.delete_resource_policy(SecretId=resource['ARN'])\n\n\n@SecretsManager.action_registry.register('set-encryption')\nclass SetEncryptionAction(Action):\n    \"\"\"\n    Set kms encryption key for secrets, key supports ARN, ID, or alias\n\n    :example:\n\n    .. code-block:: yaml\n\n        policies:\n            - name: set-secret-encryption\n              resource: aws.secrets-manager\n              actions:\n                - type: set-encryption\n                  key: alias/foo/bar\n    \"\"\"\n\n    schema = type_schema('set-encryption', key={'type': 'string'}, required=['key'])\n    permissions = ('secretsmanager:UpdateSecret', )\n\n    def process(self, resources):\n        key = self.data['key']\n        client = local_session(self.manager.session_factory).client('secretsmanager')\n        for r in resources:\n            client.update_secret(\n                SecretId=r['Name'],\n                KmsKeyId=key\n            )\n", "path": "c7n/resources/secretsmanager.py"}]}