CLI: Store OAuth credentials in local file (#626)

* CLI: Store OAuth credentials in local file

* Fix non-file path for CLI store

* Always assume oauth storage path is file

* Add --disable-oauth-store parameter

* Update CLI help

---------

Co-authored-by: Richard <rikroe@users.noreply.github.com>

bimmer_connected/account.py

@@ -93,7 +93,7 @@ async def _init_vehicles(self) -> None:
 
                     vehicle_base = dict(
                         {ATTR_ATTRIBUTES: {k: v for k, v in vehicle_profile.items() if k != "vin"}},
-                        **{"vin": vehicle_profile["vin"]}
+                        **{"vin": vehicle_profile["vin"]},
                     )
 
                     await self.add_vehicle(vehicle_base, fetched_at)
@@ -156,10 +156,15 @@ def set_observer_position(self, latitude: float, longitude: float) -> None:
         """Set the position of the observer for all vehicles."""
         self.config.observer_position = GPSPosition(latitude=latitude, longitude=longitude)
 
-    def set_refresh_token(self, refresh_token: str, gcid: Optional[str] = None) -> None:
-        """Overwrite the current value of the MyBMW refresh token and GCID (if available)."""
+    def set_refresh_token(
+        self, refresh_token: str, gcid: Optional[str] = None, access_token: Optional[str] = None
+    ) -> None:
+        """Overwrite the current value of the MyBMW tokens and GCID (if available)."""
         self.config.authentication.refresh_token = refresh_token
-        self.config.authentication.gcid = gcid
+        if gcid:
+            self.config.authentication.gcid = gcid
+        if access_token:
+            self.config.authentication.access_token = access_token
 
     @staticmethod
     def get_stored_responses() -> List[AnonymizedResponse]:

bimmer_connected/api/authentication.py

@@ -137,11 +137,11 @@ async def login(self) -> None:
             if not token_data:
                 token_data = await self._login_china()
             token_data["expires_at"] = token_data["expires_at"] - EXPIRES_AT_OFFSET
-            self.gcid = token_data["gcid"]
 
         self.access_token = token_data["access_token"]
         self.expires_at = token_data["expires_at"]
         self.refresh_token = token_data["refresh_token"]
+        self.gcid = token_data["gcid"]
 
     async def _login_row_na(self):
         """Login to Rest of World and North America."""
@@ -227,6 +227,7 @@ async def _login_row_na(self):
             "access_token": response_json["access_token"],
             "expires_at": expires_at,
             "refresh_token": response_json["refresh_token"],
+            "gcid": response_json["gcid"],
         }
 
     async def _refresh_token_row_na(self):
@@ -271,6 +272,7 @@ async def _refresh_token_row_na(self):
             "access_token": response_json["access_token"],
             "expires_at": expires_at,
             "refresh_token": response_json["refresh_token"],
+            "gcid": response_json["gcid"],
         }
 
     async def _login_china(self):

bimmer_connected/cli.py

@@ -23,8 +23,18 @@
 
 def main_parser() -> argparse.ArgumentParser:
     """Create the ArgumentParser with all relevant subparsers."""
-    parser = argparse.ArgumentParser(description="A simple executable to use and test the library.")
+    parser = argparse.ArgumentParser(description="Connect to MyBMW/MINI API and interact with your vehicle.")
     parser.add_argument("--debug", help="Print debug logs.", action="store_true")
+    parser.add_argument(
+        "--oauth-store",
+        help="Path to the OAuth2 storage file. Defaults to $HOME/.bimmer_connected.json.",
+        nargs="?",
+        metavar="FILE",
+        type=Path,
+        default=Path.home() / ".bimmer_connected.json",
+    )
+    parser.add_argument("--disable-oauth-store", help="Disable storing the OAuth2 tokens.", action="store_true")
+
     subparsers = parser.add_subparsers(dest="cmd")
     subparsers.required = True
 
@@ -125,15 +135,12 @@ def main_parser() -> argparse.ArgumentParser:
     return parser
 
 
-async def get_status(args) -> None:
+async def get_status(account: MyBMWAccount, args) -> None:
     """Get the vehicle status."""
     if args.json:
         for handler in logging.root.handlers[:]:
             logging.root.removeHandler(handler)
 
