gcpdiag.queries.network

Queries related to VPC Networks.
IPv4AddrOrIPv6Addr = typing.Union[ipaddress.IPv4Address, ipaddress.IPv6Address]
IPv4NetOrIPv6Net = typing.Union[ipaddress.IPv4Network, ipaddress.IPv6Network]
IPAddrOrNet = typing.Union[ipaddress.IPv4Address, ipaddress.IPv6Address, ipaddress.IPv4Network, ipaddress.IPv6Network]
DEFAULT_MTU = 1460
class Subnetwork(gcpdiag.models.Resource): 
34class Subnetwork(models.Resource):
35  """A VPC subnetwork."""
36
37  _resource_data: dict
38  _context: models.Context
39
40  def __init__(self, project_id, resource_data, context: models.Context):
41    super().__init__(project_id=project_id)
42    self._resource_data = resource_data
43    self._context = context
44
45  @property
46  def full_path(self) -> str:
47    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', 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    )
77    if not m:
78      raise RuntimeError(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']

A VPC subnetwork.

Subnetwork(project_id, resource_data, context: gcpdiag.models.Context) 
40  def __init__(self, project_id, resource_data, context: models.Context):
41    super().__init__(project_id=project_id)
42    self._resource_data = resource_data
43    self._context = context
full_path: str 
45  @property
46  def full_path(self) -> str:
47    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', self.self_link)
48    if result:
49      return result.group(1)
50    else:
51      return f'>> {self.self_link}'

Returns the full path of this resource.

Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'

short_path: str 
53  @property
54  def short_path(self) -> str:
55    path = self.project_id + '/' + self.name
56    return path

Returns the short name for this resource.

Note that it isn't clear from this name what kind of resource it is.

Example: 'gke1'

name: str 
58  @property
59  def name(self) -> str:
60    return self._resource_data['name']
ip_network: Union[ipaddress.IPv4Network, ipaddress.IPv6Network] 
66  @property
67  def ip_network(self) -> IPv4NetOrIPv6Net:
68    return ipaddress.ip_network(self._resource_data['ipCidrRange'])
region: str 
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    )
77    if not m:
78      raise RuntimeError(f"can't parse region URL: {self._resource_data['region']}")
79    return m.group(2)
def is_private_ip_google_access(self) -> bool: 
81  def is_private_ip_google_access(self) -> bool:
82    return self._resource_data.get('privateIpGoogleAccess', False)
network 
84  @property
85  def network(self):
86    return self._resource_data['network']
class Route(gcpdiag.models.Resource): 
 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/(.*)', self.self_link)
101    if result:
102      return result.group(1)
103    else:
104      return f'>> {self.self_link}'
105
106  @property
107  def short_path(self) -> str:
108    path = self.project_id + '/' + self.name
109    return path
110
111  @property
112  def name(self) -> str:
113    return self._resource_data['name']
114
115  @property
116  def kind(self) -> str:
117    return self._resource_data['kind']
118
119  @property
120  def self_link(self) -> str:
121    return self._resource_data['selfLink']
122
123  @property
124  def network(self) -> str:
125    return self._resource_data['network']
126
127  @property
128  def tags(self) -> List[str]:
129    if 'tags' in self._resource_data:
130      return self._resource_data['tags']
131    return []
132
133  @property
134  def dest_range(self) -> str:
135    return self._resource_data['destRange']
136
137  @property
138  def next_hop_gateway(self) -> Optional[str]:
139    if 'nextHopGateway' in self._resource_data:
140      return self._resource_data['nextHopGateway']
141    return None
142
143  @property
144  def next_hop_vpn_tunnel(self) -> Optional[str]:
145    return self._resource_data.get('nextHopVpnTunnel')
146
147  @property
148  def next_hop_hub(self) -> Optional[str]:
149    return self._resource_data.get('nextHopHub')
150
151  @property
152  def priority(self) -> int:
153    return self._resource_data['priority']
154
155  def get_next_hop(self) -> Union[Dict[str, Any], Optional[str]]:
156    hop_types = {
157      'nextHopGateway': 'nextHopGateway',
158      'nextHopVpnTunnel': 'nextHopVpnTunnel',
159      'nextHopHub': 'nextHopHub',
160      'nextHopInstance': 'nextHopInstance',
161      'nextHopAddress': 'nextHopAddress',
162      'nextHopPeering': 'nextHopPeering',
163      'nextHopIlb': 'nextHopIlb',
164      'nextHopNetwork': 'nextHopNetwork',
165      'nextHopIp': 'nextHopIp',
166    }
167
168    for hop_type, value in hop_types.items():
169      if self._resource_data.get(hop_type):
170        return {'type': value, 'link': self._resource_data[hop_type]}
171    return None
172
173  def check_route_match(self, ip1: IPAddrOrNet, ip2: str) -> bool:
174    ip2_list = [ipaddress.ip_network(ip2)]
175    if _ip_match(ip1, ip2_list, 'allow'):
176      return True
177    return False

A VPC Route.

Route(project_id, resource_data) 
94  def __init__(self, project_id, resource_data):
95    super().__init__(project_id=project_id)
96    self._resource_data = resource_data
full_path: str 
 98  @property
 99  def full_path(self) -> str:
100    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', self.self_link)
101    if result:
102      return result.group(1)
103    else:
104      return f'>> {self.self_link}'

Returns the full path of this resource.

Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'

short_path: str 
106  @property
107  def short_path(self) -> str:
108    path = self.project_id + '/' + self.name
109    return path

Returns the short name for this resource.

Note that it isn't clear from this name what kind of resource it is.

Example: 'gke1'

