gcpdiag.queries.network

35class Subnetwork(models.Resource):
36  """A VPC subnetwork."""
37
38  _resource_data: dict
39
40  def __init__(self, project_id, resource_data):
41    super().__init__(project_id=project_id)
42    self._resource_data = resource_data
43
44  @property
45  def full_path(self) -> str:
46    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)',
47                      self.self_link)
48    if result:
49      return result.group(1)
50    else:
51      return f'>> {self.self_link}'
52
53  @property
54  def short_path(self) -> str:
55    path = self.project_id + '/' + self.name
56    return path
57
58  @property
59  def name(self) -> str:
60    return self._resource_data['name']
61
62  @property
63  def self_link(self) -> str:
64    return self._resource_data.get('selfLink', '')
65
66  @property
67  def ip_network(self) -> IPv4NetOrIPv6Net:
68    return ipaddress.ip_network(self._resource_data['ipCidrRange'])
69
70  @property
71  def region(self) -> str:
72    # https://www.googleapis.com/compute/v1/projects/gcpdiag-gke1-aaaa/regions/europe-west4
73    m = re.match(
74        r'https://www.googleapis.com/compute/v1/projects/([^/]+)/regions/([^/]+)',
75        self._resource_data['region'])
76    if not m:
77      raise RuntimeError(
78          f"can't parse region URL: {self._resource_data['region']}")
79    return m.group(2)
80
81  def is_private_ip_google_access(self) -> bool:
82    return self._resource_data.get('privateIpGoogleAccess', False)
83
84  @property
85  def network(self):
86    return self._resource_data['network']

 89class Route(models.Resource):
 90  """A VPC Route."""
 91
 92  _resource_data: dict
 93
 94  def __init__(self, project_id, resource_data):
 95    super().__init__(project_id=project_id)
 96    self._resource_data = resource_data
 97
 98  @property
 99  def full_path(self) -> str:
100    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)',
101                      self.self_link)
102    if result:
103      return result.group(1)
104    else:
105      return f'>> {self.self_link}'
106
107  @property
108  def short_path(self) -> str:
109    path = self.project_id + '/' + self.name
110    return path
111
112  @property
113  def name(self) -> str:
114    return self._resource_data['name']
115
116  @property
117  def kind(self) -> str:
118    return self._resource_data['kind']
119
120  @property
121  def self_link(self) -> str:
122    return self._resource_data['selfLink']
123
124  @property
125  def network(self) -> str:
126    return self._resource_data['network']
127
128  @property
129  def tags(self) -> List[str]:
130    if 'tags' in self._resource_data:
131      return self._resource_data['tags']
132    return []
133
134  @property
135  def dest_range(self) -> str:
136    return self._resource_data['destRange']
137
138  @property
139  def next_hop_gateway(self) -> Optional[str]:
140    if 'nextHopGateway' in self._resource_data:
141      return self._resource_data['nextHopGateway']
142    return None
143
144  @property
145  def next_hop_vpn_tunnel(self) -> Optional[str]:
146    return self._resource_data.get('nextHopVpnTunnel')
147
148  @property
149  def next_hop_hub(self) -> Optional[str]:
150    return self._resource_data.get('nextHopHub')
151
152  @property
153  def priority(self) -> int:
154    return self._resource_data['priority']
155
156  def get_next_hop(self) -> Union[Dict[str, Any], Optional[str]]:
157    hop_types = {
158        'nextHopGateway': 'nextHopGateway',
159        'nextHopVpnTunnel': 'nextHopVpnTunnel',
160        'nextHopHub': 'nextHopHub',
161        'nextHopInstance': 'nextHopInstance',
162        'nextHopAddress': 'nextHopAddress',
163        'nextHopPeering': 'nextHopPeering',
164        'nextHopIlb': 'nextHopIlb',
165        'nextHopNetwork': 'nextHopNetwork',
166        'nextHopIp': 'nextHopIp'
167    }
168
169    for hop_type, value in hop_types.items():
170      if self._resource_data.get(hop_type):
171        return {'type': value, 'link': self._resource_data[hop_type]}
172    return None
173
174  def check_route_match(self, ip1: IPAddrOrNet, ip2: str) -> bool:
175    ip2_list = [ipaddress.ip_network(ip2)]
176    if _ip_match(ip1, ip2_list, 'allow'):
177      return True
178    return False

