From 99cc36e1542ce8cfc888f96257373acef9c4af1a Mon Sep 17 00:00:00 2001 From: cardosofede Date: Fri, 7 Aug 2026 23:27:39 +0300 Subject: [PATCH 1/2] (feat) improve pair deregistration and validation for OB --- services/unified_connector_service.py | 67 ++++++++++++++++++++++++++- 1 file changed, 65 insertions(+), 2 deletions(-) diff --git a/services/unified_connector_service.py b/services/unified_connector_service.py index 48f0cedf..6fe4d7af 100644 --- a/services/unified_connector_service.py +++ b/services/unified_connector_service.py @@ -268,8 +268,54 @@ async def ensure_data_connector_started( except Exception as e: logger.error(f"Error starting data connector {connector_name}: {e}") + self._purge_trading_pair_registration(connector, trading_pair) return False + async def _is_trading_pair_supported( + self, + connector: ConnectorBase, + trading_pair: str + ) -> bool: + """Check that a trading pair exists in the connector's symbol map. + + Registering an unknown pair poisons the order book WebSocket loop for every + other pair: the data source iterates over all registered pairs when + subscribing, so a single unknown pair raises and aborts the whole + subscription, which then retries and fails forever. + """ + if not hasattr(connector, "exchange_symbol_associated_to_pair"): + return True + try: + await connector.exchange_symbol_associated_to_pair(trading_pair=trading_pair) + return True + except KeyError: + return False + except Exception as e: + # Symbol map unavailable (network error, etc.) - don't block on it + logger.warning(f"Could not validate {trading_pair} against symbol map: {e}") + return True + + def _purge_trading_pair_registration( + self, + connector: ConnectorBase, + trading_pair: str + ): + """Unregister a trading pair from the connector and its order book tracker. + + Used to roll back a failed registration so a pair that never got an order + book cannot break the shared WebSocket subscription. + """ + targets = [getattr(connector, "_trading_pairs", None)] + tracker = getattr(connector, "order_book_tracker", None) + if tracker is not None: + targets.append(getattr(tracker, "_trading_pairs", None)) + + for pairs in targets: + # Connector and tracker often share the same list object + if pairs is not None and trading_pair in pairs: + pairs.remove(trading_pair) + logger.info(f"Unregistered {trading_pair} after failed order book initialization") + # ========================================================================= # Best Connector Selection (THE KEY FIX) # ========================================================================= @@ -373,6 +419,14 @@ async def initialize_order_book( except Exception: pass + # Reject pairs the exchange doesn't list before registering them anywhere + if not await self._is_trading_pair_supported(connector, trading_pair): + logger.error( + f"Trading pair {trading_pair} is not listed on {connector_name} - " + f"refusing to register it (check the quote asset)" + ) + return False + # For data connectors, ensure network is started if connector_name in self._data_connectors: if not self._data_connectors_started.get(connector_name, False): @@ -480,10 +534,12 @@ async def _add_trading_pair_to_tracker( logger.error(f"Fallback order book initialization failed: {e}") logger.error(f"Failed to add {trading_pair} to order book tracker") + self._purge_trading_pair_registration(connector, trading_pair) return False except Exception as e: logger.error(f"Error adding trading pair {trading_pair}: {e}", exc_info=True) + self._purge_trading_pair_registration(connector, trading_pair) return False async def remove_trading_pair( @@ -540,11 +596,18 @@ async def _remove_trading_pair_from_tracker( tracker = connector.order_book_tracker if trading_pair in tracker.order_books: del tracker.order_books[trading_pair] - if trading_pair in tracker._trading_pairs: - tracker._trading_pairs.remove(trading_pair) + self._purge_trading_pair_registration(connector, trading_pair) logger.info(f"Removed trading pair {trading_pair} via manual fallback") return True + # No order book, but the pair may still be registered - a pair stuck in + # the subscription list with no order book is exactly what breaks the + # WebSocket loop, so unregister it anyway. + if trading_pair in getattr(tracker, "_trading_pairs", []): + self._purge_trading_pair_registration(connector, trading_pair) + logger.info(f"Unregistered untracked trading pair {trading_pair}") + return True + logger.warning(f"Trading pair {trading_pair} not found") return False From fb2ddc98c32fcc85c76557fdbaa6178b32e39c03 Mon Sep 17 00:00:00 2001 From: cardosofede Date: Fri, 7 Aug 2026 23:27:49 +0300 Subject: [PATCH 2/2] (feat) adapt gateway certs --- main.py | 46 ++++++++++++++++++--------- services/gateway_service.py | 63 ++++++++++++++++++++++++++++++------- 2 files changed, 83 insertions(+), 26 deletions(-) diff --git a/main.py b/main.py index 54275db4..46f5f4ef 100644 --- a/main.py +++ b/main.py @@ -125,35 +125,36 @@ async def lifespan(app: FastAPI): ) # Initialize GatewayHttpClient singleton - from utils.gateway_certs import certs_present, sync_client_certs_to_root + from utils.gateway_certs import ensure_gateway_certs parsed_gateway_url = urlparse(settings.gateway.url) gateway_use_ssl = parsed_gateway_url.scheme == "https" if gateway_use_ssl: - # SEC-048: the in-process GatewayHttpClient reads its client certs only from - # root_path()/certs. Mirror the shared cert set there if the Gateway was already - # started in a previous run (no-op when certs haven't been generated yet). - sync_client_certs_to_root() + # SEC-048: generate the shared mTLS cert set up front (idempotent; an existing CA is + # reused untouched) and mirror the client certs into root_path()/certs, the only place + # hummingbot's in-process GatewayHttpClient looks. Doing this at startup rather than + # lazily on Gateway start means the client can always build its SSL context — a missing + # ca_cert.pem used to raise FileNotFoundError on every single request. + try: + ensure_gateway_certs(settings.security.config_password) + except Exception as e: + # Non-fatal: the API must boot even if the cert dir isn't writable. + logging.warning(f"Could not prepare Gateway mTLS certs: {e}") gateway_config = GatewayConfigMap( gateway_api_host=parsed_gateway_url.hostname or "localhost", gateway_api_port=str(parsed_gateway_url.port or 15888), gateway_use_ssl=gateway_use_ssl ) - gateway_client = GatewayHttpClient.get_instance(gateway_config) + GatewayHttpClient.get_instance(gateway_config) # Start the Gateway status monitor so Gateway's network connectors (e.g. # 'solana-mainnet-beta', and any newly added chain like 'ethereum-unichain') are # discovered from /config/chains and registered in AllConnectorSettings. Without # it, a Gateway network only lands in AllConnectorSettings lazily, when a connector # for it is first constructed; the monitor makes new chains enumerable without # first deploying a bot on them, and picks them up when Gateway comes online later. - # On a fresh install the shared mTLS certs don't exist until the Gateway is first - # started, and every 2s ping would fail loudly building the SSL context — defer; - # GatewayService.start() starts the monitor once the certs are generated. - if not gateway_use_ssl or certs_present(): - gateway_client.start_monitor() - else: - logging.info( - "Gateway mTLS certs not generated yet; status monitor deferred until the Gateway is started" - ) + # The monitor polls every 2s and logs loudly on every failure, so it is only started + # once there is a Gateway to poll — see the gateway_service block below, which has the + # Docker client needed to check. GatewayService.start() starts it when the Gateway + # comes up later. logging.info(f"Initialized GatewayHttpClient with URL: {settings.gateway.url}") # Initialize database @@ -268,6 +269,21 @@ async def lifespan(app: FastAPI): logging.info(f"Gateway cert reconciliation: {reconcile.get('message')}") except Exception as e: logging.warning(f"Gateway cert reconciliation skipped: {e}") + + # Start the Gateway status monitor only when there is a Gateway to poll. Gating on the + # container itself (rather than on cert presence, which is now always true) keeps the 2s + # poll loop from spamming errors for the many deployments that never run a Gateway. + # Plain-HTTP setups may point at a Gateway this API doesn't manage, so keep polling there. + try: + gateway_is_running = gateway_service.is_running() + except Exception as e: + logging.warning(f"Could not determine Gateway container state: {e}") + gateway_is_running = False + if not gateway_use_ssl or gateway_is_running: + GatewayHttpClient.get_instance().start_monitor() + else: + logging.info("Gateway container not running; status monitor deferred until it is started") + bot_archiver = BotArchiver( settings.aws.api_key, settings.aws.secret_key, diff --git a/services/gateway_service.py b/services/gateway_service.py index c1151e16..75573d26 100644 --- a/services/gateway_service.py +++ b/services/gateway_service.py @@ -80,6 +80,46 @@ def _get_gateway_container(self) -> Optional[docker.models.containers.Container] logger.error(f"Error getting Gateway container: {e}") return None + def is_running(self) -> bool: + """True when the Gateway container exists and is running. + + This is the direct signal for "is there a Gateway to talk to". Callers must not + infer it from cert presence: certs are generated up front so the mTLS client is + always usable, so their existence says nothing about whether the Gateway is up. + """ + container = self._get_gateway_container() + return container is not None and container.status == "running" + + @staticmethod + def _normalize_mount_source(source: str) -> str: + """Normalize a Docker-reported bind-mount source to a comparable host path. + + Docker Desktop reports macOS/Windows host paths through a ``/host_mnt`` prefix + (host ``/Users/x`` surfaces as ``/host_mnt/Users/x``), so a raw string compare + against a configured host path would never match there. + """ + normalized = os.path.normpath(source or "") + for prefix in ("/host_mnt", "/run/desktop/mnt/host"): + if normalized.startswith(prefix + os.sep): + normalized = normalized[len(prefix):] + return normalized + + def _mounts_shared_certs(self, container) -> bool: + """True when the container mounts *this* API's shared cert set. + + The Gateway reads its server cert from the directory we bind-mount at + ``GATEWAY_CERTS_BIND``. If that mount's source is our cert dir, the Gateway is + serving a cert signed by the CA our clients trust - consistent by construction. + A Gateway started by a different API instance (or a different checkout) mounts a + different source path, which is exactly the mismatch we need to detect. + """ + expected = self._normalize_mount_source(gateway_certs_dir(host=True)) + for mount in container.attrs.get("Mounts", []): + if mount.get("Destination") != self.GATEWAY_CERTS_BIND: + continue + return self._normalize_mount_source(mount.get("Source", "")) == expected + return False + def _get_self_container(self) -> Optional[docker.models.containers.Container]: """Best-effort lookup of the container this API process is running in. @@ -309,16 +349,17 @@ def start(self, config: GatewayConfig) -> Dict[str, Any]: def reconcile_certs(self) -> Dict[str, Any]: """Make a running Gateway usable by this API's mTLS client. - "Was the Gateway started with this API?" reduces to: does the API hold the shared client - cert set? The set lives on the bots/ volume both containers share, so a running Gateway - with the certs absent on the API side was not started by (or is inconsistent with) this - API instance — every secured request would fail the mTLS handshake. + "Was the Gateway started with this API?" reduces to: does the running container mount + *our* cert dir as its server cert source? Cert presence cannot answer this — the set is + generated up front at API startup, so it is present even for a Gateway this API never + started. A Gateway serving a cert from some other source would fail our mTLS handshake + on every request. Decision matrix: - - container not running -> nothing (start the Gateway to generate certs) - - running + certs present -> nothing (already consistent) - - running + certs missing -> regenerate the cert set and restart the Gateway so it - loads the server cert that matches our client cert + - container not running -> nothing (start the Gateway) + - running + mounts our certs -> nothing (consistent by construction) + - running + mounts other/no certs-> regenerate the cert set and restart the Gateway so it + loads the server cert that matches our client cert The restart is required: the Gateway reads its server cert/key only at startup, so writing new certs to the shared volume has no effect on an already-running container. @@ -331,15 +372,15 @@ def reconcile_certs(self) -> Dict[str, Any]: "message": "Gateway container not running; start it to generate certs", } - if certs_present(gateway_certs_dir()): + if self._mounts_shared_certs(container) and certs_present(gateway_certs_dir()): return { "success": True, "action": "none", - "message": "Gateway running and shared mTLS certs present", + "message": "Gateway running against this API's shared mTLS cert set", } logger.warning( - "Gateway container is running but the shared mTLS certs are missing on the API side; " + "Gateway container is running but is not serving this API's shared mTLS cert set; " "regenerating the cert set and restarting the Gateway so it loads a matching server cert" ) dirs = self._ensure_gateway_directories()