From a7a450786f270064df3c6963ceedfa58518b331e Mon Sep 17 00:00:00 2001 From: Steven Hosking Date: Sun, 25 Jan 2026 11:30:58 +1100 Subject: [PATCH 01/12] feat: Add INTL region support for Hello Smart International app Add support for international markets (Australia, Singapore, Israel, etc.) using the Hello Smart International app authentication flow. Changes: - Add SmartAuthenticationINTL class with 3-step auth flow - Add INTL API constants and signing secrets for sg-app-api.smart.com - Add SmartRegion enum (EU/INTL) and get_endpoint_urls_for_region() helper - Add region parameter to SmartAccount (defaults to EU for backward compat) - Add generate_intl_header() for INTL-specific request signing New vehicle data models: - Add BasicStatus dataclass (speed, engine status, park time, etc.) - Add Trailer dataclass (lamp status fields) - Add basic_status and trailer properties to Vehicle Extended battery data: - V2L discharge fields (voltage, current, power, connection status) - Charge lid status (AC/DC), scheduled charging, DCDC system status - Range at 20% SoC Extended climate data: - PM2.5 level fields (interior, secondary, exterior) Fixed: - Use ecloudeu.com for INTL data endpoints (shared with EU) - Reduce log spam for optional rear door position fields All 97 existing tests pass. EU route unchanged. --- pyproject.toml | 1 + pysmarthashtag/account.py | 114 ++++++--- pysmarthashtag/api/authentication.py | 274 +++++++++++++++++++++ pysmarthashtag/api/utils.py | 125 +++++++++- pysmarthashtag/const.py | 69 ++++++ pysmarthashtag/tests/test_endpoint_urls.py | 61 +++++ pysmarthashtag/vehicle/basic_status.py | 152 ++++++++++++ pysmarthashtag/vehicle/battery.py | 101 ++++++++ pysmarthashtag/vehicle/climate.py | 14 ++ pysmarthashtag/vehicle/safety.py | 6 +- pysmarthashtag/vehicle/trailer.py | 69 ++++++ pysmarthashtag/vehicle/vehicle.py | 10 + 12 files changed, 959 insertions(+), 37 deletions(-) create mode 100644 pysmarthashtag/vehicle/basic_status.py create mode 100644 pysmarthashtag/vehicle/trailer.py diff --git a/pyproject.toml b/pyproject.toml index c24b9d5..d44fb56 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -78,6 +78,7 @@ exclude = [ [tool.ruff.per-file-ignores] "docs/source/conf.py" = ["D100"] "pysamrthashtag/api/authentication.py" = ["D102", "D107"] +"pysmarthashtag/vehicle/battery.py" = ["C901"] # Complex data parsing function [tool.ruff.mccabe] max-complexity = 25 diff --git a/pysmarthashtag/account.py b/pysmarthashtag/account.py index de347fd..7fa056c 100644 --- a/pysmarthashtag/account.py +++ b/pysmarthashtag/account.py @@ -7,10 +7,10 @@ from typing import Optional from pysmarthashtag.api import utils -from pysmarthashtag.api.authentication import SmartAuthentication +from pysmarthashtag.api.authentication import SmartAuthentication, SmartAuthenticationINTL from pysmarthashtag.api.client import SmartClient, SmartClientConfiguration from pysmarthashtag.api.log_sanitizer import sanitize_log_data -from pysmarthashtag.const import API_CARS_URL, API_SELECT_CAR_URL, EndpointUrls +from pysmarthashtag.const import API_CARS_URL, API_SELECT_CAR_URL, EndpointUrls, SmartRegion from pysmarthashtag.models import SmartAuthError, SmartHumanCarConnectionError, SmartTokenRefreshNecessary from pysmarthashtag.vehicle.vehicle import SmartVehicle @@ -38,21 +38,72 @@ class SmartAccount: endpoint_urls: Optional[EndpointUrls] = None """Optional. Custom endpoint URLs for international API support.""" + region: Optional[SmartRegion] = None + """Optional. Region preset (EU or INTL). If set, overrides endpoint_urls.""" + vehicles: dict[str, SmartVehicle] = field(default_factory=dict, init=False) """Vehicles associated with the account.""" def __post_init__(self, password, log_responses): """Initialize the account.""" - # Ensure endpoint_urls is set - if self.endpoint_urls is None: + # Import get_endpoint_urls_for_region here to avoid circular imports + from pysmarthashtag.const import get_endpoint_urls_for_region + + # If region is specified, use it to get endpoint URLs + if self.region is not None: + self.endpoint_urls = get_endpoint_urls_for_region(self.region) + elif self.endpoint_urls is None: self.endpoint_urls = EndpointUrls() + # Choose authentication class based on region + if self.region == SmartRegion.INTL: + _LOGGER.info("Using INTL (International) authentication for region: %s", self.region) + auth = SmartAuthenticationINTL(self.username, password, endpoint_urls=self.endpoint_urls) + else: + _LOGGER.info("Using EU authentication (default)") + auth = SmartAuthentication(self.username, password, endpoint_urls=self.endpoint_urls) + if self.config is None: self.config = SmartClientConfiguration( - SmartAuthentication(self.username, password, endpoint_urls=self.endpoint_urls), + auth, log_responses=log_responses, ) + def _is_intl_region(self) -> bool: + """Check if using INTL (International) region.""" + return self.region == SmartRegion.INTL + + def _generate_api_headers(self, params: dict, method: str, url: str, body=None) -> dict[str, str]: + """Generate API headers based on region (EU or INTL). + + This method automatically selects the correct header generator + based on whether the account is configured for INTL region. + """ + auth = self.config.authentication + + if self._is_intl_region(): + # INTL region uses different headers and signing + client_id = getattr(auth, "api_client_id", None) + return utils.generate_intl_header( + device_id=auth.device_id, + access_token=auth.api_access_token, + params=params, + method=method, + url=url, + body=body, + client_id=client_id, + ) + else: + # EU region uses default headers + return utils.generate_default_header( + device_id=auth.device_id, + access_token=auth.api_access_token, + params=params, + method=method, + url=url, + body=body, + ) + async def _ensure_ssl_context(self) -> None: """Ensure SSL context is created asynchronously. @@ -90,9 +141,7 @@ async def _init_vehicles(self) -> None: # we do not know what type of car we have in our list so we fall back to the old API URL self.endpoint_urls.get_api_base_url() + API_CARS_URL + "?" + utils.join_url_params(params), headers={ - **utils.generate_default_header( - client.config.authentication.device_id, - client.config.authentication.api_access_token, + **self._generate_api_headers( params=params, method="GET", url=API_CARS_URL, @@ -152,9 +201,7 @@ async def select_active_vehicle(self, vin) -> None: r_car_info = await client.post( self.vehicles[vin].base_url + API_SELECT_CAR_URL, headers={ - **utils.generate_default_header( - client.config.authentication.device_id, - client.config.authentication.api_access_token, + **self._generate_api_headers( params={}, method="POST", url=API_SELECT_CAR_URL, @@ -176,28 +223,41 @@ async def select_active_vehicle(self, vin) -> None: async def get_vehicle_information(self, vin) -> str: """Get information about a vehicle.""" _LOGGER.debug("Getting information for vehicle") - params = { - "latest": True, - "target": "basic%2Cmore", - "userId": self.config.authentication.api_user_id, - } + + # INTL uses different API path and parameters than EU + if self._is_intl_region(): + # INTL: No /api/v1/ prefix, latest=False to get all data including climate/doors/pollution + params = { + "latest": False, + "target": "basic,more", + "userId": self.config.authentication.api_user_id, + } + url_path = "/remote-control/vehicle/status/" + vin + # INTL uses api.ecloudeu.com without /api/v1/ prefix + base_url = self.endpoint_urls.get_api_base_url() + else: + # EU: Uses /api/v1/ prefix in the base_url setting + params = { + "latest": True, + "target": "basic,more", + "userId": self.config.authentication.api_user_id, + } + url_path = "/remote-control/vehicle/status/" + vin + base_url = self.vehicles[vin].base_url + data = {} async with SmartClient(self.config) as client: for retry in range(3): try: + # Build URL with URL-encoded params for GET request + url_params = {k: str(v).replace(",", "%2C") for k, v in params.items()} r_car_info = await client.get( - self.vehicles[vin].base_url - + "/remote-control/vehicle/status/" - + vin - + "?" - + utils.join_url_params(params), + base_url + url_path + "?" + utils.join_url_params(url_params), headers={ - **utils.generate_default_header( - client.config.authentication.device_id, - client.config.authentication.api_access_token, + **self._generate_api_headers( params=params, method="GET", - url="/remote-control/vehicle/status/" + vin, + url=url_path, ) }, ) @@ -233,9 +293,7 @@ async def get_vehicle_soc(self, vin) -> str: + "?" + utils.join_url_params(params), headers={ - **utils.generate_default_header( - client.config.authentication.device_id, - client.config.authentication.api_access_token, + **self._generate_api_headers( params=params, method="GET", url="/remote-control/vehicle/status/soc/" + vin, diff --git a/pysmarthashtag/api/authentication.py b/pysmarthashtag/api/authentication.py index 4124d7b..d07aea5 100644 --- a/pysmarthashtag/api/authentication.py +++ b/pysmarthashtag/api/authentication.py @@ -19,6 +19,11 @@ from pysmarthashtag.const import ( API_SESION_URL, HTTPX_TIMEOUT, + INTL_APP_ID, + INTL_CA_KEY, + INTL_LOGIN_URL, + INTL_OAUTH_URL, + INTL_OPERATOR_CODE, EndpointUrls, ) from pysmarthashtag.models import SmartAPIError @@ -390,6 +395,275 @@ async def async_auth_flow(self, request: Request) -> AsyncGenerator[Request, Res raise +class SmartAuthenticationINTL(SmartAuthentication): + """Authentication handler for Smart International (INTL) API. + + This class handles authentication for the Hello Smart International app, + which is used in Australia, Singapore, Israel, and other international markets. + + The INTL auth flow is different from EU: + 1. POST login with email/password to sg-app-api.smart.com -> get accessToken, idToken + 2. GET oauth20/authorize with accessToken -> get authCode + 3. POST session/secure with authCode to apiv2.ecloudeu.com -> get vehicle API tokens + + Note: The sg-app-api.smart.com endpoints do not require X-Ca-Signature validation, + which simplifies the authentication process significantly. + """ + + def __init__( + self, + username: str, + password: str, + access_token: Optional[str] = None, + expires_at: Optional[datetime.datetime] = None, + refresh_token: Optional[str] = None, + ssl_context: Optional[ssl.SSLContext] = None, + endpoint_urls: Optional[EndpointUrls] = None, + ): + super().__init__( + username=username, + password=password, + access_token=access_token, + expires_at=expires_at, + refresh_token=refresh_token, + ssl_context=ssl_context, + endpoint_urls=endpoint_urls, + ) + # INTL-specific tokens + self.intl_access_token: Optional[str] = None + self.intl_refresh_token: Optional[str] = None + self.intl_id_token: Optional[str] = None + self.intl_user_id: Optional[str] = None + self.session_id: str = secrets.token_hex(16).upper() + self.device_identifier: str = secrets.token_hex(16) + self.client_id: str = "2232193363e44527" # INTL OAuth client ID + self.api_client_id: Optional[str] = None # Client ID from session response + _LOGGER.debug("SmartAuthenticationINTL initialized for INTL region") + + def _generate_intl_headers(self, auth_token: str = "") -> dict[str, str]: + """Generate headers for INTL API requests (sg-app-api.smart.com). + + The sg-app-api.smart.com endpoints do not validate X-Ca-Signature, + so we can use empty or minimal signatures. + """ + import time + import uuid + + timestamp = str(int(time.time() * 1000)) + nonce = str(uuid.uuid4()).upper() + + headers = { + "Host": "sg-app-api.smart.com", + "User-Agent": "GlobalSmart/1.0.7 (iPhone; iOS 18.6.1; Scale/3.00)", + "X-Ca-Key": INTL_CA_KEY, + "X-Ca-Timestamp": timestamp, + "X-Ca-Nonce": nonce, + "X-Ca-Signature-Method": "HmacSHA256", + "X-Ca-Signature": "", # Not validated by server + "X-Ca-Signature-Headers": ( + "Accept-Language,User-Agent,X-Ca-Nonce,X-Ca-Timestamp," + "Xs-App-Ver,Xs-Auth-Token,Xs-Client-Id,Xs-Di,Xs-Os,Xs-Session-Id,Xs-Ui" + ), + "Xs-Os": "iOS", + "Xs-App-Ver": "1.0.7", + "Xs-Session-Id": self.session_id, + "Xs-Di": self.device_identifier, + "Xs-Ui": secrets.token_hex(16), + "Xs-Auth-Token": auth_token, + "Xs-Client-Id": self.client_id, + "Accept": "*/*", + "Accept-Language": "en-AU;q=1", + "Content-Type": "application/json", + "Cache-Control": "no-cache", + } + + return headers + + def _generate_vehicle_api_headers(self, body: str = None) -> dict[str, str]: + """Generate headers for vehicle API (ecloudeu) requests. + + These headers are used for the session/secure endpoint on apiv2.ecloudeu.com. + The x-signature is calculated using the INTL signing secret. + """ + import time + import uuid + + timestamp = str(int(time.time() * 1000)) + nonce = str(uuid.uuid4()).upper() + + headers = { + "x-app-id": INTL_APP_ID, + "x-operator-code": INTL_OPERATOR_CODE, + "x-agent-type": "iOS", + "x-agent-version": "18.6.1", + "x-device-type": "mobile", + "x-device-identifier": self.device_identifier, + "x-device-manufacture": "Apple", + "x-device-brand": "Apple", + "x-device-model": "iPhone", + "x-env-type": "production", + "x-api-signature-version": "1.0", + "x-api-signature-nonce": nonce, + "x-timestamp": timestamp, + "platform": "NON-CMA", + "accept": "application/json;responseformat=3", + "accept-language": "en_US", + "content-type": "application/json", + "user-agent": "GlobalSmart/1.0.7 (iPhone; iOS 18.6.1; Scale/3.00)", + } + + # Calculate signature using INTL secret + if body is not None: + headers["x-signature"] = utils._create_sign( + nonce=nonce, + params={"identity_type": "smart-israel"}, + timestamp=timestamp, + method="POST", + url=API_SESION_URL, + body=body, + use_intl=True, + ) + + return headers + + async def _login(self): + """Login to Smart INTL web services. + + INTL login flow (uses sg-app-api.smart.com which doesn't require signatures): + 1. POST to login endpoint with email/password -> accessToken, idToken + 2. GET oauth20/authorize with accessToken -> authCode + 3. POST session/secure to apiv2.ecloudeu.com with authCode -> vehicle API tokens + """ + ssl_ctx = await self.get_ssl_context() + async with SmartLoginClient(ssl_context=ssl_ctx) as client: + _LOGGER.info("INTL: Acquiring access token via Hello Smart International API") + + # Step 1: Login with email/password (no signature required!) + _LOGGER.debug("INTL: Step 1 - Logging in with credentials") + login_data = json.dumps( + { + "email": self.username, + "password": self.password, + } + ) + + r_login = await client.post( + INTL_LOGIN_URL, + content=login_data, + headers=self._generate_intl_headers(), + ) + + try: + login_result = r_login.json() + _LOGGER.debug("INTL login response code: %s", login_result.get("code")) + + if login_result.get("code") != "200": + error_msg = login_result.get("message", "Unknown error") + _LOGGER.error("INTL login failed: %s", error_msg) + raise SmartAPIError(f"INTL login failed: {error_msg}") + + result = login_result.get("result", {}) + self.intl_access_token = result.get("accessToken") + self.intl_refresh_token = result.get("refreshToken") + self.intl_id_token = result.get("idToken") + self.intl_user_id = result.get("userId") + expires_in = result.get("expiresIn", 86400) + + expires_at = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=expires_in) + + _LOGGER.info("INTL: Login successful for user %s", self.intl_user_id) + + except (KeyError, ValueError, TypeError) as e: + raise SmartAPIError(f"Could not parse INTL login response: {e}") + + # Step 2: Exchange accessToken for authCode via OAuth + # Note: OAuth uses just accessToken in URL param, and idToken in Xs-Auth-Token header + _LOGGER.debug("INTL: Step 2 - Exchanging accessToken for authCode") + + oauth_url = f"{INTL_OAUTH_URL}?accessToken={self.intl_access_token}" + + oauth_headers = self._generate_intl_headers(self.intl_id_token) + + r_oauth = await client.get( + oauth_url, + headers=oauth_headers, + ) + + try: + oauth_result = r_oauth.json() + _LOGGER.debug("INTL OAuth response code: %s", oauth_result.get("code")) + + if oauth_result.get("code") != "200": + error_msg = oauth_result.get("message", "Unknown error") + _LOGGER.error("INTL OAuth failed: %s", error_msg) + raise SmartAPIError(f"INTL OAuth failed: {error_msg}") + + # Result is directly the authCode string (e.g., "CODE-xxx") + auth_code = oauth_result.get("result") + if isinstance(auth_code, dict): + auth_code = auth_code.get("authCode", auth_code) + + _LOGGER.debug("INTL: Got authCode: %s...", str(auth_code)[:30] if auth_code else "None") + + except (KeyError, ValueError, TypeError) as e: + raise SmartAPIError(f"Could not parse INTL OAuth response: {e}") + + # Step 3: Exchange authCode for vehicle API session + _LOGGER.debug("INTL: Step 3 - Exchanging authCode for vehicle API session") + + session_url = f"{self.endpoint_urls.get_api_base_url_v2()}{API_SESION_URL}?identity_type=smart-israel" + session_data = json.dumps({"authCode": auth_code}, separators=(",", ":")) + + session_headers = self._generate_vehicle_api_headers(body=session_data) + + r_session = await client.post( + session_url, + content=session_data, + headers=session_headers, + ) + + try: + session_result = r_session.json() + _LOGGER.debug("INTL Session response: %s", sanitize_log_data(session_result)) + + if session_result.get("code") != 1000: + error_msg = session_result.get("message", "Unknown error") + _LOGGER.error("INTL session failed: %s (code: %s)", error_msg, session_result.get("code")) + raise SmartAPIError(f"INTL session failed: {error_msg}") + + data = session_result.get("data", {}) + api_access_token = data.get("accessToken") + api_refresh_token = data.get("refreshToken") + api_user_id = data.get("userId") + api_client_id = data.get("clientId") + + # Store client_id on the auth object for use in API requests + self.api_client_id = api_client_id + + _LOGGER.info("INTL: Successfully authenticated, user ID: %s", api_user_id) + + except (KeyError, ValueError, TypeError) as e: + raise SmartAPIError(f"Could not parse INTL session response: {e}") + + return { + "access_token": api_access_token, + "refresh_token": api_refresh_token, + "api_access_token": api_access_token, + "api_refresh_token": api_refresh_token, + "api_user_id": api_user_id, + "expires_at": expires_at, + } + + async def _refresh_access_token(self): + """Refresh the INTL access token. + + For INTL, we simply re-login as the refresh flow is complex + and the login endpoint doesn't require signatures. + """ + _LOGGER.debug("INTL: Refresh requested, performing full re-login") + return {} # Return empty to trigger full login + + def get_retry_wait_time(response: httpx.Response) -> int: """Get the wait time to wait twice as long before retrying.""" try: diff --git a/pysmarthashtag/api/utils.py b/pysmarthashtag/api/utils.py index 1e32318..7aa0593 100644 --- a/pysmarthashtag/api/utils.py +++ b/pysmarthashtag/api/utils.py @@ -4,19 +4,62 @@ import logging import secrets import time +from urllib.parse import quote _LOGGER = logging.getLogger(__name__) +# EU signing secret (base64 encoded) +EU_SECRET = base64.b64decode("NzRlNzQ2OWFmZjUwNDJiYmJlZDdiYmIxYjM2YzE1ZTk=") + +# INTL (International) signing secret - used for Australia, Singapore, Israel, etc. +INTL_SECRET = b"30fb4a7b7fab4e2e8b52120d0087efdd" + def join_url_params(args: dict) -> str: """Join params for adding to URL.""" return "&".join([f"{key}={value}" for key, value in args.items()]) -def _create_sign(nonce: str, params: dict, timestamp: str, method: str, url: str, body=None) -> str: - """Create a signature for the request.""" +def _create_sign( + nonce: str, + params: dict, + timestamp: str, + method: str, + url: str, + body=None, + use_intl: bool = False, + url_encode_params: bool = False, +) -> str: + """Create a signature for the request. + + Args: + ---- + nonce: Random nonce for the request + params: URL parameters as a dict + timestamp: Timestamp in milliseconds + method: HTTP method (GET, POST, etc.) + url: API URL path (without domain) + body: Request body for POST requests + use_intl: If True, use INTL secret instead of EU secret + url_encode_params: If True, URL-encode special chars in params (required for INTL GET requests) + + Returns: + ------- + Base64-encoded HMAC-SHA1 signature. + + """ md5sum = base64.b64encode(hashlib.md5(body.encode()).digest()).decode() if body else "1B2M2Y8AsgTpgAmY7PhCfg==" - url_params = join_url_params(params) + + # URL encode params if needed (INTL GET requests require this for special chars like commas) + if url_encode_params and params: + encoded_params = {} + for k, v in params.items(): + # URL encode special characters (commas, spaces, etc.) + encoded_params[k] = quote(str(v), safe="") + url_params = join_url_params(encoded_params) + else: + url_params = join_url_params(params) if params else "" + payload = f"""application/json;responseformat=3 x-api-signature-nonce:{nonce} x-api-signature-version:1.0 @@ -26,10 +69,13 @@ def _create_sign(nonce: str, params: dict, timestamp: str, method: str, url: str {timestamp} {method} {url}""" - _LOGGER.debug("Creating signature for request") - secret = base64.b64decode("NzRlNzQ2OWFmZjUwNDJiYmJlZDdiYmIxYjM2YzE1ZTk=") - payload = payload.encode("utf-8") - hashed = hmac.new(secret, payload, hashlib.sha1).digest() + _LOGGER.debug("Creating signature for request (INTL=%s)", use_intl) + + # Choose secret based on region + secret = INTL_SECRET if use_intl else EU_SECRET + + payload_bytes = payload.encode("utf-8") + hashed = hmac.new(secret, payload_bytes, hashlib.sha1).digest() signature = base64.b64encode(hashed).decode() return signature @@ -69,6 +115,71 @@ def generate_default_header( return header +def generate_intl_header( + device_id: str, access_token: str, params: dict, method: str, url: str, body=None, client_id: str = None +) -> dict[str, str]: + """Generate a header for HTTP requests to the INTL (International) API. + + This is used for vehicles registered in Australia, Singapore, Israel, + and other international markets using the Hello Smart International app. + + Args: + ---- + device_id: Unique device identifier + access_token: API access token from session/secure + params: URL parameters as a dict + method: HTTP method (GET, POST, etc.) + url: API URL path (without domain) + body: Request body for POST requests + client_id: Client ID from session response + + Returns: + ------- + Dict of headers for the request. + + """ + import uuid + + timestamp = create_correct_timestamp() + nonce = str(uuid.uuid4()).upper() + + # INTL requires URL-encoded params in signature for GET requests with special chars + url_encode_params = method.upper() == "GET" + sign = _create_sign(nonce, params, timestamp, method, url, body, use_intl=True, url_encode_params=url_encode_params) + + header = { + "x-app-id": "SMARTAPP-ISRAEL", + "accept": "application/json;responseformat=3", + "x-agent-type": "iOS", + "x-device-type": "mobile", + "x-operator-code": "SMART-ISRAEL", + "x-device-identifier": device_id, + "x-env-type": "production", + "accept-language": "en_US", + "x-api-signature-version": "1.0", + "x-api-signature-nonce": nonce, + "x-device-manufacture": "Apple", + "x-device-brand": "Apple", + "x-device-model": "iPhone", + "x-agent-version": "18.6.1", + "content-type": "application/json", + "user-agent": "GlobalSmart/1.0.7 (iPhone; iOS 18.6.1; Scale/3.00)", + "x-signature": sign, + "x-timestamp": timestamp, + "platform": "NON-CMA", + "x-vehicle-series": "HC1H2D3B6213-01_IL", + } + + if access_token: + header["authorization"] = access_token + + if client_id: + header["x-client-id"] = client_id + + _LOGGER.debug("Constructed INTL request header for %s %s", method, url) + return header + + def create_correct_timestamp() -> str: """Create a correct timestamp for the request.""" return str(int(time.time() * 1000)) diff --git a/pysmarthashtag/const.py b/pysmarthashtag/const.py index 4ae6da1..a7da9e3 100644 --- a/pysmarthashtag/const.py +++ b/pysmarthashtag/const.py @@ -14,11 +14,80 @@ API_SELECT_CAR_URL = "/device-platform/user/session/update" API_TELEMATICS_URL = "/remote-control/vehicle/telematics/" +# INTL (International) API specific constants +# Used for Hello Smart International app (Australia, Singapore, Israel, etc.) +INTL_API_BASE = "https://sg-app-api.smart.com" +INTL_LOGIN_URL = f"{INTL_API_BASE}/iam/service/api/v1/login" +INTL_OAUTH_URL = f"{INTL_API_BASE}/iam/service/api/v1/oauth20/authorize" +INTL_REFRESH_URL = f"{INTL_API_BASE}/iam/service/api/v1/refresh/" +INTL_USER_INFO_URL = f"{INTL_API_BASE}/uc/api/user/toc/base/info" +INTL_CA_KEY = "204587190" # X-Ca-Key header value for INTL API Gateway +INTL_APP_ID = "SMARTAPP-ISRAEL" # x-app-id for vehicle API +INTL_OPERATOR_CODE = "SMART-ISRAEL" # x-operator-code for vehicle API +# Note: INTL signing secret is stored in api/utils.py + OTA_SERVER_URL = "https://ota.srv.smart.com/" HTTPX_TIMEOUT = 30.0 +class SmartRegion(str, Enum): + """Region presets for Smart API endpoints. + + Use these region presets to easily configure the API for different geographic regions. + - EU: European region (default) - for users with Hello Smart EU app + - INTL: International region (Asia-Pacific) - for users with Hello Smart International app + (Australia, Singapore, and other international markets) + """ + + EU = "eu" + INTL = "intl" + + +def get_endpoint_urls_for_region(region: SmartRegion) -> "EndpointUrls": + """Get pre-configured EndpointUrls for a specific region. + + Args: + ---- + region: The region to get endpoint URLs for. + + Returns: + ------- + EndpointUrls configured for the specified region. + + Example: + ------- + >>> from pysmarthashtag.const import SmartRegion, get_endpoint_urls_for_region + >>> from pysmarthashtag.account import SmartAccount + >>> + >>> # For Australian/International users + >>> endpoint_urls = get_endpoint_urls_for_region(SmartRegion.INTL) + >>> account = SmartAccount("user@example.com", "password", endpoint_urls=endpoint_urls) + >>> + >>> # For European users (default) + >>> endpoint_urls = get_endpoint_urls_for_region(SmartRegion.EU) + >>> account = SmartAccount("user@example.com", "password", endpoint_urls=endpoint_urls) + + """ + if region == SmartRegion.EU: + # European region - uses default endpoints + return EndpointUrls() + elif region == SmartRegion.INTL: + # International region (Asia-Pacific) - for Hello Smart International app + # Used in Australia, Singapore, Israel, and other international markets + # Note: INTL uses different auth system (sg-app-api.smart.com) but same data endpoints as EU + return EndpointUrls( + server_url="https://sg-app-api.smart.com/iam/service/api/v1/login", + auth_url="https://sg-app-api.smart.com/iam/service/api/v1/oauth20/authorize", + login_url="https://sg-app-api.smart.com/iam/service/api/v1/login", + # INTL uses same data endpoints as EU + api_base_url="https://api.ecloudeu.com", + api_base_url_v2="https://apiv2.ecloudeu.com", + ) + else: + raise ValueError(f"Unknown region: {region}") + + @dataclass class EndpointUrls: """Configuration for API endpoint URLs. diff --git a/pysmarthashtag/tests/test_endpoint_urls.py b/pysmarthashtag/tests/test_endpoint_urls.py index c42a94a..b8650c2 100644 --- a/pysmarthashtag/tests/test_endpoint_urls.py +++ b/pysmarthashtag/tests/test_endpoint_urls.py @@ -170,3 +170,64 @@ async def test_account_with_explicit_default_urls(smart_fixture: respx.Router): await account.get_vehicles() assert account is not None assert len(account.vehicles) == 2 + + +class TestSmartRegion: + """Test SmartRegion enum and get_endpoint_urls_for_region function.""" + + def test_region_eu_returns_default_endpoints(self): + """Test that EU region returns default (European) endpoints.""" + urls = get_endpoint_urls_for_region(SmartRegion.EU) + assert urls.get_api_base_url() == API_BASE_URL + assert urls.get_api_base_url_v2() == "https://apiv2.ecloudeu.com" + assert urls.get_server_url() == SERVER_URL + assert urls.get_login_url() == LOGIN_URL + assert urls.get_auth_url() == AUTH_URL + assert urls.get_ota_server_url() == OTA_SERVER_URL + + def test_region_intl_returns_asia_pacific_endpoints(self): + """Test that INTL region returns EU endpoints (same cloud infrastructure).""" + urls = get_endpoint_urls_for_region(SmartRegion.INTL) + # International region uses EU cloud endpoints (shared infrastructure) + assert urls.get_api_base_url() == "https://api.ecloudeu.com" + assert urls.get_api_base_url_v2() == "https://apiv2.ecloudeu.com" + # INTL uses different auth URLs (sg-app-api.smart.com) + assert urls.get_login_url() == "https://sg-app-api.smart.com/iam/service/api/v1/login" + assert urls.get_auth_url() == "https://sg-app-api.smart.com/iam/service/api/v1/oauth20/authorize" + assert urls.get_server_url() == "https://sg-app-api.smart.com/iam/service/api/v1/login" + + def test_region_enum_values(self): + """Test that SmartRegion enum has expected values.""" + assert SmartRegion.EU.value == "eu" + assert SmartRegion.INTL.value == "intl" + + def test_account_with_eu_region(self): + """Test creating SmartAccount with EU region preset.""" + urls = get_endpoint_urls_for_region(SmartRegion.EU) + account = SmartAccount( + username="test@example.com", + password="testpass", + endpoint_urls=urls, + ) + assert account.endpoint_urls.get_api_base_url() == API_BASE_URL + + def test_account_with_intl_region(self): + """Test creating SmartAccount with INTL region preset.""" + urls = get_endpoint_urls_for_region(SmartRegion.INTL) + account = SmartAccount( + username="test@example.com", + password="testpass", + endpoint_urls=urls, + ) + assert account.endpoint_urls.get_api_base_url() == "https://api.ecloudeu.com" + assert account.endpoint_urls.get_api_base_url_v2() == "https://apiv2.ecloudeu.com" + + def test_authentication_with_intl_region(self): + """Test SmartAuthentication with INTL region preset.""" + urls = get_endpoint_urls_for_region(SmartRegion.INTL) + auth = SmartAuthentication( + username="test@example.com", + password="testpass", + endpoint_urls=urls, + ) + assert auth.endpoint_urls.get_api_base_url() == "https://api.ecloudeu.com" diff --git a/pysmarthashtag/vehicle/basic_status.py b/pysmarthashtag/vehicle/basic_status.py new file mode 100644 index 0000000..092ba16 --- /dev/null +++ b/pysmarthashtag/vehicle/basic_status.py @@ -0,0 +1,152 @@ +"""Basic vehicle status models for pysmarthashtag.""" + +import logging +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Optional + +from pysmarthashtag.models import ValueWithUnit, VehicleDataBase, get_field_as_type + +_LOGGER = logging.getLogger(__name__) + + +@dataclass +class BasicStatus(VehicleDataBase): + """Provides an accessible version of the vehicle's basic status data. + + This includes data from basicVehicleStatus, notification, eg (eGuard), + parkTime, and configuration sections of the API response. + """ + + # Basic vehicle status + speed: Optional[ValueWithUnit] = None + """Current vehicle speed.""" + + speed_validity: Optional[bool] = None + """Whether the speed value is valid.""" + + direction: Optional[str] = None + """Vehicle heading direction.""" + + engine_status: Optional[str] = None + """Engine status (e.g., 'engine_off', 'engine_on').""" + + car_mode: Optional[int] = None + """Car mode. 0=normal.""" + + usage_mode: Optional[int] = None + """Usage mode. 0=normal.""" + + car_locator_upload_enabled: Optional[bool] = None + """Whether car locator status upload is enabled.""" + + # Notification (emergency call) status + emergency_call_status: Optional[int] = None + """Emergency call notification status. 0=no active call.""" + + notification_reason: Optional[int] = None + """Reason code for notification.""" + + notification_time: Optional[datetime] = None + """Timestamp of the notification.""" + + notification_parameters: Optional[str] = None + """JSON string of notification parameters.""" + + # eGuard status + eguard_running: Optional[bool] = None + """Whether eGuard is running.""" + + eguard_blocked_status: Optional[int] = None + """eGuard blocked status. 0=not blocked.""" + + panic_status: Optional[bool] = None + """Whether panic mode is active.""" + + # Park time + park_time: Optional[datetime] = None + """Timestamp when the vehicle was last parked.""" + + # Configuration + propulsion_type: Optional[int] = None + """Propulsion type. 4=electric.""" + + fuel_type: Optional[int] = None + """Fuel type. 4=electric.""" + + @classmethod + def from_vehicle_data(cls, vehicle_data: dict): + """Create a new instance based on data from API.""" + parsed = cls._parse_vehicle_data(vehicle_data) or {} + if len(parsed) > 0: + return cls(**parsed) + return None + + @classmethod + def _parse_vehicle_data(cls, vehicle_data: dict) -> Optional[dict]: + """Parse the basic status data based on the vehicle data dict.""" + _LOGGER.debug("Parsing basic status data") + if "vehicleStatus" not in vehicle_data: + return None + retval: dict[str, Any] = {} + try: + vehicle_status = vehicle_data["vehicleStatus"] + + # Parse basicVehicleStatus + basic_status = vehicle_status.get("basicVehicleStatus", {}) + if basic_status: + speed = get_field_as_type(basic_status, "speed", float) + if speed is not None: + retval["speed"] = ValueWithUnit(speed, "km/h") + retval["speed_validity"] = get_field_as_type(basic_status, "speedValidity", bool) + retval["direction"] = get_field_as_type(basic_status, "direction", str) + retval["engine_status"] = get_field_as_type(basic_status, "engineStatus", str) + retval["car_mode"] = get_field_as_type(basic_status, "carMode", int) + retval["usage_mode"] = get_field_as_type(basic_status, "usageMode", int) + + # Position info within basicVehicleStatus + position = basic_status.get("position", {}) + if position: + car_locator = get_field_as_type(position, "carLocatorStatUploadEn", bool) + retval["car_locator_upload_enabled"] = car_locator + + # Parse notification + notification = vehicle_status.get("notification", {}) + if notification: + retval["emergency_call_status"] = get_field_as_type(notification, "notifForEmgyCallStatus", int) + retval["notification_reason"] = get_field_as_type(notification, "reason", int) + notif_time = get_field_as_type(notification, "time", int) + if notif_time is not None: + retval["notification_time"] = datetime.fromtimestamp(notif_time) + retval["notification_parameters"] = get_field_as_type(notification, "parameters", str) + + # Parse eGuard (eg) + eg = vehicle_status.get("eg", {}) + if eg: + retval["eguard_running"] = get_field_as_type(eg, "enableRunning", bool) + retval["panic_status"] = get_field_as_type(eg, "panicStatus", bool) + blocked = eg.get("blocked", {}) + if blocked: + retval["eguard_blocked_status"] = get_field_as_type(blocked, "status", int) + + # Parse parkTime + park_time = vehicle_status.get("parkTime", {}) + if park_time: + park_status = get_field_as_type(park_time, "status", int) + if park_status is not None: + retval["park_time"] = datetime.fromtimestamp(park_status / 1000) + + # Parse configuration + config = vehicle_status.get("configuration", {}) + if config: + retval["propulsion_type"] = get_field_as_type(config, "propulsionType", int) + retval["fuel_type"] = get_field_as_type(config, "fuelType", int) + + # Timestamp + if "updateTime" in vehicle_status: + retval["timestamp"] = datetime.fromtimestamp(int(vehicle_status["updateTime"]) / 1000) + + except KeyError as e: + _LOGGER.info(f"Basic status info not available: {e}") + finally: + return retval diff --git a/pysmarthashtag/vehicle/battery.py b/pysmarthashtag/vehicle/battery.py index 7ba7daf..2e42b9a 100644 --- a/pysmarthashtag/vehicle/battery.py +++ b/pysmarthashtag/vehicle/battery.py @@ -145,6 +145,9 @@ class Battery(VehicleDataBase): remaining_range_at_full_charge: Optional[ValueWithUnit] = ValueWithUnit(None, None) """Remaining range at full charge of the vehicle.""" + remaining_range_at_20_percent: Optional[ValueWithUnit] = ValueWithUnit(None, None) + """Remaining range when battery reaches 20% SoC.""" + remaining_battery_percent: Optional[ValueWithUnit] = ValueWithUnit(None, None) """Remaining battery percent of the vehicle.""" @@ -178,6 +181,48 @@ class Battery(VehicleDataBase): average_power_consumption: Optional[ValueWithUnit] = ValueWithUnit(None, "W") """Current average consumption""" + # V2L (Vehicle-to-Load) discharge data + v2l_discharge_voltage: Optional[ValueWithUnit] = ValueWithUnit(None, None) + """V2L discharge voltage (disChargeUAct).""" + + v2l_discharge_current: Optional[ValueWithUnit] = ValueWithUnit(None, None) + """V2L discharge current (disChargeIAct).""" + + v2l_discharge_power: Optional[ValueWithUnit] = ValueWithUnit(None, None) + """V2L discharge power (calculated from voltage * current).""" + + v2l_connection_status: Optional[int] = None + """V2L connection status (disChargeConnectStatus). 0=not connected.""" + + v2l_time_remaining: Optional[ValueWithUnit] = ValueWithUnit(None, None) + """Time to target discharge (timeToTargetDisCharged). 2047=not discharging.""" + + # Charge port status + charge_lid_ac_status: Optional[int] = None + """AC charge lid status (chargeLidAcStatus). 2=closed.""" + + charge_lid_dc_status: Optional[int] = None + """DC charge lid status (chargeLidDcAcStatus). 2=closed.""" + + # Scheduled charging + scheduled_charging_status: Optional[int] = None + """Scheduled/book charging status (bookChargeSts). 0=off.""" + + # DC-DC converter status + dcdc_activated: Optional[int] = None + """DC-DC converter activated status (dcDcActvd). 0=not active.""" + + dcdc_connection_status: Optional[int] = None + """DC-DC connection status (dcDcConnectStatus). 0=not connected.""" + + # Wireless charging + wireless_charging_alignment: Optional[int] = None + """Wireless power transfer fine alignment status (wptFineAlignt). 0=not aligned.""" + + # Instantaneous power consumption + instant_power_consumption: Optional[ValueWithUnit] = ValueWithUnit(None, None) + """Instantaneous power consumption (indPowerConsumption).""" + @classmethod def from_vehicle_data(cls, vehicle_data: dict): """Create a new instance based on data from API.""" @@ -263,6 +308,62 @@ def _parse_vehicle_data(cls, vehicle_data: dict) -> Optional[dict]: if avg_consumption is not None: retval["average_power_consumption"] = ValueWithUnit(avg_consumption, "W") + # Range at 20% SoC + range_20 = get_field_as_type(evStatus, "distanceToEmptyOnBattery20Soc", int, log_missing=False) + if range_20 is not None: + retval["remaining_range_at_20_percent"] = ValueWithUnit(range_20, "km") + + # V2L (Vehicle-to-Load) discharge data + discharge_u = get_field_as_type(evStatus, "disChargeUAct", float, log_missing=False) + discharge_i = get_field_as_type(evStatus, "disChargeIAct", float, log_missing=False) + if discharge_u is not None: + retval["v2l_discharge_voltage"] = ValueWithUnit(discharge_u, "V") + if discharge_i is not None: + retval["v2l_discharge_current"] = ValueWithUnit(discharge_i, "A") + if discharge_u is not None and discharge_i is not None and discharge_u > 0 and discharge_i > 0: + retval["v2l_discharge_power"] = ValueWithUnit(discharge_u * discharge_i, "W") + + discharge_conn = get_field_as_type(evStatus, "disChargeConnectStatus", int, log_missing=False) + if discharge_conn is not None: + retval["v2l_connection_status"] = discharge_conn + + time_to_discharged = get_field_as_type(evStatus, "timeToTargetDisCharged", int, log_missing=False) + if time_to_discharged is not None and time_to_discharged != 2047: + retval["v2l_time_remaining"] = ValueWithUnit(time_to_discharged, "min") + + # Charge lid status + charge_lid_ac = get_field_as_type(evStatus, "chargeLidAcStatus", int, log_missing=False) + if charge_lid_ac is not None: + retval["charge_lid_ac_status"] = charge_lid_ac + + charge_lid_dc = get_field_as_type(evStatus, "chargeLidDcAcStatus", int, log_missing=False) + if charge_lid_dc is not None: + retval["charge_lid_dc_status"] = charge_lid_dc + + # Scheduled charging + book_charge = get_field_as_type(evStatus, "bookChargeSts", int, log_missing=False) + if book_charge is not None: + retval["scheduled_charging_status"] = book_charge + + # DC-DC converter status + dcdc_active = get_field_as_type(evStatus, "dcDcActvd", int, log_missing=False) + if dcdc_active is not None: + retval["dcdc_activated"] = dcdc_active + + dcdc_conn = get_field_as_type(evStatus, "dcDcConnectStatus", int, log_missing=False) + if dcdc_conn is not None: + retval["dcdc_connection_status"] = dcdc_conn + + # Wireless charging alignment + wpt_align = get_field_as_type(evStatus, "wptFineAlignt", int, log_missing=False) + if wpt_align is not None: + retval["wireless_charging_alignment"] = wpt_align + + # Instantaneous power consumption + instant_consumption = get_field_as_type(evStatus, "indPowerConsumption", float, log_missing=False) + if instant_consumption is not None: + retval["instant_power_consumption"] = ValueWithUnit(instant_consumption, "W") + if "vehicleStatus" in vehicle_data and "updateTime" in vehicle_data["vehicleStatus"]: retval["timestamp"] = datetime.fromtimestamp(int(vehicle_data["vehicleStatus"]["updateTime"]) / 1000) diff --git a/pysmarthashtag/vehicle/climate.py b/pysmarthashtag/vehicle/climate.py index d76c9a9..d01ee47 100644 --- a/pysmarthashtag/vehicle/climate.py +++ b/pysmarthashtag/vehicle/climate.py @@ -134,6 +134,15 @@ class Climate(VehicleDataBase): interior_PM25: Optional[ValueWithUnit] = ValueWithUnit(None, None) """The interior PM2.5 value.""" + interior_PM25_level: Optional[int] = None + """The interior PM2.5 air quality level (0-5 scale).""" + + interior_secondary_PM25_level: Optional[int] = None + """The interior secondary PM2.5 level (interiorSecondPM25Level).""" + + exterior_PM25_level: Optional[int] = None + """The exterior PM2.5 air quality level (0-5 scale).""" + relative_humidity: Optional[ValueWithUnit] = ValueWithUnit(None, None) """The relative humidity.""" @@ -205,6 +214,11 @@ def _parse_vehicle_data(cls, vehicle_data: dict) -> Optional[dict]: retval["interior_PM25"] = ValueWithUnit(interior_pm25, "μg/m³") if interior_pm25 is not None else None rel_hum = get_field_as_type(pollutionStatus, "relHumSts", float) retval["relative_humidity"] = ValueWithUnit(rel_hum, "%") if rel_hum is not None else None + retval["interior_PM25_level"] = get_field_as_type(pollutionStatus, "interiorPM25Level", int) + retval["interior_secondary_PM25_level"] = get_field_as_type( + pollutionStatus, "interiorSecondPM25Level", int + ) + retval["exterior_PM25_level"] = get_field_as_type(pollutionStatus, "exteriorPM25Level", int) retval["timestamp"] = datetime.fromtimestamp(int(vehicle_data["vehicleStatus"]["updateTime"]) / 1000) except KeyError as e: diff --git a/pysmarthashtag/vehicle/safety.py b/pysmarthashtag/vehicle/safety.py index 5e13c9d..a7354ff 100644 --- a/pysmarthashtag/vehicle/safety.py +++ b/pysmarthashtag/vehicle/safety.py @@ -123,9 +123,11 @@ def _parse_vehicle_data(cls, vehicle_data: dict) -> Optional[dict]: retval["door_open_status_passenger"] = get_field_as_type(evStatus, "doorOpenStatusPassenger", int) retval["door_open_status_passenger_rear"] = get_field_as_type(evStatus, "doorOpenStatusPassengerRear", int) retval["door_pos_driver"] = get_field_as_type(evStatus, "doorPosDriver", int) - retval["door_pos_driver_rear"] = get_field_as_type(evStatus, "doorPosDriverRear", int) + retval["door_pos_driver_rear"] = get_field_as_type(evStatus, "doorPosDriverRear", int, log_missing=False) retval["door_pos_passenger"] = get_field_as_type(evStatus, "doorPosPassenger", int) - retval["door_pos_passenger_rear"] = get_field_as_type(evStatus, "doorPosPassengerRear", int) + retval["door_pos_passenger_rear"] = get_field_as_type( + evStatus, "doorPosPassengerRear", int, log_missing=False + ) retval["electric_park_brake_status"] = get_field_as_type(evStatus, "electricParkBrakeStatus", int) retval["engine_hood_open_status"] = get_field_as_type(evStatus, "engineHoodOpenStatus", int) retval["seat_belt_status_driver"] = get_field_as_type(evStatus, "seatBeltStatusDriver", bool) diff --git a/pysmarthashtag/vehicle/trailer.py b/pysmarthashtag/vehicle/trailer.py new file mode 100644 index 0000000..1a46ae1 --- /dev/null +++ b/pysmarthashtag/vehicle/trailer.py @@ -0,0 +1,69 @@ +"""Trailer status models for pysmarthashtag.""" + +import logging +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Optional + +from pysmarthashtag.models import VehicleDataBase, get_field_as_type + +_LOGGER = logging.getLogger(__name__) + + +@dataclass +class Trailer(VehicleDataBase): + """Provides an accessible version of the vehicle's trailer status data. + + This includes all trailer lamp status information from the trailerStatus + section of the API response. + """ + + turning_lamp_status: Optional[int] = None + """Trailer turning lamp status. 0=off.""" + + fog_lamp_status: Optional[int] = None + """Trailer fog lamp status. 0=off.""" + + brake_lamp_status: Optional[int] = None + """Trailer brake lamp status. 0=off.""" + + reversing_lamp_status: Optional[int] = None + """Trailer reversing lamp status. 0=off.""" + + position_lamp_status: Optional[int] = None + """Trailer position lamp status. 0=off.""" + + @classmethod + def from_vehicle_data(cls, vehicle_data: dict): + """Create a new instance based on data from API.""" + parsed = cls._parse_vehicle_data(vehicle_data) or {} + if len(parsed) > 0: + return cls(**parsed) + return None + + @classmethod + def _parse_vehicle_data(cls, vehicle_data: dict) -> Optional[dict]: + """Parse the trailer status data based on the vehicle data dict.""" + _LOGGER.debug("Parsing trailer status data") + if "vehicleStatus" not in vehicle_data: + return None + retval: dict[str, Any] = {} + try: + trailer_status = vehicle_data["vehicleStatus"]["additionalVehicleStatus"].get("trailerStatus", {}) + if not trailer_status: + return None + + retval["turning_lamp_status"] = get_field_as_type(trailer_status, "trailerTurningLampSts", int) + retval["fog_lamp_status"] = get_field_as_type(trailer_status, "trailerFogLampSts", int) + retval["brake_lamp_status"] = get_field_as_type(trailer_status, "trailerBreakLampSts", int) + retval["reversing_lamp_status"] = get_field_as_type(trailer_status, "trailerReversingLampSts", int) + retval["position_lamp_status"] = get_field_as_type(trailer_status, "trailerPosLampSts", int) + + # Timestamp + if "updateTime" in vehicle_data["vehicleStatus"]: + retval["timestamp"] = datetime.fromtimestamp(int(vehicle_data["vehicleStatus"]["updateTime"]) / 1000) + + except KeyError as e: + _LOGGER.info(f"Trailer status info not available: {e}") + finally: + return retval diff --git a/pysmarthashtag/vehicle/vehicle.py b/pysmarthashtag/vehicle/vehicle.py index f632a01..27ffe79 100644 --- a/pysmarthashtag/vehicle/vehicle.py +++ b/pysmarthashtag/vehicle/vehicle.py @@ -6,6 +6,7 @@ from pysmarthashtag.const import API_BASE_URL, API_BASE_URL_V2 from pysmarthashtag.models import ValueWithUnit, get_element_from_dict_maybe +from pysmarthashtag.vehicle.basic_status import BasicStatus from pysmarthashtag.vehicle.battery import Battery from pysmarthashtag.vehicle.climate import Climate from pysmarthashtag.vehicle.maintenance import Maintenance @@ -13,6 +14,7 @@ from pysmarthashtag.vehicle.running import Running from pysmarthashtag.vehicle.safety import Safety from pysmarthashtag.vehicle.tires import Tires +from pysmarthashtag.vehicle.trailer import Trailer _LOGGER = logging.getLogger(__name__) @@ -56,6 +58,12 @@ class SmartVehicle: safety: Optional[Safety] = None """The safety status of the vehicle.""" + basic_status: Optional[BasicStatus] = None + """The basic status of the vehicle (speed, engine, eGuard, park time, etc.).""" + + trailer: Optional[Trailer] = None + """The trailer status of the vehicle.""" + climate_control: Optional["ClimateControll"] = None # noqa: F821 charging_control: Optional["ChargingControl"] = None # noqa: F821 @@ -121,6 +129,8 @@ def combine_data( self.running = Running.from_vehicle_data(self.data) self.climate = Climate.from_vehicle_data(self.data) self.safety = Safety.from_vehicle_data(self.data) + self.basic_status = BasicStatus.from_vehicle_data(self.data) + self.trailer = Trailer.from_vehicle_data(self.data) from pysmarthashtag.control.climate import ClimateControll From 25d1a3f10a27c4677cae49a36b77e4e802be97fd Mon Sep 17 00:00:00 2001 From: Bastian Neumann Date: Sun, 8 Feb 2026 19:14:44 +0100 Subject: [PATCH 02/12] fix: Import missing constants in tests for EndpointUrls configuration --- pysmarthashtag/const.py | 1 + pysmarthashtag/tests/test_endpoint_urls.py | 2 ++ 2 files changed, 3 insertions(+) diff --git a/pysmarthashtag/const.py b/pysmarthashtag/const.py index a7da9e3..bb05086 100644 --- a/pysmarthashtag/const.py +++ b/pysmarthashtag/const.py @@ -1,6 +1,7 @@ """URLs for different services and error code mapping.""" from dataclasses import dataclass +from enum import Enum from typing import Optional API_KEY = "3_L94eyQ-wvJhWm7Afp1oBhfTGXZArUfSHHW9p9Pncg513hZELXsxCfMWHrF8f5P5a" diff --git a/pysmarthashtag/tests/test_endpoint_urls.py b/pysmarthashtag/tests/test_endpoint_urls.py index b8650c2..5084471 100644 --- a/pysmarthashtag/tests/test_endpoint_urls.py +++ b/pysmarthashtag/tests/test_endpoint_urls.py @@ -13,6 +13,8 @@ OTA_SERVER_URL, SERVER_URL, EndpointUrls, + SmartRegion, + get_endpoint_urls_for_region, ) from pysmarthashtag.tests import TEST_PASSWORD, TEST_USERNAME from pysmarthashtag.tests.conftest import prepare_account_with_vehicles From 884a0a005e481532e8f671e7e89c1e17104e5d6f Mon Sep 17 00:00:00 2001 From: Bastian Neumann Date: Sun, 8 Feb 2026 19:21:13 +0100 Subject: [PATCH 03/12] Potential fix for code scanning alert no. 5: Clear-text logging of sensitive information Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com> --- pysmarthashtag/api/authentication.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pysmarthashtag/api/authentication.py b/pysmarthashtag/api/authentication.py index d07aea5..12114b4 100644 --- a/pysmarthashtag/api/authentication.py +++ b/pysmarthashtag/api/authentication.py @@ -591,7 +591,7 @@ async def _login(self): try: oauth_result = r_oauth.json() - _LOGGER.debug("INTL OAuth response code: %s", oauth_result.get("code")) + _LOGGER.debug("INTL OAuth response received.") if oauth_result.get("code") != "200": error_msg = oauth_result.get("message", "Unknown error") From a6136c56b1a45c02f66d45ac41e4d73c283566c3 Mon Sep 17 00:00:00 2001 From: Bastian Neumann Date: Sun, 8 Feb 2026 19:23:50 +0100 Subject: [PATCH 04/12] Potential fix for code scanning alert no. 6: Clear-text logging of sensitive information Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com> --- pysmarthashtag/api/authentication.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pysmarthashtag/api/authentication.py b/pysmarthashtag/api/authentication.py index 12114b4..dee9379 100644 --- a/pysmarthashtag/api/authentication.py +++ b/pysmarthashtag/api/authentication.py @@ -595,7 +595,8 @@ async def _login(self): if oauth_result.get("code") != "200": error_msg = oauth_result.get("message", "Unknown error") - _LOGGER.error("INTL OAuth failed: %s", error_msg) + sanitized_error_msg = sanitize_log_data(error_msg) + _LOGGER.error("INTL OAuth failed: %s", sanitized_error_msg) raise SmartAPIError(f"INTL OAuth failed: {error_msg}") # Result is directly the authCode string (e.g., "CODE-xxx") From ae14017fbe77b8f50d050f48b8b8fe85c8fd3dcd Mon Sep 17 00:00:00 2001 From: Bastian Neumann Date: Sun, 8 Feb 2026 19:25:22 +0100 Subject: [PATCH 05/12] Potential fix for code scanning alert no. 7: Clear-text logging of sensitive information Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com> --- pysmarthashtag/api/authentication.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pysmarthashtag/api/authentication.py b/pysmarthashtag/api/authentication.py index dee9379..e9f249f 100644 --- a/pysmarthashtag/api/authentication.py +++ b/pysmarthashtag/api/authentication.py @@ -604,7 +604,7 @@ async def _login(self): if isinstance(auth_code, dict): auth_code = auth_code.get("authCode", auth_code) - _LOGGER.debug("INTL: Got authCode: %s...", str(auth_code)[:30] if auth_code else "None") + _LOGGER.debug("INTL: Got authCode.") except (KeyError, ValueError, TypeError) as e: raise SmartAPIError(f"Could not parse INTL OAuth response: {e}") From bbbf07bbde27dd1e32f227d0fb083d9b1a0bc640 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 8 Feb 2026 18:27:52 +0000 Subject: [PATCH 06/12] Initial plan From 1fd8c5e14a163102dc4d53479026210be089d786 Mon Sep 17 00:00:00 2001 From: Bastian Neumann Date: Sun, 8 Feb 2026 19:28:28 +0100 Subject: [PATCH 07/12] Potential fix for code scanning alert no. 10: Clear-text logging of sensitive information Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com> --- pysmarthashtag/api/log_sanitizer.py | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/pysmarthashtag/api/log_sanitizer.py b/pysmarthashtag/api/log_sanitizer.py index 7fad5da..0a9a8d0 100644 --- a/pysmarthashtag/api/log_sanitizer.py +++ b/pysmarthashtag/api/log_sanitizer.py @@ -127,7 +127,7 @@ def _sanitize_list(data: list, depth: int = 0, max_depth: int = 10) -> list: def _sanitize_string(data: str) -> str: - """Sanitize a string by masking VINs and tokens. + """Sanitize a string by masking VINs, tokens, and potentially sensitive content. Args: ---- @@ -142,6 +142,17 @@ def _sanitize_string(data: str) -> str: result = VIN_PATTERN.sub(lambda m: _mask_value(m.group()), data) # Mask Bearer tokens result = TOKEN_PATTERN.sub(r"\1***", result) + + # As a defensive fallback, avoid logging very long or opaque strings in full. + # This helps when upstream services accidentally include secrets in generic + # "message" fields that do not match our specific patterns. + max_visible_length = 80 + if len(result) > max_visible_length: + # Preserve only a short prefix/suffix to keep logs useful while hiding content. + prefix = result[:40] + suffix = result[-10:] + return f"{prefix}***{suffix}" + return result @@ -166,7 +177,11 @@ def sanitize_log_data(data: Any) -> Any: return _sanitize_list(data) elif isinstance(data, str): return _sanitize_string(data) - return data + + # For any other type (including numbers, custom objects, etc.), avoid + # returning the raw value to prevent accidental leakage of sensitive data. + # Represent the value generically instead. + return f"" def get_data_summary(data: dict, include_keys: Union[list, None] = None) -> str: From 5af89f80b513aa3c2ac05953492fd6e0cecfd58b Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 8 Feb 2026 18:29:52 +0000 Subject: [PATCH 08/12] fix: Remove return from finally blocks to allow exception propagation Co-authored-by: DasBasti <1713093+DasBasti@users.noreply.github.com> --- pysmarthashtag/vehicle/basic_status.py | 3 +-- pysmarthashtag/vehicle/trailer.py | 3 +-- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/pysmarthashtag/vehicle/basic_status.py b/pysmarthashtag/vehicle/basic_status.py index 092ba16..f5af2c5 100644 --- a/pysmarthashtag/vehicle/basic_status.py +++ b/pysmarthashtag/vehicle/basic_status.py @@ -148,5 +148,4 @@ def _parse_vehicle_data(cls, vehicle_data: dict) -> Optional[dict]: except KeyError as e: _LOGGER.info(f"Basic status info not available: {e}") - finally: - return retval + return retval diff --git a/pysmarthashtag/vehicle/trailer.py b/pysmarthashtag/vehicle/trailer.py index 1a46ae1..0944083 100644 --- a/pysmarthashtag/vehicle/trailer.py +++ b/pysmarthashtag/vehicle/trailer.py @@ -65,5 +65,4 @@ def _parse_vehicle_data(cls, vehicle_data: dict) -> Optional[dict]: except KeyError as e: _LOGGER.info(f"Trailer status info not available: {e}") - finally: - return retval + return retval From 5bbcfc916fe8396097cb9dbe2bf701bdd2811df3 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 8 Feb 2026 18:39:18 +0000 Subject: [PATCH 09/12] test: Add exception handling tests for Trailer and BasicStatus Co-authored-by: DasBasti <1713093+DasBasti@users.noreply.github.com> --- pysmarthashtag/tests/test_missing_fields.py | 135 ++++++++++++++++++++ 1 file changed, 135 insertions(+) diff --git a/pysmarthashtag/tests/test_missing_fields.py b/pysmarthashtag/tests/test_missing_fields.py index 0f61372..cf6bb77 100644 --- a/pysmarthashtag/tests/test_missing_fields.py +++ b/pysmarthashtag/tests/test_missing_fields.py @@ -2,7 +2,10 @@ import logging +import pytest + from pysmarthashtag.models import get_field_as_type +from pysmarthashtag.vehicle.basic_status import BasicStatus from pysmarthashtag.vehicle.battery import Battery from pysmarthashtag.vehicle.climate import Climate from pysmarthashtag.vehicle.maintenance import Maintenance @@ -10,6 +13,7 @@ from pysmarthashtag.vehicle.running import Running from pysmarthashtag.vehicle.safety import Safety from pysmarthashtag.vehicle.tires import Tires +from pysmarthashtag.vehicle.trailer import Trailer class TestGetFieldAsType: @@ -270,3 +274,134 @@ def test_dc_charging_negative_value(self, caplog): assert battery.charging_voltage is not None # Should use the first element in DcChargingVoltLevels assert battery.charging_voltage.value == DcChargingVoltLevels[0] + + +class TestTrailerExceptionHandling: + """Test that Trailer exception handling works correctly after fixing finally block.""" + + def test_trailer_keyerror_is_caught_and_logged(self, caplog): + """Test that KeyError exceptions are caught and logged, not propagated.""" + # Missing 'additionalVehicleStatus' will cause KeyError on line 52 + vehicle_data = { + "vehicleStatus": { + # Missing 'additionalVehicleStatus' + "updateTime": "1706028240000", + } + } + with caplog.at_level(logging.INFO): + # Should not raise KeyError, but should log it + result = Trailer.from_vehicle_data(vehicle_data) + + # KeyError should be caught and logged + assert "Trailer status info not available" in caplog.text + # Should return None since no data was parsed + assert result is None + + def test_trailer_non_keyerror_propagates(self): + """Test that non-KeyError exceptions (e.g., ValueError) propagate correctly.""" + # Invalid updateTime that will cause ValueError in int() conversion + vehicle_data = { + "vehicleStatus": { + "additionalVehicleStatus": { + "trailerStatus": { + "trailerTurningLampSts": "0", + } + }, + "updateTime": "not_a_number", # Will cause ValueError in int() + } + } + + # ValueError should propagate, not be swallowed + with pytest.raises(ValueError): + Trailer.from_vehicle_data(vehicle_data) + + def test_trailer_with_valid_data_returns_object(self): + """Test that valid trailer data parses correctly.""" + vehicle_data = { + "vehicleStatus": { + "additionalVehicleStatus": { + "trailerStatus": { + "trailerTurningLampSts": "0", + "trailerFogLampSts": "1", + "trailerBreakLampSts": "0", + "trailerReversingLampSts": "0", + "trailerPosLampSts": "1", + } + }, + "updateTime": "1706028240000", + } + } + + trailer = Trailer.from_vehicle_data(vehicle_data) + assert trailer is not None + assert trailer.turning_lamp_status == 0 + assert trailer.fog_lamp_status == 1 + + +class TestBasicStatusExceptionHandling: + """Test that BasicStatus exception handling works correctly after fixing finally block.""" + + def test_basic_status_keyerror_is_caught_and_logged(self, caplog): + """Test that KeyError exceptions are caught and logged, not propagated.""" + # Create a scenario where KeyError might occur + # Note: BasicStatus uses .get() so KeyError is less likely, but we can still test the handler + vehicle_data = { + "vehicleStatus": {} # Minimal data that might cause issues + } + with caplog.at_level(logging.INFO): + # Should not raise KeyError + result = BasicStatus.from_vehicle_data(vehicle_data) + + # Should return None since no meaningful data was parsed + assert result is None + + def test_basic_status_non_keyerror_propagates(self): + """Test that non-KeyError exceptions (e.g., ValueError) propagate correctly.""" + # Invalid updateTime that will cause ValueError in int() conversion + vehicle_data = { + "vehicleStatus": { + "basicVehicleStatus": { + "speed": "50.0", + }, + "updateTime": "invalid_timestamp", # Will cause ValueError in int() + } + } + + # ValueError should propagate, not be swallowed + with pytest.raises(ValueError): + BasicStatus.from_vehicle_data(vehicle_data) + + def test_basic_status_with_invalid_notification_time_propagates(self): + """Test that ValueError from datetime.fromtimestamp propagates.""" + # Very large timestamp value that will cause OSError/ValueError + vehicle_data = { + "vehicleStatus": { + "notification": { + "time": 999999999999999999999, # Absurdly large timestamp + }, + "updateTime": "1706028240000", + } + } + + # Should propagate the exception from datetime.fromtimestamp + with pytest.raises((ValueError, OSError, OverflowError)): + BasicStatus.from_vehicle_data(vehicle_data) + + def test_basic_status_with_valid_data_returns_object(self): + """Test that valid basic status data parses correctly.""" + vehicle_data = { + "vehicleStatus": { + "basicVehicleStatus": { + "speed": "50.0", + "direction": "N", + "engineStatus": "engine_off", + }, + "updateTime": "1706028240000", + } + } + + basic_status = BasicStatus.from_vehicle_data(vehicle_data) + assert basic_status is not None + assert basic_status.speed is not None + assert basic_status.speed.value == 50.0 + assert basic_status.direction == "N" From 379e42fced994da9a859771e468b2cab37c485af Mon Sep 17 00:00:00 2001 From: Bastian Neumann Date: Sun, 8 Feb 2026 19:58:12 +0100 Subject: [PATCH 10/12] update ignore file --- .gitignore | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.gitignore b/.gitignore index 592f206..b728a17 100644 --- a/.gitignore +++ b/.gitignore @@ -74,3 +74,6 @@ venv.bak/ # Generated response files response-*.json + +.copilot* +external From ddfaf5ac5f97b22c1ae77500617202d99d88eee1 Mon Sep 17 00:00:00 2001 From: Bastian Neumann Date: Sun, 8 Feb 2026 20:02:56 +0100 Subject: [PATCH 11/12] fix: Correct minor issues in authentication and utils, update vehicle API header generation --- pyproject.toml | 2 +- pysmarthashtag/api/authentication.py | 8 ++++---- pysmarthashtag/api/utils.py | 20 ++++++++++++++++---- pysmarthashtag/const.py | 10 ++++------ pysmarthashtag/vehicle/basic_status.py | 2 +- 5 files changed, 26 insertions(+), 16 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index d44fb56..fcdda6e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -77,7 +77,7 @@ exclude = [ [tool.ruff.per-file-ignores] "docs/source/conf.py" = ["D100"] -"pysamrthashtag/api/authentication.py" = ["D102", "D107"] +"pysmarthashtag/api/authentication.py" = ["D102", "D107"] "pysmarthashtag/vehicle/battery.py" = ["C901"] # Complex data parsing function [tool.ruff.mccabe] diff --git a/pysmarthashtag/api/authentication.py b/pysmarthashtag/api/authentication.py index e9f249f..e915a8a 100644 --- a/pysmarthashtag/api/authentication.py +++ b/pysmarthashtag/api/authentication.py @@ -479,7 +479,7 @@ def _generate_intl_headers(self, auth_token: str = "") -> dict[str, str]: return headers - def _generate_vehicle_api_headers(self, body: str = None) -> dict[str, str]: + def _generate_vehicle_api_headers(self, body: Optional[str] = None) -> dict[str, str]: """Generate headers for vehicle API (ecloudeu) requests. These headers are used for the session/secure endpoint on apiv2.ecloudeu.com. @@ -574,7 +574,7 @@ async def _login(self): _LOGGER.info("INTL: Login successful for user %s", self.intl_user_id) except (KeyError, ValueError, TypeError) as e: - raise SmartAPIError(f"Could not parse INTL login response: {e}") + raise SmartAPIError(f"Could not parse INTL login response: {e}") from e # Step 2: Exchange accessToken for authCode via OAuth # Note: OAuth uses just accessToken in URL param, and idToken in Xs-Auth-Token header @@ -607,7 +607,7 @@ async def _login(self): _LOGGER.debug("INTL: Got authCode.") except (KeyError, ValueError, TypeError) as e: - raise SmartAPIError(f"Could not parse INTL OAuth response: {e}") + raise SmartAPIError(f"Could not parse INTL OAuth response: {e}") from e # Step 3: Exchange authCode for vehicle API session _LOGGER.debug("INTL: Step 3 - Exchanging authCode for vehicle API session") @@ -644,7 +644,7 @@ async def _login(self): _LOGGER.info("INTL: Successfully authenticated, user ID: %s", api_user_id) except (KeyError, ValueError, TypeError) as e: - raise SmartAPIError(f"Could not parse INTL session response: {e}") + raise SmartAPIError(f"Could not parse INTL session response: {e}") from e return { "access_token": api_access_token, diff --git a/pysmarthashtag/api/utils.py b/pysmarthashtag/api/utils.py index 7aa0593..cc65064 100644 --- a/pysmarthashtag/api/utils.py +++ b/pysmarthashtag/api/utils.py @@ -4,6 +4,7 @@ import logging import secrets import time +from typing import Optional from urllib.parse import quote _LOGGER = logging.getLogger(__name__) @@ -26,7 +27,7 @@ def _create_sign( timestamp: str, method: str, url: str, - body=None, + body: Optional[str] = None, use_intl: bool = False, url_encode_params: bool = False, ) -> str: @@ -81,7 +82,7 @@ def _create_sign( def generate_default_header( - device_id: str, access_token: str, params: dict, method: str, url: str, body=None + device_id: str, access_token: str, params: dict, method: str, url: str, body: Optional[str] = None ) -> dict[str, str]: """Generate a header for HTTP requests to the server.""" timestamp = create_correct_timestamp() @@ -116,7 +117,13 @@ def generate_default_header( def generate_intl_header( - device_id: str, access_token: str, params: dict, method: str, url: str, body=None, client_id: str = None + device_id: str, + access_token: str, + params: dict, + method: str, + url: str, + body: Optional[str] = None, + client_id: Optional[str] = None, ) -> dict[str, str]: """Generate a header for HTTP requests to the INTL (International) API. @@ -147,6 +154,11 @@ def generate_intl_header( url_encode_params = method.upper() == "GET" sign = _create_sign(nonce, params, timestamp, method, url, body, use_intl=True, url_encode_params=url_encode_params) + # Get vehicle series from params if available (passed as _vehicle_series) + vehicle_series = params.pop("_vehicle_series", None) if params else None + if not vehicle_series: + raise ValueError("vehicle_series is required for INTL API requests. Pass it in params as '_vehicle_series'.") + header = { "x-app-id": "SMARTAPP-ISRAEL", "accept": "application/json;responseformat=3", @@ -167,7 +179,7 @@ def generate_intl_header( "x-signature": sign, "x-timestamp": timestamp, "platform": "NON-CMA", - "x-vehicle-series": "HC1H2D3B6213-01_IL", + "x-vehicle-series": vehicle_series, } if access_token: diff --git a/pysmarthashtag/const.py b/pysmarthashtag/const.py index bb05086..4385efa 100644 --- a/pysmarthashtag/const.py +++ b/pysmarthashtag/const.py @@ -78,12 +78,10 @@ def get_endpoint_urls_for_region(region: SmartRegion) -> "EndpointUrls": # Used in Australia, Singapore, Israel, and other international markets # Note: INTL uses different auth system (sg-app-api.smart.com) but same data endpoints as EU return EndpointUrls( - server_url="https://sg-app-api.smart.com/iam/service/api/v1/login", - auth_url="https://sg-app-api.smart.com/iam/service/api/v1/oauth20/authorize", - login_url="https://sg-app-api.smart.com/iam/service/api/v1/login", - # INTL uses same data endpoints as EU - api_base_url="https://api.ecloudeu.com", - api_base_url_v2="https://apiv2.ecloudeu.com", + server_url=INTL_LOGIN_URL, + auth_url=INTL_OAUTH_URL, + login_url=INTL_LOGIN_URL, + # INTL uses same data endpoints as EU (leave as None to use defaults) ) else: raise ValueError(f"Unknown region: {region}") diff --git a/pysmarthashtag/vehicle/basic_status.py b/pysmarthashtag/vehicle/basic_status.py index f5af2c5..2dc37f2 100644 --- a/pysmarthashtag/vehicle/basic_status.py +++ b/pysmarthashtag/vehicle/basic_status.py @@ -117,7 +117,7 @@ def _parse_vehicle_data(cls, vehicle_data: dict) -> Optional[dict]: retval["notification_reason"] = get_field_as_type(notification, "reason", int) notif_time = get_field_as_type(notification, "time", int) if notif_time is not None: - retval["notification_time"] = datetime.fromtimestamp(notif_time) + retval["notification_time"] = datetime.fromtimestamp(notif_time / 1000.0) retval["notification_parameters"] = get_field_as_type(notification, "parameters", str) # Parse eGuard (eg) From 9ce1f331fd5c05eb201ae806d5926fa8983b5db9 Mon Sep 17 00:00:00 2001 From: Bastian Neumann Date: Sun, 8 Feb 2026 20:56:55 +0100 Subject: [PATCH 12/12] Add token validation, improve log sanitization, and fix test expectations - authentication.py: Add validation for required INTL tokens after extraction - authentication.py: Sanitize log outputs for session errors and user IDs - authentication.py: Fix return dict to use OAuth tokens for access_token/refresh_token - log_sanitizer.py: Allowlist safe primitives (int, float, bool, None) - log_sanitizer.py: Add secret detection heuristic before truncating long strings - test_missing_fields.py: Rename test to match actual behavior --- pysmarthashtag/api/authentication.py | 25 ++++++++++-- pysmarthashtag/api/log_sanitizer.py | 44 +++++++++++++++++++-- pysmarthashtag/tests/test_missing_fields.py | 12 +++--- 3 files changed, 68 insertions(+), 13 deletions(-) diff --git a/pysmarthashtag/api/authentication.py b/pysmarthashtag/api/authentication.py index e915a8a..88472d1 100644 --- a/pysmarthashtag/api/authentication.py +++ b/pysmarthashtag/api/authentication.py @@ -567,6 +567,21 @@ async def _login(self): self.intl_refresh_token = result.get("refreshToken") self.intl_id_token = result.get("idToken") self.intl_user_id = result.get("userId") + + # Validate that required tokens are present + missing_tokens = [] + if not self.intl_access_token: + missing_tokens.append("accessToken") + if not self.intl_id_token: + missing_tokens.append("idToken") + if not self.intl_user_id: + missing_tokens.append("userId") + + if missing_tokens: + error_msg = f"INTL login response missing required tokens: {', '.join(missing_tokens)}" + _LOGGER.error(error_msg) + raise SmartAPIError(error_msg) + expires_in = result.get("expiresIn", 86400) expires_at = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=expires_in) @@ -629,7 +644,8 @@ async def _login(self): if session_result.get("code") != 1000: error_msg = session_result.get("message", "Unknown error") - _LOGGER.error("INTL session failed: %s (code: %s)", error_msg, session_result.get("code")) + sanitized_code = sanitize_log_data(session_result.get("code")) + _LOGGER.error("INTL session failed: %s (code: %s)", sanitize_log_data(error_msg), sanitized_code) raise SmartAPIError(f"INTL session failed: {error_msg}") data = session_result.get("data", {}) @@ -641,14 +657,15 @@ async def _login(self): # Store client_id on the auth object for use in API requests self.api_client_id = api_client_id - _LOGGER.info("INTL: Successfully authenticated, user ID: %s", api_user_id) + _LOGGER.info("INTL: Successfully authenticated, user ID: %s", sanitize_log_data(api_user_id)) except (KeyError, ValueError, TypeError) as e: raise SmartAPIError(f"Could not parse INTL session response: {e}") from e + # Return OAuth tokens for standard access, and session API tokens for vehicle API return { - "access_token": api_access_token, - "refresh_token": api_refresh_token, + "access_token": self.intl_access_token, + "refresh_token": self.intl_refresh_token, "api_access_token": api_access_token, "api_refresh_token": api_refresh_token, "api_user_id": api_user_id, diff --git a/pysmarthashtag/api/log_sanitizer.py b/pysmarthashtag/api/log_sanitizer.py index 0a9a8d0..ae2cdf9 100644 --- a/pysmarthashtag/api/log_sanitizer.py +++ b/pysmarthashtag/api/log_sanitizer.py @@ -126,6 +126,39 @@ def _sanitize_list(data: list, depth: int = 0, max_depth: int = 10) -> list: return result +def _looks_like_secret(s: str) -> bool: + """Check if a string looks like a secret (high entropy, no whitespace, base64/hex-like). + + Args: + ---- + s: String to check + + Returns: + ------- + True if the string appears to be a secret/token + + """ + # If it has whitespace, it's likely a human-readable message, not a secret + if any(c.isspace() for c in s): + return False + + # Long strings with only alphanumeric characters and common encoding chars + # (likely base64, hex, or URL-safe tokens) + alphanumeric_chars = sum(1 for c in s if c.isalnum()) + special_chars = sum(1 for c in s if c in "-_=+/") + total_chars = len(s) + + # If mostly alphanumeric with some special chars, and no whitespace, looks like a secret + if total_chars > 20 and (alphanumeric_chars + special_chars) / total_chars > 0.95: + return True + + # Check for common secret patterns: long hex strings, base64-like strings + if len(s) >= 32 and all(c in "0123456789abcdefABCDEF" for c in s): + return True # Hex string + + return False + + def _sanitize_string(data: str) -> str: """Sanitize a string by masking VINs, tokens, and potentially sensitive content. @@ -146,8 +179,9 @@ def _sanitize_string(data: str) -> str: # As a defensive fallback, avoid logging very long or opaque strings in full. # This helps when upstream services accidentally include secrets in generic # "message" fields that do not match our specific patterns. - max_visible_length = 80 - if len(result) > max_visible_length: + # Only truncate if it looks like a secret and is long enough + max_visible_length = 512 + if len(result) > max_visible_length and _looks_like_secret(result): # Preserve only a short prefix/suffix to keep logs useful while hiding content. prefix = result[:40] suffix = result[-10:] @@ -180,7 +214,11 @@ def sanitize_log_data(data: Any) -> Any: # For any other type (including numbers, custom objects, etc.), avoid # returning the raw value to prevent accidental leakage of sensitive data. - # Represent the value generically instead. + # Allowlist safe primitives while keeping generic sanitization for other types. + if isinstance(data, (int, float, bool)): + return str(data) + if data is None: + return "None" return f"" diff --git a/pysmarthashtag/tests/test_missing_fields.py b/pysmarthashtag/tests/test_missing_fields.py index cf6bb77..0c62ae4 100644 --- a/pysmarthashtag/tests/test_missing_fields.py +++ b/pysmarthashtag/tests/test_missing_fields.py @@ -341,15 +341,15 @@ def test_trailer_with_valid_data_returns_object(self): class TestBasicStatusExceptionHandling: """Test that BasicStatus exception handling works correctly after fixing finally block.""" - def test_basic_status_keyerror_is_caught_and_logged(self, caplog): - """Test that KeyError exceptions are caught and logged, not propagated.""" - # Create a scenario where KeyError might occur - # Note: BasicStatus uses .get() so KeyError is less likely, but we can still test the handler + def test_basic_status_from_vehicle_data_returns_none_when_empty(self, caplog): + """Test that BasicStatus.from_vehicle_data returns None when no data can be parsed.""" + # With empty vehicleStatus, BasicStatus uses .get() so it won't raise KeyError + # Instead, it should return None since no meaningful data was parsed vehicle_data = { - "vehicleStatus": {} # Minimal data that might cause issues + "vehicleStatus": {} # Minimal data that won't trigger parsing } with caplog.at_level(logging.INFO): - # Should not raise KeyError + # Should not raise KeyError, should return None result = BasicStatus.from_vehicle_data(vehicle_data) # Should return None since no meaningful data was parsed