16981563750

Committed 14 Aug 2025 10:49PM UTC coverage: 86.896% (+0.04%) from 86.852%

Build # 16981563750

Build Type

push

github

Committed by

web-flow

Commit Message

add support for Fn::Tranform in CFnV2 (#12966)

Co-authored-by: Simon Walker <simon.walker@localstack.cloud>

Coverage Stats

181 of 195 new or added lines in 6 files covered. (92.82%)

348 existing lines in 22 files now uncovered.

66915 of 77006 relevant lines covered (86.9%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.27

/localstack-core/localstack/utils/net.py

import logging
import random
import re
import socket
import threading
from collections.abc import MutableMapping
from contextlib import closing
from typing import Any, NamedTuple, Optional, Union
from urllib.parse import urlparse

import dns.resolver
from dnslib import DNSRecord

from localstack import config, constants

from .collections import CustomExpiryTTLCache
from .numbers import is_number
from .objects import singleton_factory
from .sync import retry

LOG = logging.getLogger(__name__)

# regular expression for IPv4 addresses
IP_REGEX = (
    r"^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$"
)

# many linux kernels use 32768-60999, RFC 6335 is 49152-65535, so we use a mix here
DYNAMIC_PORT_RANGE_START = 32768
DYNAMIC_PORT_RANGE_END = 65536

DEFAULT_PORT_RESERVED_SECONDS = 6
"""Default nuber of seconds a port is reserved in a PortRange."""


class Port(NamedTuple):
    """Represents a network port, with port number and protocol (TCP/UDP)"""

    port: int
    """the port number"""
    protocol: str
    """network protocol name (usually 'tcp' or 'udp')"""

    @classmethod
    def wrap(cls, port: "IntOrPort") -> "Port":
        """Return the given port as a Port object, using 'tcp' as the default protocol."""
        if isinstance(port, Port):
            return port
        return Port(port=port, protocol="tcp")


# simple helper type to encapsulate int/Port argument types
IntOrPort = Union[int, Port]


def is_port_open(
    port_or_url: Union[int, str],
    http_path: str = None,
    expect_success: bool = True,
    protocols: Optional[Union[str, list[str]]] = None,
    quiet: bool = True,
):
    from localstack.utils.http import safe_requests

    protocols = protocols or ["tcp"]
    port = port_or_url
    if is_number(port):
        port = int(port)
    host = "localhost"
    protocol = "http"
    protocols = protocols if isinstance(protocols, list) else [protocols]
    if isinstance(port, str):
        url = urlparse(port_or_url)
        port = url.port
        host = url.hostname
        protocol = url.scheme
    nw_protocols = []
    nw_protocols += [socket.SOCK_STREAM] if "tcp" in protocols else []
    nw_protocols += [socket.SOCK_DGRAM] if "udp" in protocols else []
    for nw_protocol in nw_protocols:
        with closing(
            socket.socket(socket.AF_INET if ":" not in host else socket.AF_INET6, nw_protocol)
        ) as sock:
            sock.settimeout(1)
            if nw_protocol == socket.SOCK_DGRAM:
                try:
                    if port == 53:
                        dnshost = "127.0.0.1" if host == "localhost" else host
                        resolver = dns.resolver.Resolver()
                        resolver.nameservers = [dnshost]
                        resolver.timeout = 1
                        resolver.lifetime = 1
                        answers = resolver.query("google.com", "A")
                        assert len(answers) > 0
                    else:
                        sock.sendto(b"", (host, port))
                        sock.recvfrom(1024)
                except Exception:
                    if not quiet:
                        LOG.error(
                            "Error connecting to UDP port %s:%s",
                            host,
                            port,
                            exc_info=LOG.isEnabledFor(logging.DEBUG),
                        )
                    return False
            elif nw_protocol == socket.SOCK_STREAM:
                result = sock.connect_ex((host, port))
                if result != 0:
                    if not quiet:
                        LOG.warning(
                            "Error connecting to TCP port %s:%s (result=%s)", host, port, result
                        )
                    return False
    if "tcp" not in protocols or not http_path:
        return True
    host = f"[{host}]" if ":" in host else host
    url = f"{protocol}://{host}:{port}{http_path}"
    try:
        response = safe_requests.get(url, verify=False)
        return not expect_success or response.status_code < 400
    except Exception:
        return False


def wait_for_port_open(
    port: int, http_path: str = None, expect_success=True, retries=10, sleep_time=0.5
):
    """Ping the given TCP network port until it becomes available (for a given number of retries).
    If 'http_path' is set, make a GET request to this path and assert a non-error response."""
    return wait_for_port_status(
        port,
        http_path=http_path,
        expect_success=expect_success,
        retries=retries,
        sleep_time=sleep_time,
    )


def wait_for_port_closed(
    port: int, http_path: str = None, expect_success=True, retries=10, sleep_time=0.5
):
    return wait_for_port_status(
        port,
        http_path=http_path,
        expect_success=expect_success,
        retries=retries,
        sleep_time=sleep_time,
        expect_closed=True,
    )


def wait_for_port_status(
    port: int,
    http_path: str = None,
    expect_success=True,
    retries=10,
    sleep_time=0.5,
    expect_closed=False,
):
    """Ping the given TCP network port until it becomes (un)available (for a given number of retries)."""

    def check():
        status = is_port_open(port, http_path=http_path, expect_success=expect_success)
        if bool(status) != (not expect_closed):
            raise Exception(
                "Port %s (path: %s) was not %s"
                % (port, http_path, "closed" if expect_closed else "open")
            )

    return retry(check, sleep=sleep_time, retries=retries)


def port_can_be_bound(port: IntOrPort, address: str = "") -> bool:
    """
    Return whether a local port (TCP or UDP) can be bound to. Note that this is a stricter check
    than is_port_open(...) above, as is_port_open() may return False if the port is
    not accessible (i.e., does not respond), yet cannot be bound to.
    """
    try:
        port = Port.wrap(port)
        if port.protocol == "tcp":
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        elif port.protocol == "udp":
            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        else:
            LOG.debug("Unsupported network protocol '%s' for port check", port.protocol)
            return False
        sock.bind((address, port.port))
        return True
    except OSError:
        # either the port is used or we don't have permission to bind it
        return False
    except Exception:
        LOG.error("cannot bind port %s", port, exc_info=LOG.isEnabledFor(logging.DEBUG))
        return False


def get_free_udp_port(blocklist: list[int] = None) -> int:
    blocklist = blocklist or []
    for i in range(10):
        udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        udp.bind(("", 0))
        addr, port = udp.getsockname()
        udp.close()
        if port not in blocklist:
            return port
    raise Exception(f"Unable to determine free UDP port with blocklist {blocklist}")


def get_free_tcp_port(blocklist: list[int] = None) -> int:
    """
    Tries to bind a socket to port 0 and returns the port that was assigned by the system. If the port is
    in the given ``blocklist``, or the port is marked as reserved in ``dynamic_port_range``, the procedure
    is repeated for up to 50 times.

    :param blocklist: an optional list of ports that are not allowed as random ports
    :return: a free TCP port
    """
    blocklist = blocklist or []
    for i in range(50):
        tcp = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        tcp.bind(("", 0))
        addr, port = tcp.getsockname()
        tcp.close()
        if port not in blocklist and not dynamic_port_range.is_port_reserved(port):
            try:
                dynamic_port_range.mark_reserved(port)
            except ValueError:
                # depending on the ephemeral port range of the system, the allocated port may be outside what
                # we defined as dynamic port range
                pass
            return port
    raise Exception(f"Unable to determine free TCP port with blocklist {blocklist}")


def get_free_tcp_port_range(num_ports: int, max_attempts: int = 50) -> "PortRange":
    """
    Attempts to get a contiguous range of free ports from the dynamic port range. For instance,
    ``get_free_tcp_port_range(4)`` may return the following result: ``PortRange(44000:44004)``.

    :param num_ports: the number of ports in the range
    :param max_attempts: the number of times to retry if a contiguous range was not found
    :return: a port range of free TCP ports
    :raises PortNotAvailableException: if max_attempts was reached to re-try
    """
    if num_ports < 2:
        raise ValueError(f"invalid number of ports {num_ports}")

    def _is_port_range_free(_range: PortRange):
        for _port in _range:
            if dynamic_port_range.is_port_reserved(_port) or not port_can_be_bound(_port):
                return False
        return True

    for _ in range(max_attempts):
        # try to find a suitable starting point (leave enough space at the end)
        port_range_start = random.randint(
            dynamic_port_range.start, dynamic_port_range.end - num_ports - 1
        )
        port_range = PortRange(port_range_start, port_range_start + num_ports - 1)

        # check that each port in the range is available (has not been reserved and can be bound)
        # we don't use dynamic_port_range.reserve_port because in case the port range check fails at some port
        # all ports up until then would be reserved
        if not _is_port_range_free(port_range):
            continue

        # port range found! mark them as reserved in the dynamic port range and return
        for port in port_range:
            dynamic_port_range.mark_reserved(port)
        return port_range

    raise PortNotAvailableException("reached max_attempts when trying to find port range")


def resolve_hostname(hostname: str) -> Optional[str]:
    """Resolve the given hostname and return its IP address, or None if it cannot be resolved."""
    try:
        return socket.gethostbyname(hostname)
    except OSError:
        return None


def is_ip_address(addr: str) -> bool:
    try:
        socket.inet_aton(addr)
        return True
    except OSError:
        return False


def is_ipv4_address(address: str) -> bool:
    """
    Checks if passed string looks like an IPv4 address
    :param address: Possible IPv4 address
    :return: True if string looks like IPv4 address, False otherwise
    """
    return bool(re.match(IP_REGEX, address))


class PortNotAvailableException(Exception):
    """Exception which indicates that the PortRange could not reserve a port."""

    pass


class PortRange:
    """Manages a range of ports that can be reserved and requested."""

    def __init__(self, start: int, end: int):
        """
        Create a new port range. The port range is inclusive, meaning ``PortRange(5000,5005)`` is 6 ports
        including both 5000 and 5005. This is different from ``range`` which is not inclusive, i.e.::

            PortRange(5000, 5005).as_range() == range(5000, 5005 + 1)

        :param start: the start port (inclusive)
        :param end: the end of the range (inclusive).
        """
        self.start = start
        self.end = end

        # cache for locally available ports (ports are reserved for a short period of a few seconds)
        self._ports_cache: MutableMapping[Port, Any] = CustomExpiryTTLCache(
            maxsize=len(self),
            ttl=DEFAULT_PORT_RESERVED_SECONDS,
        )
        self._ports_lock = threading.RLock()

    def subrange(self, start: int = None, end: int = None) -> "PortRange":
        """
        Creates a new PortRange object from this range which is a sub-range of this port range. The new
        object will use the same port cache and locks of this port range, so you can constrain port
        reservations of an existing port range but have reservations synced between them.

        :param start: the start of the subrange
        :param end: the end of the subrange
        :raises ValueError: if start or end are outside the current port range
        :return: a new PortRange object synced to this one
        """
        start = start if start is not None else self.start
        end = end if end is not None else self.end

        if start < self.start:
            raise ValueError(f"start not in range ({start} < {self.start})")
        if end > self.end:
            raise ValueError(f"end not in range ({end} < {self.end})")

        # ensures that we return an instance of a subclass
        port_range = type(self)(start, end)
        port_range._ports_cache = self._ports_cache
        port_range._ports_lock = self._ports_lock
        return port_range

    def as_range(self) -> range:
        """
        Returns a ``range(start, end+1)`` object representing this port range.

        :return: a range
        """
        return range(self.start, self.end + 1)

    def reserve_port(self, port: Optional[IntOrPort] = None, duration: Optional[int] = None) -> int:
        """
        Reserves the given port (if it is still free). If the given port is None, it reserves a free port from the
        configured port range for external services. If a port is given, it has to be within the configured
        range of external services (i.e., in the range [self.start, self.end)).

        :param port: explicit port to check or None if a random port from the configured range should be selected
        :param duration: the time in seconds the port is reserved for (defaults to a few seconds)
        :return: reserved, free port number (int)
        :raises PortNotAvailableException: if the given port is outside the configured range, it is already bound or
                    reserved, or if the given port is none and there is no free port in the configured service range.
        """
        ports_range = self.as_range()
        port = Port.wrap(port) if port is not None else port
        if port is not None and port.port not in ports_range:
            raise PortNotAvailableException(
                f"The requested port ({port}) is not in the port range ({ports_range})."
            )
        with self._ports_lock:
            if port is not None:
                return self._try_reserve_port(port, duration=duration)
            else:
                for port_in_range in ports_range:
                    try:
                        return self._try_reserve_port(port_in_range, duration=duration)
                    except PortNotAvailableException:
                        # We ignore the fact that this single port is reserved, we just check the next one
                        pass
        raise PortNotAvailableException(
            f"No free network ports available in {self!r} (currently reserved: %s)",
            list(self._ports_cache.keys()),
        )

    def is_port_reserved(self, port: IntOrPort) -> bool:
        """
        Checks whether the port has been reserved in this PortRange. Does not check whether the port can be
        bound or not, and does not check whether the port is in range.

        :param port: the port to check
        :return: true if the port is reserved within the range
        """
        port = Port.wrap(port)
        return self._ports_cache.get(port) is not None

    def mark_reserved(self, port: IntOrPort, duration: int = None):
        """
        Marks the given port as reserved for the given duration, regardless of whether it is free for not.

        :param port: the port to reserve
        :param duration: the duration
        :raises ValueError: if the port is not in this port range
        """
        port = Port.wrap(port)

        if port.port not in self.as_range():
            raise ValueError(f"port {port} not in {self!r}")

        with self._ports_lock:
            # reserve the port for a short period of time
            self._ports_cache[port] = "__reserved__"
            if duration:
                self._ports_cache.set_expiry(port, duration)

    def _try_reserve_port(self, port: IntOrPort, duration: int) -> int:
        """Checks if the given port is currently not reserved and can be bound."""
        port = Port.wrap(port)

        if self.is_port_reserved(port):
            raise PortNotAvailableException(f"The given port ({port}) is already reserved.")
        if not self._port_can_be_bound(port):
            raise PortNotAvailableException(f"The given port ({port}) is already in use.")

        self.mark_reserved(port, duration)
        return port.port

    def _port_can_be_bound(self, port: IntOrPort) -> bool:
        """
        Internal check whether the port can be bound. Will open a socket connection and see if the port is
        available. Can be overwritten by subclasses to provide a custom implementation.

        :param port: the port to check
        :return: true if the port is free on the system
        """
        return port_can_be_bound(port)

    def __len__(self):
        return self.end - self.start + 1

    def __iter__(self):
        return self.as_range().__iter__()

    def __repr__(self):
        return f"PortRange({self.start}:{self.end})"


@singleton_factory
def get_docker_host_from_container() -> str:
    """
    Get the hostname/IP to connect to the host from within a Docker container (e.g., Lambda function).
    The logic is roughly as follows:
      1. return `host.docker.internal` if we're running in host mode, in a non-Linux OS
      2. return the IP address that `host.docker.internal` (or alternatively `host.containers.internal`)
        resolves to, if we're inside Docker
      3. return the Docker bridge IP (config.DOCKER_BRIDGE_IP) as a fallback, if option (2) fails
    """
    result = config.DOCKER_BRIDGE_IP
    try:
        if not config.is_in_docker and not config.is_in_linux:
            # If we're running outside Docker (in host mode), and would like the Lambda containers to be able
            # to access services running on the local machine, return `host.docker.internal` accordingly
            result = "host.docker.internal"
        if config.is_in_docker:
            try:
                result = socket.gethostbyname("host.docker.internal")
            except OSError:
                result = socket.gethostbyname("host.containers.internal")
    except OSError:
        # TODO if neither host resolves, we might be in linux. We could just use the default gateway then
        pass
    return result


def get_addressable_container_host(default_local_hostname: str = None) -> str:
    """
    Return the target host to address endpoints exposed by Docker containers, depending on
    the current execution context.

    If we're currently executing within Docker, then return get_docker_host_from_container(); otherwise, return
    the value of `LOCALHOST_HOSTNAME`, assuming that container endpoints are exposed and accessible under localhost.

    :param default_local_hostname: local hostname to return, if running outside Docker (defaults to LOCALHOST_HOSTNAME)
    """
    default_local_hostname = default_local_hostname or constants.LOCALHOST_HOSTNAME
    return get_docker_host_from_container() if config.is_in_docker else default_local_hostname


def send_dns_query(
    name: str,
    port: int = 53,
    ip_address: str = "127.0.0.1",
    qtype: str = "A",
    timeout: float = 1.0,
    tcp: bool = False,
) -> DNSRecord:
    LOG.debug("querying %s:%d for name %s", ip_address, port, name)
    request = DNSRecord.question(qname=name, qtype=qtype)
    reply_bytes = request.send(dest=ip_address, port=port, tcp=tcp, timeout=timeout, ipv6=False)
    return DNSRecord.parse(reply_bytes)


dynamic_port_range = PortRange(DYNAMIC_PORT_RANGE_START, DYNAMIC_PORT_RANGE_END)
"""The dynamic port range."""

1	import logging	1✔
2	import random	1✔
3	import re	1✔
4	import socket	1✔
5	import threading	1✔
6	from collections.abc import MutableMapping	1✔
7	from contextlib import closing	1✔
8	from typing import Any, NamedTuple, Optional, Union	1✔
9	from urllib.parse import urlparse	1✔
10
11	import dns.resolver	1✔
12	from dnslib import DNSRecord	1✔
13
14	from localstack import config, constants	1✔
15
16	from .collections import CustomExpiryTTLCache	1✔
17	from .numbers import is_number	1✔
18	from .objects import singleton_factory	1✔
19	from .sync import retry	1✔
20
21	LOG = logging.getLogger(__name__)	1✔
22
23	# regular expression for IPv4 addresses
24	IP_REGEX = (	1✔
25	r"^(?:(?:25[0-5]\|2[0-4][0-9]\|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]\|2[0-4][0-9]\|[01]?[0-9][0-9]?)$"
26	)
27
28	# many linux kernels use 32768-60999, RFC 6335 is 49152-65535, so we use a mix here
29	DYNAMIC_PORT_RANGE_START = 32768	1✔
30	DYNAMIC_PORT_RANGE_END = 65536	1✔
31
32	DEFAULT_PORT_RESERVED_SECONDS = 6	1✔
33	"""Default nuber of seconds a port is reserved in a PortRange."""	1✔
34
35
36	class Port(NamedTuple):	1✔
37	"""Represents a network port, with port number and protocol (TCP/UDP)"""
38
39	port: int	1✔
40	"""the port number"""	1✔
41	protocol: str	1✔
42	"""network protocol name (usually 'tcp' or 'udp')"""	1✔
43
44	@classmethod	1✔
45	def wrap(cls, port: "IntOrPort") -> "Port":	1✔
46	"""Return the given port as a Port object, using 'tcp' as the default protocol."""
47	if isinstance(port, Port):	1✔
48	return port	1✔
49	return Port(port=port, protocol="tcp")	1✔
50
51
52	# simple helper type to encapsulate int/Port argument types
53	IntOrPort = Union[int, Port]	1✔
54
55
56	def is_port_open(	1✔
57	port_or_url: Union[int, str],
58	http_path: str = None,
59	expect_success: bool = True,
60	protocols: Optional[Union[str, list[str]]] = None,
61	quiet: bool = True,
62	):
63	from localstack.utils.http import safe_requests	1✔
64
65	protocols = protocols or ["tcp"]	1✔
66	port = port_or_url	1✔
67	if is_number(port):	1✔
68	port = int(port)	1✔
69	host = "localhost"	1✔
70	protocol = "http"	1✔
71	protocols = protocols if isinstance(protocols, list) else [protocols]	1✔
72	if isinstance(port, str):	1✔
73	url = urlparse(port_or_url)	1✔
74	port = url.port	1✔
75	host = url.hostname	1✔
76	protocol = url.scheme	1✔
77	nw_protocols = []	1✔
78	nw_protocols += [socket.SOCK_STREAM] if "tcp" in protocols else []	1✔
79	nw_protocols += [socket.SOCK_DGRAM] if "udp" in protocols else []	1✔
80	for nw_protocol in nw_protocols:	1✔
81	with closing(	1✔
82	socket.socket(socket.AF_INET if ":" not in host else socket.AF_INET6, nw_protocol)
83	) as sock:
84	sock.settimeout(1)	1✔
85	if nw_protocol == socket.SOCK_DGRAM:	1✔
86	try:	×
87	if port == 53:	×
88	dnshost = "127.0.0.1" if host == "localhost" else host	×
89	resolver = dns.resolver.Resolver()	×
90	resolver.nameservers = [dnshost]	×
91	resolver.timeout = 1	×
92	resolver.lifetime = 1	×
93	answers = resolver.query("google.com", "A")	×
94	assert len(answers) > 0	×
95	else:
96	sock.sendto(b"", (host, port))	×
97	sock.recvfrom(1024)	×
98	except Exception:	×
99	if not quiet:	×
100	LOG.error(	×
101	"Error connecting to UDP port %s:%s",
102	host,
103	port,
104	exc_info=LOG.isEnabledFor(logging.DEBUG),
105	)
106	return False	×
107	elif nw_protocol == socket.SOCK_STREAM:	1✔
108	result = sock.connect_ex((host, port))	1✔
109	if result != 0:	1✔
110	if not quiet:	1✔
UNCOV 111	LOG.warning(	×
112	"Error connecting to TCP port %s:%s (result=%s)", host, port, result
113	)
114	return False	1✔
115	if "tcp" not in protocols or not http_path:	1✔
116	return True	1✔
117	host = f"[{host}]" if ":" in host else host	×
118	url = f"{protocol}://{host}:{port}{http_path}"	×
UNCOV 119	try:	×
UNCOV 120	response = safe_requests.get(url, verify=False)	×
UNCOV 121	return not expect_success or response.status_code < 400	×
UNCOV 122	except Exception:	×
UNCOV 123	return False	×
124
125
126	def wait_for_port_open(	1✔
127	port: int, http_path: str = None, expect_success=True, retries=10, sleep_time=0.5
128	):
129	"""Ping the given TCP network port until it becomes available (for a given number of retries).
130	If 'http_path' is set, make a GET request to this path and assert a non-error response."""
131	return wait_for_port_status(	1✔
132	port,
133	http_path=http_path,
134	expect_success=expect_success,
135	retries=retries,
136	sleep_time=sleep_time,
137	)
138
139
140	def wait_for_port_closed(	1✔
141	port: int, http_path: str = None, expect_success=True, retries=10, sleep_time=0.5
142	):
143	return wait_for_port_status(	1✔
144	port,
145	http_path=http_path,
146	expect_success=expect_success,
147	retries=retries,
148	sleep_time=sleep_time,
149	expect_closed=True,
150	)
151
152
153	def wait_for_port_status(	1✔
154	port: int,
155	http_path: str = None,
156	expect_success=True,
157	retries=10,
158	sleep_time=0.5,
159	expect_closed=False,
160	):
161	"""Ping the given TCP network port until it becomes (un)available (for a given number of retries)."""
162
163	def check():	1✔
164	status = is_port_open(port, http_path=http_path, expect_success=expect_success)	1✔
165	if bool(status) != (not expect_closed):	1✔
UNCOV 166	raise Exception(	×
167	"Port %s (path: %s) was not %s"
168	% (port, http_path, "closed" if expect_closed else "open")
169	)
170
171	return retry(check, sleep=sleep_time, retries=retries)	1✔
172
173
174	def port_can_be_bound(port: IntOrPort, address: str = "") -> bool:	1✔
175	"""
176	Return whether a local port (TCP or UDP) can be bound to. Note that this is a stricter check
177	than is_port_open(...) above, as is_port_open() may return False if the port is
178	not accessible (i.e., does not respond), yet cannot be bound to.
179	"""
180	try:	1✔
181	port = Port.wrap(port)	1✔
182	if port.protocol == "tcp":	1✔
183	sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)	1✔
184	elif port.protocol == "udp":	1✔
185	sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)	1✔
186	else:
UNCOV 187	LOG.debug("Unsupported network protocol '%s' for port check", port.protocol)	×
UNCOV 188	return False	×
189	sock.bind((address, port.port))	1✔
190	return True	1✔
191	except OSError:	1✔
192	# either the port is used or we don't have permission to bind it
193	return False	1✔
194	except Exception:	1✔
195	LOG.error("cannot bind port %s", port, exc_info=LOG.isEnabledFor(logging.DEBUG))	1✔
196	return False	1✔
197
198
199	def get_free_udp_port(blocklist: list[int] = None) -> int:	1✔
200	blocklist = blocklist or []	1✔
201	for i in range(10):	1✔
202	udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)	1✔
203	udp.bind(("", 0))	1✔
204	addr, port = udp.getsockname()	1✔
205	udp.close()	1✔
206	if port not in blocklist:	1✔
207	return port	1✔
UNCOV 208	raise Exception(f"Unable to determine free UDP port with blocklist {blocklist}")	×
209
210
211	def get_free_tcp_port(blocklist: list[int] = None) -> int:	1✔
212	"""
213	Tries to bind a socket to port 0 and returns the port that was assigned by the system. If the port is
214	in the given ``blocklist``, or the port is marked as reserved in ``dynamic_port_range``, the procedure
215	is repeated for up to 50 times.
216
217	:param blocklist: an optional list of ports that are not allowed as random ports
218	:return: a free TCP port
219	"""
220	blocklist = blocklist or []	1✔
221	for i in range(50):	1✔
222	tcp = socket.socket(socket.AF_INET, socket.SOCK_STREAM)	1✔
223	tcp.bind(("", 0))	1✔
224	addr, port = tcp.getsockname()	1✔
225	tcp.close()	1✔
226	if port not in blocklist and not dynamic_port_range.is_port_reserved(port):	1✔
227	try:	1✔
228	dynamic_port_range.mark_reserved(port)	1✔
UNCOV 229	except ValueError:	×
230	# depending on the ephemeral port range of the system, the allocated port may be outside what
231	# we defined as dynamic port range
UNCOV 232	pass	×
233	return port	1✔
234	raise Exception(f"Unable to determine free TCP port with blocklist {blocklist}")	1✔
235
236
237	def get_free_tcp_port_range(num_ports: int, max_attempts: int = 50) -> "PortRange":	1✔
238	"""
239	Attempts to get a contiguous range of free ports from the dynamic port range. For instance,
240	``get_free_tcp_port_range(4)`` may return the following result: ``PortRange(44000:44004)``.
241
242	:param num_ports: the number of ports in the range
243	:param max_attempts: the number of times to retry if a contiguous range was not found
244	:return: a port range of free TCP ports
245	:raises PortNotAvailableException: if max_attempts was reached to re-try
246	"""
247	if num_ports < 2:	1✔
UNCOV 248	raise ValueError(f"invalid number of ports {num_ports}")	×
249
250	def _is_port_range_free(_range: PortRange):	1✔
251	for _port in _range:	1✔
252	if dynamic_port_range.is_port_reserved(_port) or not port_can_be_bound(_port):	1✔
253	return False	1✔
254	return True	1✔
255
256	for _ in range(max_attempts):	1✔
257	# try to find a suitable starting point (leave enough space at the end)
258	port_range_start = random.randint(	1✔
259	dynamic_port_range.start, dynamic_port_range.end - num_ports - 1
260	)
261	port_range = PortRange(port_range_start, port_range_start + num_ports - 1)	1✔
262
263	# check that each port in the range is available (has not been reserved and can be bound)
264	# we don't use dynamic_port_range.reserve_port because in case the port range check fails at some port
265	# all ports up until then would be reserved
266	if not _is_port_range_free(port_range):	1✔
267	continue	1✔
268
269	# port range found! mark them as reserved in the dynamic port range and return
270	for port in port_range:	1✔
271	dynamic_port_range.mark_reserved(port)	1✔
272	return port_range	1✔
273
274	raise PortNotAvailableException("reached max_attempts when trying to find port range")	1✔
275
276
277	def resolve_hostname(hostname: str) -> Optional[str]:	1✔
278	"""Resolve the given hostname and return its IP address, or None if it cannot be resolved."""
279	try:	1✔
280	return socket.gethostbyname(hostname)	1✔
281	except OSError:	1✔
282	return None	1✔
283
284
285	def is_ip_address(addr: str) -> bool:	1✔
286	try:	1✔
287	socket.inet_aton(addr)	1✔
288	return True	1✔
289	except OSError:	1✔
290	return False	1✔
291
292
293	def is_ipv4_address(address: str) -> bool:	1✔
294	"""
295	Checks if passed string looks like an IPv4 address
296	:param address: Possible IPv4 address
297	:return: True if string looks like IPv4 address, False otherwise
298	"""
299	return bool(re.match(IP_REGEX, address))	1✔
300
301
302	class PortNotAvailableException(Exception):	1✔
303	"""Exception which indicates that the PortRange could not reserve a port."""
304
305	pass	1✔
306
307
308	class PortRange:	1✔
309	"""Manages a range of ports that can be reserved and requested."""
310
311	def __init__(self, start: int, end: int):	1✔
312	"""
313	Create a new port range. The port range is inclusive, meaning ``PortRange(5000,5005)`` is 6 ports
314	including both 5000 and 5005. This is different from ``range`` which is not inclusive, i.e.::
315
316	PortRange(5000, 5005).as_range() == range(5000, 5005 + 1)
317
318	:param start: the start port (inclusive)
319	:param end: the end of the range (inclusive).
320	"""
321	self.start = start	1✔
322	self.end = end	1✔
323
324	# cache for locally available ports (ports are reserved for a short period of a few seconds)
325	self._ports_cache: MutableMapping[Port, Any] = CustomExpiryTTLCache(	1✔
326	maxsize=len(self),
327	ttl=DEFAULT_PORT_RESERVED_SECONDS,
328	)
329	self._ports_lock = threading.RLock()	1✔
330
331	def subrange(self, start: int = None, end: int = None) -> "PortRange":	1✔
332	"""
333	Creates a new PortRange object from this range which is a sub-range of this port range. The new
334	object will use the same port cache and locks of this port range, so you can constrain port
335	reservations of an existing port range but have reservations synced between them.
336
337	:param start: the start of the subrange
338	:param end: the end of the subrange
339	:raises ValueError: if start or end are outside the current port range
340	:return: a new PortRange object synced to this one
341	"""
342	start = start if start is not None else self.start	1✔
343	end = end if end is not None else self.end	1✔
344
345	if start < self.start:	1✔
UNCOV 346	raise ValueError(f"start not in range ({start} < {self.start})")	×
347	if end > self.end:	1✔
UNCOV 348	raise ValueError(f"end not in range ({end} < {self.end})")	×
349
350	# ensures that we return an instance of a subclass
351	port_range = type(self)(start, end)	1✔
352	port_range._ports_cache = self._ports_cache	1✔
353	port_range._ports_lock = self._ports_lock	1✔
354	return port_range	1✔
355
356	def as_range(self) -> range:	1✔
357	"""
358	Returns a ``range(start, end+1)`` object representing this port range.
359
360	:return: a range
361	"""
362	return range(self.start, self.end + 1)	1✔
363
364	def reserve_port(self, port: Optional[IntOrPort] = None, duration: Optional[int] = None) -> int:	1✔
365	"""
366	Reserves the given port (if it is still free). If the given port is None, it reserves a free port from the
367	configured port range for external services. If a port is given, it has to be within the configured
368	range of external services (i.e., in the range [self.start, self.end)).
369
370	:param port: explicit port to check or None if a random port from the configured range should be selected
371	:param duration: the time in seconds the port is reserved for (defaults to a few seconds)
372	:return: reserved, free port number (int)
373	:raises PortNotAvailableException: if the given port is outside the configured range, it is already bound or
374	reserved, or if the given port is none and there is no free port in the configured service range.
375	"""
376	ports_range = self.as_range()	1✔
377	port = Port.wrap(port) if port is not None else port	1✔
378	if port is not None and port.port not in ports_range:	1✔
379	raise PortNotAvailableException(	1✔
380	f"The requested port ({port}) is not in the port range ({ports_range})."
381	)
382	with self._ports_lock:	1✔
383	if port is not None:	1✔
384	return self._try_reserve_port(port, duration=duration)	1✔
385	else:
386	for port_in_range in ports_range:	1✔
387	try:	1✔
388	return self._try_reserve_port(port_in_range, duration=duration)	1✔
389	except PortNotAvailableException:	1✔
390	# We ignore the fact that this single port is reserved, we just check the next one
391	pass	1✔
392	raise PortNotAvailableException(	1✔
393	f"No free network ports available in {self!r} (currently reserved: %s)",
394	list(self._ports_cache.keys()),
395	)
396
397	def is_port_reserved(self, port: IntOrPort) -> bool:	1✔
398	"""
399	Checks whether the port has been reserved in this PortRange. Does not check whether the port can be
400	bound or not, and does not check whether the port is in range.
401
402	:param port: the port to check
403	:return: true if the port is reserved within the range
404	"""
405	port = Port.wrap(port)	1✔
406	return self._ports_cache.get(port) is not None	1✔
407
408	def mark_reserved(self, port: IntOrPort, duration: int = None):	1✔
409	"""
410	Marks the given port as reserved for the given duration, regardless of whether it is free for not.
411
412	:param port: the port to reserve
413	:param duration: the duration
414	:raises ValueError: if the port is not in this port range
415	"""
416	port = Port.wrap(port)	1✔
417
418	if port.port not in self.as_range():	1✔
UNCOV 419	raise ValueError(f"port {port} not in {self!r}")	×
420
421	with self._ports_lock:	1✔
422	# reserve the port for a short period of time
423	self._ports_cache[port] = "__reserved__"	1✔
424	if duration:	1✔
425	self._ports_cache.set_expiry(port, duration)	1✔
426
427	def _try_reserve_port(self, port: IntOrPort, duration: int) -> int:	1✔
428	"""Checks if the given port is currently not reserved and can be bound."""
429	port = Port.wrap(port)	1✔
430
431	if self.is_port_reserved(port):	1✔
432	raise PortNotAvailableException(f"The given port ({port}) is already reserved.")	1✔
433	if not self._port_can_be_bound(port):	1✔
UNCOV 434	raise PortNotAvailableException(f"The given port ({port}) is already in use.")	×
435
436	self.mark_reserved(port, duration)	1✔
437	return port.port	1✔
438
439	def _port_can_be_bound(self, port: IntOrPort) -> bool:	1✔
440	"""
441	Internal check whether the port can be bound. Will open a socket connection and see if the port is
442	available. Can be overwritten by subclasses to provide a custom implementation.
443
444	:param port: the port to check
445	:return: true if the port is free on the system
446	"""
447	return port_can_be_bound(port)	1✔
448
449	def __len__(self):	1✔
450	return self.end - self.start + 1	1✔
451
452	def __iter__(self):	1✔
453	return self.as_range().__iter__()	1✔
454
455	def __repr__(self):
456	return f"PortRange({self.start}:{self.end})"
457
458
459	@singleton_factory	1✔
460	def get_docker_host_from_container() -> str:	1✔
461	"""
462	Get the hostname/IP to connect to the host from within a Docker container (e.g., Lambda function).
463	The logic is roughly as follows:
464	1. return `host.docker.internal` if we're running in host mode, in a non-Linux OS
465	2. return the IP address that `host.docker.internal` (or alternatively `host.containers.internal`)
466	resolves to, if we're inside Docker
467	3. return the Docker bridge IP (config.DOCKER_BRIDGE_IP) as a fallback, if option (2) fails
468	"""
469	result = config.DOCKER_BRIDGE_IP	1✔
470	try:	1✔
471	if not config.is_in_docker and not config.is_in_linux:	1✔
472	# If we're running outside Docker (in host mode), and would like the Lambda containers to be able
473	# to access services running on the local machine, return `host.docker.internal` accordingly
UNCOV 474	result = "host.docker.internal"	×
475	if config.is_in_docker:	1✔
476	try:	1✔
477	result = socket.gethostbyname("host.docker.internal")	1✔
478	except OSError:	1✔
479	result = socket.gethostbyname("host.containers.internal")	1✔
480	except OSError:	1✔
481	# TODO if neither host resolves, we might be in linux. We could just use the default gateway then
482	pass	1✔
483	return result	1✔
484
485
486	def get_addressable_container_host(default_local_hostname: str = None) -> str:	1✔
487	"""
488	Return the target host to address endpoints exposed by Docker containers, depending on
489	the current execution context.
490
491	If we're currently executing within Docker, then return get_docker_host_from_container(); otherwise, return
492	the value of `LOCALHOST_HOSTNAME`, assuming that container endpoints are exposed and accessible under localhost.
493
494	:param default_local_hostname: local hostname to return, if running outside Docker (defaults to LOCALHOST_HOSTNAME)
495	"""
496	default_local_hostname = default_local_hostname or constants.LOCALHOST_HOSTNAME	1✔
497	return get_docker_host_from_container() if config.is_in_docker else default_local_hostname	1✔
498
499
500	def send_dns_query(	1✔
501	name: str,
502	port: int = 53,
503	ip_address: str = "127.0.0.1",
504	qtype: str = "A",
505	timeout: float = 1.0,
506	tcp: bool = False,
507	) -> DNSRecord:
UNCOV 508	LOG.debug("querying %s:%d for name %s", ip_address, port, name)	×
UNCOV 509	request = DNSRecord.question(qname=name, qtype=qtype)	×
UNCOV 510	reply_bytes = request.send(dest=ip_address, port=port, tcp=tcp, timeout=timeout, ipv6=False)	×
UNCOV 511	return DNSRecord.parse(reply_bytes)	×
512
513
514	dynamic_port_range = PortRange(DYNAMIC_PORT_RANGE_START, DYNAMIC_PORT_RANGE_END)	1✔
515	"""The dynamic port range."""	1✔

localstack / localstack / 16981563750

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous