gcpdiag.queries.gke

39class NodeConfig:
40  """Represents a GKE node pool configuration."""
41
42  def __init__(self, resource_data):
43    self._resource_data = resource_data
44
45  def has_accelerators(self) -> bool:
46    if 'accelerators' in self._resource_data:
47      return True
48    return False
49
50  @property
51  def machine_type(self) -> str:
52    return self._resource_data['machineType']
53
54  @property
55  def image_type(self) -> str:
56    return self._resource_data['imageType']
57
58  @property
59  def oauth_scopes(self) -> list:
60    return self._resource_data['oauthScopes']
61
62  @property
63  def has_serial_port_logging_enabled(self) -> bool:
64    """Check if serial port logging is enabled in the node config.
65
66    Returns:
67      bool: True if serial port logging is enabled or not explicitly disabled.
68            False if explicitly disabled.
69    """
70    metadata = self._resource_data.get('metadata', {})
71    return metadata.get('serial-port-logging-enable', 'true').lower() == 'true'

 74class NodePool(models.Resource):
 75  """Represents a GKE node pool."""
 76
 77  version: Version
 78
 79  def __init__(self, cluster, resource_data):
 80    super().__init__(project_id=cluster.project_id)
 81    self._cluster = cluster
 82    self._resource_data = resource_data
 83    self.version = Version(self._resource_data['version'])
 84    self._migs = None
 85
 86  def _get_service_account(self) -> str:
 87    return self._resource_data.get('config', {}).get('serviceAccount', None)
 88
 89  @property
 90  def full_path(self) -> str:
 91    # https://container.googleapis.com/v1/projects/gcpdiag-gke1-aaaa/
 92    #   locations/europe-west1/clusters/gke2/nodePools/default-pool
 93    m = re.match(
 94      r'https://container.googleapis.com/v1/(.*)',
 95      self._resource_data.get('selfLink', ''),
 96    )
 97    if not m:
 98      raise RuntimeError("can't parse selfLink of nodepool resource")
 99    return m.group(1)
100
101  @property
102  def short_path(self) -> str:
103    path = self.full_path
104    path = re.sub(r'^projects/', '', path)
105    path = re.sub(r'/locations/', '/', path)
106    path = re.sub(r'/zones/', '/', path)
107    path = re.sub(r'/clusters/', '/', path)
108    path = re.sub(r'/nodePools/', '/', path)
109    return path
110
111  @property
112  def name(self) -> str:
113    return self._resource_data['name']
114
115  @property
116  def config(self) -> NodeConfig:
117    return NodeConfig(self._resource_data['config'])
118
119  @property
120  def node_count(self) -> int:
121    return self._resource_data.get('initialNodeCount', 0)
122
123  def has_default_service_account(self) -> bool:
124    sa = self._get_service_account()
125    return sa == 'default'
126
127  def has_image_streaming_enabled(self) -> bool:
128    return get_path(self._resource_data, ('config', 'gcfsConfig', 'enabled'), default=False)
129
130  def has_md_concealment_enabled(self) -> bool:
131    # Empty ({}) workloadMetadataConfig means that 'Metadata concealment'
132    # (predecessor of Workload Identity) is enabled.
133    # https://cloud.google.com/kubernetes-engine/docs/how-to/protecting-cluster-metadata#concealment
134    return (
135      get_path(
136        self._resource_data,
137        ('config', 'workloadMetadataConfig'),
138        default=None,
139      )
140      == {}
141    )
142
143  def has_workload_identity_enabled(self) -> bool:
144    # 'Metadata concealment' (workloadMetadataConfig == {}) doesn't protect the
145    # default SA's token
146    return bool(
147      get_path(
148        self._resource_data,
149        ('config', 'workloadMetadataConfig'),
150        default=None,
151      )
152    )
153
154  @property
155  def service_account(self) -> str:
156    sa = self._get_service_account()
157    if sa == 'default':
158      project_nr = crm.get_project(self.project_id).number
159      return f'{project_nr}-compute@developer.gserviceaccount.com'
160    else:
161      return sa
162
163  @property
164  def pod_ipv4_cidr_size(self) -> int:
165    return self._resource_data['podIpv4CidrSize']
166
167  @property
168  def pod_ipv4_cidr_block(self) -> Optional[IPv4NetOrIPv6Net]:
169    # Get the pod cidr range in use by the nodepool
170    pod_cidr = get_path(self._resource_data, ('networkConfig', 'podIpv4CidrBlock'), default=None)
171
172    if pod_cidr:
173      return ipaddress.ip_network(pod_cidr)
174    else:
175      return None
176
177  @property
178  def max_pod_per_node(self) -> int:
179    return int(
180      get_path(
181        self._resource_data,
182        ('maxPodsConstraint', 'maxPodsPerNode'),
183        default=DEFAULT_MAX_PODS_PER_NODE,
184      )
185    )
186
187  @property
188  def cluster(self) -> 'Cluster':
189    return self._cluster
190
191  @property
192  def instance_groups(self) -> List[gce.ManagedInstanceGroup]:
193    if self._migs is None:
194      project_migs_by_selflink = {}
195      for m in gce.get_managed_instance_groups(models.Context(project_id=self.project_id)).values():
196        project_migs_by_selflink[m.self_link] = m
197
198      self._migs = []
199      for url in self._resource_data.get('instanceGroupUrls', []):
200        try:
201          self._migs.append(project_migs_by_selflink[url])
202        except KeyError:
203          continue
204    return self._migs
205
206  @property
207  def node_tags(self) -> List[str]:
208    """Returns the firewall tags used for nodes in this cluster.
209
210    If the node tags can't be determined, [] is returned.
211    """
212    migs = self.instance_groups
213    if not migs:
214      return []
215    return migs[0].template.tags
216
217  def get_machine_type(self) -> str:
218    """Returns the machine type of the nodepool nodes"""
219    return self.config.machine_type