name: str 
111  @property
112  def name(self) -> str:
113    return self._resource_data['name']
kind: str 
115  @property
116  def kind(self) -> str:
117    return self._resource_data['kind']
network: str 
123  @property
124  def network(self) -> str:
125    return self._resource_data['network']
tags: List[str] 
127  @property
128  def tags(self) -> List[str]:
129    if 'tags' in self._resource_data:
130      return self._resource_data['tags']
131    return []
dest_range: str 
133  @property
134  def dest_range(self) -> str:
135    return self._resource_data['destRange']
next_hop_gateway: Optional[str] 
137  @property
138  def next_hop_gateway(self) -> Optional[str]:
139    if 'nextHopGateway' in self._resource_data:
140      return self._resource_data['nextHopGateway']
141    return None
next_hop_vpn_tunnel: Optional[str] 
143  @property
144  def next_hop_vpn_tunnel(self) -> Optional[str]:
145    return self._resource_data.get('nextHopVpnTunnel')
next_hop_hub: Optional[str] 
147  @property
148  def next_hop_hub(self) -> Optional[str]:
149    return self._resource_data.get('nextHopHub')
priority: int 
151  @property
152  def priority(self) -> int:
153    return self._resource_data['priority']
def get_next_hop(self) -> Union[Dict[str, Any], str, NoneType]: 
155  def get_next_hop(self) -> Union[Dict[str, Any], Optional[str]]:
156    hop_types = {
157      'nextHopGateway': 'nextHopGateway',
158      'nextHopVpnTunnel': 'nextHopVpnTunnel',
159      'nextHopHub': 'nextHopHub',
160      'nextHopInstance': 'nextHopInstance',
161      'nextHopAddress': 'nextHopAddress',
162      'nextHopPeering': 'nextHopPeering',
163      'nextHopIlb': 'nextHopIlb',
164      'nextHopNetwork': 'nextHopNetwork',
165      'nextHopIp': 'nextHopIp',
166    }
167
168    for hop_type, value in hop_types.items():
169      if self._resource_data.get(hop_type):
170        return {'type': value, 'link': self._resource_data[hop_type]}
171    return None
def check_route_match( self, ip1: Union[ipaddress.IPv4Address, ipaddress.IPv6Address, ipaddress.IPv4Network, ipaddress.IPv6Network], ip2: str) -> bool: 
173  def check_route_match(self, ip1: IPAddrOrNet, ip2: str) -> bool:
174    ip2_list = [ipaddress.ip_network(ip2)]
175    if _ip_match(ip1, ip2_list, 'allow'):
176      return True
177    return False
class ManagedZone(gcpdiag.models.Resource): 
180class ManagedZone(models.Resource):
181  """
182  Represent a DNS zone (public or private
183
184  https://cloud.google.com/dns/docs/reference/v1beta2/managedZones
185  """
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
208    ) or self._resource_data['privateVisibilityConfig'].get('gkeClusters', False)
209
210  @property
211  def dnssec_config_state(self) -> bool:
212    if 'dnssecConfig' not in self._resource_data:
213      self._resource_data['dnssecConfig'] = {}
214
215    return self._resource_data['dnssecConfig'].get('state', False)
216
217  @property
218  def name(self) -> str:
219    return self._resource_data['name']
220
221  @property
222  def full_path(self) -> str:
223    result = re.match(r'https://dns.googleapis.com/dns/v1beta2/(.*)', self.self_link)
224    if result:
225      return result.group(1)
226    else:
227      return f'>> {self.self_link}'
228
229  @property
230  def short_path(self) -> str:
231    path = self.project_id + '/' + self.name
232    return path
233
234  @property
235  def self_link(self) -> str:
236    return self._resource_data.get('selfLink', '')

Represent a DNS zone (public or private

https://cloud.google.com/dns/docs/reference/v1beta2/managedZones

ManagedZone(project_id, resource_data) 
189  def __init__(self, project_id, resource_data):
190    super().__init__(project_id=project_id)
191    self._resource_data = resource_data
cloud_logging_config: bool 
193  @property
194  def cloud_logging_config(self) -> bool:
195    return self._resource_data['cloudLoggingConfig'].get('enableLogging', False)
is_public: bool 
197  @property
198  def is_public(self) -> bool:
199    return self._resource_data['visibility'] == 'public'
vpc_attached: bool 
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
208    ) or self._resource_data['privateVisibilityConfig'].get('gkeClusters', False)
dnssec_config_state: bool 
210  @property
211  def dnssec_config_state(self) -> bool:
212    if 'dnssecConfig' not in self._resource_data:
213      self._resource_data['dnssecConfig'] = {}
214
215    return self._resource_data['dnssecConfig'].get('state', False)
name: str 
217  @property
218  def name(self) -> str:
219    return self._resource_data['name']
full_path: str 
221  @property
222  def full_path(self) -> str:
223    result = re.match(r'https://dns.googleapis.com/dns/v1beta2/(.*)', self.self_link)
224    if result:
225      return result.group(1)
226    else:
227      return f'>> {self.self_link}'

Returns the full path of this resource.

Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'

short_path: str 
229  @property
230  def short_path(self) -> str:
231    path = self.project_id + '/' + self.name
232    return path

Returns the short name for this resource.

Note that it isn't clear from this name what kind of resource it is.

Example: 'gke1'

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

A VPC Router.

Router(project_id, resource_data) 
244  def __init__(self, project_id, resource_data):
245    super().__init__(project_id=project_id)
246    self._resource_data = resource_data
247    self._nats = None
full_path: str 
249  @property
250  def full_path(self) -> str:
251    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', self.self_link)
252    if result:
253      return result.group(1)
254    else:
255      return f'>> {self.self_link}'

Returns the full path of this resource.

Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'

short_path: str 
257  @property
258  def short_path(self) -> str:
259    path = self.project_id + '/' + self.name
260    return path

Returns the short name for this resource.

Note that it isn't clear from this name what kind of resource it is.

Example: 'gke1'