181class ManagedZone(models.Resource):
182  """
183  Represent a DNS zone (public or private
184
185  https://cloud.google.com/dns/docs/reference/v1beta2/managedZones
186  """
187  _resource_data: dict
188
189  def __init__(self, project_id, resource_data):
190    super().__init__(project_id=project_id)
191    self._resource_data = resource_data
192
193  @property
194  def cloud_logging_config(self) -> bool:
195    return self._resource_data['cloudLoggingConfig'].get('enableLogging', False)
196
197  @property
198  def is_public(self) -> bool:
199    return self._resource_data['visibility'] == 'public'
200
201  @property
202  def vpc_attached(self) -> bool:
203    if 'privateVisibilityConfig' not in self._resource_data:
204      self._resource_data['privateVisibilityConfig'] = {}
205
206    return (self._resource_data['privateVisibilityConfig'].get(
207        'networks', False) or
208            self._resource_data['privateVisibilityConfig'].get(
209                'gkeClusters', False))
210
211  @property
212  def dnssec_config_state(self) -> bool:
213    if 'dnssecConfig' not in self._resource_data:
214      self._resource_data['dnssecConfig'] = {}
215
216    return self._resource_data['dnssecConfig'].get('state', False)
217
218  @property
219  def name(self) -> str:
220    return self._resource_data['name']
221
222  @property
223  def full_path(self) -> str:
224    result = re.match(r'https://dns.googleapis.com/dns/v1beta2/(.*)',
225                      self.self_link)
226    if result:
227      return result.group(1)
228    else:
229      return f'>> {self.self_link}'
230
231  @property
232  def short_path(self) -> str:
233    path = self.project_id + '/' + self.name
234    return path
235
236  @property
237  def self_link(self) -> str:
238    return self._resource_data.get('selfLink', '')

241class Router(models.Resource):
242  """A VPC Router."""
243
244  _resource_data: dict
245
246  def __init__(self, project_id, resource_data):
247    super().__init__(project_id=project_id)
248    self._resource_data = resource_data
249    self._nats = None
250
251  @property
252  def full_path(self) -> str:
253    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)',
254                      self.self_link)
255    if result:
256      return result.group(1)
257    else:
258      return f'>> {self.self_link}'
259
260  @property
261  def short_path(self) -> str:
262    path = self.project_id + '/' + self.name
263    return path
264
265  @property
266  def name(self) -> str:
267    return self._resource_data['name']
268
269  @property
270  def self_link(self) -> str:
271    return self._resource_data.get('selfLink', '')
272
273  @property
274  def network(self) -> str:
275    return self._resource_data['network']
276
277  def get_network_name(self) -> str:
278    logging.info('inside get_network_name function')
279    if self._resource_data['network']:
280      return self._resource_data['network'].split('/')[-1]
281    return ''
282
283  @property
284  def nats(self):
285    return self._resource_data.get('nats', [])
286
287  def get_nat_ip_allocate_option(self, nat_gateway) -> str:
288    nats = self._resource_data.get('nats', [])
289    nat = [n for n in nats if n['name'] == nat_gateway]
290    return nat[0].get('natIpAllocateOption', '')
291
292  def get_enable_dynamic_port_allocation(self, nat_gateway) -> str:
293    nats = self._resource_data.get('nats', [])
294    nat = [n for n in nats if n['name'] == nat_gateway]
295    return nat[0].get('enableDynamicPortAllocation', '')
296
297  def subnet_has_nat(self, subnetwork):
298    if not self._resource_data.get('nats', []):
299      return False
300    for n in self._resource_data.get('nats', []):
301      if n['sourceSubnetworkIpRangesToNat'] == 'LIST_OF_SUBNETWORKS':
302        # Cloud NAT configure for specific subnets
303        if 'subnetworks' in n and subnetwork.self_link in [
304            s['name'] for s in n['subnetworks']
305        ]:
306          return True
307      else:
308        # Cloud NAT configured for all subnets
309        return True
310    return False

313class RouterStatus(models.Resource):
314  """NAT Router Status"""
315
316  _resource_data: dict
317
318  def __init__(self, project_id, resource_data):
319    super().__init__(project_id=project_id)
320    self._resource_data = resource_data
321
322  @property
323  def full_path(self) -> str:
324    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)',
325                      self.self_link)
326    if result:
327      return result.group(1)
328    else:
329      return f'>> {self.self_link}'
330
331  @property
332  def short_path(self) -> str:
333    path = self.project_id + '/' + self.name
334    return path
335
336  @property
337  def name(self) -> str:
338    return self._resource_data.get('name', '')
339
340  @property
341  def self_link(self) -> str:
342    return self._resource_data.get('selfLink', '')
343
344  @property
345  def min_extra_nat_ips_needed(self) -> str:
346    nat_status = self._resource_data.get('result', {}).get('natStatus', {})
347    return nat_status[0].get('minExtraNatIpsNeeded', None)
348
349  @property
350  def num_vms_with_nat_mappings(self) -> str:
351    nat_status = self._resource_data.get('result', {}).get('natStatus', {})
352    return nat_status[0].get('numVmEndpointsWithNatMappings', None)
353
354  @property
355  def bgp_peer_status(self) -> str:
356    bgp_peer_status = self._resource_data.get('result',
357                                              {}).get('bgpPeerStatus', {})
358    return bgp_peer_status

