diff --git a/packages/modules/vehicles/nissanconnect/__init__.py b/packages/modules/vehicles/nissanconnect/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/packages/modules/vehicles/nissanconnect/api.py b/packages/modules/vehicles/nissanconnect/api.py new file mode 100644 index 0000000000..04ea89e78e --- /dev/null +++ b/packages/modules/vehicles/nissanconnect/api.py @@ -0,0 +1,308 @@ +"""Passive MyNISSAN EU client; tested with Leaf ZE1. No vehicle commands. + +Adapted by wippofax from dan-r/HomeAssistant-NissanConnect, revision +b0bafe9108d4574e243aa182324584dbc68369ad (MIT), kamereon/kamereon.py. +Upstream credits @mitchellrj and @Tobiaswk; Apache-2.0 portions relicensed with permission. +https://github.com/dan-r/HomeAssistant-NissanConnect +""" +# MIT License -- Copyright (c) 2024 Daniel Raper +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +import base64 +from contextvars import ContextVar +from datetime import datetime +import hashlib +from html.parser import HTMLParser +from http import HTTPStatus +import logging +import math +import re +import secrets +import time +from typing import Optional, Tuple +from urllib.parse import parse_qs, quote, urlencode, urljoin, urlsplit + +from requests import RequestException +from modules.common.req import get_http_session + +log = logging.getLogger(__name__) +AUTH_BASE_URL = "https://login.mynissan-account.com/" +BFF_BASE_URL = "https://nci-bff-web-prod.apps.eu2.kamereon.io/bff-web/" +USERS_BASE_URL = "https://alliance-platform-usersadapter-prod.apps.eu2.kamereon.io/user-adapter/" +CARS_BASE_URL = "https://alliance-platform-caradapter-prod.apps.eu2.kamereon.io/car-adapter/" +OAUTH_CALLBACK_URI = "com://wso2.service.nci" +OAUTH_CLIENT_ID = "ZM3WK7ax1OtQKYQ8Qqzcv5VgiA8a" # Public application identifier, not a secret. +OAUTH_LOGIN_SCOPE = "openid name profile email offline_access" +JSON_API_CONTENT_TYPE = "application/vnd.api+json" +REQUEST_TIMEOUT_SECONDS = 10 +DEFAULT_TOKEN_LIFETIME_SECONDS = 3600 +TOKEN_REFRESH_MARGIN_SECONDS = 30 +MAX_MEASUREMENT_FUTURE_SKEW_SECONDS = 300 +MAX_SOC_PERCENT = 100 +MAX_LOGIN_REDIRECTS = 10 +PKCE_VERIFIER_RANDOM_BYTES = 64 +OAUTH_STATE_RANDOM_BYTES = 32 +VIN_PATTERN = r"[A-HJ-NPR-Z0-9]{17}" +_PRIVATE_HTTP_REQUEST = ContextVar("nissan_private_http_request", default=False) + + +def _allow_transport_log(record): + """Suppress URL/payload diagnostics only inside this context's Nissan HTTP calls.""" + return not _PRIVATE_HTTP_REQUEST.get() + + +# Parent logger filters do not cover child records; register once on the emitters. +for _logger_name in ("urllib3.connection", "urllib3.connectionpool", "urllib3.poolmanager", + "urllib3.response", "urllib3.util.retry"): + logging.getLogger(_logger_name).addFilter(_allow_transport_log) + + +class NissanError(Exception): + """Errors omit response bodies, credentials and account identifiers.""" + + +class AuthenticationError(NissanError): + pass + + +class _LoginForm(HTMLParser): + def __init__(self): + super().__init__() + self.forms = [] + self.current = None + + def handle_starttag(self, tag, attrs): + attrs = dict(attrs) + if tag == "form": + self.current = {"action": attrs.get("action"), "inputs": {}} + elif tag == "input" and self.current is not None and attrs.get("name"): + self.current["inputs"][attrs["name"]] = attrs.get("value", "") + + def handle_endtag(self, tag): + if tag == "form" and self.current is not None: + self.forms.append(self.current) + self.current = None + + +def _login_target(url): + """Only follow HTTPS login redirects and submit credentials to the login host.""" + parsed = urlsplit(url) + if (parsed.scheme != "https" or parsed.netloc != urlsplit(AUTH_BASE_URL).netloc or parsed.fragment): + raise NissanError("Unexpected Nissan login destination.") + return url + + +def _number(value, maximum=math.inf): + if isinstance(value, bool) or not isinstance(value, (int, float)) or not 0 <= value < math.inf or value > maximum: + raise NissanError("Nissan returned an invalid vehicle measurement.") + try: + return float(value) + except OverflowError: + raise NissanError("Nissan returned an invalid vehicle measurement.") from None + + +def parse_battery(body) -> Tuple[float, Optional[float], float]: + """Preserve the actual measurement time and valid zero values; never invent a SoC.""" + try: + attributes = body["data"]["attributes"] + soc_percent = _number(attributes["batteryLevel"], MAX_SOC_PERCENT) + measured = datetime.fromisoformat(attributes["lastUpdateTime"].replace("Z", "+00:00")) + measurement_unix_seconds = measured.timestamp() + if (measured.tzinfo is None + or not 0 < measurement_unix_seconds <= time.time() + MAX_MEASUREMENT_FUTURE_SKEW_SECONDS): + raise ValueError + range_km = attributes.get("rangeHvacOn") + if range_km is None: + range_km = attributes.get("rangeHvacOff") + try: + range_km = _number(range_km) + except NissanError: + range_km = None + return soc_percent, range_km, measurement_unix_seconds + except (KeyError, TypeError, ValueError, AttributeError, OverflowError): + raise NissanError("Nissan did not return a valid SoC with a measurement timestamp.") from None + + +class NissanClient: + def __init__(self, user_id, password, vin=None): + self._user = user_id + self._password = password + self._configured_vin = (vin or "").strip().upper() + self._vin = None + self._access_token = self._refresh_token = None + self._expires_at_monotonic_seconds = 0 + self._session = get_http_session() + # The shared response hook logs full tokens/payloads. Check status locally instead. + self._session.hooks["response"] = [] + self._session.trust_env = False # Do not load unrelated netrc credentials or proxies. + + def _request(self, method, url, auth_request=False, **kwargs): + log_context = _PRIVATE_HTTP_REQUEST.set(True) + try: + response = self._session.request(method, url, timeout=REQUEST_TIMEOUT_SECONDS, + allow_redirects=False, **kwargs) + except RequestException: + raise NissanError("Nissan could not be reached.") from None + finally: + _PRIVATE_HTTP_REQUEST.reset(log_context) + if response.status_code in (HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN) or ( + auth_request and response.status_code == HTTPStatus.BAD_REQUEST): + raise AuthenticationError("Nissan rejected the login or access authorization.") + if response.status_code >= HTTPStatus.BAD_REQUEST: + raise NissanError(f"Nissan service returned HTTP {response.status_code}.") + return response + + def _json(self, method, url, **kwargs): + response = self._request(method, url, **kwargs) + try: + body = response.json() + if not isinstance(body, dict) or body.get("error") or body.get("errors"): + raise ValueError + return body + except ValueError: + raise NissanError("Nissan returned an invalid response.") from None + + def _redirects(self, response, url, callback=False): + for _ in range(MAX_LOGIN_REDIRECTS): + if not response.is_redirect: + return response, url + target = urljoin(url, response.headers["Location"]) + if callback and target.split("?", 1)[0] == OAUTH_CALLBACK_URI: + return response, target + url = _login_target(target) + response = self._request("GET", url) + raise NissanError("Too many Nissan login redirects.") + + def _login(self): + """Sign in through MyNISSAN, then exchange its identity token for Nissan API tokens.""" + if not self._user or not self._password: + raise AuthenticationError("MyNISSAN username and password are required.") + self._session.cookies.clear() + self._access_token = self._refresh_token = None + # Request a PKCE-bound authorization code through the hosted login form. + verifier = secrets.token_urlsafe(PKCE_VERIFIER_RANDOM_BYTES) + challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode() + state = secrets.token_urlsafe(OAUTH_STATE_RANDOM_BYTES) + url = AUTH_BASE_URL + "oauth2/authorize?" + urlencode({ + "response_type": "code", "redirect_uri": OAUTH_CALLBACK_URI, "client_id": OAUTH_CLIENT_ID, + "state": state, "scope": OAUTH_LOGIN_SCOPE, "code_challenge": challenge, "code_challenge_method": "S256", + "locale": "en_GB", "brand": "Nissan", "client": "mynissanapp"}) + response, url = self._redirects(self._request("GET", url), url) + parser = _LoginForm() + parser.feed(response.text) + form = next((f for f in parser.forms if {"sessionDataKey", "password"}.issubset(f["inputs"])), None) + if not form or not form["action"]: + raise AuthenticationError("Nissan login form is unavailable. Check sign-in through MyNISSAN.") + target = _login_target(urljoin(url, form["action"])) + data = form["inputs"] + region = data.get("regionCode", "") + data.update(userName=self._user, username=region + "/" + self._user if region else self._user, + password=self._password) + response = self._request("POST", target, data=data, headers={ + "Origin": AUTH_BASE_URL.rstrip("/"), "Referer": url}) + _, callback = self._redirects(response, target, callback=True) + values = parse_qs(urlsplit(callback).query) + if (callback.split("?", 1)[0] != OAUTH_CALLBACK_URI or urlsplit(callback).fragment + or values.get("state") != [state] or len(values.get("code", [])) != 1 or values.get("error")): + raise AuthenticationError("Nissan login did not complete. Check credentials and notices in MyNISSAN.") + # Redeem the authorization code at the identity provider. + token = self._json("POST", AUTH_BASE_URL + "oauth2/token", auth_request=True, data={ + "grant_type": "authorization_code", "redirect_uri": OAUTH_CALLBACK_URI, "client_id": OAUTH_CLIENT_ID, + "code": values["code"][0], "code_verifier": verifier, "scope": OAUTH_LOGIN_SCOPE}) + if not isinstance(token.get("id_token"), str) or not token["id_token"]: + raise AuthenticationError("Nissan did not return an identity token.") + # The vehicle API requires separate access/refresh tokens from the Nissan backend. + self._install_token(self._json("POST", BFF_BASE_URL + "v1/oauth2/access_token?platform=Android", + auth_request=True, headers={"Authorization": token["id_token"], + "Content-Type": JSON_API_CONTENT_TYPE})) + + def _install_token(self, token): + if not isinstance(token.get("access_token"), str) or not token["access_token"]: + raise AuthenticationError("Nissan did not return an access token.") + try: + lifetime_seconds = float(token.get("expires_in", DEFAULT_TOKEN_LIFETIME_SECONDS)) + if not math.isfinite(lifetime_seconds) or lifetime_seconds <= 0: + raise ValueError + except (TypeError, ValueError, OverflowError): + raise AuthenticationError("Nissan returned an invalid token lifetime.") from None + self._access_token = token["access_token"] + self._refresh_token = token.get("refresh_token") or self._refresh_token + self._expires_at_monotonic_seconds = time.monotonic() + max(0, lifetime_seconds - TOKEN_REFRESH_MARGIN_SECONDS) + + def _authenticate(self, force=False): + if self._access_token and not force and time.monotonic() < self._expires_at_monotonic_seconds: + return + if self._refresh_token: + try: + self._install_token(self._json("POST", BFF_BASE_URL + "v1/oauth2/refresh-token?platform=Android", + auth_request=True, headers={"Authorization": self._refresh_token, + "Content-Type": JSON_API_CONTENT_TYPE}, + json={"scope": "openid profile vehicles"})) + return + except AuthenticationError: + pass # A rejected refresh token requires a new login. + self._login() + + def _get(self, url): + # Retry authorization once; scheduling and service-error retries belong to Core. + for force in (False, True): + self._authenticate(force=force) + try: + return self._json("GET", url, headers={"Authorization": "Bearer " + self._access_token, + "Content-Type": JSON_API_CONTENT_TYPE}) + except AuthenticationError: + if force: + self._access_token = self._refresh_token = None + raise + + def _select_vehicle(self): + user_id = self._get(USERS_BASE_URL + "v1/users/current").get("userId") + if not isinstance(user_id, str) or not user_id: + raise NissanError("Nissan did not return a user identifier.") + vehicles = self._get(BFF_BASE_URL + "v5/users/" + quote(user_id, safe="") + "/cars").get("data") + if not isinstance(vehicles, list): + raise NissanError("Nissan did not return a vehicle list.") + candidates = {v["vin"].upper() for v in vehicles if isinstance(v, dict) and isinstance(v.get("vin"), str) + and re.fullmatch(VIN_PATTERN, v["vin"].upper())} + if self._configured_vin: + if self._configured_vin not in candidates: + raise NissanError("The configured VIN was not found in the Nissan account.") + self._vin = self._configured_vin + elif len(candidates) == 1: + self._vin = candidates.pop() + else: + raise NissanError("Could not select a unique vehicle. Please configure the VIN.") + + def fetch_battery(self) -> Tuple[float, Optional[float], float, Optional[float]]: + """Return SoC (%), range (km), battery measurement time (Unix seconds), odometer (km). + + Range and odometer may be None; valid zero values are preserved. The timestamp + belongs only to the battery data, not the separately retrieved odometer. + Battery or required login failures raise NissanError. Odometer failures do not + invalidate a valid SoC and return None for mileage. + """ + if not self._vin: + self._select_vehicle() + url = CARS_BASE_URL + "v1/cars/" + self._vin + soc_percent, range_km, measurement_unix_seconds = parse_battery(self._get(url + "/battery-status")) + # Cockpit availability differs by vehicle. Never discard a valid SoC or invent mileage. + odometer_km = None + try: + odometer_km = _number(self._get(url + "/cockpit")["data"]["attributes"]["totalMileage"]) + except (NissanError, KeyError, TypeError): + log.debug("Nissan odometer unavailable.") + log.debug("Nissan battery measurement_unix_seconds=%s", measurement_unix_seconds) + return soc_percent, range_km, measurement_unix_seconds, odometer_km diff --git a/packages/modules/vehicles/nissanconnect/api_test.py b/packages/modules/vehicles/nissanconnect/api_test.py new file mode 100644 index 0000000000..3a21a45ee3 --- /dev/null +++ b/packages/modules/vehicles/nissanconnect/api_test.py @@ -0,0 +1,244 @@ +"""Synthetic protocol tests; unmatched requests cannot reach Nissan.""" +import base64 +from concurrent.futures import ThreadPoolExecutor +import hashlib +import logging +import re +from unittest.mock import Mock +from urllib.parse import parse_qs, urlsplit + +import pytest +import requests +import requests_mock +from urllib3.connectionpool import HTTPSConnectionPool +from urllib3.response import HTTPResponse +from urllib3.util.retry import Retry + +from modules.vehicles.nissanconnect import api +from modules.vehicles.nissanconnect.config import NissanConnect, NissanConnectConfiguration + +VIN = "SJNFAAZE1U0000001" +OTHER_VIN = "SJNFAAZE1U0000002" +MEASUREMENT_UNIX_SECONDS = 1640995200.0 +BATTERY_URL = api.CARS_BASE_URL + "v1/cars/" + VIN + "/battery-status" +COCKPIT_URL = api.CARS_BASE_URL + "v1/cars/" + VIN + "/cockpit" +REFRESH_URL = api.BFF_BASE_URL + "v1/oauth2/refresh-token?platform=Android" +FORM = ('
' + '
') + + +@pytest.fixture(autouse=True) +def http(): + with requests_mock.Mocker() as mock: + yield mock + + +def battery(soc=47, timestamp="2022-01-01T00:00:00Z", **extra): + return {"data": {"attributes": dict(batteryLevel=soc, lastUpdateTime=timestamp, **extra)}} + + +def client(): + result = api.NissanClient("synthetic@example.invalid", "synthetic-password") + result._vin = VIN + result._install_token({"access_token": "synthetic-access", "refresh_token": "synthetic-refresh"}) + return result + + +def login(http): + http.get(re.compile(re.escape(api.AUTH_BASE_URL + "oauth2/authorize?")), text=FORM) + + def callback(request, context): + state = parse_qs(urlsplit(http.request_history[-2].url).query)["state"][0] + context.headers["Location"] = api.OAUTH_CALLBACK_URI + "?code=synthetic-code&state=" + state + return "" + + http.post(api.AUTH_BASE_URL + "commonauth", status_code=302, text=callback) + http.post(api.AUTH_BASE_URL + "oauth2/token", json={"id_token": "synthetic-id"}) + http.post(api.BFF_BASE_URL + "v1/oauth2/access_token?platform=Android", + json={"access_token": "synthetic-access", "refresh_token": "synthetic-refresh"}) + + +def test_login_selection_measurements_and_session_reuse(http, caplog): + caplog.set_level(logging.DEBUG) + login(http) + http.get(api.USERS_BASE_URL + "v1/users/current", json={"userId": "synthetic-user"}) + http.get(api.BFF_BASE_URL + "v5/users/synthetic-user/cars", json={"data": [{"vin": VIN}]}) + http.get(BATTERY_URL, json=battery(rangeHvacOn=123.4)) + http.get(COCKPIT_URL, json={"data": {"attributes": {"totalMileage": 12345.6}}}) + instance = api.NissanClient("synthetic@example.invalid", "synthetic-password") + assert instance.fetch_battery() == (47, 123.4, MEASUREMENT_UNIX_SECONDS, 12345.6) + calls = list(http.request_history) + authorize = parse_qs(urlsplit(calls[0].url).query) + verifier = parse_qs(calls[2].text)["code_verifier"][0] + challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode() + assert authorize["code_challenge"] == [challenge] + assert parse_qs(calls[1].text)["username"] == ["EU/synthetic@example.invalid"] + assert calls[3].headers["Authorization"] == "synthetic-id" + assert calls[-1].headers["Authorization"] == "Bearer synthetic-access" + assert all(call.timeout == api.REQUEST_TIMEOUT_SECONDS for call in calls) + assert all("/actions/" not in call.url for call in calls) + assert not any(record.name == "modules.common.req" for record in caplog.records) + instance.fetch_battery() + assert len(http.request_history) == len(calls) + 2 + + +@pytest.mark.parametrize("soc", [0, 47.5, 100]) +def test_valid_zero_and_old_measurement(soc): + assert api.parse_battery(battery(soc, rangeHvacOn=0)) == (soc, 0, MEASUREMENT_UNIX_SECONDS) + + +@pytest.mark.parametrize("soc", [None, True, -1, 101, "47", float("nan"), float("inf")]) +def test_invalid_soc_never_becomes_zero(soc): + with pytest.raises(api.NissanError): + api.parse_battery(battery(soc)) + + +@pytest.mark.parametrize("timestamp", [None, "bad", "2022-01-01T00:00:00", "2099-01-01T00:00:00Z"]) +def test_invalid_timestamp_never_becomes_now(timestamp): + with pytest.raises(api.NissanError): + api.parse_battery(battery(timestamp=timestamp)) + + +@pytest.mark.parametrize("odometer", [None, -1, True, "12345", float("inf")]) +def test_unavailable_optional_data_preserves_soc(http, odometer): + http.get(BATTERY_URL, json=battery(rangeHvacOn=-1)) + http.get(COCKPIT_URL, json={"data": {"attributes": {"totalMileage": odometer}}}) + assert client().fetch_battery() == (47, None, MEASUREMENT_UNIX_SECONDS, None) + + +def test_optional_endpoint_failure_preserves_soc(http): + http.get(BATTERY_URL, json=battery(rangeHvacOff=150)) + http.get(COCKPIT_URL, status_code=503) + assert client().fetch_battery() == (47, 150, MEASUREMENT_UNIX_SECONDS, None) + + +@pytest.mark.parametrize("status", [429, 503]) +def test_service_errors_raise_and_next_core_update_can_retry(http, status): + http.get(BATTERY_URL, [{"status_code": status, "text": "synthetic-private-body"}, {"json": battery()}]) + http.get(COCKPIT_URL, json={"data": {"attributes": {"totalMileage": 0}}}) + instance = client() + with pytest.raises(api.NissanError, match=f"HTTP {status}"): + instance.fetch_battery() + assert http.call_count == 1 + assert instance.fetch_battery()[-1] == 0 + + +@pytest.mark.parametrize("expired", [False, True]) +def test_authentication_refreshes_once(http, expired): + http.get(BATTERY_URL, [{"json": battery()}] if expired else [{"status_code": 401}, {"json": battery()}]) + http.get(COCKPIT_URL, json={}) + refresh = http.post(REFRESH_URL, json={"access_token": "synthetic-new"}) + instance = client() + if expired: + instance._expires_at_monotonic_seconds = 0 + assert instance.fetch_battery()[0] == 47 + assert refresh.call_count == 1 + assert http.last_request.headers["Authorization"] == "Bearer synthetic-new" + + +def test_rejected_refresh_uses_login_and_repeated_401_stops(http): + login(http) + http.get(BATTERY_URL, status_code=401) + refresh = http.post(REFRESH_URL, status_code=400) + with pytest.raises(api.AuthenticationError): + client().fetch_battery() + assert refresh.call_count == 1 + assert sum(request.url == BATTERY_URL for request in http.request_history) == 2 + + +@pytest.mark.parametrize("target", [ + "https://example.invalid/steal", "http://login.mynissan-account.com/", + "https://user@login.mynissan-account.com/", "https://login.mynissan-account.com:444/"]) +@pytest.mark.parametrize("form", [False, True]) +def test_rejects_unsafe_login_destinations(http, target, form): + http.get(re.compile(re.escape(api.AUTH_BASE_URL)), + text=FORM.replace("/commonauth", target) if form else "", status_code=200 if form else 302, + headers={} if form else {"Location": target}) + with pytest.raises(api.NissanError, match="destination"): + api.NissanClient("synthetic@example.invalid", "synthetic-password").fetch_battery() + assert http.call_count == 1 + + +def test_oauth_state_mismatch_stops_before_token_exchange(http): + login(http) + http.post(api.AUTH_BASE_URL + "commonauth", status_code=302, + headers={"Location": api.OAUTH_CALLBACK_URI + "?code=synthetic-code&state=wrong"}) + with pytest.raises(api.AuthenticationError): + api.NissanClient("synthetic@example.invalid", "synthetic-password").fetch_battery() + assert http.call_count == 2 + + +@pytest.mark.parametrize("configured_vin,vehicles,expected", [ + ("", [VIN, OTHER_VIN], None), (OTHER_VIN, [VIN], None), ("", [], None), + (OTHER_VIN.lower(), [VIN, OTHER_VIN], OTHER_VIN), ("", [VIN], VIN)]) +def test_vehicle_selection(http, configured_vin, vehicles, expected): + http.get(api.USERS_BASE_URL + "v1/users/current", json={"userId": "synthetic-user"}) + http.get(api.BFF_BASE_URL + "v5/users/synthetic-user/cars", json={"data": [{"vin": vin} for vin in vehicles]}) + instance = client() + instance._configured_vin = configured_vin.upper() + if expected: + instance._select_vehicle() + assert instance._vin == expected + else: + with pytest.raises(api.NissanError): + instance._select_vehicle() + + +def test_transport_and_json_errors_omit_private_data(http): + http.get(BATTERY_URL, exc=requests.Timeout("synthetic-private-url")) + instance = client() + with pytest.raises(api.NissanError, match="^Nissan could not be reached.$"): + instance.fetch_battery() + http.get(BATTERY_URL, text="synthetic-private-body") + with pytest.raises(api.NissanError, match="^Nissan returned an invalid response.$"): + instance.fetch_battery() + + +def test_native_configuration_roundtrip_and_private_repr(): + from dataclass_utils import asdict, dataclass_from_dict + + config = NissanConnect(configuration=NissanConnectConfiguration("synthetic@example.invalid", "synthetic-password")) + decoded = dataclass_from_dict(NissanConnect, asdict(config)) + assert decoded == config + assert "synthetic" not in repr(config) + assert NissanConnect().configuration is not NissanConnect().configuration + + +@pytest.mark.parametrize("path", [ + "/car-adapter/v1/cars/" + VIN + "/battery-status", + "/bff-web/v5/users/synthetic-user/cars", + "/commonauth?code=synthetic-code&sessionDataKey=synthetic-session", +]) +@pytest.mark.parametrize("fail", [False, True]) +def test_transport_logs_hide_private_urls_without_muting_other_requests(monkeypatch, caplog, path, fail): + caplog.set_level(logging.DEBUG) + instance = client() + pool = HTTPSConnectionPool("example.invalid") + connection = Mock(is_closed=False, is_verified=True) + connection.getresponse.return_value = HTTPResponse(status=200, headers={}, preload_content=False) + monkeypatch.setattr(pool, "_validate_conn", lambda connection: None) + transport_log = logging.getLogger("urllib3.connectionpool") + + def request(*args, **kwargs): + # Real urllib3 log paths, but a fake connection: no socket or Nissan access. + pool._make_request(connection, "GET", path) + Retry(total=1).increment("GET", path) + with ThreadPoolExecutor(max_workers=1) as executor: + executor.submit(transport_log.debug, "Other vehicle request remains visible").result() + api.log.debug("Nissan diagnostic remains visible") + if fail: + raise requests.Timeout("synthetic-private-url") + return Mock(status_code=200) + + monkeypatch.setattr(instance._session, "request", request) + if fail: + with pytest.raises(api.NissanError, match="^Nissan could not be reached.$"): + instance._request("GET", "https://example.invalid" + path) + else: + assert instance._request("GET", "https://example.invalid" + path).status_code == 200 + transport_log.debug("Subsequent request remains visible") + assert path not in caplog.text + assert "synthetic-private-url" not in caplog.text + assert "Other vehicle request remains visible" in caplog.text + assert "Nissan diagnostic remains visible" in caplog.text + assert "Subsequent request remains visible" in caplog.text diff --git a/packages/modules/vehicles/nissanconnect/config.py b/packages/modules/vehicles/nissanconnect/config.py new file mode 100644 index 0000000000..3c625227df --- /dev/null +++ b/packages/modules/vehicles/nissanconnect/config.py @@ -0,0 +1,21 @@ +from dataclasses import dataclass, field +from typing import Optional + + +@dataclass +class NissanConnectConfiguration: + user_id: Optional[str] = field(default=None, repr=False) + password: Optional[str] = field(default=None, repr=False) + vin: Optional[str] = field(default=None, repr=False) + + +@dataclass +class NissanConnect: + name: str = "Nissan – MyNISSAN EU (experimental)" + type: str = "nissanconnect" + official: bool = False + configuration: NissanConnectConfiguration = None + + def __post_init__(self): + if self.configuration is None: + self.configuration = NissanConnectConfiguration() diff --git a/packages/modules/vehicles/nissanconnect/soc.py b/packages/modules/vehicles/nissanconnect/soc.py new file mode 100644 index 0000000000..e505b59c1a --- /dev/null +++ b/packages/modules/vehicles/nissanconnect/soc.py @@ -0,0 +1,28 @@ +from modules.common.abstract_device import DeviceDescriptor +from modules.common.abstract_vehicle import VehicleUpdateData +from modules.common.component_state import CarState +from modules.common.configurable_vehicle import ConfigurableVehicle +from modules.vehicles.nissanconnect.api import NissanClient +from modules.vehicles.nissanconnect.config import NissanConnect + + +def create_vehicle(vehicle_config: NissanConnect, vehicle: int): + client = None + configuration_key = None + + def updater(vehicle_update_data: VehicleUpdateData) -> CarState: + nonlocal client, configuration_key + config = vehicle_config.configuration + current = (config.user_id, config.password, config.vin) + # Reuse tokens until credentials or vehicle selection change. + if client is None or current != configuration_key: + client = NissanClient(*current) + configuration_key = current + soc_percent, range_km, measurement_unix_seconds, odometer_km = client.fetch_battery() + return CarState(soc=soc_percent, range=range_km, soc_timestamp=measurement_unix_seconds, odometer=odometer_km) + + return ConfigurableVehicle(vehicle_config=vehicle_config, component_updater=updater, + vehicle=vehicle, calc_while_charging=False) + + +device_descriptor = DeviceDescriptor(configuration_factory=NissanConnect)