name: str 
262  @property
263  def name(self) -> str:
264    return self._resource_data['name']
network: str 
270  @property
271  def network(self) -> str:
272    return self._resource_data['network']
def get_network_name(self) -> str: 
274  def get_network_name(self) -> str:
275    logging.info('inside get_network_name function')
276    if self._resource_data['network']:
277      return self._resource_data['network'].split('/')[-1]
278    return ''
nats 
280  @property
281  def nats(self):
282    return self._resource_data.get('nats', [])
def get_nat_ip_allocate_option(self, nat_gateway) -> str: 
284  def get_nat_ip_allocate_option(self, nat_gateway) -> str:
285    nats = self._resource_data.get('nats', [])
286    nat = [n for n in nats if n['name'] == nat_gateway]
287    return nat[0].get('natIpAllocateOption', '')
def get_enable_dynamic_port_allocation(self, nat_gateway) -> str: 
289  def get_enable_dynamic_port_allocation(self, nat_gateway) -> str:
290    nats = self._resource_data.get('nats', [])
291    nat = [n for n in nats if n['name'] == nat_gateway]
292    return nat[0].get('enableDynamicPortAllocation', '')
def subnet_has_nat(self, subnetwork): 
294  def subnet_has_nat(self, subnetwork):
295    if not self._resource_data.get('nats', []):
296      return False
297    for n in self._resource_data.get('nats', []):
298      if n['sourceSubnetworkIpRangesToNat'] == 'LIST_OF_SUBNETWORKS':
299        # Cloud NAT configure for specific subnets
300        if 'subnetworks' in n and subnetwork.self_link in [s['name'] for s in n['subnetworks']]:
301          return True
302      else:
303        # Cloud NAT configured for all subnets
304        return True
305    return False
class RouterStatus(gcpdiag.models.Resource): 
308class RouterStatus(models.Resource):
309  """NAT Router Status"""
310
311  _resource_data: dict
312
313  def __init__(self, project_id, resource_data):
314    super().__init__(project_id=project_id)
315    self._resource_data = resource_data
316
317  @property
318  def full_path(self) -> str:
319    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', self.self_link)
320    if result:
321      return result.group(1)
322    else:
323      return f'>> {self.self_link}'
324
325  @property
326  def short_path(self) -> str:
327    path = self.project_id + '/' + self.name
328    return path
329
330  @property
331  def name(self) -> str:
332    return self._resource_data.get('name', '')
333
334  @property
335  def self_link(self) -> str:
336    return self._resource_data.get('selfLink', '')
337
338  @property
339  def min_extra_nat_ips_needed(self) -> str:
340    nat_status = self._resource_data.get('result', {}).get('natStatus', {})
341    return nat_status[0].get('minExtraNatIpsNeeded', None)
342
343  @property
344  def num_vms_with_nat_mappings(self) -> str:
345    nat_status = self._resource_data.get('result', {}).get('natStatus', {})
346    return nat_status[0].get('numVmEndpointsWithNatMappings', None)
347
348  @property
349  def bgp_peer_status(self) -> str:
350    bgp_peer_status = self._resource_data.get('result', {}).get('bgpPeerStatus', {})
351    return bgp_peer_status

NAT Router Status

RouterStatus(project_id, resource_data) 
313  def __init__(self, project_id, resource_data):
314    super().__init__(project_id=project_id)
315    self._resource_data = resource_data
full_path: str 
317  @property
318  def full_path(self) -> str:
319    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', self.self_link)
320    if result:
321      return result.group(1)
322    else:
323      return f'>> {self.self_link}'

Returns the full path of this resource.

Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'

short_path: str 
325  @property
326  def short_path(self) -> str:
327    path = self.project_id + '/' + self.name
328    return path

Returns the short name for this resource.

Note that it isn't clear from this name what kind of resource it is.

Example: 'gke1'

name: str 
330  @property
331  def name(self) -> str:
332    return self._resource_data.get('name', '')
min_extra_nat_ips_needed: str 
338  @property
339  def min_extra_nat_ips_needed(self) -> str:
340    nat_status = self._resource_data.get('result', {}).get('natStatus', {})
341    return nat_status[0].get('minExtraNatIpsNeeded', None)
num_vms_with_nat_mappings: str 
343  @property
344  def num_vms_with_nat_mappings(self) -> str:
345    nat_status = self._resource_data.get('result', {}).get('natStatus', {})
346    return nat_status[0].get('numVmEndpointsWithNatMappings', None)
bgp_peer_status: str 
348  @property
349  def bgp_peer_status(self) -> str:
350    bgp_peer_status = self._resource_data.get('result', {}).get('bgpPeerStatus', {})
351    return bgp_peer_status
class RouterNatIpInfo(gcpdiag.models.Resource): 
354class RouterNatIpInfo(models.Resource):
355  """NAT IP Info"""
356
357  _resource_data: dict
358
359  def __init__(self, project_id, resource_data):
360    super().__init__(project_id=project_id)
361    self._resource_data = resource_data
362
363  @property
364  def full_path(self) -> str:
365    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', self.self_link)
366    if result:
367      return result.group(1)
368    else:
369      return f'>> {self.self_link}'
370
371  @property
372  def short_path(self) -> str:
373    path = self.project_id + '/' + self.name
374    return path
375
376  @property
377  def self_link(self) -> str:
378    return self._resource_data.get('selfLink', '')
379
380  @property
381  def name(self) -> str:
382    return self._resource_data.get('name', '')
383
384  @property
385  def result(self) -> str:
386    return self._resource_data.get('result', [])

NAT IP Info

RouterNatIpInfo(project_id, resource_data) 
359  def __init__(self, project_id, resource_data):
360    super().__init__(project_id=project_id)
361    self._resource_data = resource_data
full_path: str 
363  @property
364  def full_path(self) -> str:
365    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', self.self_link)
366    if result:
367      return result.group(1)
368    else:
369      return f'>> {self.self_link}'

Returns the full path of this resource.

Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'

short_path: str 
371  @property
372  def short_path(self) -> str:
373    path = self.project_id + '/' + self.name
374    return path

Returns the short name for this resource.

Note that it isn't clear from this name what kind of resource it is.

Example: 'gke1'

name: str 
380  @property
381  def name(self) -> str:
382    return self._resource_data.get('name', '')
result: str 
384  @property
385  def result(self) -> str:
386    return self._resource_data.get('result', [])
@dataclasses.dataclass
class Peering: 
389@dataclasses.dataclass
390class Peering:
391  """VPC Peerings"""
392
393  name: str
394  url: str
395  state: str
396  exports_custom_routes: bool
397  imports_custom_routes: bool
398  auto_creates_routes: bool
399
400  def __str__(self):
401    return self.name

VPC Peerings

Peering( name: str, url: str, state: str, exports_custom_routes: bool, imports_custom_routes: bool, auto_creates_routes: bool)
name: str
url: str
state: str
exports_custom_routes: bool
imports_custom_routes: bool
auto_creates_routes: bool
class Network(gcpdiag.models.Resource): 
404class Network(models.Resource):
405  """A VPC network."""
406
407  _resource_data: dict
408  _subnetworks: Optional[Dict[str, Subnetwork]]
409  _context: models.Context
410
411  def __init__(self, project_id, resource_data, context: models.Context):
412    super().__init__(project_id=project_id)
413    self._resource_data = resource_data
414    self._subnetworks = None
415    self._context = context
416
417  @property
418  def full_path(self) -> str:
419    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', 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', [])), self._context
452    )
453
454  @property
455  def peerings(self) -> List[Peering]:
456    return [
457      Peering(
458        peer['name'],
459        peer['network'],
460        peer['state'],
461        peer['exportCustomRoutes'],
462        peer['importCustomRoutes'],
463        peer['autoCreateRoutes'],
464      )
465      for peer in self._resource_data.get('peerings', [])
466    ]