222class UndefinedClusterPropertyError(Exception):
223  """Thrown when a property of a cluster can't be determined for some reason.
224
225  For example, the cluster_hash can't be determined because there are no
226  nodepools defined.
227  """
228
229  pass

232class Cluster(models.Resource):
233  """Represents a GKE cluster.
234
235  https://cloud.google.com/kubernetes-engine/docs/reference/rest/v1/projects.locations.clusters#Cluster
236  """
237
238  _resource_data: dict
239  master_version: Version
240  _context: models.Context
241  _nodepools: Optional[List[NodePool]]
242
243  def __init__(self, project_id, resource_data, context: models.Context):
244    super().__init__(project_id=project_id)
245    self._resource_data = resource_data
246    self.master_version = Version(self._resource_data['currentMasterVersion'])
247    self._nodepools = None
248    self._context = context
249
250  @property
251  def full_path(self) -> str:
252    if utils.is_region(self._resource_data['location']):
253      return f'projects/{self.project_id}/locations/{self.location}/clusters/{self.name}'
254    else:
255      return f'projects/{self.project_id}/zones/{self.location}/clusters/{self.name}'
256
257  @property
258  def short_path(self) -> str:
259    path = self.full_path
260    path = re.sub(r'^projects/', '', path)
261    path = re.sub(r'/locations/', '/', path)
262    path = re.sub(r'/zones/', '/', path)
263    path = re.sub(r'/clusters/', '/', path)
264    return path
265
266  @property
267  def name(self) -> str:
268    return self._resource_data['name']
269
270  @property
271  def location(self) -> str:
272    return self._resource_data['location']
273
274  @property
275  def pod_ipv4_cidr(self) -> IPv4NetOrIPv6Net:
276    cidr = self._resource_data['clusterIpv4Cidr']
277    return ipaddress.ip_network(cidr)
278
279  @property
280  def current_node_count(self) -> int:
281    return self._resource_data.get('currentNodeCount', 0)
282
283  @property
284  def release_channel(self) -> Optional[str]:
285    try:
286      return self._resource_data['releaseChannel']['channel']
287    except KeyError:
288      return None
289
290  @property
291  def nap_node_image_type(self) -> Optional[str]:
292    return get_path(
293      self._resource_data,
294      ('autoscaling', 'autoprovisioningNodePoolDefaults', 'imageType'),
295      default=None,
296    )
297
298  @property
299  def app_layer_sec_key(self) -> str:
300    return self._resource_data['databaseEncryption'].get('keyName')
301
302  @property
303  def status(self) -> str:
304    return self._resource_data['status']
305
306  @property
307  def status_message(self) -> str:
308    return self._resource_data.get('statusMessage', None)
309
310  def has_app_layer_enc_enabled(self) -> bool:
311    # state := 'DECRYPTED' | 'ENCRYPTED', keyName := 'full_path_to_key_resouce'
312    return (
313      get_path(self._resource_data, ('databaseEncryption', 'state'), default=None) == 'ENCRYPTED'
314    )
315
316  def has_logging_enabled(self) -> bool:
317    return self._resource_data['loggingService'] != 'none'
318
319  def enabled_logging_components(self) -> List[str]:
320    return self._resource_data['loggingConfig']['componentConfig']['enableComponents']
321
322  def has_monitoring_enabled(self) -> bool:
323    return self._resource_data['monitoringService'] != 'none'
324
325  def enabled_monitoring_components(self) -> List[str]:
326    return self._resource_data['monitoringConfig']['componentConfig']['enableComponents']
327
328  def has_control_plane_logging_enabled(self) -> bool:
329    if not self.has_logging_enabled():
330      return False
331    components = get_path(
332      self._resource_data, ('loggingConfig', 'componentConfig', 'enableComponents'), default=[]
333    )
334    return all(c in components for c in ['API_SERVER', 'SCHEDULER', 'CONTROLLER_MANAGER'])
335
336  def has_control_plane_monitoring_enabled(self) -> bool:
337    if not self.has_monitoring_enabled():
338      return False
339    components = get_path(
340      self._resource_data, ('monitoringConfig', 'componentConfig', 'enableComponents'), default=[]
341    )
342    return all(c in components for c in ['API_SERVER', 'SCHEDULER', 'CONTROLLER_MANAGER'])
343
344  def has_authenticator_group_enabled(self) -> bool:
345    return len(self._resource_data.get('authenticatorGroupsConfig', {})) > 0
346
347  def has_workload_identity_enabled(self) -> bool:
348    return len(self._resource_data.get('workloadIdentityConfig', {})) > 0
349
350  def has_http_load_balancing_enabled(self) -> bool:
351    # HTTP load balancing needs to be enabled to use GKE ingress
352    return (
353      get_path(
354        self._resource_data,
355        ('addonsConfig', 'httpLoadBalancing', 'disabled'),
356        default=None,
357      )
358      is not True
359    )
360
361  def has_network_policy_enabled(self) -> bool:
362    # Network policy enforcement
363    return (
364      get_path(
365        self._resource_data,
366        ('addonsConfig', 'networkPolicyConfig', 'disabled'),
367        default=False,
368      )
369      is not True
370    )
371
372  def has_dpv2_enabled(self) -> bool:
373    # Checks whether dataplane V2 is enabled in clusters
374    return (
375      get_path(
376        self._resource_data,
377        ('networkConfig', 'datapathProvider'),
378        default=None,
379      )
380      == 'ADVANCED_DATAPATH'
381    )
382
383  def has_intra_node_visibility_enabled(self) -> bool:
384    if (
385      'networkConfig' in self._resource_data
386      and 'enableIntraNodeVisibility' in self._resource_data['networkConfig']
387    ):
388      return self._resource_data['networkConfig']['enableIntraNodeVisibility']
389    return False
390
391  def has_maintenance_window(self) -> bool:
392    # 'e3b0c442' is a hexadecimal string that represents the value of an empty
393    # string ('') in cryptography. If the maintenance windows are defined, the
394    # value of 'resourceVersion' is not empty ('e3b0c442').
395    return self._resource_data['maintenancePolicy']['resourceVersion'] != 'e3b0c442'
396
397  @property
398  def maintenance_policy(self) -> dict:
399    """Returns the maintenance policy of the cluster."""
400    return self._resource_data.get('maintenancePolicy', {})
401
402  def has_image_streaming_enabled(self) -> bool:
403    """Check if cluster has Image Streaming (aka  Google Container File System)
404
405    enabled
406    """
407    global_gcsfs = get_path(
408      self._resource_data,
409      ('nodePoolDefaults', 'nodeConfigDefaults', 'gcfsConfig', 'enabled'),
410      default=False,
411    )
412    # Check nodePoolDefaults settings
413    if global_gcsfs:
414      return True
415    for np in self.nodepools:
416      # Check if any nodepool has image streaming enabled
417      if np.has_image_streaming_enabled():
418        return True
419    return False
420
421  @property
422  def nodepools(self) -> List[NodePool]:
423    if self._nodepools is None:
424      self._nodepools = []
425      for n in self._resource_data.get('nodePools', []):
426        self._nodepools.append(NodePool(self, n))
427    return self._nodepools
428
429  @property
430  def network(self) -> network.Network:
431    # projects/gcpdiag-gke1-aaaa/global/networks/default
432    network_string = self._resource_data['networkConfig']['network']
433    m = re.match(r'projects/([^/]+)/global/networks/([^/]+)$', network_string)
434    if not m:
435      raise RuntimeError("can't parse network string: %s" % network_string)
436    return network.get_network(m.group(1), m.group(2), self._context)
437
438  @property
439  def subnetwork(self) -> Optional[models.Resource]:
440    # 'projects/gcpdiag-gke1-aaaa/regions/europe-west4/subnetworks/default'
441    if 'subnetwork' not in self._resource_data['networkConfig']:
442      return None
443
444    subnetwork_string = self._resource_data['networkConfig']['subnetwork']
445    m = re.match(
446      r'projects/([^/]+)/regions/([^/]+)/subnetworks/([^/]+)$',
447      subnetwork_string,
448    )
449    if not m:
450      raise RuntimeError("can't parse network string: %s" % subnetwork_string)
451    return network.get_subnetwork(m.group(1), m.group(2), m.group(3))
452
453  @property
454  def get_subnet_name(self) -> Optional[models.Resource]:
455    if 'subnetwork' not in self._resource_data:
456      return None
457    return self._resource_data['subnetwork']
458
459  @property
460  def get_nodepool_config(self) -> Optional[models.Resource]:
461    if 'nodePools' not in self._resource_data:
462      return None
463    return self._resource_data['nodePools']
464
465  @property
466  def get_network_string(self) -> str:
467    if 'networkConfig' not in self._resource_data:
468      return ''
469    if 'network' not in self._resource_data['networkConfig']:
470      return ''
471    return self._resource_data['networkConfig']['network']
472
473  @property
474  def is_private(self) -> bool:
475    if 'privateClusterConfig' not in self._resource_data:
476      return False
477
478    return self._resource_data['privateClusterConfig'].get('enablePrivateNodes', False)
479
480  @property
481  def is_vpc_native(self) -> bool:
482    return get_path(
483      self._resource_data,
484      ('ipAllocationPolicy', 'useIpAliases'),
485      default=False,
486    )
487
488  @property
489  def is_regional(self) -> bool:
490    return len(self._resource_data['locations']) > 1
491
492  @property
493  def cluster_ca_certificate(self) -> str:
494    return self._resource_data['masterAuth']['clusterCaCertificate']
495
496  @property
497  def endpoint(self) -> Optional[str]:
498    if 'endpoint' not in self._resource_data:
499      return None
500    return self._resource_data['endpoint']
501
502  @property
503  def is_autopilot(self) -> bool:
504    if 'autopilot' not in self._resource_data:
505      return False
506    return self._resource_data['autopilot'].get('enabled', False)
507
508  @property
509  def masters_cidr_list(self) -> Iterable[IPv4NetOrIPv6Net]:
510    if get_path(
511      self._resource_data,
512      ('privateClusterConfig', 'masterIpv4CidrBlock'),
513      default=None,
514    ):
515      return [
516        ipaddress.ip_network(self._resource_data['privateClusterConfig']['masterIpv4CidrBlock'])
517      ]
518    else:
519      # only older clusters still have ssh firewall rules
520      if self.current_node_count and not self.cluster_hash:
521        logging.warning("couldn't retrieve cluster hash for cluster %s.", self.name)
522        return []
523      fw_rule_name = f'gke-{self.name}-{self.cluster_hash}-ssh'
524      rule = self.network.firewall.get_vpc_ingress_rules(name=fw_rule_name)
525      if rule and rule[0].is_enabled():
526        return rule[0].source_ranges
527      return []
528
529  @property
530  def cluster_hash(self) -> Optional[str]:
531    """Returns the "cluster hash" as used in automatic firewall rules for GKE clusters.
532
533    Cluster hash is the first 8 characters of cluster id. See also:
534    https://cloud.google.com/kubernetes-engine/docs/concepts/firewall-rules
535    """
536    if 'id' in self._resource_data:
537      return self._resource_data['id'][:8]
538    raise UndefinedClusterPropertyError('no id')
539
540  @property
541  def is_nodelocal_dnscache_enabled(self) -> bool:
542    """Returns True if NodeLocal DNSCache is enabled for the cluster."""
543    addons_config = self._resource_data.get('addonsConfig', {})
544    dns_cache_config = addons_config.get('dnsCacheConfig', {})
545    return dns_cache_config.get('enabled', False)