fix: improve proxy liveness and graceful shutdown

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

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.

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:
+class ConfigurationError(ValueError):
-    """Retourne False tant que l'option n'est pas explicitement vraie."""
+    """Configuration de l'add-on invalide."""
-    return value is not None and value.strip().lower() == "true"
+
+
+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)
+                try:
-                await self.send_keepalive()
+                    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))
+        self.pac_reader_task = asyncio.create_task(self.pac_reader(reader))
-        keepalive_task = asyncio.create_task(self.pac_keepalive())
+        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()
+            if self.keepalive_task is not None:
-            await asyncio.gather(keepalive_task, return_exceptions=True)
+                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()
+            self.request_stop()
-            server.close()
+            if self.server is not None:
-            await server.wait_closed()
+                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:
+async def run_proxy(proxy: ArkteosProxy) -> None:
-    pac_host = sys.argv[1] if len(sys.argv) > 1 else "192.168.X.X"
+    loop = asyncio.get_running_loop()
-    pac_port = int(sys.argv[2]) if len(sys.argv) > 2 else 9641
+    installed_signals: list[signal.Signals] = []
-    proxy_port = int(sys.argv[3]) if len(sys.argv) > 3 else 9641
+    for signal_name in (signal.SIGTERM, signal.SIGINT):
-    allow_client_writes = parse_allow_client_writes(sys.argv[4] if len(sys.argv) > 4 else None)
+        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())

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"

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}"

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()