A VPC network.

Network(project_id, resource_data, context: gcpdiag.models.Context) 
411  def __init__(self, project_id, resource_data, context: models.Context):
412    super().__init__(project_id=project_id)
413    self._resource_data = resource_data
414    self._subnetworks = None
415    self._context = context
full_path: str 
417  @property
418  def full_path(self) -> str:
419    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', self.self_link)
420    if result:
421      return result.group(1)
422    else:
423      return f'>> {self.self_link}'

Returns the full path of this resource.

Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'

short_path: str 
425  @property
426  def short_path(self) -> str:
427    path = self.project_id + '/' + self.name
428    return path

Returns the short name for this resource.

Note that it isn't clear from this name what kind of resource it is.

Example: 'gke1'

name: str 
430  @property
431  def name(self) -> str:
432    return self._resource_data['name']
firewall: EffectiveFirewalls 
438  @property
439  def firewall(self) -> 'EffectiveFirewalls':
440    return _get_effective_firewalls(self)
mtu: int 
442  @property
443  def mtu(self) -> int:
444    if 'mtu' in self._resource_data:
445      return self._resource_data['mtu']
446    return DEFAULT_MTU
subnetworks: Dict[str, Subnetwork] 
448  @property
449  def subnetworks(self) -> Dict[str, Subnetwork]:
450    return _batch_get_subnetworks(
451      self._project_id, frozenset(self._resource_data.get('subnetworks', [])), self._context
452    )
peerings: List[Peering] 
454  @property
455  def peerings(self) -> List[Peering]:
456    return [
457      Peering(
458        peer['name'],
459        peer['network'],
460        peer['state'],
461        peer['exportCustomRoutes'],
462        peer['importCustomRoutes'],
463        peer['autoCreateRoutes'],
464      )
465      for peer in self._resource_data.get('peerings', [])
466    ]
@dataclasses.dataclass
class FirewallCheckResult: 
573@dataclasses.dataclass
574class FirewallCheckResult:
575  """The result of a firewall connectivity check."""
576
577  action: str
578  firewall_policy_name: Optional[str] = None
579  firewall_policy_rule_description: Optional[str] = None
580  vpc_firewall_rule_id: Optional[str] = None
581  vpc_firewall_rule_name: Optional[str] = None
582
583  def __str__(self):
584    return self.action
585
586  @property
587  def matched_by_str(self):
588    if self.firewall_policy_name:
589      if self.firewall_policy_rule_description:
590        return f'policy: {self.firewall_policy_name}, rule: {self.firewall_policy_rule_description}'
591      else:
592        return f'policy: {self.firewall_policy_name}'
593    elif self.vpc_firewall_rule_name:
594      return f'vpc firewall rule: {self.vpc_firewall_rule_name}'

The result of a firewall connectivity check.

FirewallCheckResult( action: str, firewall_policy_name: Optional[str] = None, firewall_policy_rule_description: Optional[str] = None, vpc_firewall_rule_id: Optional[str] = None, vpc_firewall_rule_name: Optional[str] = None)
action: str
firewall_policy_name: Optional[str] = None
firewall_policy_rule_description: Optional[str] = None
vpc_firewall_rule_id: Optional[str] = None
vpc_firewall_rule_name: Optional[str] = None
matched_by_str 
586  @property
587  def matched_by_str(self):
588    if self.firewall_policy_name:
589      if self.firewall_policy_rule_description:
590        return f'policy: {self.firewall_policy_name}, rule: {self.firewall_policy_rule_description}'
591      else:
592        return f'policy: {self.firewall_policy_name}'
593    elif self.vpc_firewall_rule_name:
594      return f'vpc firewall rule: {self.vpc_firewall_rule_name}'
class FirewallRuleNotFoundError(builtins.Exception): 
597class FirewallRuleNotFoundError(Exception):
598  rule_name: str
599
600  def __init__(self, name, disabled=False):
601    # Call the base class constructor with the parameters it needs
602    super().__init__(f'firewall rule not found: {name}')
603    self.rule_name = name
604    self.disabled = disabled

Common base class for all non-exit exceptions.

FirewallRuleNotFoundError(name, disabled=False) 
600  def __init__(self, name, disabled=False):
601    # Call the base class constructor with the parameters it needs
602    super().__init__(f'firewall rule not found: {name}')
603    self.rule_name = name
604    self.disabled = disabled
rule_name: str
disabled
class VpcFirewallRule: 
607class VpcFirewallRule:
608  """Represents firewall rule"""
609
610  def __init__(self, resource_data):
611    self._resource_data = resource_data
612
613  @property
614  def name(self) -> str:
615    return self._resource_data['name']
616
617  @property
618  def source_ranges(self) -> List[ipaddress.IPv4Network]:
619    return self._resource_data['sourceRanges']
620
621  @property
622  def target_tags(self) -> set:
623    return self._resource_data['targetTags']
624
625  @property
626  def allowed(self) -> List[dict]:
627    return self._resource_data['allowed']
628
629  def is_enabled(self) -> bool:
630    return not self._resource_data['disabled']

Represents firewall rule

VpcFirewallRule(resource_data) 
610  def __init__(self, resource_data):
611    self._resource_data = resource_data
name: str 
613  @property
614  def name(self) -> str:
615    return self._resource_data['name']
source_ranges: List[ipaddress.IPv4Network] 
617  @property
618  def source_ranges(self) -> List[ipaddress.IPv4Network]:
619    return self._resource_data['sourceRanges']
target_tags: set 
621  @property
622  def target_tags(self) -> set:
623    return self._resource_data['targetTags']
allowed: List[dict] 
625  @property
626  def allowed(self) -> List[dict]:
627    return self._resource_data['allowed']
def is_enabled(self) -> bool: 
629  def is_enabled(self) -> bool:
630    return not self._resource_data['disabled']
class EffectiveFirewalls: 
 993class EffectiveFirewalls:
 994  """Effective firewall rules for a VPC network or Instance.
 995
 996  Includes org/folder firewall policies)."""
 997
 998  _resource_data: dict
 999  _policies: List[_FirewallPolicy]
