diff options
Diffstat (limited to 'cloudinit/net/__init__.py')
| -rw-r--r-- | cloudinit/net/__init__.py | 37 |
1 files changed, 27 insertions, 10 deletions
diff --git a/cloudinit/net/__init__.py b/cloudinit/net/__init__.py index 017c50c5..7558745f 100644 --- a/cloudinit/net/__init__.py +++ b/cloudinit/net/__init__.py @@ -11,6 +11,7 @@ import ipaddress import logging import os import re +from typing import Any, Dict from cloudinit import subp from cloudinit import util @@ -971,18 +972,33 @@ def get_ib_hwaddrs_by_interface(): return ret -def has_url_connectivity(url): - """Return true when the instance has access to the provided URL +def has_url_connectivity(url_data: Dict[str, Any]) -> bool: + """Return true when the instance has access to the provided URL. Logs a warning if url is not the expected format. + + url_data is a dictionary of kwargs to send to readurl. E.g.: + + has_url_connectivity({ + "url": "http://example.invalid", + "headers": {"some": "header"}, + "timeout": 10 + }) """ + if 'url' not in url_data: + LOG.warning( + "Ignoring connectivity check. No 'url' to check in %s", url_data) + return False + url = url_data['url'] if not any([url.startswith('http://'), url.startswith('https://')]): LOG.warning( "Ignoring connectivity check. Expected URL beginning with http*://" " received '%s'", url) return False + if 'timeout' not in url_data: + url_data['timeout'] = 5 try: - readurl(url, timeout=5) + readurl(**url_data) except UrlError: return False return True @@ -1025,14 +1041,15 @@ class EphemeralIPv4Network(object): No operations are performed if the provided interface already has the specified configuration. - This can be verified with the connectivity_url. + This can be verified with the connectivity_url_data. If unconnected, bring up the interface with valid ip, prefix and broadcast. If router is provided setup a default route for that interface. Upon context exit, clean up the interface leaving no configuration behind. """ def __init__(self, interface, ip, prefix_or_mask, broadcast, router=None, - connectivity_url=None, static_routes=None): + connectivity_url_data: Dict[str, Any] = None, + static_routes=None): """Setup context manager and validate call signature. @param interface: Name of the network interface to bring up. @@ -1041,7 +1058,7 @@ class EphemeralIPv4Network(object): prefix. @param broadcast: Broadcast address for the IPv4 network. @param router: Optionally the default gateway IP. - @param connectivity_url: Optionally, a URL to verify if a usable + @param connectivity_url_data: Optionally, a URL to verify if a usable connection already exists. @param static_routes: Optionally a list of static routes from DHCP """ @@ -1056,7 +1073,7 @@ class EphemeralIPv4Network(object): 'Cannot setup network: {0}'.format(e) ) from e - self.connectivity_url = connectivity_url + self.connectivity_url_data = connectivity_url_data self.interface = interface self.ip = ip self.broadcast = broadcast @@ -1066,11 +1083,11 @@ class EphemeralIPv4Network(object): def __enter__(self): """Perform ephemeral network setup if interface is not connected.""" - if self.connectivity_url: - if has_url_connectivity(self.connectivity_url): + if self.connectivity_url_data: + if has_url_connectivity(self.connectivity_url_data): LOG.debug( 'Skip ephemeral network setup, instance has connectivity' - ' to %s', self.connectivity_url) + ' to %s', self.connectivity_url_data['url']) return self._bringup_device() |