-    account = MyBMWAccount(args.username, args.password, get_region_from_name(args.region))
-    if args.lat and args.lng:
-        account.set_observer_position(args.lat, args.lng)
     await account.get_vehicles()
 
     if args.json:
@@ -162,59 +169,47 @@ def get_vehicle_or_return(account: MyBMWAccount, vin: str) -> MyBMWVehicle:
     return vehicle
 
 
-async def fingerprint(args) -> None:
+async def fingerprint(account: MyBMWAccount, args) -> None:
     """Save the vehicle fingerprint."""
     time_dir = Path.home() / "vehicle_fingerprint" / time.strftime("%Y-%m-%d_%H-%M-%S")
     time_dir.mkdir(parents=True)
 
-    account = MyBMWAccount(
-        args.username,
-        args.password,
-        get_region_from_name(args.region),
-        log_responses=True,
-    )
-    if args.lat and args.lng:
-        account.set_observer_position(args.lat, args.lng)
+    account.config.log_responses = True
     await account.get_vehicles()
 
     log_response_store_to_file(account.get_stored_responses(), time_dir)
     print(f"fingerprint of the vehicles written to {time_dir}")
 
 
-async def light_flash(args) -> None:
+async def light_flash(account: MyBMWAccount, args) -> None:
     """Trigger the vehicle to flash its lights."""
-    account = MyBMWAccount(args.username, args.password, get_region_from_name(args.region))
     await account.get_vehicles()
     vehicle = get_vehicle_or_return(account, args.vin)
     status = await vehicle.remote_services.trigger_remote_light_flash()
     print(status.state)
 
 
-async def horn(args) -> None:
+async def horn(account: MyBMWAccount, args) -> None:
     """Trigger the vehicle to horn."""
-    account = MyBMWAccount(args.username, args.password, get_region_from_name(args.region))
     await account.get_vehicles()
     vehicle = get_vehicle_or_return(account, args.vin)
     status = await vehicle.remote_services.trigger_remote_horn()
     print(status.state)
 
 
-async def vehicle_finder(args) -> None:
+async def vehicle_finder(account: MyBMWAccount, args) -> None:
     """Trigger the vehicle finder to locate it."""
-    account = MyBMWAccount(args.username, args.password, get_region_from_name(args.region))
-    account.set_observer_position(args.lat, args.lng)
     await account.get_vehicles()
     vehicle = get_vehicle_or_return(account, args.vin)
     status = await vehicle.remote_services.trigger_remote_vehicle_finder()
     print(status.state)
     print({"gps_position": vehicle.vehicle_location.location, "heading": vehicle.vehicle_location.heading})
 
 
-async def chargingsettings(args) -> None:
+async def chargingsettings(account: MyBMWAccount, args) -> None:
     """Trigger a change to charging settings."""
     if not args.target_soc and not args.ac_limit:
         raise ValueError("At least one of 'charging-target' and 'ac-limit' has to be provided.")