1000  _vpc_firewall: _VpcFirewall
1001
1002  def __init__(self, resource_data):
1003    self._resource_data = resource_data
1004    self._policies = []
1005    if 'firewallPolicys' in resource_data:
1006      for policy in resource_data['firewallPolicys']:
1007        self._policies.append(_FirewallPolicy(policy))
1008    self._vpc_firewall = _VpcFirewall(resource_data.get('firewalls', {}))
1009
1010  def check_connectivity_ingress(
1011    self,  #
1012    *,
1013    src_ip: IPAddrOrNet,
1014    ip_protocol: str,
1015    port: Optional[int] = None,
1016    source_service_account: Optional[str] = None,
1017    source_tags: Optional[List[str]] = None,
1018    target_service_account: Optional[str] = None,
1019    target_tags: Optional[List[str]] = None,
1020  ) -> FirewallCheckResult:
1021    if ip_protocol != 'ICMP' and port is None:
1022      raise ValueError('TCP and UDP must have port numbers')
1023
1024    # Firewall policies (organization, folders)
1025    for p in self._policies:
1026      result = p.check_connectivity_ingress(
1027        src_ip=src_ip,
1028        ip_protocol=ip_protocol,
1029        port=port,
1030        # target_network=self._network,
1031        target_service_account=target_service_account,
1032      )
1033      if result.action != 'goto_next':
1034        return result
1035
1036    # VPC firewall rules
1037    return self._vpc_firewall.check_connectivity_ingress(
1038      src_ip=src_ip,
1039      ip_protocol=ip_protocol,
1040      port=port,
1041      source_service_account=source_service_account,
1042      source_tags=source_tags,
1043      target_service_account=target_service_account,
1044      target_tags=target_tags,
1045    )
1046
1047  def check_connectivity_egress(
1048    self,  #
1049    *,
1050    src_ip: IPAddrOrNet,
1051    ip_protocol: str,
1052    port: Optional[int] = None,
1053    source_service_account: Optional[str] = None,
1054    source_tags: Optional[List[str]] = None,
1055    target_service_account: Optional[str] = None,
1056    target_tags: Optional[List[str]] = None,
1057  ) -> FirewallCheckResult:
1058    if ip_protocol != 'ICMP' and port is None:
1059      raise ValueError('TCP and UDP must have port numbers')
1060
1061    # Firewall policies (organization, folders)
1062    for p in self._policies:
1063      result = p.check_connectivity_egress(
1064        src_ip=src_ip,
1065        ip_protocol=ip_protocol,
1066        port=port,
1067        # target_network=self._network,
1068        target_service_account=target_service_account,
1069      )
1070      if result.action != 'goto_next':
1071        return result
1072
1073    # VPC firewall rules
1074    return self._vpc_firewall.check_connectivity_egress(
1075      src_ip=src_ip,
1076      ip_protocol=ip_protocol,
1077      port=port,
1078      source_service_account=source_service_account,
1079      source_tags=source_tags,
1080      target_service_account=target_service_account,
1081      target_tags=target_tags,
1082    )
1083
1084  def get_vpc_ingress_rules(
1085    self,
1086    name: Optional[str] = None,
1087    name_pattern: Optional[re.Pattern] = None,
1088    target_tags: Optional[List[str]] = None,
1089  ) -> List[VpcFirewallRule]:
1090    """Retrieve the list of ingress firewall rules matching name or name pattern and target tags.
1091
1092    Args:
1093        name (Optional[str], optional): firewall rune name. Defaults to None.
1094        name_pattern (Optional[re.Pattern], optional): firewall rule name pattern. Defaults to None.
1095        target_tags (Optional[List[str]], optional): firewall target tags
1096          (if not specified any tag will match). Defaults to None.
1097
1098    Returns:
1099        List[VpcFirewallRule]: List of ingress firewall rules
1100    """
1101    rules = self._vpc_firewall.get_vpc_ingress_rules(name, name_pattern, target_tags)
1102    return rules
1103
1104  def get_vpc_egress_rules(
1105    self,
1106    name: Optional[str] = None,
1107    name_pattern: Optional[re.Pattern] = None,
1108    target_tags: Optional[List[str]] = None,
1109  ) -> List[VpcFirewallRule]:
1110    """Retrieve the list of egress firewall rules matching name or name pattern and target tags.
1111
1112    Args:
1113        name (Optional[str], optional): firewall rune name. Defaults to None.
1114        name_pattern (Optional[re.Pattern], optional): firewall rule name pattern. Defaults to None.
1115        target_tags (Optional[List[str]], optional): firewall target tags
1116          (if not specified any tag will match). Defaults to None.
1117
1118    Returns:
1119        List[VpcFirewallRule]: List of egress firewall rules
1120    """
1121    rules = self._vpc_firewall.get_vpc_egress_rules(name, name_pattern, target_tags)
1122    return rules
1123
1124  def verify_ingress_rule_exists(self, name: str):
1125    """Verify that a certain VPC rule exists. This is useful to verify
1126    whether maybe a permission was missing on a shared VPC and an
1127    automatic rule couldn't be created."""
1128    return self._vpc_firewall.verify_ingress_rule_exists(name)
1129
1130  def verify_egress_rule_exists(self, name: str):
1131    """Verify that a certain VPC rule exists. This is useful to verify
1132    whether maybe a permission was missing on a shared VPC and an
1133    automatic rule couldn't be created."""
1134    return self._vpc_firewall.verify_egress_rule_exists(name)

Effective firewall rules for a VPC network or Instance.

Includes org/folder firewall policies).

