Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/c7n/resources/s3control.py: 63%

1# Copyright The Cloud Custodian Authors. 

2# SPDX-License-Identifier: Apache-2.0 

3from c7n.actions import Action 

4from c7n.filters.iamaccess import CrossAccountAccessFilter 

5from c7n.manager import resources 

6from c7n.resources.aws import Arn 

7from c7n.query import QueryResourceManager, TypeInfo, DescribeSource 

8from c7n.utils import local_session, type_schema 

9 

10 

11class AccessPointDescribe(DescribeSource): 

12 def get_query_params(self, query_params): 

13 query_params = query_params or {} 

14 query_params['AccountId'] = self.manager.config.account_id 

15 return query_params 

16 

17 def augment(self, resources): 

18 client = local_session(self.manager.session_factory).client('s3control') 

19 results = [] 

20 for r in resources: 

21 arn = Arn.parse(r['AccessPointArn']) 

22 ap = client.get_access_point(AccountId=arn.account_id, Name=r['Name']) 

23 ap.pop('ResponseMetadata', None) 

24 ap['AccessPointArn'] = arn.arn 

25 results.append(ap) 

26 return results 

27 

28 

29@resources.register('s3-access-point') 

30class AccessPoint(QueryResourceManager): 

31 class resource_type(TypeInfo): 

32 service = 's3control' 

33 id = name = 'Name' 

34 enum_spec = ('list_access_points', 'AccessPointList', None) 

35 arn = 'AccessPointArn' 

36 arn_service = 's3' 

37 arn_type = 'accesspoint' 

38 cfn_type = 'AWS::S3::AccessPoint' 

39 permission_prefix = 's3' 

40 

41 source_mapping = {'describe': AccessPointDescribe} 

42 

43 

44@AccessPoint.filter_registry.register('cross-account') 

45class AccessPointCrossAccount(CrossAccountAccessFilter): 

46 

47 policy_attribute = 'c7n:Policy' 

48 permissions = ('s3:GetAccessPointPolicy',) 

49 

50 def process(self, resources, event=None): 

51 client = local_session(self.manager.session_factory).client('s3control') 

52 for r in resources: 

53 if self.policy_attribute in r: 

54 continue 

55 arn = Arn.parse(r['AccessPointArn']) 

56 r[self.policy_attribute] = client.get_access_point_policy( 

57 AccountId=arn.account_id, Name=r['Name'] 

58 ).get('Policy') 

59 

60 return super().process(resources, event) 

61 

62 

63@AccessPoint.action_registry.register('delete') 

64class Delete(Action): 

65 

66 schema = type_schema('delete') 

67 permissions = ('s3:DeleteAccessPoint',) 

68 

69 def process(self, resources): 

70 client = local_session(self.manager.session_factory).client('s3control') 

71 for r in resources: 

72 arn = Arn.parse(r['AccessPointArn']) 

73 try: 

74 client.delete_access_point(AccountId=arn.account_id, Name=r['Name']) 

75 except client.NotFoundException: 

76 continue 

77 

78 

79class MultiRegionAccessPointDescribe(DescribeSource): 

80 def get_query_params(self, query_params): 

81 query_params = query_params or {} 

82 query_params['AccountId'] = self.manager.config.account_id 

83 return query_params 

84 

85 

86@resources.register('s3-access-point-multi') 

87class MultiRegionAccessPoint(QueryResourceManager): 

88 class resource_type(TypeInfo): 

89 service = 's3control' 

90 id = name = 'Name' 

91 enum_spec = ('list_multi_region_access_points', 'AccessPoints', None) 

92 arn_service = 's3' 

93 arn_type = 'accesspoint' 

94 cfn_type = 'AWS::S3::MultiRegionAccessPoint' 

95 permission_prefix = 's3' 

96 

97 source_mapping = {'describe': MultiRegionAccessPointDescribe}