Allow passing external SSL context

bimmer_connected/account.py

@@ -6,6 +6,8 @@
 from dataclasses import InitVar, dataclass, field
 from typing import List, Optional
 
+import httpx
+
 from bimmer_connected.api.authentication import MyBMWAuthentication
 from bimmer_connected.api.client import RESPONSE_STORE, MyBMWClient, MyBMWClientConfiguration
 from bimmer_connected.api.regions import Regions
@@ -45,12 +47,15 @@ class MyBMWAccount:
     observer_position: InitVar[GPSPosition] = None
     """Optional. Required for getting a position on older cars."""
 
+    verify: InitVar[httpx._types.VerifyTypes] = True
+    """Optional. Specify SSL context (required for Home Assistant)."""
+
     use_metric_units: InitVar[Optional[bool]] = None
     """Deprecated. All returned values are metric units (km, l)."""
 
     vehicles: List[MyBMWVehicle] = field(default_factory=list, init=False)
 
-    def __post_init__(self, password, log_responses, observer_position, use_metric_units):
+    def __post_init__(self, password, log_responses, observer_position, verify, use_metric_units):
         """Initialize the account."""
 
         if use_metric_units is not None:
@@ -61,9 +66,10 @@ def __post_init__(self, password, log_responses, observer_position, use_metric_u
 
         if self.config is None:
             self.config = MyBMWClientConfiguration(
-                MyBMWAuthentication(self.username, password, self.region),
+                MyBMWAuthentication(self.username, password, self.region, verify=verify),
                 log_responses=log_responses,
                 observer_position=observer_position,
+                verify=verify,
             )
 
     async def _init_vehicles(self) -> None:

bimmer_connected/api/authentication.py

@@ -53,6 +53,7 @@ def __init__(
         expires_at: Optional[datetime.datetime] = None,
         refresh_token: Optional[str] = None,
         gcid: Optional[str] = None,
+        verify: httpx._types.VerifyTypes = True,
     ):
         self.username: str = username
         self.password: str = password
@@ -63,6 +64,9 @@ def __init__(
         self.session_id: str = str(uuid4())
         self._lock: Optional[asyncio.Lock] = None
         self.gcid: Optional[str] = gcid
+        # Use external SSL context. Required in Home Assistant due to event loop blocking when httpx loads
+        # SSL certificates from disk. If not given, uses httpx defaults.
+        self.verify: Optional[httpx._types.VerifyTypes] = verify
 
     @property
     def login_lock(self) -> asyncio.Lock:
@@ -145,7 +149,7 @@ async def login(self) -> None:
 
     async def _login_row_na(self):
         """Login to Rest of World and North America."""
-        async with MyBMWLoginClient(region=self.region) as client:
+        async with MyBMWLoginClient(region=self.region, verify=self.verify) as client:
             _LOGGER.debug("Authenticating with MyBMW flow for North America & Rest of World.")
 
             # Get OAuth2 settings from BMW API
@@ -233,7 +237,7 @@ async def _login_row_na(self):
     async def _refresh_token_row_na(self):
         """Login to Rest of World and North America using existing refresh_token."""
         try:
-            async with MyBMWLoginClient(region=self.region) as client:
+            async with MyBMWLoginClient(region=self.region, verify=self.verify) as client:
                 _LOGGER.debug("Authenticating with refresh token for North America & Rest of World.")
 
                 # Get OAuth2 settings from BMW API
@@ -276,7 +280,7 @@ async def _refresh_token_row_na(self):
         }
 
     async def _login_china(self):
-        async with MyBMWLoginClient(region=self.region) as client:
+        async with MyBMWLoginClient(region=self.region, verify=self.verify) as client:
             _LOGGER.debug("Authenticating with MyBMW flow for China.")
 
             # While PIL.Image is only needed in `get_capture_position`, we test it here to avoid
@@ -329,7 +333,7 @@ async def _login_china(self):
 
     async def _refresh_token_china(self):
         try:
-            async with MyBMWLoginClient(region=self.region) as client:
+            async with MyBMWLoginClient(region=self.region, verify=self.verify) as client:
                 _LOGGER.debug("Authenticating with refresh token for China.")
 
                 current_utc_time = datetime.datetime.now(tz=datetime.timezone.utc)

bimmer_connected/api/client.py

@@ -25,6 +25,7 @@ class MyBMWClientConfiguration:
     authentication: MyBMWAuthentication
     log_responses: Optional[bool] = False
     observer_position: Optional[GPSPosition] = None
+    verify: httpx._types.VerifyTypes = True
 
     def set_log_responses(self, log_responses: bool) -> None:
         """Set if responses are logged and clear response store."""
@@ -45,6 +46,10 @@ def __init__(self, config: MyBMWClientConfiguration, *args, brand: Optional[CarB
         # Increase timeout
         kwargs["timeout"] = httpx.Timeout(HTTPX_TIMEOUT)
 
+        # Use external SSL context stored in MyBMWClientConfiguration. Required in Home Assistant due to event loop
+        # blocking when httpx loads SSL certificates from disk. If not given, uses httpx defaults.
+        kwargs["verify"] = self.config.verify
+
         # Set default values
         kwargs["base_url"] = kwargs.get("base_url") or get_server_url(config.authentication.region)
         kwargs["headers"] = kwargs.get("headers") or self.generate_default_header(brand)