EffectiveFirewalls(resource_data) 
1002  def __init__(self, resource_data):
1003    self._resource_data = resource_data
1004    self._policies = []
1005    if 'firewallPolicys' in resource_data:
1006      for policy in resource_data['firewallPolicys']:
1007        self._policies.append(_FirewallPolicy(policy))
1008    self._vpc_firewall = _VpcFirewall(resource_data.get('firewalls', {}))
def check_connectivity_ingress( self, *, src_ip: Union[ipaddress.IPv4Address, ipaddress.IPv6Address, ipaddress.IPv4Network, ipaddress.IPv6Network], ip_protocol: str, port: Optional[int] = None, source_service_account: Optional[str] = None, source_tags: Optional[List[str]] = None, target_service_account: Optional[str] = None, target_tags: Optional[List[str]] = None) -> FirewallCheckResult: 
1010  def check_connectivity_ingress(
1011    self,  #
1012    *,
1013    src_ip: IPAddrOrNet,
1014    ip_protocol: str,
1015    port: Optional[int] = None,
1016    source_service_account: Optional[str] = None,
1017    source_tags: Optional[List[str]] = None,
1018    target_service_account: Optional[str] = None,
1019    target_tags: Optional[List[str]] = None,
1020  ) -> FirewallCheckResult:
1021    if ip_protocol != 'ICMP' and port is None:
1022      raise ValueError('TCP and UDP must have port numbers')
1023
1024    # Firewall policies (organization, folders)
1025    for p in self._policies:
1026      result = p.check_connectivity_ingress(
1027        src_ip=src_ip,
1028        ip_protocol=ip_protocol,
1029        port=port,
1030        # target_network=self._network,
1031        target_service_account=target_service_account,
1032      )
1033      if result.action != 'goto_next':
1034        return result
1035
1036    # VPC firewall rules
1037    return self._vpc_firewall.check_connectivity_ingress(
1038      src_ip=src_ip,
1039      ip_protocol=ip_protocol,
1040      port=port,
1041      source_service_account=source_service_account,
1042      source_tags=source_tags,
1043      target_service_account=target_service_account,
1044      target_tags=target_tags,
1045    )
def check_connectivity_egress( self, *, src_ip: Union[ipaddress.IPv4Address, ipaddress.IPv6Address, ipaddress.IPv4Network, ipaddress.IPv6Network], ip_protocol: str, port: Optional[int] = None, source_service_account: Optional[str] = None, source_tags: Optional[List[str]] = None, target_service_account: Optional[str] = None, target_tags: Optional[List[str]] = None) -> FirewallCheckResult: 
1047  def check_connectivity_egress(
1048    self,  #
1049    *,
1050    src_ip: IPAddrOrNet,
1051    ip_protocol: str,
1052    port: Optional[int] = None,
1053    source_service_account: Optional[str] = None,
1054    source_tags: Optional[List[str]] = None,
1055    target_service_account: Optional[str] = None,
1056    target_tags: Optional[List[str]] = None,
1057  ) -> FirewallCheckResult:
1058    if ip_protocol != 'ICMP' and port is None:
1059      raise ValueError('TCP and UDP must have port numbers')
1060
1061    # Firewall policies (organization, folders)
1062    for p in self._policies:
1063      result = p.check_connectivity_egress(
1064        src_ip=src_ip,
1065        ip_protocol=ip_protocol,
1066        port=port,
1067        # target_network=self._network,
1068        target_service_account=target_service_account,
1069      )
1070      if result.action != 'goto_next':
1071        return result
1072
1073    # VPC firewall rules
1074    return self._vpc_firewall.check_connectivity_egress(
1075      src_ip=src_ip,
1076      ip_protocol=ip_protocol,
1077      port=port,
1078      source_service_account=source_service_account,
1079      source_tags=source_tags,
1080      target_service_account=target_service_account,
1081      target_tags=target_tags,
1082    )
def get_vpc_ingress_rules( self, name: Optional[str] = None, name_pattern: Optional[re.Pattern] = None, target_tags: Optional[List[str]] = None) -> List[VpcFirewallRule]: 
1084  def get_vpc_ingress_rules(
1085    self,
1086    name: Optional[str] = None,
1087    name_pattern: Optional[re.Pattern] = None,
1088    target_tags: Optional[List[str]] = None,
1089  ) -> List[VpcFirewallRule]:
1090    """Retrieve the list of ingress firewall rules matching name or name pattern and target tags.
1091
1092    Args:
1093        name (Optional[str], optional): firewall rune name. Defaults to None.
1094        name_pattern (Optional[re.Pattern], optional): firewall rule name pattern. Defaults to None.
1095        target_tags (Optional[List[str]], optional): firewall target tags
1096          (if not specified any tag will match). Defaults to None.
1097
1098    Returns:
1099        List[VpcFirewallRule]: List of ingress firewall rules
1100    """
1101    rules = self._vpc_firewall.get_vpc_ingress_rules(name, name_pattern, target_tags)
1102    return rules

Retrieve the list of ingress firewall rules matching name or name pattern and target tags.

Arguments:
  • name (Optional[str], optional): firewall rune name. Defaults to None.
  • name_pattern (Optional[re.Pattern], optional): firewall rule name pattern. Defaults to None.
  • target_tags (Optional[List[str]], optional): firewall target tags (if not specified any tag will match). Defaults to None.
Returns:

List[VpcFirewallRule]: List of ingress firewall rules

def get_vpc_egress_rules( self, name: Optional[str] = None, name_pattern: Optional[re.Pattern] = None, target_tags: Optional[List[str]] = None) -> List[VpcFirewallRule]: 
1104  def get_vpc_egress_rules(
1105    self,
1106    name: Optional[str] = None,
1107    name_pattern: Optional[re.Pattern] = None,
1108    target_tags: Optional[List[str]] = None,
1109  ) -> List[VpcFirewallRule]:
1110    """Retrieve the list of egress firewall rules matching name or name pattern and target tags.
1111
1112    Args:
1113        name (Optional[str], optional): firewall rune name. Defaults to None.
1114        name_pattern (Optional[re.Pattern], optional): firewall rule name pattern. Defaults to None.
1115        target_tags (Optional[List[str]], optional): firewall target tags
1116          (if not specified any tag will match). Defaults to None.
1117
1118    Returns:
1119        List[VpcFirewallRule]: List of egress firewall rules
1120    """
1121    rules = self._vpc_firewall.get_vpc_egress_rules(name, name_pattern, target_tags)
1122    return rules

Retrieve the list of egress firewall rules matching name or name pattern and target tags.

Arguments:
  • name (Optional[str], optional): firewall rune name. Defaults to None.
  • name_pattern (Optional[re.Pattern], optional): firewall rule name pattern. Defaults to None.
  • target_tags (Optional[List[str]], optional): firewall target tags (if not specified any tag will match). Defaults to None.
Returns:

List[VpcFirewallRule]: List of egress firewall rules

def verify_ingress_rule_exists(self, name: str): 
1124  def verify_ingress_rule_exists(self, name: str):
1125    """Verify that a certain VPC rule exists. This is useful to verify
1126    whether maybe a permission was missing on a shared VPC and an
1127    automatic rule couldn't be created."""
1128    return self._vpc_firewall.verify_ingress_rule_exists(name)

Verify that a certain VPC rule exists. This is useful to verify whether maybe a permission was missing on a shared VPC and an automatic rule couldn't be created.

def verify_egress_rule_exists(self, name: str): 
1130  def verify_egress_rule_exists(self, name: str):
1131    """Verify that a certain VPC rule exists. This is useful to verify
1132    whether maybe a permission was missing on a shared VPC and an
1133    automatic rule couldn't be created."""
1134    return self._vpc_firewall.verify_egress_rule_exists(name)

