20565403496

Committed 29 Dec 2025 05:11AM UTC coverage: 84.103% (-2.8%) from 86.921%

Build # 20565403496

Build Type

Pull #13567

github

Committed by

web-flow

Commit Message

Merge 4816837a5 into 2417384aa

Pull Request Pull Request #13567: Update ASF APIs

Run Details

67166 of 79862 relevant lines covered (84.1%)

0.84 hits per line

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

44.9

/localstack-core/localstack/services/edge.py

import argparse
import logging
import shlex
import subprocess
import sys
from typing import TypeVar

from localstack import config, constants
from localstack.config import HostAndPort
from localstack.constants import (
    LOCALSTACK_ROOT_FOLDER,
)
from localstack.http import Router
from localstack.http.dispatcher import Handler, handler_dispatcher
from localstack.http.router import GreedyPathConverter
from localstack.utils.collections import split_list_by
from localstack.utils.net import get_free_tcp_port
from localstack.utils.run import is_root, run
from localstack.utils.server.tcp_proxy import TCPProxy
from localstack.utils.threads import start_thread

T = TypeVar("T")

LOG = logging.getLogger(__name__)


ROUTER: Router[Handler] = Router(
    dispatcher=handler_dispatcher(), converters={"greedy_path": GreedyPathConverter}
)
"""This special Router is part of the edge proxy. Use the router to inject custom handlers that are handled before
the actual AWS service call is made."""


def do_start_edge(
    listen: HostAndPort | list[HostAndPort], use_ssl: bool, asynchronous: bool = False
):
    from localstack.aws.serving.edge import serve_gateway

    return serve_gateway(listen, use_ssl, asynchronous)


def can_use_sudo():
    try:
        run("sudo -n -v", print_error=False)
        return True
    except Exception:
        return False


def ensure_can_use_sudo():
    if not is_root() and not can_use_sudo():
        if not sys.stdin.isatty():
            raise OSError("cannot get sudo password from non-tty input")
        print("Please enter your sudo password (required to configure local network):")
        run("sudo -v", stdin=True)


def start_component(
    component: str, listen_str: str | None = None, target_address: str | None = None
):
    if component == "edge":
        return start_edge(listen_str=listen_str)
    if component == "proxy":
        if target_address is None:
            raise ValueError("no target address specified")

        return start_proxy(
            listen_str=listen_str,
            target_address=HostAndPort.parse(
                target_address,
                default_host=config.default_ip,
                default_port=constants.DEFAULT_PORT_EDGE,
            ),
        )
    raise Exception(f"Unexpected component name '{component}' received during start up")


def start_proxy(
    listen_str: str, target_address: HostAndPort, asynchronous: bool = False
) -> TCPProxy:
    """
    Starts a TCP proxy to perform a low-level forwarding of incoming requests.

    :param listen_str: address to listen on
    :param target_address: target address to proxy requests to
    :param asynchronous: False if the function should join the proxy thread and block until it terminates.
    :return: created thread executing the proxy
    """
    listen_hosts = parse_gateway_listen(
        listen_str,
        default_host=constants.LOCALHOST_IP,
        default_port=constants.DEFAULT_PORT_EDGE,
    )
    listen = listen_hosts[0]
    return do_start_tcp_proxy(listen, target_address, asynchronous)


def do_start_tcp_proxy(
    listen: HostAndPort, target_address: HostAndPort, asynchronous: bool = False
) -> TCPProxy:
    src = str(listen)
    dst = str(target_address)

    LOG.debug("Starting Local TCP Proxy: %s -> %s", src, dst)
    proxy = TCPProxy(
        target_address=target_address.host,
        target_port=target_address.port,
        host=listen.host,
        port=listen.port,
    )
    proxy.start()
    if not asynchronous:
        proxy.join()
    return proxy


def start_edge(listen_str: str, use_ssl: bool = True, asynchronous: bool = False):
    if listen_str:
        listen = parse_gateway_listen(
            listen_str, default_host=config.default_ip, default_port=constants.DEFAULT_PORT_EDGE
        )
    else:
        listen = config.GATEWAY_LISTEN

    if len(listen) == 0:
        raise ValueError("no listen addresses provided")

    # separate privileged and unprivileged addresses
    unprivileged, privileged = split_list_by(listen, lambda addr: addr.is_unprivileged() or False)

    # if we are root, we can directly bind to privileged ports as well
    if is_root():
        unprivileged = unprivileged + privileged
        privileged = []

    # check that we are actually started the gateway server
    if not unprivileged:
        unprivileged = parse_gateway_listen(
            f":{get_free_tcp_port()}",
            default_host=config.default_ip,
            default_port=constants.DEFAULT_PORT_EDGE,
        )

    # bind the gateway server to unprivileged addresses
    edge_thread = do_start_edge(unprivileged, use_ssl=use_ssl, asynchronous=True)

    # start TCP proxies for the remaining addresses
    proxy_destination = unprivileged[0]
    for address in privileged:
        # escalate to root
        args = [
            "proxy",
            "--gateway-listen",
            str(address),
            "--target-address",
            str(proxy_destination),
        ]
        run_module_as_sudo(
            module="localstack.services.edge",
            arguments=args,
            asynchronous=True,
        )

    if edge_thread is not None:
        edge_thread.join()


