ignore urllib3 InsecureRequestWarning for http requests

client/starwhale/base/cloud.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import typing as t
 from copy import deepcopy
 from http import HTTPStatus
@@ -32,6 +34,12 @@
 _DEFAULT_TIMEOUT_SECS = 90
 _UPLOAD_CHUNK_SIZE = 20 * 1024 * 1024
 
+# TODO: support users to set ssl_verify
+# Current http request disable ssl verify, because of the self-signed certificates, so disable the warning.
+# https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html#ssl-warnings
+_urllib3 = requests.packages.urllib3  # type: ignore
+_urllib3.disable_warnings(_urllib3.exceptions.InsecureRequestWarning)
+
 
 class CloudRequestMixed:
     def fmt_timestamp(self, ts: t.Union[float, str]) -> str:
@@ -131,17 +139,26 @@ def do_http_request_simple_ret(
     @http_retry
     def do_http_request(
         path: str,
-        instance: Instance,
+        instance: Instance | str,
         method: str = HTTPMethod.GET,
         timeout: int = _DEFAULT_TIMEOUT_SECS,
         headers: t.Optional[t.Dict[str, t.Any]] = None,
         disable_default_content_type: bool = False,
         **kw: t.Any,
     ) -> requests.Response:
-        _url = f"{instance.url}/api/{SW_API_VERSION}/{path.lstrip('/')}"
-        _headers = {
-            "Authorization": instance.token,
-        }
+        if isinstance(instance, Instance):
+            server = instance.url
+            token = instance.token
+        else:
+            server = instance
+            token = None
+
+        _url = f"{server}/api/{SW_API_VERSION}/{path.lstrip('/')}"
+        _headers = {}
+
+        if token:
+            _headers["Authorization"] = token
+
         if not disable_default_content_type:
             _headers["Content-Type"] = "application/json"
 

client/starwhale/core/instance/view.py

@@ -8,15 +8,16 @@
 from rich.table import Table
 
 from starwhale.utils import console, fmt_http_server
-from starwhale.consts import UserRoleType, SW_API_VERSION, STANDALONE_INSTANCE
+from starwhale.consts import HTTPMethod, UserRoleType, STANDALONE_INSTANCE
 from starwhale.base.view import BaseTermView
+from starwhale.base.cloud import CloudRequestMixed
 from starwhale.utils.http import wrap_sw_error_resp
 from starwhale.base.uri.instance import Instance
 
 DEFAULT_HTTP_TIMEOUT = 90
 
 
-class InstanceTermView(BaseTermView):
+class InstanceTermView(BaseTermView, CloudRequestMixed):
     def __init__(self) -> None:
         super().__init__()
 
@@ -64,22 +65,22 @@ def login(self, instance: str, alias: str, **kw: str) -> None:
             wrap_sw_error_resp(r, "login failed!", exit=True)
 
     def _login_request_by_token(self, server: str, token: str) -> requests.Response:
-        url = f"{server}/api/{SW_API_VERSION}/user/current"
-        return requests.get(
-            url,
+        return self.do_http_request(  # type: ignore[no-any-return]
+            path="/user/current",
+            instance=Instance(uri=server, token=token),
+            method=HTTPMethod.GET,
             timeout=DEFAULT_HTTP_TIMEOUT,
-            verify=False,
-            headers={"Authorization": token},
         )
 
     def _login_request_by_username(
         self, server: str, auth_request: t.Dict[str, str]
     ) -> requests.Response:
-        url = f"{server}/api/{SW_API_VERSION}/login"
-        return requests.post(
-            url,
-            verify=False,
+        return self.do_http_request(  # type: ignore[no-any-return]
+            path="/login",
+            instance=server,
+            method=HTTPMethod.POST,
             timeout=DEFAULT_HTTP_TIMEOUT,
+            disable_default_content_type=True,
             data={
                 "userName": auth_request["username"],
                 "userPwd": auth_request["password"],