Feat/gateway certs improvements and improve pair registration

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,

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

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