Verify that a certain VPC rule exists. This is useful to verify whether maybe a permission was missing on a shared VPC and an automatic rule couldn't be created.

class VPCEffectiveFirewalls(EffectiveFirewalls): 
1137class VPCEffectiveFirewalls(EffectiveFirewalls):
1138  """Effective firewall rules for a VPC network.
1139
1140  Includes org/folder firewall policies).
1141  """
1142
1143  _network: Network
1144
1145  def __init__(self, network, resource_data):
1146    super().__init__(resource_data)
1147    self._network = network

Effective firewall rules for a VPC network.

Includes org/folder firewall policies).

VPCEffectiveFirewalls(network, resource_data) 
1145  def __init__(self, network, resource_data):
1146    super().__init__(resource_data)
1147    self._network = network
Inherited Members
@caching.cached_api_call(in_memory=True)
def get_network( project_id: str, network_name: str, context: gcpdiag.models.Context) -> Network: 
1160@caching.cached_api_call(in_memory=True)
1161def get_network(project_id: str, network_name: str, context: models.Context) -> Network:
1162  logging.debug('fetching network: %s/%s', project_id, network_name)
1163  compute = apis.get_api('compute', 'v1', project_id)
1164  request = compute.networks().get(project=project_id, network=network_name)
1165  response = request.execute(num_retries=config.API_RETRIES)
1166  return Network(project_id, response, context)
def get_subnetwork_from_url(url: str) -> Subnetwork: 
1169def get_subnetwork_from_url(url: str) -> Subnetwork:
1170  """Returns Subnetwork object given subnetwork url"""
1171  m = re.match(
1172    (
1173      r'https://www.googleapis.com/compute/v1/projects/'
1174      r'([^/]+)/regions/([^/]+)/subnetworks/([^/]+)$'
1175    ),
1176    url,
1177  )
1178  if not m:
1179    raise ValueError(f"can't parse network url: {url}")
1180  (project_id, region, subnetwork_name) = (m.group(1), m.group(2), m.group(3))
1181  return get_subnetwork(project_id, region, subnetwork_name)

Returns Subnetwork object given subnetwork url

def get_network_from_url(url: str) -> Network: 
1184def get_network_from_url(url: str) -> Network:
1185  m = re.match(
1186    r'https://www.googleapis.com/compute/v1/projects/([^/]+)/global/networks/([^/]+)', url
1187  )
1188  if not m:
1189    raise ValueError(f"can't parse network url: {url}")
1190  (project_id, network_name) = (m.group(1), m.group(2))
1191  return get_network(project_id, network_name, models.Context(project_id=project_id))
@caching.cached_api_call(in_memory=True)
def get_networks(context: gcpdiag.models.Context) -> List[Network]: 
1194@caching.cached_api_call(in_memory=True)
1195def get_networks(context: models.Context) -> List[Network]:
1196  logging.debug('fetching network: %s', context.project_id)
1197  compute = apis.get_api('compute', 'v1', context.project_id)
1198  request = compute.networks().list(project=context.project_id)
1199  response = request.execute(num_retries=config.API_RETRIES)
1200  return [Network(context.project_id, item, context) for item in response.get('items', [])]
@caching.cached_api_call(in_memory=True)
def get_subnetwork( project_id: str, region: str, subnetwork_name: str) -> Subnetwork: 
1203@caching.cached_api_call(in_memory=True)
1204def get_subnetwork(project_id: str, region: str, subnetwork_name: str) -> Subnetwork:
1205  logging.debug('fetching network: %s/%s', project_id, subnetwork_name)
1206  compute = apis.get_api('compute', 'v1', project_id)
1207  request = compute.subnetworks().get(project=project_id, region=region, subnetwork=subnetwork_name)
1208  response = request.execute(num_retries=config.API_RETRIES)
1209  if response is None:
1210    raise RuntimeError(f'failed to fetch subnetwork: {project_id}/{region}/{subnetwork_name}')
1211  return Subnetwork(project_id, response, context=models.Context(project_id=project_id))
@caching.cached_api_call(in_memory=True)
def get_routes(project_id: str) -> List[Route]: 
1250@caching.cached_api_call(in_memory=True)
1251def get_routes(project_id: str) -> List[Route]:
1252  logging.debug('fetching routes: %s', project_id)
1253  compute = apis.get_api('compute', 'v1', project_id)
1254  request = compute.routes().list(project=project_id)
1255  response = request.execute(num_retries=config.API_RETRIES)
1256  return [Route(project_id, item) for item in response.get('items', [])]
@caching.cached_api_call(in_memory=True)
def get_zones(project_id: str) -> List[ManagedZone]: 
1259@caching.cached_api_call(in_memory=True)
1260def get_zones(project_id: str) -> List[ManagedZone]:
1261  logging.debug('fetching DNS zones: %s', project_id)
1262  dns = apis.get_api('dns', 'v1beta2', project_id)
1263  request = dns.managedZones().list(project=project_id)
1264  response = request.execute(num_retries=config.API_RETRIES)
1265  zones = []
1266  for zone in response.get('managedZones', []):
1267    request2 = dns.managedZones().get(project=project_id, managedZone=zone['name'])
1268    response2 = request2.execute(num_retries=config.API_RETRIES)
1269    zones.append(ManagedZone(project_id, response2))
1270  return zones
@caching.cached_api_call(in_memory=True)
def get_routers( project_id: str, region: str, network) -> List[Router]: 
1273@caching.cached_api_call(in_memory=True)
1274def get_routers(project_id: str, region: str, network) -> List[Router]:
1275  logging.debug('fetching routers: %s/%s', project_id, region)
1276  compute = apis.get_api('compute', 'v1', project_id)
1277  request = compute.routers().list(
1278    project=project_id, region=region, filter=f'network="{network.self_link}"'
1279  )
1280  response = request.execute(num_retries=config.API_RETRIES)
1281  return [Router(project_id, item) for item in response.get('items', [])]
@caching.cached_api_call(in_memory=True)
def get_router(project_id: str, region: str, network) -> Router: 
1284@caching.cached_api_call(in_memory=True)
1285def get_router(project_id: str, region: str, network) -> Router:
1286  logging.debug('fetching routers: %s/%s', project_id, region)
1287  compute = apis.get_api('compute', 'v1', project_id)
1288  request = compute.routers().list(
1289    project=project_id, region=region, filter=f'network="{network.self_link}"'
1290  )
1291  response = request.execute(num_retries=config.API_RETRIES)
1292  return Router(project_id, next(iter(response.get('items', [{}]))))
@caching.cached_api_call(in_memory=True)
def get_router_by_name( project_id: str, region: str, router_name: str) -> Router: 
1295@caching.cached_api_call(in_memory=True)
1296def get_router_by_name(project_id: str, region: str, router_name: str) -> Router:
1297  logging.debug('fetching router list: %s/%s in region %s', project_id, router_name, region)
1298  compute = apis.get_api('compute', 'v1', project_id)
1299  request = compute.routers().list(project=project_id, region=region)
1300  response = request.execute(num_retries=config.API_RETRIES)
1301  return next(
1302    Router(project_id, item) for item in response.get('items', []) if item['name'] == router_name
1303  )
@caching.cached_api_call(in_memory=True)
def nat_router_status( project_id: str, router_name: str, region: str) -> RouterStatus: 
1306@caching.cached_api_call(in_memory=True)
1307def nat_router_status(project_id: str, router_name: str, region: str) -> RouterStatus:
1308  logging.debug('fetching router status: %s/%s in region %s', project_id, router_name, region)
1309  compute = apis.get_api('compute', 'v1', project_id)
1310  request = compute.routers().getRouterStatus(project=project_id, router=router_name, region=region)
1311  response = request.execute(num_retries=config.API_RETRIES)
1312  if 'result' in str(response):
1313    return RouterStatus(project_id, response)
1314  else:
1315    logging.debug(
1316      'unable to fetch router status: %s/%s in region %s', project_id, router_name, region
1317    )
1318    return RouterStatus(project_id, {})
@caching.cached_api_call(in_memory=True)
def get_nat_ip_info( project_id: str, router_name: str, region: str) -> RouterNatIpInfo: 
1321@caching.cached_api_call(in_memory=True)
1322def get_nat_ip_info(project_id: str, router_name: str, region: str) -> RouterNatIpInfo:
1323  logging.debug(
1324    'fetching NAT IP info for router: %s/%s in region %s', project_id, router_name, region
1325  )
1326  compute = apis.get_api('compute', 'v1', project_id)
1327  request = compute.routers().getNatIpInfo(project=project_id, router=router_name, region=region)
1328  response = request.execute(num_retries=config.API_RETRIES)
1329  if 'result' in str(response):
1330    return RouterNatIpInfo(project_id, response)
1331  else:
1332    logging.debug(
1333      'unable to fetch Nat IP Info for router: %s/%s in region %s', project_id, router_name, region
1334    )
1335    return RouterNatIpInfo(project_id, {})
class VPCSubnetworkIAMPolicy(gcpdiag.queries.iam.BaseIAMPolicy): 
1338class VPCSubnetworkIAMPolicy(iam.BaseIAMPolicy):
1339  def _is_resource_permission(self, permission):
1340    return True