361class RouterNatIpInfo(models.Resource):
362  """NAT IP Info"""
363
364  _resource_data: dict
365
366  def __init__(self, project_id, resource_data):
367    super().__init__(project_id=project_id)
368    self._resource_data = resource_data
369
370  @property
371  def full_path(self) -> str:
372    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)',
373                      self.self_link)
374    if result:
375      return result.group(1)
376    else:
377      return f'>> {self.self_link}'
378
379  @property
380  def short_path(self) -> str:
381    path = self.project_id + '/' + self.name
382    return path
383
384  @property
385  def self_link(self) -> str:
386    return self._resource_data.get('selfLink', '')
387
388  @property
389  def name(self) -> str:
390    return self._resource_data.get('name', '')
391
392  @property
393  def result(self) -> str:
394    return self._resource_data.get('result', [])

397@dataclasses.dataclass
398class Peering:
399  """VPC Peerings"""
400
401  name: str
402  url: str
403  state: str
404  exports_custom_routes: bool
405  imports_custom_routes: bool
406  auto_creates_routes: bool
407
408  def __str__(self):
409    return self.name

412class Network(models.Resource):
413  """A VPC network."""
414  _resource_data: dict
415  _subnetworks: Optional[Dict[str, Subnetwork]]
416
417  def __init__(self, project_id, resource_data):
418    super().__init__(project_id=project_id)
419    self._resource_data = resource_data
420    self._subnetworks = None
421
422  @property
423  def full_path(self) -> str:
424    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)',
425                      self.self_link)
426    if result:
427      return result.group(1)
428    else:
429      return f'>> {self.self_link}'
430
431  @property
432  def short_path(self) -> str:
433    path = self.project_id + '/' + self.name
434    return path
435
436  @property
437  def name(self) -> str:
438    return self._resource_data['name']
439
440  @property
441  def self_link(self) -> str:
442    return self._resource_data.get('selfLink', '')
443
444  @property
445  def firewall(self) -> 'EffectiveFirewalls':
446    return _get_effective_firewalls(self)
447
448  @property
449  def mtu(self) -> int:
450    if 'mtu' in self._resource_data:
451      return self._resource_data['mtu']
452    return DEFAULT_MTU
453
454  @property
455  def subnetworks(self) -> Dict[str, Subnetwork]:
456    return _batch_get_subnetworks(
457        self._project_id, frozenset(self._resource_data.get('subnetworks', [])))
458
459  @property
460  def peerings(self) -> List[Peering]:
461    return [
462        Peering(peer['name'], peer['network'], peer['state'],
463                peer['exportCustomRoutes'], peer['importCustomRoutes'],
464                peer['autoCreateRoutes'])
465        for peer in self._resource_data.get('peerings', [])
466    ]

574@dataclasses.dataclass
575class FirewallCheckResult:
576  """The result of a firewall connectivity check."""
577
578  action: str
579  firewall_policy_name: Optional[str] = None
580  firewall_policy_rule_description: Optional[str] = None
581  vpc_firewall_rule_id: Optional[str] = None
582  vpc_firewall_rule_name: Optional[str] = None
583
584  def __str__(self):
585    return self.action
586
587  @property
588  def matched_by_str(self):
589    if self.firewall_policy_name:
590      if self.firewall_policy_rule_description:
591        return f'policy: {self.firewall_policy_name}, rule: {self.firewall_policy_rule_description}'
592      else:
593        return f'policy: {self.firewall_policy_name}'
594    elif self.vpc_firewall_rule_name:
595      return f'vpc firewall rule: {self.vpc_firewall_rule_name}'

598class FirewallRuleNotFoundError(Exception):
599  rule_name: str
600
601  def __init__(self, name, disabled=False):
602    # Call the base class constructor with the parameters it needs
603    super().__init__(f'firewall rule not found: {name}')
604    self.rule_name = name
605    self.disabled = disabled

608class VpcFirewallRule:
609  """Represents firewall rule"""
610
611  def __init__(self, resource_data):
612    self._resource_data = resource_data
613
614  @property
615  def name(self) -> str:
616    return self._resource_data['name']
617
618  @property
619  def source_ranges(self) -> List[ipaddress.IPv4Network]:
620    return self._resource_data['sourceRanges']
621
622  @property
623  def target_tags(self) -> set:
624    return self._resource_data['targetTags']
625
626  @property
627  def allowed(self) -> List[dict]:
628    return self._resource_data['allowed']
629
630  def is_enabled(self) -> bool:
631    return not self._resource_data['disabled']