diff --git a/.gitignore b/.gitignore index b9c1ab8..32e8c98 100644 --- a/.gitignore +++ b/.gitignore @@ -77,6 +77,7 @@ target/ # Jupyter Notebook .ipynb_checkpoints +*.ipynb # IPython profile_default/ diff --git a/pyproject.toml b/pyproject.toml index 19a4f31..4894dcd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "python-chargepoint" -version = "1.10.0" +version = "1.11.0" description = "A simple, Pythonic wrapper for the ChargePoint API." authors = ["Marc Billow "] license = "MIT" diff --git a/python_chargepoint/client.py b/python_chargepoint/client.py index f065a53..a4bdecc 100644 --- a/python_chargepoint/client.py +++ b/python_chargepoint/client.py @@ -1,9 +1,9 @@ -from uuid import uuid4 from typing import List, Optional from functools import wraps from time import sleep +from importlib.metadata import version, PackageNotFoundError -from requests import Session, codes, post +from requests import Session, codes from .types import ( ChargePointAccount, @@ -21,17 +21,14 @@ from .global_config import ChargePointGlobalConfiguration from .session import ChargingSession from .constants import _LOGGER, DISCOVERY_API +from . import __name__ as MODULE_NAME +try: + MODULE_VERSION = version(MODULE_NAME) +except PackageNotFoundError: + MODULE_VERSION = "unknown" -def _dict_for_query(device_data: dict) -> dict: - """ - GET requests send device data as a nested object. - To avoid storing the device data block in two - formats, we are just going to compute the flat - dictionary. - """ - return {f"deviceData[{key}]": value for key, value in device_data.items()} - +USER_AGENT = f"{MODULE_NAME}/{MODULE_VERSION}" def _require_login(func): @wraps(func) @@ -57,33 +54,28 @@ def __init__( self, username: str, password: str, - session_token: str = "", - app_version: str = "5.97.0", + auth_token: Optional[str] = "", + session_token: Optional[str] = "", ): self._session = Session() - self._app_version = app_version - self._device_data = { - "appId": "com.coulomb.ChargePoint", - "manufacturer": "Apple", - "model": "iPhone", - "notificationId": "", - "notificationIdType": "", - "type": "IOS", - "udid": str(uuid4()), - "version": app_version, - } - self._device_query_params = _dict_for_query(self._device_data) self._user_id = None self._logged_in = False - self._session_token = None + + self._session.headers = { + "user-agent": USER_AGENT, + } + self._global_config = self._get_configuration(username) - if session_token: + if session_token or auth_token: self._set_session_token(session_token) + self._set_auth_token(auth_token) self._logged_in = True try: + self._get_initial_session_token() account: ChargePointAccount = self.get_account() self._user_id = str(account.user.user_id) + self.refresh_session_token() return except ChargePointCommunicationException: _LOGGER.warning( @@ -103,11 +95,8 @@ def session(self) -> Session: @property def session_token(self) -> Optional[str]: - return self._session_token + return self._get_session_token() - @property - def device_data(self) -> dict: - return self._device_data @property def global_config(self) -> ChargePointGlobalConfiguration: @@ -120,31 +109,28 @@ def login(self, username: str, password: str) -> None: :param password: Account password """ login_url = ( - f"{self._global_config.endpoints.accounts}v2/driver/profile/account/login" + f"{self._global_config.endpoints.sso}v1/user/login" ) - headers = { - "User-Agent": f"com.coulomb.ChargePoint/{self._app_version} CFNetwork/1329 Darwin/21.3.0" - } + request = { - "deviceData": self._device_data, "username": username, "password": password, } _LOGGER.debug("Attempting client login with user: %s", username) - login = post(login_url, json=request, headers=headers) + login = self._session.post(login_url, json=request) _LOGGER.debug(login.cookies.get_dict()) _LOGGER.debug(login.headers) if login.status_code == codes.ok: - req = login.json() - self._user_id = req["user"]["userId"] - _LOGGER.debug("Authentication success! User ID: %s", self._user_id) - self._set_session_token(req["sessionId"]) self._logged_in = True + self._get_initial_session_token() + account: ChargePointAccount = self.get_account() + self._user_id = str(account.user.user_id) + self.refresh_session_token() return _LOGGER.error( - "Failed to get account information! status_code=%s err=%s", + "Failed to get auth token! status_code=%s err=%s", login.status_code, login.text, ) @@ -152,8 +138,7 @@ def login(self, username: str, password: str) -> None: def logout(self): response = self._session.post( - f"{self._global_config.endpoints.accounts}v1/driver/profile/account/logout", - json={"deviceData": self._device_data}, + f"{self._global_config.endpoints.sso}v1/user/logout", ) if response.status_code != codes.ok: @@ -161,14 +146,12 @@ def logout(self): response=response, message="Failed to log out!" ) - self._session.headers = {} self._session.cookies.clear_session_cookies() - self._session_token = None self._logged_in = False def _get_configuration(self, username: str) -> ChargePointGlobalConfiguration: _LOGGER.debug("Discovering account region for username %s", username) - request = {"deviceData": self._device_data, "username": username} + request = {"username": username} response = self._session.post(DISCOVERY_API, json=request) if response.status_code != codes.ok: raise ChargePointCommunicationException( @@ -184,28 +167,65 @@ def _get_configuration(self, username: str) -> ChargePointGlobalConfiguration: ) return config + def _get_initial_session_token(self): + _LOGGER.debug("Requesting inital session token") + response = self._session.post( + f"{self._global_config.endpoints.portal_domain}index.php/nghelper/getSession", json={"user_id": self.user_id} + ) + + if (response.status_code != codes.ok): + _LOGGER.error( + "Failed to get session! status_code=%s err=%s", + response.status_code, + response.text, + ) + raise ChargePointCommunicationException( + response=response, message="Failed to retrieve session." + ) + + def refresh_session_token(self): + _LOGGER.debug("Requesting long lived token") + response = self._session.post( + f"{self._global_config.endpoints.webservices}mobileapi/v5", json={"user_id": self.user_id} + ) + + if (response.status_code != codes.ok): + _LOGGER.error( + "Failed to get long lived token! status_code=%s err=%s", + response.status_code, + response.text, + ) + raise ChargePointCommunicationException( + response=response, message="Failed to retrieve long lived token." + ) + + self._session.cookies.clear(domain='') + + def _get_session_token(self) -> str: + out ='' + token = [cookie for cookie in self._session.cookies if cookie.name == 'coulomb_sess'] + + if token: + out = token[0].value + + return out + def _set_session_token(self, session_token: str): - try: - self._session.headers = { - "cp-session-type": "CP_SESSION_TOKEN", - "cp-session-token": session_token, - # Data: |------------------Token Data------------------||---?---||-Reg-| - # Session ID: rAnDomBaSe64EnCodEdDaTaToKeNrAnDomBaSe64EnCodEdD#D???????#RNA-US - "cp-region": session_token.split("#R")[1], - "user-agent": "ChargePoint/236 (iPhone; iOS 15.3; Scale/3.00)", - } - except IndexError: - raise ChargePointBaseException("Invalid session token format.") + if session_token: + if len(session_token) != 32: + raise ChargePointBaseException("Invalid session token format.") + + self._session.cookies.set("coulomb_sess", session_token) - self._session_token = session_token - self._session.cookies.set("coulomb_sess", session_token) + def _set_auth_token(self, auth_token: str): + if auth_token: + self._session.cookies.set("auth-session", auth_token) @_require_login def get_account(self) -> ChargePointAccount: _LOGGER.debug("Getting ChargePoint Account Details") response = self._session.get( f"{self._global_config.endpoints.accounts}v1/driver/profile/user", - params=self._device_query_params, ) if response.status_code != codes.ok: @@ -226,7 +246,6 @@ def get_vehicles(self) -> List[ElectricVehicle]: _LOGGER.debug("Listing vehicles") response = self._session.get( f"{self._global_config.endpoints.accounts}v1/driver/vehicle", - params=self._device_query_params, ) if response.status_code != codes.ok: @@ -334,7 +353,7 @@ def get_home_charger_technical_info( @_require_login def get_user_charging_status(self) -> Optional[UserChargingStatus]: _LOGGER.debug("Checking account charging status") - request = {"deviceData": self._device_data, "user_status": {"mfhs": {}}} + request = {"user_status": {"mfhs": {}}} response = self._session.post( f"{self._global_config.endpoints.mapcache}v2", json=request ) @@ -363,13 +382,21 @@ def set_amperage_limit( self, charger_id: int, amperage_limit: int, max_retry: int = 5 ) -> None: _LOGGER.debug(f"Setting amperage limit for {charger_id} to {amperage_limit}") + + headers = { + "cp-session-type": "CP_SESSION_TOKEN", + "cp-session-token": self._get_session_token(), + "cp-region": self._global_config.region, + } + headers.update(self._session.headers) + request = { - "deviceData": self._device_data, "chargeAmperageLimit": amperage_limit, } response = self._session.post( f"{self._global_config.endpoints.internal_api}/driver/charger/{charger_id}/config/v1/charge-amperage-limit", json=request, + headers=headers, ) if response.status_code != codes.ok: