gcpdiag.queries.pubsub

30class Topic(models.Resource):
31  """Represent a Topic"""
32
33  _resource_data: dict
34
35  def __init__(self, project_id, resource_data):
36    super().__init__(project_id=project_id)
37    self._resource_data = resource_data
38    self._metadata_dict = None
39
40  @property
41  def name(self) -> str:
42    m = re.search(r'/topics/([^/]+)$', self._resource_data['name'])
43    if not m:
44      raise RuntimeError("can't determine name of topic %s" % (self._resource_data['name']))
45    return m.group(1)
46
47  @property
48  def full_path(self) -> str:
49    return self._resource_data['name']
50
51  @property
52  def short_path(self) -> str:
53    path = self.project_id + '/' + self.name
54    return path
55
56  @property
57  def kms_key_name(self) -> str:
58    return self._resource_data['kmsKeyName']

61@caching.cached_api_call
62def get_topics(context: models.Context) -> Mapping[str, Topic]:
63  """Get all topics(Does not include deleted topics)."""
64  topics: Dict[str, Topic] = {}
65  if not apis.is_enabled(context.project_id, 'pubsub'):
66    return topics
67  pubsub_api = apis.get_api('pubsub', 'v1', context.project_id)
68  logging.debug('fetching list of PubSub topics in project %s', context.project_id)
69  query = pubsub_api.projects().topics().list(project=f'projects/{context.project_id}')
70  try:
71    resp = query.execute(num_retries=config.API_RETRIES)
72    if 'topics' not in resp:
73      return topics
74    for t in resp['topics']:
75      # verify that we have some minimal data that we expect
76      if 'name' not in t:
77        raise RuntimeError('missing data in topics response')
78        # projects/{project}/topics/{topic}
79      result = re.match(r'projects/[^/]+/topics/([^/]+)', t['name'])
80      if not result:
81        logging.error('invalid topic data: %s', t['name'])
82        continue
83
84      if not context.match_project_resource(resource=result.group(1), labels=t.get('labels', {})):
85        continue
86
87      topics[t['name']] = Topic(project_id=context.project_id, resource_data=t)
88  except googleapiclient.errors.HttpError as err:
89    raise utils.GcpApiError(err) from err
90  return topics

93class TopicIAMPolicy(iam.BaseIAMPolicy):
94  def _is_resource_permission(self, permission):
95    return True

 98@caching.cached_api_call(in_memory=True)
 99def get_topic_iam_policy(context: models.Context, name: str) -> TopicIAMPolicy:
100  project_id = utils.get_project_by_res_name(name)
101
102  pubsub_api = apis.get_api('pubsub', 'v1', project_id)
103  request = pubsub_api.projects().topics().getIamPolicy(resource=name)
104
105  return iam.fetch_iam_policy(request, TopicIAMPolicy, project_id, name, context)

