[P/D]Add ssl cert for metaserver proxy (#5875)

### What this PR does / why we need it?
When the P node accesses the proxy meteserver, add the SSL certificate
and the CA certificate path to improve security.

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
By ci

- vLLM version: v0.13.0
- vLLM main:
bde38c11df

---------

Signed-off-by: wangxiaoteng <wangxiaoteng@huawei.com>

vllm_ascend/distributed/kv_transfer/kv_p2p/mooncake_layerwise_connector.py

@@ -567,10 +567,26 @@ class MooncakeLayerwiseConnectorScheduler:
         self._reqs_need_recv: dict[str, tuple[Request, list[int],
                                               list[int]]] = {}
         self._reqs_need_send_layerwise: dict[str, SendReqInfo] = {}
-
         self.executor = ThreadPoolExecutor(32)
-        self.metaserver_client = httpx.Client(
-            limits=httpx.Limits(max_connections=100000), timeout=None)
+        tls_config: dict[
+            str, Any] = vllm_config.kv_transfer_config.get_from_extra_config(
+                "tls_config", {})
+        ssl_keyfile = tls_config.get("ssl_keyfile", None)
+        ssl_certfile = tls_config.get("ssl_certfile", None)
+        ssl_ca_certs = tls_config.get("ssl_ca_certs", False)
+        ssl_keyfile_password = tls_config.get("ssl_keyfile_password", None)
+        self.cert_path = (ssl_certfile, ssl_keyfile, ssl_keyfile_password)
+        self.ssl_enable = tls_config.get("ssl_enable", False)
+        self.ca_path = ssl_ca_certs
+        if self.ssl_enable:
+            self.metaserver_client = httpx.Client(
+                limits=httpx.Limits(max_connections=100000),
+                timeout=None,
+                cert=self.cert_path,
+                verify=self.ca_path)
+        else:
+            self.metaserver_client = httpx.Client(
+                limits=httpx.Limits(max_connections=100000), timeout=None)
 
     def get_num_new_matched_tokens(
             self, request: "Request",
@@ -645,11 +661,10 @@ class MooncakeLayerwiseConnectorScheduler:
                 remote_host=self.side_channel_host,
                 remote_port=self.side_channel_port,
             )
-            future = self.executor.submit(
-                self._access_metaserver,
-                url=params.get("metaserver", None),
-                message=kv_transfer_params,
-            )
+
+            future = self.executor.submit(self._access_metaserver,
+                                          url=params.get("metaserver", None),
+                                          message=kv_transfer_params)
 
             def handle_exception(future):
                 if future.exception():