Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/c7n_gcp/resources/storage.py: 78%

1# Copyright The Cloud Custodian Authors. 

2# SPDX-License-Identifier: Apache-2.0 

3from c7n.utils import type_schema 

4from c7n_gcp.actions import MethodAction 

5from c7n_gcp.provider import resources 

6from c7n_gcp.query import QueryResourceManager, TypeInfo 

7from c7n_gcp.filters import IamPolicyFilter 

8 

9 

10@resources.register('bucket') 

11class Bucket(QueryResourceManager): 

12 

13 class resource_type(TypeInfo): 

14 service = 'storage' 

15 version = 'v1' 

16 component = 'buckets' 

17 scope = 'project' 

18 enum_spec = ('list', 'items[]', {'projection': 'full'}) 

19 name = id = 'name' 

20 default_report_fields = [ 

21 "name", "timeCreated", "location", "storageClass"] 

22 asset_type = "storage.googleapis.com/Bucket" 

23 scc_type = "google.cloud.storage.Bucket" 

24 metric_key = 'resource.labels.bucket_name' 

25 urn_component = "bucket" 

26 labels = True 

27 labels_op = 'patch' 

28 

29 @staticmethod 

30 def get(client, resource_info): 

31 # pull mode passes the bucket name in the info using bucket_name. 

32 # gcp-scc mode passes in the full resourceName 

33 if not (bucket_name := resource_info.get("bucket_name")): 

34 # There is no nice way to return no resource, so if there is no 

35 # resourceName in the info, we will raise a KeyError. 

36 prefix = "//storage.googleapis.com/" 

37 bucket_name = resource_info["resourceName"].removeprefix(prefix) 

38 

39 return client.execute_command("get", {"bucket": bucket_name}) 

40 

41 @staticmethod 

42 def get_label_params(resource, all_labels): 

43 return {'bucket': resource['name'], 'body': {'labels': all_labels}} 

44 

45 

46@Bucket.filter_registry.register('iam-policy') 

47class BucketIamPolicyFilter(IamPolicyFilter): 

48 """ 

49 Overrides the base implementation to process bucket resources correctly. 

50 """ 

51 permissions = ('storage.buckets.getIamPolicy',) 

52 

53 def _verb_arguments(self, resource): 

54 verb_arguments = {{"bucket": resource["name"]}} 

55 return verb_arguments 

56 

57 

58@Bucket.action_registry.register('set-uniform-access') 

59class BucketLevelAccess(MethodAction): 

60 '''Uniform access disables object ACLs on a bucket. 

61 

62 Enabling this means only bucket policies (and organization bucket 

63 policies) govern access to a bucket. 

64 

65 When enabled, users can only specify bucket level IAM policies 

66 and not Object level ACL's. 

67 

68 Example Policy: 

69 

70 .. code-block:: yaml 

71 

72 policies: 

73 - name: enforce-uniform-bucket-level-access 

74 resource: gcp.bucket 

75 filters: 

76 - iamConfiguration.uniformBucketLevelAccess.enable: false 

77 actions: 

78 - type: set-uniform-access 

79 # The following is also the default 

80 state: true 

81 ''' 

82 

83 schema = type_schema('set-uniform-access', state={'type': 'boolean'}) 

84 method_spec = {'op': 'patch'} 

85 method_perm = 'update' 

86 

87 # the google docs and example on this api appear to broken. 

88 # https://cloud.google.com/storage/docs/using-uniform-bucket-level-access#rest-apis 

89 # 

90 # instead we observe the behavior gsutil interaction to effect the same. 

91 # the key seems to be the undocumented projection parameter 

92 # 

93 def get_resource_params(self, model, resource): 

94 enabled = self.data.get('state', True) 

95 return {'bucket': resource['name'], 

96 'fields': 'iamConfiguration', 

97 'projection': 'noAcl', # not documented but 

98 'body': {'iamConfiguration': {'uniformBucketLevelAccess': {'enabled': enabled}}}}