108class Subscription(models.Resource):
109  """Represent a Subscription."""
110
111  _resource_data: dict
112
113  def __init__(self, project_id, resource_data):
114    super().__init__(project_id=project_id)
115    self._resource_data = resource_data
116    self._metadata_dict = None
117
118  @property
119  def name(self) -> str:
120    m = re.search(r'/subscriptions/([^/]+)$', self._resource_data['name'])
121    if not m:
122      raise RuntimeError("can't determine name of subscription %s" % (self._resource_data['name']))
123    return m.group(1)
124
125  @property
126  def full_path(self) -> str:
127    return self._resource_data['name']
128
129  @property
130  def short_path(self) -> str:
131    path = self.project_id + '/' + self.name
132    return path
133
134  @property
135  def topic(self) -> Union[Topic, str]:
136    """
137    Return subscription's topic as a Topic object,
138    or String '_deleted-topic_' if topic is deleted.
139    """
140    if 'topic' not in self._resource_data:
141      raise RuntimeError('topic not set for subscription {self.name}')
142    elif self._resource_data['topic'] == '_deleted-topic_':
143      return '_deleted_topic_'
144
145    m = re.match(r'projects/([^/]+)/topics/([^/]+)', self._resource_data['topic'])
146    if not m:
147      raise RuntimeError("can't parse topic: %s" % self._resource_data['topic'])
148    (project_id, topic_name) = (m.group(1), self._resource_data['topic'])
149    topics = get_topics(models.Context(project_id))
150    if topic_name not in topics:
151      raise RuntimeError(f'Topic {topic_name} for Subscription {self.name} not found')
152    return topics[topic_name]
153
154  @property
155  def push_config(self) -> dict:
156    return self._resource_data.get('pushConfig', {})
157
158  @property
159  def push_oidc_service_account_email(self) -> str:
160    """Return the OIDC service account email for a push subscription."""
161    return self.push_config.get('oidcToken', {}).get('serviceAccountEmail', '')
162
163  def is_detached(self) -> bool:
164    """Return if subscription is detached."""
165    if 'detached' in self._resource_data:
166      return bool(self._resource_data['detached'])
167    return False
168
169  def is_big_query_subscription(self) -> bool:
170    """Return Boolean value if subscription is a big query subscription."""
171    if 'bigqueryConfig' in self._resource_data:
172      return True
173    return False
174
175  def is_gcs_subscription(self) -> bool:
176    """Return Boolean value if subscription is a gcs subscription."""
177    if 'cloudStorageConfig' in self._resource_data:
178      return True
179    return False
180
181  def is_push_subscription(self) -> bool:
182    """Return Boolean value if subscription is a push subscription."""
183    if (
184      self._resource_data['pushConfig']
185      or self.is_big_query_subscription()
186      or self.is_gcs_subscription()
187    ):
188      return True
189    return False
190
191  def is_active(self) -> bool:
192    """Return Boolean value if subscription is active."""
193    return self._resource_data['state'] == 'ACTIVE'
194
195  def has_dead_letter_topic(self) -> bool:
196    """Return Truthy value if subscription has a dead-letter topic."""
197    if 'deadLetterPolicy' in self._resource_data:
198      return bool(self._resource_data['deadLetterPolicy']['deadLetterTopic'])
199    return False
200
201  def dead_letter_topic(self) -> str:
202    """Return the dead-letter topic."""
203    if self.has_dead_letter_topic():
204      return self._resource_data.get('deadLetterPolicy', {}).get('deadLetterTopic', '')
205    return ''
206
207  def gcs_subscription_bucket(self) -> str:
208    """Return the name of the bucket attached to GCS subscription."""
209    if self.is_gcs_subscription():
210      return get_path(self._resource_data, ('cloudStorageConfig', 'bucket'))
211    return ''  # acts as a null return that can be evaluated as a false value

214@caching.cached_api_call
215def get_subscriptions(context: models.Context) -> Mapping[str, Subscription]:
216  subscriptions: Dict[str, Subscription] = {}
217  if not apis.is_enabled(context.project_id, 'pubsub'):
218    return subscriptions
219  pubsub_api = apis.get_api('pubsub', 'v1', context.project_id)
220  logging.debug('fetching list of PubSub subscriptions in project %s', context.project_id)
221  query = pubsub_api.projects().subscriptions().list(project=f'projects/{context.project_id}')
222  try:
223    resp = query.execute(num_retries=config.API_RETRIES)
224    if 'subscriptions' not in resp:
225      return subscriptions
226    for s in resp['subscriptions']:
227      # verify that we have some minimal data that we expect
228      if 'name' not in s:
229        raise RuntimeError('missing data in topics response')
230
231      # projects/{project}/subscriptions/{sub}
232      result = re.match(r'projects/[^/]+/subscriptions/([^/]+)', s['name'])
233      if not result:
234        logging.error('invalid subscription data: %s', s['name'])
235        continue
236
237      if not context.match_project_resource(resource=result.group(1), labels=s.get('labels', {})):
238        continue
239
240      subscriptions[s['name']] = Subscription(project_id=context.project_id, resource_data=s)
241  except googleapiclient.errors.HttpError as err:
242    raise utils.GcpApiError(err) from err
243  return subscriptions

246@caching.cached_api_call
247def get_subscription(project_id: str, subscription_name: str) -> Union[None, Subscription]:
248  if not apis.is_enabled(project_id, 'pubsub'):
249    return None
250  pubsub_api = apis.get_api('pubsub', 'v1', project_id)
251  logging.debug('fetching PubSub subscription in project %s', project_id)
252  query = (
253    pubsub_api.projects()
254    .subscriptions()
255    .get(subscription=f'projects/{project_id}/subscriptions/{subscription_name}')
256  )
257  try:
258    resp = query.execute(num_retries=config.API_RETRIES)
259    return Subscription(project_id=project_id, resource_data=resp)
260  except googleapiclient.errors.HttpError as err:
261    raise utils.GcpApiError(err) from err

264class SubscriptionIAMPolicy(iam.BaseIAMPolicy):
265  def _is_resource_permission(self, permission):
266    return True

269@caching.cached_api_call(in_memory=True)
270def get_subscription_iam_policy(context: models.Context, name: str) -> SubscriptionIAMPolicy:
271  project_id = utils.get_project_by_res_name(name)
272
273  pubsub_api = apis.get_api('pubsub', 'v1', project_id)
274  request = pubsub_api.projects().subscriptions().getIamPolicy(resource=name)
275
276  return iam.fetch_iam_policy(request, SubscriptionIAMPolicy, project_id, name, context)