Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/c7n/resources/cloudtrail.py: 52%
139 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-08 06:51 +0000
1# Copyright The Cloud Custodian Authors.
2# SPDX-License-Identifier: Apache-2.0
3import logging
5from c7n.actions import Action, BaseAction
6from c7n.exceptions import PolicyValidationError
7from c7n.filters import ValueFilter, Filter
8from c7n.manager import resources
9from c7n.tags import universal_augment
10from c7n.query import ConfigSource, DescribeSource, QueryResourceManager, TypeInfo
11from c7n.utils import local_session, type_schema
13from .aws import shape_validate, Arn
15log = logging.getLogger('c7n.resources.cloudtrail')
18class DescribeTrail(DescribeSource):
20 def augment(self, resources):
21 return universal_augment(self.manager, resources)
24def get_trail_groups(session_factory, trails):
25 # returns a dictionary -> key: region value: (client, trails)
26 grouped = {}
27 for t in trails:
28 region = Arn.parse(t['TrailARN']).region
29 client, trails = grouped.setdefault(region, (None, []))
30 trails.append(t)
31 if client is None:
32 client = local_session(session_factory).client(
33 'cloudtrail', region_name=region)
34 grouped[region] = client, trails
35 return grouped
38@resources.register('cloudtrail')
39class CloudTrail(QueryResourceManager):
40 class resource_type(TypeInfo):
41 service = 'cloudtrail'
42 enum_spec = ('describe_trails', 'trailList', None)
43 filter_name = 'trailNameList'
44 filter_type = 'list'
45 arn_type = 'trail'
46 arn = id = 'TrailARN'
47 name = 'Name'
48 cfn_type = config_type = "AWS::CloudTrail::Trail"
49 universal_taggable = object()
51 source_mapping = {
52 'describe': DescribeTrail,
53 'config': ConfigSource
54 }
57@CloudTrail.filter_registry.register('is-shadow')
58class IsShadow(Filter):
59 """Identify shadow trails (secondary copies), shadow trails
60 can't be modified directly, the origin trail needs to be modified.
62 Shadow trails are created for multi-region trails as well for
63 organizational trails.
64 """
65 schema = type_schema('is-shadow', state={'type': 'boolean'})
66 permissions = ('cloudtrail:DescribeTrails',)
67 embedded = False
69 def process(self, resources, event=None):
70 rcount = len(resources)
71 trails = [t for t in resources if (self.is_shadow(t) == self.data.get('state', True))]
72 if len(trails) != rcount and self.embedded:
73 self.log.info("implicitly filtering shadow trails %d -> %d",
74 rcount, len(trails))
75 return trails
77 def is_shadow(self, t):
78 if t.get('IsOrganizationTrail') and self.manager.config.account_id not in t['TrailARN']:
79 return True
80 if t.get('IsMultiRegionTrail') and t['HomeRegion'] != self.manager.config.region:
81 return True
82 return False
85@CloudTrail.filter_registry.register('status')
86class Status(ValueFilter):
87 """Filter a cloudtrail by its status.
89 :Example:
91 .. code-block:: yaml
93 policies:
94 - name: cloudtrail-check-status
95 resource: aws.cloudtrail
96 filters:
97 - type: status
98 key: IsLogging
99 value: False
100 """
102 schema = type_schema('status', rinherit=ValueFilter.schema)
103 schema_alias = False
104 permissions = ('cloudtrail:GetTrailStatus',)
105 annotation_key = 'c7n:TrailStatus'
107 def process(self, resources, event=None):
108 grouped_trails = get_trail_groups(self.manager.session_factory, resources)
109 for region, (client, trails) in grouped_trails.items():
110 for t in trails:
111 if self.annotation_key in t:
112 continue
113 status = client.get_trail_status(Name=t['TrailARN'])
114 status.pop('ResponseMetadata')
115 t[self.annotation_key] = status
116 return super(Status, self).process(resources)
118 def __call__(self, r):
119 return self.match(r[self.annotation_key])
122@CloudTrail.filter_registry.register('event-selectors')
123class EventSelectors(ValueFilter):
124 """Filter a cloudtrail by its related Event Selectors.
126 :example:
128 .. code-block:: yaml
130 policies:
131 - name: cloudtrail-event-selectors
132 resource: aws.cloudtrail
133 filters:
134 - type: event-selectors
135 key: EventSelectors[].IncludeManagementEvents
136 op: contains
137 value: True
138 """
140 schema = type_schema('event-selectors', rinherit=ValueFilter.schema)
141 schema_alias = False
142 permissions = ('cloudtrail:GetEventSelectors',)
143 annotation_key = 'c7n:TrailEventSelectors'
145 def process(self, resources, event=None):
146 grouped_trails = get_trail_groups(self.manager.session_factory, resources)
147 for region, (client, trails) in grouped_trails.items():
148 for t in trails:
149 if self.annotation_key in t:
150 continue
151 selectors = client.get_event_selectors(TrailName=t['TrailARN'])
152 selectors.pop('ResponseMetadata')
153 t[self.annotation_key] = selectors
154 return super(EventSelectors, self).process(resources)
156 def __call__(self, r):
157 return self.match(r[self.annotation_key])
160@CloudTrail.action_registry.register('update-trail')
161class UpdateTrail(Action):
162 """Update trail attributes.
164 :Example:
166 .. code-block:: yaml
168 policies:
169 - name: cloudtrail-set-log
170 resource: aws.cloudtrail
171 filters:
172 - or:
173 - KmsKeyId: empty
174 - LogFileValidationEnabled: false
175 actions:
176 - type: update-trail
177 attributes:
178 KmsKeyId: arn:aws:kms:us-west-2:111122223333:key/1234abcd-12ab-34cd-56ef
179 EnableLogFileValidation: true
180 """
181 schema = type_schema(
182 'update-trail',
183 attributes={'type': 'object'},
184 required=('attributes',))
185 shape = 'UpdateTrailRequest'
186 permissions = ('cloudtrail:UpdateTrail',)
188 def validate(self):
189 attrs = dict(self.data['attributes'])
190 if 'Name' in attrs:
191 raise PolicyValidationError(
192 "Can't include Name in update-trail action")
193 attrs['Name'] = 'PolicyValidation'
194 return shape_validate(
195 attrs,
196 self.shape,
197 self.manager.resource_type.service)
199 def process(self, resources):
200 client = local_session(self.manager.session_factory).client('cloudtrail')
201 shadow_check = IsShadow({'state': False}, self.manager)
202 shadow_check.embedded = True
203 resources = shadow_check.process(resources)
205 for r in resources:
206 client.update_trail(
207 Name=r['Name'],
208 **self.data['attributes'])
211@CloudTrail.action_registry.register('set-logging')
212class SetLogging(Action):
213 """Set the logging state of a trail
215 :Example:
217 .. code-block:: yaml
219 policies:
220 - name: cloudtrail-set-active
221 resource: aws.cloudtrail
222 filters:
223 - type: status
224 key: IsLogging
225 value: False
226 actions:
227 - type: set-logging
228 enabled: True
229 """
230 schema = type_schema(
231 'set-logging', enabled={'type': 'boolean'})
233 def get_permissions(self):
234 enable = self.data.get('enabled', True)
235 if enable is True:
236 return ('cloudtrail:StartLogging',)
237 else:
238 return ('cloudtrail:StopLogging',)
240 def process(self, resources):
241 client = local_session(self.manager.session_factory).client('cloudtrail')
242 shadow_check = IsShadow({'state': False}, self.manager)
243 shadow_check.embedded = True
244 resources = shadow_check.process(resources)
245 enable = self.data.get('enabled', True)
247 for r in resources:
248 if enable:
249 client.start_logging(Name=r['Name'])
250 else:
251 client.stop_logging(Name=r['Name'])
254@CloudTrail.action_registry.register('delete')
255class DeleteTrail(BaseAction):
256 """ Delete a cloud trail
258 :example:
260 .. code-block:: yaml
262 policies:
263 - name: delete-cloudtrail
264 resource: aws.cloudtrail
265 filters:
266 - type: value
267 key: Name
268 value: delete-me
269 op: eq
270 actions:
271 - type: delete
272 """
274 schema = type_schema('delete')
275 permissions = ('cloudtrail:DeleteTrail',)
277 def process(self, resources):
278 client = local_session(self.manager.session_factory).client('cloudtrail')
279 shadow_check = IsShadow({'state': False}, self.manager)
280 shadow_check.embedded = True
281 resources = shadow_check.process(resources)
282 for r in resources:
283 try:
284 client.delete_trail(Name=r['Name'])
285 except client.exceptions.TrailNotFoundException:
286 continue