From 97c82b38e458f00c156b27463f4725fd57cc2c54 Mon Sep 17 00:00:00 2001 From: raph666 Date: Wed, 22 Jul 2026 14:37:05 +0200 Subject: [PATCH] fix: improve proxy liveness and graceful shutdown --- CHANGELOG.md | 13 +++ README.md | 6 ++ arkteos_proxy.py | 196 +++++++++++++++++++++++++++++++----- config.yaml | 2 +- run.sh | 2 +- tests/test_arkteos_proxy.py | 129 +++++++++++++++++++++++- 6 files changed, 318 insertions(+), 30 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index eff973c..3f7c2eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,19 @@ Toutes les modifications importantes de ce projet sont documentées ici. ## [Unreleased] +## [1.0.6] - 2026-07-22 + +### Fixed + +- Ajout d’un délai maximal pour l’établissement de la connexion à la PAC. +- Détection d’une connexion PAC silencieuse malgré une socket toujours ouverte. +- Fermeture propre du serveur, des clients et de la connexion PAC lors de l’arrêt de l’add-on. +- Validation plus stricte de la configuration au démarrage. + +### Tests + +- Ajout de tests pour les timeouts, la reconnexion, l’arrêt gracieux et les configurations invalides. + ## [1.0.5] - 2026-07-22 ### Changed diff --git a/README.md b/README.md index 22ec3bf..ffe31a3 100644 --- a/README.md +++ b/README.md @@ -30,6 +30,12 @@ Active `allow_client_writes` uniquement si un client de confiance doit écrire v Un keepalive composé d’un octet nul est envoyé toutes les 300 secondes à la PAC afin de maintenir la connexion. Le code n’établit pas le rôle exact de cet octet pour la PAC. +## Robustesse de la connexion + +L’établissement de la connexion vers la PAC est limité dans le temps. Le proxy se reconnecte automatiquement après une coupure ou lorsqu’aucun octet n’est reçu de la PAC pendant une période prolongée, même si la socket reste ouverte. + +Lors d’un arrêt, d’une mise à jour ou d’un redémarrage de l’add-on, le proxy arrête les reconnexions et ferme proprement le serveur, les clients et la connexion PAC. Ces mécanismes ne garantissent pas la disponibilité de la PAC ou du réseau. + ## Clients TCP Les clients TCP utilisent le port du proxy pour recevoir le flux binaire de la PAC. Aucun client particulier n’est requis pour le fonctionnement de l’add-on. diff --git a/arkteos_proxy.py b/arkteos_proxy.py index 57b2c10..857675a 100644 --- a/arkteos_proxy.py +++ b/arkteos_proxy.py @@ -3,11 +3,14 @@ import asyncio import logging +import signal import sys KEEPALIVE_INTERVAL = 300 RECONNECT_DELAY = 10 +PAC_CONNECT_TIMEOUT = 10.0 +PAC_READ_TIMEOUT = 600.0 logging.basicConfig( level=logging.INFO, @@ -17,9 +20,53 @@ logging.basicConfig( logger = logging.getLogger(__name__) -def parse_allow_client_writes(value: str | None) -> bool: - """Retourne False tant que l'option n'est pas explicitement vraie.""" - return value is not None and value.strip().lower() == "true" +class ConfigurationError(ValueError): + """Configuration de l'add-on invalide.""" + + +def parse_allow_client_writes(value: bool | str | None) -> bool: + """Accepte uniquement un booléen ou les chaînes explicites true et false.""" + if value is None: + return False + if isinstance(value, bool): + return value + if isinstance(value, str): + normalized = value.strip().lower() + if normalized == "true": + return True + if normalized == "false": + return False + raise ConfigurationError("allow_client_writes doit être true ou false") + + +def parse_port(value: int | str, option_name: str) -> int: + """Valide un port TCP sans accepter de valeur ambiguë.""" + if isinstance(value, bool): + raise ConfigurationError(f"{option_name} doit être un entier entre 1 et 65535") + if isinstance(value, str): + if not value.strip().isdigit(): + raise ConfigurationError(f"{option_name} doit être un entier entre 1 et 65535") + value = int(value.strip()) + if not isinstance(value, int) or not 1 <= value <= 65535: + raise ConfigurationError(f"{option_name} doit être un entier entre 1 et 65535") + return value + + +def validate_configuration( + pac_host: str, + pac_port: int | str, + proxy_port: int | str, + allow_client_writes: bool | str | None, +) -> tuple[str, int, int, bool]: + """Retourne une configuration validée avant toute ouverture réseau.""" + if not isinstance(pac_host, str) or not pac_host.strip(): + raise ConfigurationError("pac_host doit être une chaîne non vide") + return ( + pac_host.strip(), + parse_port(pac_port, "pac_port"), + parse_port(proxy_port, "proxy_port"), + parse_allow_client_writes(allow_client_writes), + ) class ArkteosProxy: @@ -35,9 +82,30 @@ class ArkteosProxy: self.proxy_port = proxy_port self.allow_client_writes = allow_client_writes self.clients: set[asyncio.StreamWriter] = set() + self.client_tasks: set[asyncio.Task[None]] = set() self.pac_writer: asyncio.StreamWriter | None = None self.pac_write_lock = asyncio.Lock() self.stop_event = asyncio.Event() + self.server: asyncio.AbstractServer | None = None + self.pac_reader_task: asyncio.Task[None] | None = None + self.keepalive_task: asyncio.Task[None] | None = None + + def request_stop(self) -> None: + """Déclenche l'arrêt sans bloquer un gestionnaire de signal.""" + if self.stop_event.is_set(): + return + logger.info("Arrêt du proxy demandé") + self.stop_event.set() + if self.server is not None: + self.server.close() + if self.keepalive_task is not None: + self.keepalive_task.cancel() + if self.pac_reader_task is not None: + self.pac_reader_task.cancel() + if self.pac_writer is not None: + self.pac_writer.close() + for writer in tuple(self.clients): + writer.close() async def write_to_pac(self, data: bytes) -> None: """Écrit un bloc complet vers la PAC sans l'altérer.""" @@ -55,13 +123,21 @@ class ArkteosProxy: logger.info("Démarrage lecture PAC") try: while not self.stop_event.is_set(): - data = await reader.read(4096) + try: + async with asyncio.timeout(PAC_READ_TIMEOUT): + data = await reader.read(4096) + except TimeoutError: + logger.warning("PAC restée silencieuse pendant %.0f secondes", PAC_READ_TIMEOUT) + return if not data: logger.info("PAC a fermé la connexion") return await self.broadcast_to_clients(data) except (ConnectionError, OSError) as error: logger.info("Erreur lecture PAC : %s", error) + except asyncio.CancelledError: + if not self.stop_event.is_set(): + raise finally: logger.info("Arrêt lecture PAC") @@ -80,10 +156,17 @@ class ArkteosProxy: logger.info("Démarrage keepalive PAC") try: while not self.stop_event.is_set(): - await asyncio.sleep(KEEPALIVE_INTERVAL) - await self.send_keepalive() + try: + async with asyncio.timeout(KEEPALIVE_INTERVAL): + await self.stop_event.wait() + return + except TimeoutError: + await self.send_keepalive() except (ConnectionError, OSError) as error: logger.info("Erreur keepalive PAC : %s", error) + except asyncio.CancelledError: + if not self.stop_event.is_set(): + raise finally: logger.info("Arrêt keepalive PAC") @@ -92,6 +175,9 @@ class ArkteosProxy: reader: asyncio.StreamReader, writer: asyncio.StreamWriter, ) -> None: + task = asyncio.current_task() + if task is not None: + self.client_tasks.add(task) peername = writer.get_extra_info("peername") logger.info("Nouveau client : %s", peername) self.clients.add(writer) @@ -113,8 +199,13 @@ class ArkteosProxy: return except (ConnectionError, OSError) as error: logger.info("Erreur client %s : %s", peername, error) + except asyncio.CancelledError: + if not self.stop_event.is_set(): + raise finally: await self.close_client(writer) + if task is not None: + self.client_tasks.discard(task) async def close_client(self, writer: asyncio.StreamWriter) -> None: if writer not in self.clients: @@ -131,17 +222,38 @@ class ArkteosProxy: for writer in tuple(self.clients): await self.close_client(writer) + async def stop_client_tasks(self) -> None: + current_task = asyncio.current_task() + tasks = [task for task in self.client_tasks if task is not current_task] + for task in tasks: + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + async def run_pac_connection(self) -> None: - reader, writer = await asyncio.open_connection(self.pac_host, self.pac_port) + try: + async with asyncio.timeout(PAC_CONNECT_TIMEOUT): + reader, writer = await asyncio.open_connection(self.pac_host, self.pac_port) + except TimeoutError as error: + logger.warning("Délai de connexion PAC dépassé après %.0f secondes", PAC_CONNECT_TIMEOUT) + raise ConnectionError("Délai de connexion PAC dépassé") from error + self.pac_writer = writer logger.info("Connecté à la PAC %s:%s", self.pac_host, self.pac_port) - reader_task = asyncio.create_task(self.pac_reader(reader)) - keepalive_task = asyncio.create_task(self.pac_keepalive()) + self.pac_reader_task = asyncio.create_task(self.pac_reader(reader)) + self.keepalive_task = asyncio.create_task(self.pac_keepalive()) try: - await reader_task + try: + await self.pac_reader_task + except asyncio.CancelledError: + if not self.stop_event.is_set(): + raise finally: - keepalive_task.cancel() - await asyncio.gather(keepalive_task, return_exceptions=True) + if self.keepalive_task is not None: + self.keepalive_task.cancel() + await asyncio.gather(self.keepalive_task, return_exceptions=True) + self.keepalive_task = None + self.pac_reader_task = None if self.pac_writer is writer: self.pac_writer = None writer.close() @@ -150,6 +262,14 @@ class ArkteosProxy: except (ConnectionError, OSError): pass await self.close_clients() + await self.stop_client_tasks() + + async def wait_for_reconnect_delay(self) -> None: + try: + async with asyncio.timeout(RECONNECT_DELAY): + await self.stop_event.wait() + except TimeoutError: + pass async def serve(self) -> None: mode = ( @@ -158,36 +278,62 @@ class ArkteosProxy: else "Mode lecture seule : écritures des clients bloquées" ) logger.info(mode) - server = await asyncio.start_server(self.handle_client, "0.0.0.0", self.proxy_port) + self.server = await asyncio.start_server(self.handle_client, "0.0.0.0", self.proxy_port) logger.info("Proxy en écoute sur 0.0.0.0:%s", self.proxy_port) try: while not self.stop_event.is_set(): try: await self.run_pac_connection() except (ConnectionError, OSError) as error: - logger.info("Échec connexion PAC : %s", error) + if not self.stop_event.is_set(): + logger.info("Échec connexion PAC : %s", error) if not self.stop_event.is_set(): logger.info("Nouvelle tentative de connexion PAC dans 10s...") - await asyncio.sleep(RECONNECT_DELAY) + await self.wait_for_reconnect_delay() finally: - self.stop_event.set() - server.close() - await server.wait_closed() + self.request_stop() + if self.server is not None: + await self.server.wait_closed() + self.server = None await self.close_clients() + await self.stop_client_tasks() logger.info("Proxy arrêté") -def main() -> None: - pac_host = sys.argv[1] if len(sys.argv) > 1 else "192.168.X.X" - pac_port = int(sys.argv[2]) if len(sys.argv) > 2 else 9641 - proxy_port = int(sys.argv[3]) if len(sys.argv) > 3 else 9641 - allow_client_writes = parse_allow_client_writes(sys.argv[4] if len(sys.argv) > 4 else None) +async def run_proxy(proxy: ArkteosProxy) -> None: + loop = asyncio.get_running_loop() + installed_signals: list[signal.Signals] = [] + for signal_name in (signal.SIGTERM, signal.SIGINT): + try: + loop.add_signal_handler(signal_name, proxy.request_stop) + installed_signals.append(signal_name) + except (NotImplementedError, RuntimeError): + pass + try: + await proxy.serve() + finally: + for signal_name in installed_signals: + loop.remove_signal_handler(signal_name) + + +def main() -> int: + try: + pac_host, pac_port, proxy_port, allow_client_writes = validate_configuration( + sys.argv[1] if len(sys.argv) > 1 else "192.168.X.X", + sys.argv[2] if len(sys.argv) > 2 else "9641", + sys.argv[3] if len(sys.argv) > 3 else "9641", + sys.argv[4] if len(sys.argv) > 4 else None, + ) + except ConfigurationError as error: + logger.error("Configuration invalide : %s", error) + return 2 proxy = ArkteosProxy(pac_host, pac_port, proxy_port, allow_client_writes) try: - asyncio.run(proxy.serve()) + asyncio.run(run_proxy(proxy)) except KeyboardInterrupt: logger.info("Arrêt demandé") + return 0 if __name__ == "__main__": - main() + raise SystemExit(main()) diff --git a/config.yaml b/config.yaml index 1522cab..f08b0d6 100644 --- a/config.yaml +++ b/config.yaml @@ -1,5 +1,5 @@ name: Arkteos Proxy -version: "1.0.5" +version: "1.0.6" slug: arkteos_proxy description: Proxy TCP pour PAC Arkteos REG3 — maintient une connexion unique vers la PAC et partage le flux avec plusieurs clients locaux. Les écritures client sont désactivées par défaut. url: "https://github.com/raph666/arkteos-proxy-addon" diff --git a/run.sh b/run.sh index d708ce7..687661f 100755 --- a/run.sh +++ b/run.sh @@ -9,4 +9,4 @@ bashio::log.info "Démarrage du proxy Arkteos" bashio::log.info "PAC : ${PAC_HOST}:${PAC_PORT}" bashio::log.info "Proxy en écoute sur port : ${PROXY_PORT}" -exec python3 /arkteos_proxy.py "${PAC_HOST}" "${PAC_PORT}" "${PROXY_PORT}" "${ALLOW_CLIENT_WRITES}" +exec python3 /arkteos_proxy.py "${PAC_HOST}" "${PAC_PORT}" "${PROXY_PORT}" "${ALLOW_CLIENT_WRITES:-false}" diff --git a/tests/test_arkteos_proxy.py b/tests/test_arkteos_proxy.py index c4a0604..e4a65c7 100644 --- a/tests/test_arkteos_proxy.py +++ b/tests/test_arkteos_proxy.py @@ -2,7 +2,13 @@ import asyncio import unittest from unittest.mock import AsyncMock, patch -from arkteos_proxy import ArkteosProxy, KEEPALIVE_INTERVAL, RECONNECT_DELAY +from arkteos_proxy import ( + ArkteosProxy, + ConfigurationError, + KEEPALIVE_INTERVAL, + RECONNECT_DELAY, + validate_configuration, +) class QueueReader: @@ -149,18 +155,135 @@ class ArkteosProxyTests(unittest.IsolatedAsyncioTestCase): proxy.stop_event.set() proxy.run_pac_connection = fake_pac_connection + proxy.wait_for_reconnect_delay = AsyncMock() with ( patch("arkteos_proxy.asyncio.start_server", new_callable=AsyncMock, return_value=FakeServer()), - patch("arkteos_proxy.asyncio.sleep", new_callable=AsyncMock) as sleep, ): await proxy.serve() self.assertEqual(attempts, 2) - sleep.assert_awaited_once_with(RECONNECT_DELAY) + proxy.wait_for_reconnect_delay.assert_awaited_once_with() def test_missing_option_defaults_to_read_only(self): self.assertFalse(ArkteosProxy("pac", 9641, 9641).allow_client_writes) + async def test_connection_timeout_stops_a_single_attempt(self): + proxy = ArkteosProxy("pac", 9641, 9641) + + async def never_connect(*_args, **_kwargs): + await asyncio.Event().wait() + + with ( + patch("arkteos_proxy.PAC_CONNECT_TIMEOUT", 0.01), + patch("arkteos_proxy.asyncio.open_connection", new_callable=AsyncMock, side_effect=never_connect) as connect, + ): + with self.assertRaises(ConnectionError): + await proxy.run_pac_connection() + + connect.assert_awaited_once_with("pac", 9641) + + async def test_silent_pac_closes_writer_and_cancels_keepalive(self): + proxy = ArkteosProxy("pac", 9641, 9641) + reader = QueueReader() + writer = FakeWriter() + keepalive_cancelled = asyncio.Event() + + async def tracked_keepalive(): + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + keepalive_cancelled.set() + raise + + proxy.pac_keepalive = tracked_keepalive + with ( + patch("arkteos_proxy.PAC_READ_TIMEOUT", 0.01), + patch("arkteos_proxy.asyncio.open_connection", new_callable=AsyncMock, return_value=(reader, writer)), + ): + await proxy.run_pac_connection() + + self.assertTrue(writer.closed) + self.assertTrue(keepalive_cancelled.is_set()) + self.assertIsNone(proxy.pac_writer) + self.assertIsNone(proxy.keepalive_task) + + async def test_pac_activity_is_distributed_before_connection_closes(self): + proxy = ArkteosProxy("pac", 9641, 9641) + reader = QueueReader() + writer = FakeWriter() + client = FakeWriter() + proxy.clients.add(client) + await reader.items.put(b"pac-data") + await reader.items.put(b"") + + with ( + patch("arkteos_proxy.PAC_READ_TIMEOUT", 0.01), + patch("arkteos_proxy.asyncio.open_connection", new_callable=AsyncMock, return_value=(reader, writer)), + ): + await proxy.run_pac_connection() + + self.assertEqual(client.writes, [b"pac-data"]) + self.assertTrue(writer.closed) + + async def test_graceful_stop_closes_server_clients_and_pac_writer(self): + proxy = ArkteosProxy("pac", 9641, 0) + server = FakeServer() + pac_writer = FakeWriter() + client = FakeWriter() + proxy.pac_writer = pac_writer + proxy.clients.add(client) + + async def wait_for_stop(): + await proxy.stop_event.wait() + + proxy.run_pac_connection = wait_for_stop + with patch("arkteos_proxy.asyncio.start_server", new_callable=AsyncMock, return_value=server): + task = asyncio.create_task(proxy.serve()) + await asyncio.sleep(0) + proxy.request_stop() + await task + + self.assertTrue(server.closed) + self.assertTrue(client.closed) + self.assertTrue(pac_writer.closed) + self.assertEqual(proxy.client_tasks, set()) + + async def test_stop_during_reconnect_does_not_wait_for_full_delay(self): + proxy = ArkteosProxy("pac", 9641, 0) + server = FakeServer() + + async def failed_connection(): + raise ConnectionError("PAC indisponible") + + async def stop_during_delay(): + proxy.request_stop() + + proxy.run_pac_connection = failed_connection + proxy.wait_for_reconnect_delay = stop_during_delay + with patch("arkteos_proxy.asyncio.start_server", new_callable=AsyncMock, return_value=server): + await asyncio.wait_for(proxy.serve(), timeout=0.1) + + self.assertTrue(server.closed) + + def test_configuration_validation_rejects_invalid_values(self): + invalid_configurations = ( + (" ", "9641", "9641", "false"), + ("pac", "0", "9641", "false"), + ("pac", "65536", "9641", "false"), + ("pac", "9641", "invalid", "false"), + ("pac", "9641", "9641", "maybe"), + ) + for configuration in invalid_configurations: + with self.subTest(configuration=configuration): + with self.assertRaises(ConfigurationError): + validate_configuration(*configuration) + + def test_configuration_validation_accepts_supported_values(self): + self.assertEqual( + validate_configuration(" pac ", "9641", 9641, "false"), + ("pac", 9641, 9641, False), + ) + if __name__ == "__main__": unittest.main()