Common class for IAM policies

Inherited Members
@caching.cached_api_call(in_memory=True)
def get_subnetwork_iam_policy( context: gcpdiag.models.Context, region: str, subnetwork_name: str) -> VPCSubnetworkIAMPolicy: 
1343@caching.cached_api_call(in_memory=True)
1344def get_subnetwork_iam_policy(
1345  context: models.Context, region: str, subnetwork_name: str
1346) -> VPCSubnetworkIAMPolicy:
1347  project_id = context.project_id
1348  resource_name = f'projects/{project_id}/regions/{region}/subnetworks/{subnetwork_name}'
1349
1350  compute = apis.get_api('compute', 'v1', project_id)
1351  request = compute.subnetworks().getIamPolicy(
1352    project=project_id, region=region, resource=subnetwork_name
1353  )
1354
1355  return iam.fetch_iam_policy(request, VPCSubnetworkIAMPolicy, project_id, resource_name, context)
class Address(gcpdiag.models.Resource): 
1358class Address(models.Resource):
1359  """IP Addresses."""
1360
1361  _resource_data: dict
1362
1363  def __init__(self, project_id, resource_data):
1364    super().__init__(project_id=project_id)
1365    self._resource_data = resource_data
1366
1367  @property
1368  def full_path(self) -> str:
1369    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', self.self_link)
1370    if result:
1371      return result.group(1)
1372    else:
1373      return f'>> {self.self_link}'
1374
1375  @property
1376  def short_path(self) -> str:
1377    path = self.project_id + '/' + self.name
1378    return path
1379
1380  @property
1381  def name(self) -> str:
1382    return self._resource_data['name']
1383
1384  @property
1385  def self_link(self) -> str:
1386    return self._resource_data.get('selfLink', '')
1387
1388  @property
1389  def subnetwork(self) -> str:
1390    return self._resource_data['subnetwork']
1391
1392  @property
1393  def status(self) -> str:
1394    return self._resource_data['status']

IP Addresses.

Address(project_id, resource_data) 
1363  def __init__(self, project_id, resource_data):
1364    super().__init__(project_id=project_id)
1365    self._resource_data = resource_data
full_path: str 
1367  @property
1368  def full_path(self) -> str:
1369    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', self.self_link)
1370    if result:
1371      return result.group(1)
1372    else:
1373      return f'>> {self.self_link}'

Returns the full path of this resource.

Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'

short_path: str 
1375  @property
1376  def short_path(self) -> str:
1377    path = self.project_id + '/' + self.name
1378    return path

Returns the short name for this resource.

Note that it isn't clear from this name what kind of resource it is.

Example: 'gke1'

name: str 
1380  @property
1381  def name(self) -> str:
1382    return self._resource_data['name']
subnetwork: str 
1388  @property
1389  def subnetwork(self) -> str:
1390    return self._resource_data['subnetwork']
status: str 
1392  @property
1393  def status(self) -> str:
1394    return self._resource_data['status']
@caching.cached_api_call(in_memory=True)
def get_addresses(project_id: str) -> List[Address]: 
1397@caching.cached_api_call(in_memory=True)
1398def get_addresses(project_id: str) -> List[Address]:
1399  logging.debug('fetching addresses list: %s', project_id)
1400  compute = apis.get_api('compute', 'v1', project_id)
1401  addresses = []
1402  request = compute.addresses().aggregatedList(project=project_id)
1403  response = request.execute(num_retries=config.API_RETRIES)
1404  addresses_by_regions = response['items']
1405  for _, data_ in addresses_by_regions.items():
1406    if 'addresses' not in data_:
1407      continue
1408    addresses.extend([Address(project_id, address) for address in data_['addresses']])
1409  return addresses
built with pdoc logo