def run_module_as_sudo(
    module: str, arguments: list[str] | None = None, asynchronous=False, env_vars=None
):
    # prepare environment
    env_vars = env_vars or {}
    env_vars["PYTHONPATH"] = f".:{LOCALSTACK_ROOT_FOLDER}"

    # start the process as sudo
    python_cmd = sys.executable
    cmd = ["sudo", "-n", "--preserve-env", python_cmd, "-m", module]
    arguments = arguments or []
    shell_cmd = shlex.join(cmd + arguments)

    # make sure we can run sudo commands
    try:
        ensure_can_use_sudo()
    except Exception as e:
        LOG.error("cannot run command as root (%s): %s ", str(e), shell_cmd)
        return

    def run_command(*_):
        run(shell_cmd, outfile=subprocess.PIPE, print_error=False, env_vars=env_vars)

    LOG.debug("Running command as sudo: %s", shell_cmd)
    result = (
        start_thread(run_command, quiet=True, name="sudo-edge") if asynchronous else run_command()
    )
    return result


def parse_gateway_listen(listen: str, default_host: str, default_port: int) -> list[HostAndPort]:
    addresses = []
    for address in listen.split(","):
        addresses.append(HostAndPort.parse(address, default_host, default_port))
    return addresses


if __name__ == "__main__":
    logging.basicConfig()
    parser = argparse.ArgumentParser()
    parser.add_argument("component")
    parser.add_argument("-l", "--gateway-listen", required=False, type=str)
    parser.add_argument("-t", "--target-address", required=False, type=str)
    args = parser.parse_args()

    start_component(args.component, args.gateway_listen, args.target_address)

1	import argparse	1✔
2	import logging	1✔
3	import shlex	1✔
4	import subprocess	1✔
5	import sys	1✔
6	from typing import TypeVar	1✔
7
8	from localstack import config, constants	1✔
9	from localstack.config import HostAndPort	1✔
10	from localstack.constants import (	1✔
11	LOCALSTACK_ROOT_FOLDER,
12	)
13	from localstack.http import Router	1✔
14	from localstack.http.dispatcher import Handler, handler_dispatcher	1✔
15	from localstack.http.router import GreedyPathConverter	1✔
16	from localstack.utils.collections import split_list_by	1✔
17	from localstack.utils.net import get_free_tcp_port	1✔
18	from localstack.utils.run import is_root, run	1✔
19	from localstack.utils.server.tcp_proxy import TCPProxy	1✔
20	from localstack.utils.threads import start_thread	1✔
21
22	T = TypeVar("T")	1✔
23
24	LOG = logging.getLogger(__name__)	1✔
25
26
27	ROUTER: Router[Handler] = Router(	1✔
28	dispatcher=handler_dispatcher(), converters={"greedy_path": GreedyPathConverter}
29	)
30	"""This special Router is part of the edge proxy. Use the router to inject custom handlers that are handled before	1✔
31	the actual AWS service call is made."""
32
33
34	def do_start_edge(	1✔
35	listen: HostAndPort \| list[HostAndPort], use_ssl: bool, asynchronous: bool = False
36	):
37	from localstack.aws.serving.edge import serve_gateway	×
38
39	return serve_gateway(listen, use_ssl, asynchronous)	×
40
41
42	def can_use_sudo():	1✔
43	try:	×
44	run("sudo -n -v", print_error=False)	×
45	return True	×
46	except Exception:	×
47	return False	×
48
49
50	def ensure_can_use_sudo():	1✔
51	if not is_root() and not can_use_sudo():	×
52	if not sys.stdin.isatty():	×
53	raise OSError("cannot get sudo password from non-tty input")	×
54	print("Please enter your sudo password (required to configure local network):")	×
55	run("sudo -v", stdin=True)	×
56
57
58	def start_component(	1✔
59	component: str, listen_str: str \| None = None, target_address: str \| None = None
60	):
61	if component == "edge":	×
62	return start_edge(listen_str=listen_str)	×
63	if component == "proxy":	×
64	if target_address is None:	×
65	raise ValueError("no target address specified")	×
66
67	return start_proxy(	×
68	listen_str=listen_str,
69	target_address=HostAndPort.parse(
70	target_address,
71	default_host=config.default_ip,
72	default_port=constants.DEFAULT_PORT_EDGE,
73	),
74	)
75	raise Exception(f"Unexpected component name '{component}' received during start up")	×
76
77
78	def start_proxy(	1✔
79	listen_str: str, target_address: HostAndPort, asynchronous: bool = False
80	) -> TCPProxy:
81	"""
82	Starts a TCP proxy to perform a low-level forwarding of incoming requests.
83
84	:param listen_str: address to listen on
85	:param target_address: target address to proxy requests to
86	:param asynchronous: False if the function should join the proxy thread and block until it terminates.
87	:return: created thread executing the proxy
88	"""
89	listen_hosts = parse_gateway_listen(	1✔
90	listen_str,
91	default_host=constants.LOCALHOST_IP,
92	default_port=constants.DEFAULT_PORT_EDGE,
93	)
94	listen = listen_hosts[0]	1✔
95	return do_start_tcp_proxy(listen, target_address, asynchronous)	1✔
96
97
98	def do_start_tcp_proxy(	1✔
99	listen: HostAndPort, target_address: HostAndPort, asynchronous: bool = False
100	) -> TCPProxy:
101	src = str(listen)	1✔
102	dst = str(target_address)	1✔
103
104	LOG.debug("Starting Local TCP Proxy: %s -> %s", src, dst)	1✔
105	proxy = TCPProxy(	1✔
106	target_address=target_address.host,
107	target_port=target_address.port,
108	host=listen.host,
109	port=listen.port,
110	)
111	proxy.start()	1✔
112	if not asynchronous:	1✔
113	proxy.join()	×
114	return proxy	1✔
115
116
117	def start_edge(listen_str: str, use_ssl: bool = True, asynchronous: bool = False):	1✔
118	if listen_str:	×
119	listen = parse_gateway_listen(	×
120	listen_str, default_host=config.default_ip, default_port=constants.DEFAULT_PORT_EDGE
121	)
122	else:
123	listen = config.GATEWAY_LISTEN	×
124
125	if len(listen) == 0:	×
126	raise ValueError("no listen addresses provided")	×
127
128	# separate privileged and unprivileged addresses
129	unprivileged, privileged = split_list_by(listen, lambda addr: addr.is_unprivileged() or False)	×
130
131	# if we are root, we can directly bind to privileged ports as well
132	if is_root():	×
133	unprivileged = unprivileged + privileged	×
134	privileged = []	×
135
136	# check that we are actually started the gateway server
137	if not unprivileged:	×
138	unprivileged = parse_gateway_listen(	×
139	f":{get_free_tcp_port()}",
140	default_host=config.default_ip,
141	default_port=constants.DEFAULT_PORT_EDGE,
142	)
143
144	# bind the gateway server to unprivileged addresses
145	edge_thread = do_start_edge(unprivileged, use_ssl=use_ssl, asynchronous=True)	×
146
147	# start TCP proxies for the remaining addresses
148	proxy_destination = unprivileged[0]	×
149	for address in privileged:	×
150	# escalate to root
151	args = [	×
152	"proxy",
153	"--gateway-listen",
154	str(address),
155	"--target-address",
156	str(proxy_destination),
157	]
158	run_module_as_sudo(	×
159	module="localstack.services.edge",
160	arguments=args,
161	asynchronous=True,
162	)
163
164	if edge_thread is not None:	×
165	edge_thread.join()	×
166
167
168	def run_module_as_sudo(	1✔
169	module: str, arguments: list[str] \| None = None, asynchronous=False, env_vars=None
170	):
171	# prepare environment
172	env_vars = env_vars or {}	×
173	env_vars["PYTHONPATH"] = f".:{LOCALSTACK_ROOT_FOLDER}"	×
174
175	# start the process as sudo
176	python_cmd = sys.executable	×
177	cmd = ["sudo", "-n", "--preserve-env", python_cmd, "-m", module]	×
178	arguments = arguments or []	×
179	shell_cmd = shlex.join(cmd + arguments)	×
180
181	# make sure we can run sudo commands
182	try:	×
183	ensure_can_use_sudo()	×
184	except Exception as e:	×
185	LOG.error("cannot run command as root (%s): %s ", str(e), shell_cmd)	×
186	return	×
187
188	def run_command(*_):	×
189	run(shell_cmd, outfile=subprocess.PIPE, print_error=False, env_vars=env_vars)	×
190
191	LOG.debug("Running command as sudo: %s", shell_cmd)	×
192	result = (	×
193	start_thread(run_command, quiet=True, name="sudo-edge") if asynchronous else run_command()
194	)
195	return result	×
196
197
198	def parse_gateway_listen(listen: str, default_host: str, default_port: int) -> list[HostAndPort]:	1✔
199	addresses = []	1✔
200	for address in listen.split(","):	1✔
201	addresses.append(HostAndPort.parse(address, default_host, default_port))	1✔
202	return addresses	1✔
203
204
205	if __name__ == "__main__":
206	logging.basicConfig()
207	parser = argparse.ArgumentParser()
208	parser.add_argument("component")
209	parser.add_argument("-l", "--gateway-listen", required=False, type=str)
210	parser.add_argument("-t", "--target-address", required=False, type=str)
211	args = parser.parse_args()
212
213	start_component(args.component, args.gateway_listen, args.target_address)

localstack / localstack / 20565403496

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