gcpdiag.queries.gcs

31@dataclasses.dataclass(frozen=True)
32class RetentionPolicy:
33  """Bucket's retention policy."""
34
35  retention_period: int

38class RetentionPolicyBuilder:
39  """Builds Bucket's retention policy from dict representation."""
40
41  def __init__(self, retention_policy):
42    self._retention_policy = retention_policy
43
44  def build(self) -> RetentionPolicy:
45    return RetentionPolicy(retention_period=self._get_retention_period())
46
47  def _get_retention_period(self) -> int:
48    try:
49      return int(self._retention_policy['retentionPeriod'])
50    except (KeyError, ValueError):
51      return 0

54class Bucket(models.Resource):
55  """Represents a GCS Bucket."""
56
57  _resource_data: dict
58
59  def __init__(self, project_id, resource_data):
60    super().__init__(project_id=project_id)
61    self._resource_data = resource_data
62    self._metadata_dict = None
63
64  @property
65  def id(self) -> str:
66    return self._resource_data['id']
67
68  @property
69  def name(self) -> str:
70    return self._resource_data['name']
71
72  def is_uniform_access(self) -> bool:
73    return get_path(
74      self._resource_data,
75      ('iamConfiguration', 'uniformBucketLevelAccess', 'enabled'),
76      default=False,
77    )
78
79  @property
80  def full_path(self) -> str:
81    result = re.match(
82      r'https://www.googleapis.com/storage/v1/(.*)', self._resource_data['selfLink']
83    )
84    if result:
85      return result.group(1)
86    else:
87      return '>> ' + self._resource_data['selfLink']
88
89  @property
90  def short_path(self) -> str:
91    return self.name
92
93  @property
94  def labels(self) -> dict:
95    return self._resource_data.get('labels', {})
96
97  @property
98  def retention_policy(self) -> RetentionPolicy:
99    return RetentionPolicyBuilder(self._resource_data.get('retentionPolicy', {})).build()

102class BucketIAMPolicy(iam.BaseIAMPolicy):
103  def _is_resource_permission(self, permission):
104    return True

107@caching.cached_api_call(in_memory=True)
108def get_bucket_iam_policy(context: models.Context, bucket: str) -> BucketIAMPolicy:
109  project_id = context.project_id
110  gcs_api = apis.get_api('storage', 'v1', project_id)
111  request = gcs_api.buckets().getIamPolicy(bucket=bucket)
112
113  return iam.fetch_iam_policy(request, BucketIAMPolicy, project_id, bucket, context)

116@caching.cached_api_call(in_memory=True)
117def get_bucket(context: models.Context, bucket: str) -> Bucket:
118  gcs_api = apis.get_api('storage', 'v1', context.project_id)
119  logging.debug('fetching GCS bucket %s', bucket)
120  query = gcs_api.buckets().get(bucket=bucket)
121  try:
122    response = query.execute(num_retries=config.API_RETRIES)
123  except googleapiclient.errors.HttpError as err:
124    print(err)
125    raise utils.GcpApiError(err) from err
126  print(response)
127  # Resource data only provides us project number.
128  # We don't know project id at this point.
129  return Bucket(project_id=None, resource_data=response)

132@caching.cached_api_call(in_memory=True)
133def get_buckets(context: models.Context) -> Mapping[str, Bucket]:
134  buckets: Dict[str, Bucket] = {}
135  if not apis.is_enabled(context.project_id, 'storage'):
136    return buckets
137  gcs_api = apis.get_api('storage', 'v1', context.project_id)
138  logging.debug('fetching list of GCS buckets in project %s', context.project_id)
139  query = gcs_api.buckets().list(project=context.project_id)
140  try:
141    resp = query.execute(num_retries=config.API_RETRIES)
142    if 'items' not in resp:
143      return buckets
144    for b in resp['items']:
145      # verify that we have some minimal data that we expect
146      if 'id' not in b:
147        raise RuntimeError('missing data in bucket response')
148      # Does not support matching for location for buckets
149      # names are globally unique and should suffice
150      if not context.match_project_resource(
151        resource=b.get('name'),
152        labels=b.get('labels', {}),
153      ):
154        continue
155
156      buckets[b['name']] = Bucket(project_id=context.project_id, resource_data=b)
157  except googleapiclient.errors.HttpError as err:
158    raise utils.GcpApiError(err) from err
159  return buckets