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)

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

391@dataclasses.dataclass
392class Peering:
393  """VPC Peerings"""
394
395  name: str
396  url: str
397  state: str
398  exports_custom_routes: bool
399  imports_custom_routes: bool
400  auto_creates_routes: bool
401
402  def __str__(self):
403    return self.name

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

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

592class FirewallRuleNotFoundError(Exception):
593  rule_name: str
594
595  def __init__(self, name, disabled=False):
596    # Call the base class constructor with the parameters it needs
597    super().__init__(f'firewall rule not found: {name}')
598    self.rule_name = name
599    self.disabled = disabled

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