-    account = MyBMWAccount(args.username, args.password, get_region_from_name(args.region))
     await account.get_vehicles()
     vehicle = get_vehicle_or_return(account, args.vin)
     status = await vehicle.remote_services.trigger_charging_settings_update(
@@ -223,11 +218,10 @@ async def chargingsettings(args) -> None:
     print(status.state)
 
 
-async def chargingprofile(args) -> None:
+async def chargingprofile(account: MyBMWAccount, args) -> None:
     """Trigger a change to charging profile."""
     if not args.charging_mode and not args.precondition_climate:
         raise ValueError("At least one of 'charging-mode' and 'precondition-climate' has to be provided.")
-    account = MyBMWAccount(args.username, args.password, get_region_from_name(args.region))
     await account.get_vehicles()
     vehicle = get_vehicle_or_return(account, args.vin)
     status = await vehicle.remote_services.trigger_charging_profile_update(
@@ -236,18 +230,16 @@ async def chargingprofile(args) -> None:
     print(status.state)
 
 
-async def charge(args) -> None:
+async def charge(account: MyBMWAccount, args) -> None:
     """Trigger a vehicle to start or stop charging."""
-    account = MyBMWAccount(args.username, args.password, get_region_from_name(args.region))
     await account.get_vehicles()
     vehicle = get_vehicle_or_return(account, args.vin)
     status = await getattr(vehicle.remote_services, f"trigger_charge_{args.action.lower()}")()
     print(status.state)
 
 
-async def image(args) -> None:
+async def image(account: MyBMWAccount, args) -> None:
     """Download a rendered image of the vehicle."""
-    account = MyBMWAccount(args.username, args.password, get_region_from_name(args.region))
     await account.get_vehicles()
     vehicle = get_vehicle_or_return(account, args.vin)
 
@@ -261,9 +253,8 @@ async def image(args) -> None:
         print(f"vehicle image saved to {filename}")
 
 
-async def send_poi(args) -> None:
+async def send_poi(account: MyBMWAccount, args) -> None:
     """Send Point Of Interest to car."""
-    account = MyBMWAccount(args.username, args.password, get_region_from_name(args.region))
     await account.get_vehicles()
     vehicle = get_vehicle_or_return(account, args.vin)
 
@@ -279,9 +270,8 @@ async def send_poi(args) -> None:
     await vehicle.remote_services.trigger_send_poi(poi_data)
 
 
-async def send_poi_from_address(args) -> None:
+async def send_poi_from_address(account: MyBMWAccount, args) -> None:
     """Create Point of Interest from OSM Nominatim and send to car."""
-    account = MyBMWAccount(args.username, args.password, get_region_from_name(args.region))
     await account.get_vehicles()
     vehicle = get_vehicle_or_return(account, args.vin)
 
@@ -337,13 +327,37 @@ def main():
         logging.basicConfig(level=logging.DEBUG)
         logging.getLogger("asyncio").setLevel(logging.WARNING)
 
+    account = MyBMWAccount(args.username, args.password, get_region_from_name(args.region))
+    if args.lat and args.lng:
+        account.set_observer_position(args.lat, args.lng)
+
+    if args.oauth_store.exists():
+        try:
+            account.set_refresh_token(**json.load(args.oauth_store.open()))
+        except json.JSONDecodeError:
+            pass
+
     loop = asyncio.get_event_loop()
     try:
-        loop.run_until_complete(args.func(args))
+        loop.run_until_complete(args.func(account, args))
     except Exception as ex:  # pylint: disable=broad-except
         sys.stderr.write(f"{type(ex).__name__}: {ex}\n")
         sys.exit(1)
 
+    if args.disable_oauth_store:
+        return
+
+    args.oauth_store.parent.mkdir(parents=True, exist_ok=True)
+    args.oauth_store.write_text(
+        json.dumps(
+            {
+                "refresh_token": account.config.authentication.refresh_token,
+                "gcid": account.config.authentication.gcid,
+                "access_token": account.config.authentication.access_token,
+            }
+        ),
+    )
+
 
 if __name__ == "__main__":
     main()

bimmer_connected/tests/conftest.py

@@ -44,6 +44,15 @@ def bmw_log_all_responses(monkeypatch: pytest.MonkeyPatch):
     monkeypatch.setattr("bimmer_connected.account.RESPONSE_STORE", temp_store)
 
 
+@pytest.fixture
+def cli_home_dir(tmp_path_factory: pytest.TempPathFactory, monkeypatch: pytest.MonkeyPatch):
+    """Create a temporary home directory for the CLI tests."""
+    tmp_path = tmp_path_factory.mktemp("cli-home-")
+    monkeypatch.setattr("pathlib.Path.home", lambda: tmp_path)
+
+    return tmp_path
+
+
 async def prepare_account_with_vehicles(region: Optional[Regions] = None):
     """Initialize account and get vehicles."""
     account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, region or TEST_REGION)

bimmer_connected/tests/test_account.py

@@ -373,11 +373,11 @@ async def test_refresh_token_getset(bmw_fixture: respx.Router):
     assert account.refresh_token is None
     await account.get_vehicles()
     assert account.refresh_token == "another_token_string"
-    assert account.gcid is None
+    assert account.gcid == "DUMMY"
 
     account.set_refresh_token("new_refresh_token")
     assert account.refresh_token == "new_refresh_token"
-    assert account.gcid is None
+    assert account.gcid == "DUMMY"
 
     account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name("china"))
     account.set_refresh_token("new_refresh_token", "dummy_gcid")