gcpdiag.queries.gke

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

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

233class UndefinedClusterPropertyError(Exception):
234  """Thrown when a property of a cluster can't be determined for some reason.
235
236  For example, the cluster_hash can't be determined because there are no
237  nodepools defined.
238  """
239
240  pass

243class Cluster(models.Resource):
244  """Represents a GKE cluster.
245
246  https://cloud.google.com/kubernetes-engine/docs/reference/rest/v1/projects.locations.clusters#Cluster
247  """
248
249  _resource_data: dict
250  master_version: Version
251  _context: models.Context
252  _nodepools: Optional[List[NodePool]]
253
254  def __init__(self, project_id, resource_data, context: models.Context):
255    super().__init__(project_id=project_id)
256    self._resource_data = resource_data
257    self.master_version = Version(self._resource_data['currentMasterVersion'])
258    self._nodepools = None
259    self._context = context
260
261  @property
262  def full_path(self) -> str:
263    if utils.is_region(self._resource_data['location']):
264      return f'projects/{self.project_id}/locations/{self.location}/clusters/{self.name}'
265    else:
266      return f'projects/{self.project_id}/zones/{self.location}/clusters/{self.name}'
267
268  @property
269  def short_path(self) -> str:
270    path = self.full_path
271    path = re.sub(r'^projects/', '', path)
272    path = re.sub(r'/locations/', '/', path)
273    path = re.sub(r'/zones/', '/', path)
274    path = re.sub(r'/clusters/', '/', path)
275    return path
276
277  @property
278  def name(self) -> str:
279    return self._resource_data['name']
280
281  @property
282  def location(self) -> str:
283    return self._resource_data['location']
284
285  @property
286  def pod_ipv4_cidr(self) -> IPv4NetOrIPv6Net:
287    cidr = self._resource_data['clusterIpv4Cidr']
288    return ipaddress.ip_network(cidr)
289
290  @property
291  def current_node_count(self) -> int:
292    return self._resource_data.get('currentNodeCount', 0)
293
294  @property
295  def release_channel(self) -> Optional[str]:
296    try:
297      return self._resource_data['releaseChannel']['channel']
298    except KeyError:
299      return None
300
301  @property
302  def nap_node_image_type(self) -> Optional[str]:
303    return get_path(
304      self._resource_data,
305      ('autoscaling', 'autoprovisioningNodePoolDefaults', 'imageType'),
306      default=None,
307    )
308
309  @property
310  def app_layer_sec_key(self) -> str:
311    return self._resource_data['databaseEncryption'].get('keyName')
312
313  @property
314  def status(self) -> str:
315    return self._resource_data['status']
316
317  @property
318  def status_message(self) -> str:
319    return self._resource_data.get('statusMessage', None)
320
321  def has_app_layer_enc_enabled(self) -> bool:
322    # state := 'DECRYPTED' | 'ENCRYPTED', keyName := 'full_path_to_key_resouce'
323    return (
324      get_path(self._resource_data, ('databaseEncryption', 'state'), default=None) == 'ENCRYPTED'
325    )
326
327  def has_logging_enabled(self) -> bool:
328    return self._resource_data['loggingService'] != 'none'
329
330  def enabled_logging_components(self) -> List[str]:
331    return self._resource_data['loggingConfig']['componentConfig']['enableComponents']
332
333  def has_monitoring_enabled(self) -> bool:
334    return self._resource_data['monitoringService'] != 'none'
335
336  def enabled_monitoring_components(self) -> List[str]:
337    return self._resource_data['monitoringConfig']['componentConfig']['enableComponents']
338
339  def has_control_plane_logging_enabled(self) -> bool:
340    if not self.has_logging_enabled():
341      return False
342    components = get_path(
343      self._resource_data, ('loggingConfig', 'componentConfig', 'enableComponents'), default=[]
344    )
345    return all(c in components for c in ['API_SERVER', 'SCHEDULER', 'CONTROLLER_MANAGER'])
346
347  def has_control_plane_monitoring_enabled(self) -> bool:
348    if not self.has_monitoring_enabled():
349      return False
350    components = get_path(
351      self._resource_data, ('monitoringConfig', 'componentConfig', 'enableComponents'), default=[]
352    )
353    return all(c in components for c in ['API_SERVER', 'SCHEDULER', 'CONTROLLER_MANAGER'])
354
355  def has_authenticator_group_enabled(self) -> bool:
356    return len(self._resource_data.get('authenticatorGroupsConfig', {})) > 0
357
358  def has_workload_identity_enabled(self) -> bool:
359    return len(self._resource_data.get('workloadIdentityConfig', {})) > 0
360
361  def has_http_load_balancing_enabled(self) -> bool:
362    # HTTP load balancing needs to be enabled to use GKE ingress
363    return (
364      get_path(
365        self._resource_data,
366        ('addonsConfig', 'httpLoadBalancing', 'disabled'),
367        default=None,
368      )
369      is not True
370    )
371
372  def has_network_policy_enabled(self) -> bool:
373    # Network policy enforcement
374    return (
375      get_path(
376        self._resource_data,
377        ('addonsConfig', 'networkPolicyConfig', 'disabled'),
378        default=False,
379      )
380      is not True
381    )
382
383  def has_dpv2_enabled(self) -> bool:
384    # Checks whether dataplane V2 is enabled in clusters
385    return (
386      get_path(
387        self._resource_data,
388        ('networkConfig', 'datapathProvider'),
389        default=None,
390      )
391      == 'ADVANCED_DATAPATH'
392    )
393
394  def has_intra_node_visibility_enabled(self) -> bool:
395    if (
396      'networkConfig' in self._resource_data
397      and 'enableIntraNodeVisibility' in self._resource_data['networkConfig']
398    ):
399      return self._resource_data['networkConfig']['enableIntraNodeVisibility']
400    return False
401
402  def has_maintenance_window(self) -> bool:
403    # 'e3b0c442' is a hexadecimal string that represents the value of an empty
404    # string ('') in cryptography. If the maintenance windows are defined, the
405    # value of 'resourceVersion' is not empty ('e3b0c442').
406    return self._resource_data['maintenancePolicy']['resourceVersion'] != 'e3b0c442'
407
408  @property
409  def maintenance_policy(self) -> dict:
410    """Returns the maintenance policy of the cluster."""
411    return self._resource_data.get('maintenancePolicy', {})
412
413  def has_image_streaming_enabled(self) -> bool:
414    """Check if cluster has Image Streaming (aka  Google Container File System)
415
416    enabled
417    """
418    global_gcsfs = get_path(
419      self._resource_data,
420      ('nodePoolDefaults', 'nodeConfigDefaults', 'gcfsConfig', 'enabled'),
421      default=False,
422    )
423    # Check nodePoolDefaults settings
424    if global_gcsfs:
425      return True
426    for np in self.nodepools:
427      # Check if any nodepool has image streaming enabled
428      if np.has_image_streaming_enabled():
429        return True
430    return False
431
432  @property
433  def nodepools(self) -> List[NodePool]:
434    if self._nodepools is None:
435      self._nodepools = []
436      for n in self._resource_data.get('nodePools', []):
437        self._nodepools.append(NodePool(self, n))
438    return self._nodepools
439
440  @property
441  def network(self) -> network.Network:
442    # projects/gcpdiag-gke1-aaaa/global/networks/default
443    network_string = self._resource_data['networkConfig']['network']
444    m = re.match(r'projects/([^/]+)/global/networks/([^/]+)$', network_string)
445    if not m:
446      raise RuntimeError("can't parse network string: %s" % network_string)
447    return network.get_network(m.group(1), m.group(2), self._context)
448
449  @property
450  def subnetwork(self) -> Optional[models.Resource]:
451    # 'projects/gcpdiag-gke1-aaaa/regions/europe-west4/subnetworks/default'
452    if 'subnetwork' not in self._resource_data['networkConfig']:
453      return None
454
455    subnetwork_string = self._resource_data['networkConfig']['subnetwork']
456    m = re.match(
457      r'projects/([^/]+)/regions/([^/]+)/subnetworks/([^/]+)$',
458      subnetwork_string,
459    )
460    if not m:
461      raise RuntimeError("can't parse network string: %s" % subnetwork_string)
462    return network.get_subnetwork(m.group(1), m.group(2), m.group(3))
463
464  @property
465  def get_subnet_name(self) -> Optional[models.Resource]:
466    if 'subnetwork' not in self._resource_data:
467      return None
468    return self._resource_data['subnetwork']
469
470  @property
471  def get_nodepool_config(self) -> Optional[models.Resource]:
472    if 'nodePools' not in self._resource_data:
473      return None
474    return self._resource_data['nodePools']
475
476  @property
477  def get_network_string(self) -> str:
478    if 'networkConfig' not in self._resource_data:
479      return ''
480    if 'network' not in self._resource_data['networkConfig']:
481      return ''
482    return self._resource_data['networkConfig']['network']
483
484  @property
485  def is_private(self) -> bool:
486    if 'privateClusterConfig' not in self._resource_data:
487      return False
488
489    return self._resource_data['privateClusterConfig'].get('enablePrivateNodes', False)
490
491  @property
492  def is_vpc_native(self) -> bool:
493    return get_path(
494      self._resource_data,
495      ('ipAllocationPolicy', 'useIpAliases'),
496      default=False,
497    )
498
499  @property
500  def is_regional(self) -> bool:
501    return len(self._resource_data['locations']) > 1
502
503  @property
504  def cluster_ca_certificate(self) -> str:
505    return self._resource_data['masterAuth']['clusterCaCertificate']
506
507  @property
508  def cluster_ca_certificate_info(self) -> Optional[Dict]:
509    """Parses and returns details about the cluster CA certificates.
510
511    Returns:
512      A dictionary containing raw certs, parsed certificate list, and a summary
513      of validity dates. Returns None if no certificate is available or if
514      parsing fails.
515    """
516    cert_str = self.cluster_ca_certificate
517    if not cert_str:
518      return None
519
520    try:
521      if '-----BEGIN CERTIFICATE-----' in cert_str:
522        pem_str = cert_str
523      else:
524        decoded_bytes = base64.b64decode(cert_str)
525        try:
526          pem_str = decoded_bytes.decode('utf-8')
527        except UnicodeDecodeError:
528          cert = x509.load_der_x509_certificate(decoded_bytes)
529          certs = [cert]
530          pem_str = None
531
532      if pem_str is not None:
533        certs = []
534        for block in pem_str.split('-----BEGIN CERTIFICATE-----'):
535          if '-----END CERTIFICATE-----' in block:
536            pem_block = f'-----BEGIN CERTIFICATE-----{block}'
537            certs.append(x509.load_pem_x509_certificate(pem_block.encode('utf-8')))
538
539      if not certs:
540        return None
541
542      parsed_certs = []
543      for cert in certs:
544        pem_bytes = cert.public_bytes(serialization.Encoding.PEM)
545        parsed_certs.append(
546          {
547            'pem': pem_bytes.decode('utf-8'),
548            'subject': cert.subject.rfc4514_string(),
549            'issuer': cert.issuer.rfc4514_string(),
550            'notBefore': cert.not_valid_before_utc.isoformat(),
551            'notAfter': cert.not_valid_after_utc.isoformat(),
552          }
553        )
554
555      earliest_not_before = min(certs, key=lambda c: c.not_valid_before_utc).not_valid_before_utc
556      latest_not_after = max(certs, key=lambda c: c.not_valid_after_utc).not_valid_after_utc
557
558      return {
559        'raw_certs': cert_str,
560        'certificates': parsed_certs,
561        'summary': {
562          'earliestNotBefore': earliest_not_before.isoformat(),
563          'latestNotAfter': latest_not_after.isoformat(),
564        },
565      }
566    except Exception as e:  # pylint: disable=broad-exception-caught
567      logging.warning('Failed to parse CA certificate for cluster %s: %s', self.name, e)
568      return None
569
570  @property
571  def endpoint(self) -> Optional[str]:
572    if 'endpoint' not in self._resource_data:
573      return None
574    return self._resource_data['endpoint']
575
576  @property
577  def is_autopilot(self) -> bool:
578    if 'autopilot' not in self._resource_data:
579      return False
580    return self._resource_data['autopilot'].get('enabled', False)
581
582  @property
583  def masters_cidr_list(self) -> Iterable[IPv4NetOrIPv6Net]:
584    if get_path(
585      self._resource_data,
586      ('privateClusterConfig', 'masterIpv4CidrBlock'),
587      default=None,
588    ):
589      return [
590        ipaddress.ip_network(self._resource_data['privateClusterConfig']['masterIpv4CidrBlock'])
591      ]
592    else:
593      # only older clusters still have ssh firewall rules
594      if self.current_node_count and not self.cluster_hash:
595        logging.warning("couldn't retrieve cluster hash for cluster %s.", self.name)
596        return []
597      fw_rule_name = f'gke-{self.name}-{self.cluster_hash}-ssh'
598      rule = self.network.firewall.get_vpc_ingress_rules(name=fw_rule_name)
599      if rule and rule[0].is_enabled():
600        return rule[0].source_ranges
601      return []
602
603  @property
604  def cluster_hash(self) -> Optional[str]:
605    """Returns the "cluster hash" as used in automatic firewall rules for GKE clusters.
606
607    Cluster hash is the first 8 characters of cluster id. See also:
608    https://cloud.google.com/kubernetes-engine/docs/concepts/firewall-rules
609    """
610    if 'id' in self._resource_data:
611      return self._resource_data['id'][:8]
612    raise UndefinedClusterPropertyError('no id')
613
614  @property
615  def is_nodelocal_dnscache_enabled(self) -> bool:
616    """Returns True if NodeLocal DNSCache is enabled for the cluster."""
617    addons_config = self._resource_data.get('addonsConfig', {})
618    dns_cache_config = addons_config.get('dnsCacheConfig', {})
619    return dns_cache_config.get('enabled', False)
620
621  @property
622  def dns_provider(self) -> str:
623    """Returns GKE DNS provider ('CLOUD_DNS', 'KUBE_DNS', etc.)."""
624    dns_config = self._resource_data.get('networkConfig', {}).get('dnsConfig', {})
625    cluster_dns = dns_config.get('clusterDns')
626    if cluster_dns:
627      return cluster_dns
628
629    if self.is_autopilot:
630      version_tuple = (
631        self.master_version.major,
632        self.master_version.minor,
633        self.master_version.patch,
634      )
635      if version_tuple >= (1, 25, 9):
636        return 'CLOUD_DNS'
637
638    return 'KUBE_DNS'