SSO Login Endpoint and Session Token Refresh

.gitignore

@@ -77,6 +77,7 @@ target/
 
 # Jupyter Notebook
 .ipynb_checkpoints
+*.ipynb
 
 # IPython
 profile_default/

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 <[email protected]>"]
 license = "MIT"

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,55 +109,49 @@ 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,
         )
         raise ChargePointLoginError(login, "Failed to authenticate to ChargePoint!")
 
     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:
             raise ChargePointCommunicationException(
                 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: