diff --git a/WebHostLib/__init__.py b/WebHostLib/__init__.py index b3dc203a3d..90bd0c2fdd 100644 --- a/WebHostLib/__init__.py +++ b/WebHostLib/__init__.py @@ -42,6 +42,7 @@ app.config["SELFLAUNCH"] = True # application process is in charge of launching app.config["SELFLAUNCHCERT"] = None # can point to a SSL Certificate to encrypt Room websocket connections app.config["SELFLAUNCHKEY"] = None # can point to a SSL Certificate Key to encrypt Room websocket connections app.config["SELFGEN"] = True # application process is in charge of scheduling Generations. +app.config["GAME_PORTS"] = ["49152-65535", 0] # at what amount of worlds should scheduling be used, instead of rolling in the web-thread app.config["JOB_THRESHOLD"] = 1 # after what time in seconds should generation be aborted, freeing the queue slot. Can be set to None to disable. diff --git a/WebHostLib/autolauncher.py b/WebHostLib/autolauncher.py index 165e08a103..d101f149ef 100644 --- a/WebHostLib/autolauncher.py +++ b/WebHostLib/autolauncher.py @@ -193,6 +193,7 @@ class MultiworldInstance(): self.cert = config["SELFLAUNCHCERT"] self.key = config["SELFLAUNCHKEY"] self.host = config["HOST_ADDRESS"] + self.game_ports = config["GAME_PORTS"] self.rooms_to_start = multiprocessing.Queue() self.rooms_shutting_down = multiprocessing.Queue() self.name = f"MultiHoster{id}" @@ -203,7 +204,7 @@ class MultiworldInstance(): process = multiprocessing.Process(group=None, target=run_server_process, args=(self.name, self.ponyconfig, get_static_server_data(), - self.cert, self.key, self.host, + self.cert, self.key, self.host, self.game_ports, self.rooms_to_start, self.rooms_shutting_down), name=self.name) process.start() diff --git a/WebHostLib/customserver.py b/WebHostLib/customserver.py index 4257c6aff3..85a720169e 100644 --- a/WebHostLib/customserver.py +++ b/WebHostLib/customserver.py @@ -4,6 +4,7 @@ import asyncio import collections import datetime import functools +import itertools import logging import multiprocessing import pickle @@ -13,7 +14,9 @@ import threading import time import typing import sys +from collections.abc import Iterable +import psutil import websockets from pony.orm import commit, db_session, select @@ -24,6 +27,7 @@ from MultiServer import ( server_per_message_deflate_factory, ) from Utils import restricted_loads, cache_argsless + from .locker import Locker from .models import Command, GameDataPackage, Room, db @@ -76,12 +80,8 @@ class WebHostContext(Context): self.tags = ["AP", "WebHost"] def __del__(self): - try: - import psutil - from Utils import format_SI_prefix - self.logger.debug(f"Context destroyed, Mem: {format_SI_prefix(psutil.Process().memory_info().rss, 1024)}iB") - except ImportError: - self.logger.debug("Context destroyed") + from Utils import format_SI_prefix + self.logger.debug(f"Context destroyed, Mem: {format_SI_prefix(psutil.Process().memory_info().rss, 1024)}iB") def _load_game_data(self): for key, value in self.static_server_data.items(): @@ -115,7 +115,7 @@ class WebHostContext(Context): if room.last_port: self.port = room.last_port else: - self.port = get_random_port() + self.port = 0 multidata = self.decompress(room.seed.multidata) game_data_packages = {} @@ -181,8 +181,97 @@ class WebHostContext(Context): return d -def get_random_port(): - return random.randint(49152, 65535) +class GameRangePorts(typing.NamedTuple): + valid_ports: list[int] + ephemeral_allowed: bool + + +class RandomPortSocketCreator: + """ Creates server sockets on random available ports from a configured range. """ + + _next_port_index: int + _used_ports_cache: tuple[frozenset[int], int] | None + _parsed_ports: GameRangePorts + + def __init__(self, game_ports: Iterable[str | int]) -> None: + self._next_port_index = 0 + self._used_ports_cache = None + self._parsed_ports = self._parse_game_ports(game_ports) + + @staticmethod + def _parse_game_ports(game_ports: Iterable[str | int]) -> GameRangePorts: + """ Parse the game ports configuration into a structured format. """ + valid_ports: list[int] = [] + ephemeral_allowed = False + + for item in game_ports: + if isinstance(item, str) and "-" in item: + start, end = map(int, item.split("-")) + x = range(start, end + 1) + valid_ports.extend(x) + elif int(item) == 0: + ephemeral_allowed = True + else: + valid_ports.append(int(item)) + + random.shuffle(valid_ports) + return GameRangePorts(valid_ports, ephemeral_allowed) + + @staticmethod + def _try_conns_per_process(p: psutil.Process) -> Iterable[int]: + """ Get ports from a single process's connections. """ + try: + return (c.laddr.port for c in p.net_connections("tcp4") if c.laddr) + except psutil.AccessDenied: + return () + + @staticmethod + def _get_active_net_connections() -> Iterable[int]: + """ Get all active TCP4 connections on the system. """ + # Don't even try to check if system using AIX + if psutil.AIX: + return () + + try: + return (c.laddr.port for c in psutil.net_connections("tcp4") if c.laddr) + # raises AccessDenied when done on macOS + except psutil.AccessDenied: + # flatten the list of iterables + return itertools.chain.from_iterable(map( + RandomPortSocketCreator._try_conns_per_process, + psutil.process_iter(["net_connections"]) + )) + + def _get_used_ports(self) -> frozenset[int]: + """ Get currently used ports with 90-second caching. """ + t_hash = round(time.monotonic() / 90) + if self._used_ports_cache is None or self._used_ports_cache[1] != t_hash: + self._used_ports_cache = (frozenset(self._get_active_net_connections()), t_hash) + + return self._used_ports_cache[0] + + def create(self, host: str) -> socket.socket: + """ Create a server socket on an available port. """ + valid_ports, ephemeral_allowed = self._parsed_ports + used_ports = self._get_used_ports() + + next_index = self._next_port_index + for i, port in enumerate(itertools.chain(valid_ports[next_index:], valid_ports[:next_index])): + if port in used_ports: + continue + + try: + res = socket.create_server((host, port)) + next_index = (next_index + i + 1) % len(valid_ports) + self._next_port_index = next_index + return res + except OSError: + pass + + if ephemeral_allowed: + return socket.create_server((host, 0)) + + raise OSError(98, "No available ports") @cache_argsless @@ -247,7 +336,8 @@ def tear_down_logging(room_id): def run_server_process(name: str, ponyconfig: dict, static_server_data: dict, cert_file: typing.Optional[str], cert_key_file: typing.Optional[str], - host: str, rooms_to_run: multiprocessing.Queue, rooms_shutting_down: multiprocessing.Queue): + host: str, game_ports: Iterable[str | int], + rooms_to_run: multiprocessing.Queue, rooms_shutting_down: multiprocessing.Queue): from setproctitle import setproctitle setproctitle(name) @@ -291,6 +381,7 @@ def run_server_process(name: str, ponyconfig: dict, static_server_data: dict, gc.collect() # free intermediate objects used during setup loop = asyncio.get_event_loop() + socket_creator = RandomPortSocketCreator(game_ports) async def start_room(room_id): with Locker(f"RoomLocker {room_id}"): @@ -300,20 +391,26 @@ def run_server_process(name: str, ponyconfig: dict, static_server_data: dict, ctx.load(room_id) ctx.init_save() assert ctx.server is None - try: + if ctx.port != 0: + try: + ctx.server = websockets.serve( + functools.partial(server, ctx=ctx), + ctx.host, + ctx.port, + ssl=get_ssl_context(), + extensions=[server_per_message_deflate_factory], + ) + await ctx.server + except OSError: + ctx.port = 0 + if ctx.port == 0: ctx.server = websockets.serve( functools.partial(server, ctx=ctx), - ctx.host, - ctx.port, + sock=socket_creator.create(ctx.host), ssl=get_ssl_context(), extensions=[server_per_message_deflate_factory], ) await ctx.server - except OSError: # likely port in use - ctx.server = websockets.serve( - functools.partial(server, ctx=ctx), ctx.host, 0, ssl=get_ssl_context()) - - await ctx.server port = 0 for wssocket in ctx.server.ws_server.sockets: socketname = wssocket.getsockname() @@ -388,7 +485,7 @@ def run_server_process(name: str, ponyconfig: dict, static_server_data: dict, def run(self): while 1: - next_room = rooms_to_run.get(block=True, timeout=None) + next_room = rooms_to_run.get(block=True, timeout=None) gc.collect() task = asyncio.run_coroutine_threadsafe(start_room(next_room), loop) self._tasks.append(task) diff --git a/WebHostLib/requirements.txt b/WebHostLib/requirements.txt index 28d6760db0..886d759d9c 100644 --- a/WebHostLib/requirements.txt +++ b/WebHostLib/requirements.txt @@ -12,3 +12,4 @@ markupsafe==3.0.3 setproctitle==1.3.7 mistune==3.2.1 docutils==0.22.4 +psutil==7.2.2 diff --git a/docs/webhost configuration sample.yaml b/docs/webhost configuration sample.yaml index 93094f1ce7..a250e8eae7 100644 --- a/docs/webhost configuration sample.yaml +++ b/docs/webhost configuration sample.yaml @@ -17,6 +17,12 @@ # Web hosting port #PORT: 80 +# Ports used for game hosting. Values can be specific ports, port ranges or both. Default is: [49152-65535, 0] +# Zero means it will use a random free port if there is no free port in the ranges specified +# Examples of valid values: [40000-41000, 49152-65535] +# If ports within the range(s) are already in use, the WebHost will fallback to the default [49152-65535, 0] range. +#GAME_PORTS: [49152-65535, 0] + # Place where uploads go. #UPLOAD_FOLDER: uploads diff --git a/test/webhost/test_port_allocation.py b/test/webhost/test_port_allocation.py new file mode 100644 index 0000000000..914e7b2c87 --- /dev/null +++ b/test/webhost/test_port_allocation.py @@ -0,0 +1,89 @@ +import os +import unittest +from socket import socket as Socket # noqa: N812 + +from Utils import is_macos +from WebHostLib.customserver import RandomPortSocketCreator + +ci = bool(os.environ.get("CI")) + + +class TestPortAllocating(unittest.TestCase): + def test_parse_game_ports(self) -> None: + """Ensure that game ports with ranges are parsed correctly""" + val = RandomPortSocketCreator._parse_game_ports(("1000-2000", "2000-5000", "1000-2000", 20, 40, "20", "0")) + + self.assertCountEqual(val.valid_ports, + [*range(1000, 2001), *range(2000, 5001), *range(1000, 2001), 20, 40, 20], + "The parsed game ports are not the expected length") + self.assertTrue(val.ephemeral_allowed, "The ephemeral allowed flag is not set even though it was passed") + + val = RandomPortSocketCreator._parse_game_ports(()) + self.assertListEqual(val.valid_ports, [], "Empty list of game port returned something") + self.assertFalse(val.ephemeral_allowed, "Empty list returned that ephemeral is allowed") + + val = RandomPortSocketCreator._parse_game_ports((0,)) + self.assertListEqual(val.valid_ports, [], "Empty list of ranges returned something") + self.assertTrue(val.ephemeral_allowed, "List with just 0 is not allowing ephemeral ports") + + val = RandomPortSocketCreator._parse_game_ports((1,)) + self.assertListEqual(val.valid_ports, [1], "Valid ports doesn't contain the expected values") + self.assertFalse(val.ephemeral_allowed, "List with just single port returned that ephemeral is allowed") + + def test_parse_game_port_errors(self) -> None: + """Ensure that game ports with incorrect values raise the expected error""" + with self.assertRaises(ValueError, msg="Negative numbers didn't get interpreted as an invalid range"): + RandomPortSocketCreator._parse_game_ports(tuple("-50215")) + with self.assertRaises(ValueError, msg="Text got interpreted as a valid number"): + RandomPortSocketCreator._parse_game_ports(tuple("dwafawg")) + with self.assertRaises( + ValueError, + msg="A range with an extra dash at the end didn't get interpreted as an invalid number because of it's end dash" + ): + RandomPortSocketCreator._parse_game_ports(tuple("20-21215-")) + with self.assertRaises(ValueError, msg="Text got interpreted as a valid number for the start of a range"): + RandomPortSocketCreator._parse_game_ports(tuple("f-21215")) + + def test_random_port_socket_edge_cases(self) -> None: + """Verify if edge cases on creation of random port socket is working fine""" + # Try giving an empty tuple and fail over it + creator = RandomPortSocketCreator(()) + with self.assertRaises(OSError) as err: + creator.create("127.0.0.1") + self.assertEqual(err.exception.errno, 98, "Raised an unexpected error code") + self.assertEqual(err.exception.strerror, "No available ports", "Raised an unexpected error string") + + # Try only having ephemeral ports enabled + creator = RandomPortSocketCreator(("0",)) + try: + creator.create("127.0.0.1").close() + except OSError as err: + self.assertEqual(err.errno, 98, "Raised an unexpected error code") + # If it returns our error string that means something is wrong with our code + self.assertNotEqual(err.strerror, "No available ports", + "Raised an unexpected error string") + + @unittest.skipUnless(ci, "can't guarantee free ports outside of CI") + def test_random_port_socket(self) -> None: + """Verify if returned sockets use the correct port ranges""" + creator = RandomPortSocketCreator(("8080-8085",)) + sockets: list[Socket] = [] + for _ in range(6): + socket = creator.create("127.0.0.1") + sockets.append(socket) + _, port = socket.getsockname() + self.assertIn(port, range(8080, 8086), "Port of socket was not inside the expected range") + for s in sockets: + s.close() + + sockets.clear() + creator = RandomPortSocketCreator(("30000-65535",)) + length = 5_000 if is_macos else (30_000 - len(creator._get_used_ports())) + for _ in range(length): + socket = creator.create("127.0.0.1") + sockets.append(socket) + _, port = socket.getsockname() + self.assertIn(port, range(30_000, 65536), "Port of socket was not inside the expected range") + + for s in sockets: + s.close()