Feature vw euda refactoring

packages/modules/vehicles/vweuda/api.py

@@ -0,0 +1,238 @@
+"""Synchronous VW EU Data Act login and dataset downloads."""
+
+import io
+import json
+import re
+import zipfile
+from html.parser import HTMLParser
+from typing import Optional, Union
+from urllib.parse import urlencode, urljoin, urlparse
+
+from requests import HTTPError, RequestException, Response, Session
+
+BASE_URL = "https://eu-data-act.drivesomethinggreater.com"
+OIDC_AUTHORIZE_URL = "https://identity.vwgroup.io/oidc/v1/authorize"
+CALLBACK_LOGIN_PATH = "/services/callbacklogin"
+METADATA_PATH = "/proxy_api/euda-apim/datarequest/vehicles/{vin}/metadata/partial"
+LIST_PATH = "/proxy_api/euda-apim/datadelivery/vehicles/{vin}/{identifier}/list"
+DOWNLOAD_PATH = "/proxy_api/euda-apim/datadelivery/vehicles/{vin}/{identifier}/download"
+NO_CONTENT_SUFFIX = "_no_content_found.zip"
+HTTP_TIMEOUT = (5, 15)
+USER_AGENT = "vw_eu_data_act/0.3.3"
+
+# OIDC client ID and portal brand state. SEAT and CUPRA share a client ID.
+BRANDS = {
+    "volkswagen": ("9b58543e-1c15-4193-91d5-8a14145bebb0@apps_vw-dilab_com", "VOLKSWAGEN_PASSENGER_CARS"),
+    "audi": ("cc29b87a-5e9a-4362-aecf-5adea6b01bbb@apps_vw-dilab_com", "AUDI"),
+    "skoda": ("3ea88bf9-1d4e-4a68-b3ad-4098c1f1d246@apps_vw-dilab_com", "SKODA"),
+    "seat": ("f85e5b69-e3b2-43aa-9c0d-1b7d0e0b576f@apps_vw-dilab_com", "SEAT"),
+    "cupra": ("f85e5b69-e3b2-43aa-9c0d-1b7d0e0b576f@apps_vw-dilab_com", "CUPRA"),
+}
+BRAND_PREFIXES = {
+    "audi": ("WAU", "WA1", "WUA", "WU1", "99A", "AAA", "TRU"),
+    "skoda": ("NAD", "TMB", "Y6U"),
+    "cupra": ("VSS",),  # SEAT/CUPRA cannot be distinguished by this prefix.
+}
+
+
+def brand_for_vin(vin: str) -> str:
+    return next((brand for brand, prefixes in BRAND_PREFIXES.items() if vin[:3] in prefixes), "volkswagen")
+
+
+class ApiError(Exception):
+    """Portal/network failure, with HTTP status when available."""
+
+    def __init__(self, message: str, *, status: Optional[int] = None) -> None:
+        super().__init__(message)
+        self.status = status
+
+
+class AuthError(ApiError):
+    """Login failed or the identity provider's form changed."""
+
+
+class DatasetError(ApiError):
+    """Malformed archive; it must not prevent processing newer datasets."""
+
+
+class _LoginForm(HTMLParser):
+    """Collect the first HTML form and its input fields."""
+
+    def __init__(self, html: str) -> None:
+        super().__init__()
+        self.action = None
+        self.fields = {}
+        self._inside = False
+        self._done = False
+        self.feed(html)
+
+    def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> None:
+        attrs = dict(attrs)
+        if tag == "form" and not self._done and not self._inside:
+            self.action = attrs.get("action")
+            self._inside = True
+        elif tag == "input" and self._inside and attrs.get("name"):
+            self.fields[attrs["name"]] = attrs.get("value") or ""
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag == "form" and self._inside:
+            self._inside, self._done = False, True
+
+
+def _template_model(html: str) -> dict:
+    # raw_decode also handles braces inside JSON strings, unlike brace counting.
+    index = html.find("templateModel")
+    start = html.find("{", index) if index >= 0 else -1
+    try:
+        return json.JSONDecoder().raw_decode(html[start:])[0] if start >= 0 else {}
+    except ValueError:
+        return {}
+
+
+def _login_error(html: str) -> Optional[str]:
+    model = _template_model(html)
+    error = model.get("error") or model.get("errorCode")
+    if isinstance(error, dict):
+        return error.get("text") or error.get("errorCode") or str(error)
+    return str(error) if error else None
+
+
+def _login_fields(html: str) -> tuple[dict, Optional[str]]:
+    form = _LoginForm(html)
+    model = _template_model(html)
+    for key in ("hmac", "relayState"):
+        if model.get(key):
+            form.fields[key] = model[key]
+    csrf = re.search(r"csrf_token\s*[:=]\s*['\"]([^'\"]+)['\"]", html)
+    if csrf:
+        form.fields.setdefault("_csrf", csrf.group(1))
+    if not all(key in form.fields for key in ("hmac", "_csrf")):
+        raise AuthError(_login_error(html) or "Could not parse identity login form")
+    return form.fields, form.action
+
+
+def _check_login_status(response: Response) -> None:
+    status = response.status_code
+    if status >= 500 or status in (404, 429):
+        raise ApiError(f"Identity provider error (HTTP {status})", status=status)
+    if status >= 400:
+        raise AuthError(_login_error(response.text) or f"Login rejected (HTTP {status})")
+
+
+class EudaApiClient:
+    """The synchronous caller owns and closes the session."""
+
+    def __init__(self, session: Session, email: str, password: str, brand: str = "volkswagen") -> None:
+        self._session = session
+        self._email, self._password, self._brand = email, password, brand
+        self._logged_in = False
+
+    def _request(self, method: str, url: str, **kwargs) -> Response:
+        headers = {"User-Agent": USER_AGENT, **kwargs.pop("headers", {})}
+        try:
+            return self._session.request(method, url, headers=headers, timeout=HTTP_TIMEOUT, **kwargs)
+        except HTTPError as error:
+            # req.get_http_session() raises in its response hook. Keep the full
+            # response for VW login status handling and the 401/403 retry below.
+            if error.response is None:
+                raise ApiError(f"HTTP error during {method}: {error}") from error
+            return error.response
+        except RequestException as error:
+            raise ApiError(f"Network error during {method}: {error}") from error
+
+    @staticmethod
+    def _build_authorize_url(brand: str = "volkswagen") -> str:
+        client_id, state = BRANDS.get(brand, BRANDS["volkswagen"])
+        # The portal's redirect/authentication servlet fails for non-browser clients.
+        return OIDC_AUTHORIZE_URL + "?" + urlencode({
+            "client_id": client_id, "state": f"de__en__{state}", "response_type": "code",
+            "scope": "openid cars profile", "redirect_uri": BASE_URL + "/login", "prompt": "login",
+        })
+
+    def _post_login_form(self, url: str, *, data: dict, referer: str) -> Response:
+        """Follow at most ten redirects, stopping at the session-setting callback."""
+        response = self._request("POST", url, data=data, headers={"Referer": referer}, allow_redirects=False)
+        for hop in range(11):
+            location = response.headers.get("Location")
+            parsed = urlparse(response.url)
+            callback = parsed.netloc == urlparse(BASE_URL).netloc and parsed.path.startswith(CALLBACK_LOGIN_PATH)
+            if callback or response.status_code not in (301, 302, 303) or not location:
+                return response
+            previous_url = response.url
+            next_url = urljoin(previous_url, location)
+            response.close()
+            if hop == 10:
+                raise ApiError("Login redirect chain exceeded 10 hops")
+            response = self._request("GET", next_url, headers={"Referer": previous_url}, allow_redirects=False)
+
+    def login(self) -> None:
+        """Authorize → email form → password form → portal callback."""
+        self._logged_in = False
+        with self._request("GET", self._build_authorize_url(self._brand)) as response:
+            _check_login_status(response)
+            signin_url = response.url
+            fields, action = _login_fields(response.text)
+
+        fields["email"] = self._email
+        with self._request("POST", urljoin(signin_url, action or ""), data=fields,
+                           headers={"Referer": signin_url}) as response:
+            _check_login_status(response)
+            authenticate_url = response.url
+            fields, action = _login_fields(response.text)
+
+        fields.update(email=self._email, password=self._password)
+        # relayState belongs in the POST body, not in the action URL as well.
+        action_url = urljoin(authenticate_url, action) if action else authenticate_url.split("?", 1)[0]
+        with self._post_login_form(action_url, data=fields, referer=authenticate_url) as response:
+            _check_login_status(response)
+            parsed = urlparse(response.url)
+            if parsed.netloc != urlparse(BASE_URL).netloc or not parsed.path.startswith(CALLBACK_LOGIN_PATH):
+                raise AuthError(_login_error(response.text) or "Login did not reach the portal callback")
+            if not response.cookies.get("access_token"):
+                raise AuthError("Login callback did not establish a session (access_token cookie missing)")
+        self._logged_in = True
+
+    def _authenticated_get(self, url: str, headers: Optional[dict] = None) -> Response:
+        """Share the same one-time reauthentication for JSON and ZIP requests."""
+        if not self._logged_in:
+            self.login()
+        for attempt in range(2):
+            response = self._request("GET", url, headers=headers or {})
+            status = response.status_code
+            if status in (401, 403) and attempt == 0:
+                response.close()
+                self.login()
+                continue
+            if status >= 400:
+                response.close()
+                raise ApiError(f"Portal request failed (HTTP {status})", status=status)
+            return response
+
+    def _get_json(self, url: str, headers: Optional[dict] = None) -> Union[dict, list]:
+        with self._authenticated_get(url, headers) as response:
+            try:
+                return response.json()
+            except ValueError as error:
+                raise ApiError("Invalid JSON from portal") from error
+
+    def get_metadata(self, vin: str) -> dict:
+        return self._get_json(BASE_URL + METADATA_PATH.format(vin=vin))
+
+    def list_datasets(self, vin: str, identifier: str) -> list[dict]:
+        url = BASE_URL + LIST_PATH.format(vin=vin, identifier=identifier)
+        # The endpoint returns HTTP 500 without the matching request type.
+        data = self._get_json(url, headers={"type": "partial"})
+        return data if isinstance(data, list) else data.get("files", [])
+
+    def download_dataset(self, vin: str, identifier: str, name: str) -> tuple[str, dict]:
+        url = BASE_URL + DOWNLOAD_PATH.format(vin=vin, identifier=identifier)
+        with self._authenticated_get(url, {"filename": name, "type": "partial"}) as response:
+            try:
+                with zipfile.ZipFile(io.BytesIO(response.content)) as archive:
+                    member = next((name for name in archive.namelist() if name.lower().endswith(".json")), None)
+                    if member is None:
+                        raise DatasetError("No JSON inside downloaded dataset")
+                    data = json.loads(archive.read(member).decode("utf-8"))
+            except (zipfile.BadZipFile, ValueError) as error:
+                raise DatasetError("Could not read downloaded dataset") from error
+        return name.removesuffix(".zip") + ".json", data

packages/modules/vehicles/vweuda/api_test.py

@@ -0,0 +1,102 @@
+import io
+import json
+import zipfile
+from unittest.mock import Mock
+from urllib.parse import parse_qs
+
+import pytest
+from modules.common.req import get_http_session
+
+from modules.vehicles.vweuda import api
+
+
+def test_login_stops_at_callback_and_posts_relay_state(requests_mock):
+    signin = "https://identity.vwgroup.io/signin-service/login"
+    authenticate = signin + "/authenticate"
+    callback = api.BASE_URL + api.CALLBACK_LOGIN_PATH + "?code=secret"
+    cms = api.BASE_URL + "/missing-cms-page"
+    form = '<form action="/signin-service/login"><input name="hmac" value="h"></form>'
+    form += '<script>csrf_token: "csrf"</script>'
+    requests_mock.get(api.OIDC_AUTHORIZE_URL, status_code=302, headers={"Location": signin})
+    requests_mock.get(signin, text=form)
+    password_page = '<script>templateModel: {"hmac":"h2","relayState":"relay"}; csrf_token: "csrf2"</script>'
+    requests_mock.post(signin, status_code=302, headers={"Location": authenticate + "?relayState=relay"})
+    requests_mock.get(authenticate, text=password_page)
+    credentials = requests_mock.post(authenticate, status_code=302, headers={"Location": callback})
+    requests_mock.get(callback, status_code=302, headers={
+        "Location": cms, "Set-Cookie": "access_token=token; Path=/; Secure",
+    })
+    with get_http_session() as session:
+        client = api.EudaApiClient(session, "test@example.com", "password")
+        client.login()
+        assert client._logged_in
+    assert all(request.url != cms for request in requests_mock.request_history)
+    assert credentials.last_request.query == ""
+    fields = parse_qs(credentials.last_request.text)
+    assert fields["password"] == ["password"]
+    assert fields["relayState"] == ["relay"]
+
+
+def test_callback_without_cookie_is_not_successful(requests_mock, monkeypatch):
+    callback = api.BASE_URL + api.CALLBACK_LOGIN_PATH
+    with get_http_session() as session:
+        client = api.EudaApiClient(session, "test@example.com", "password")
+        form = '<form><input name="hmac" value="h"></form><script>csrf_token: "csrf"</script>'
+        requests_mock.get(api.OIDC_AUTHORIZE_URL, text=form)
+        requests_mock.post(api.OIDC_AUTHORIZE_URL, text=form)
+        requests_mock.get(callback, status_code=302)
+        monkeypatch.setattr(client, "_post_login_form", lambda *args, **kwargs: session.get(callback))
+        with pytest.raises(api.AuthError, match="cookie missing"):
+            client.login()
+        assert not client._logged_in
+
+
+@pytest.mark.parametrize("status", [404, 429, 500, 503])
+def test_transient_login_errors_remain_retryable(requests_mock, status):
+    requests_mock.get(api.OIDC_AUTHORIZE_URL, status_code=status)
+    with get_http_session() as session:
+        with pytest.raises(api.ApiError) as error:
+            api.EudaApiClient(session, "test@example.com", "password").login()
+        assert not isinstance(error.value, api.AuthError)
+        assert error.value.status == status
+
+
+def test_json_session_expiry_retries_only_once(requests_mock, monkeypatch):
+    url = api.BASE_URL + "/test-json"
+    response = requests_mock.get(url, [{"status_code": 401}, {"status_code": 403}])
+    with get_http_session() as session:
+        client = api.EudaApiClient(session, "test@example.com", "password")
+        client._logged_in = True
+        login = Mock()
+        monkeypatch.setattr(client, "login", login)
+        with pytest.raises(api.ApiError) as error:
+            client._get_json(url)
+        assert error.value.status == 403
+        login.assert_called_once()
+        assert response.call_count == 2
+
+
+def test_helper_http_error_preserves_login_error_body(requests_mock):
+    requests_mock.get(api.OIDC_AUTHORIZE_URL, status_code=400,
+                      text='<script>templateModel: {"error": {"text": "Account locked"}}</script>')
+    with get_http_session() as session:
+        with pytest.raises(api.AuthError, match="Account locked"):
+            api.EudaApiClient(session, "test@example.com", "password").login()
+
+
+def test_download_session_expiry_preserves_zip_decoding(requests_mock, monkeypatch):
+    payload = {"vin": "WVW00000000000001", "Data": []}
+    archive = io.BytesIO()
+    with zipfile.ZipFile(archive, "w") as handle:
+        handle.writestr("data.json", json.dumps(payload))
+    url = api.BASE_URL + api.DOWNLOAD_PATH.format(vin=payload["vin"], identifier="id")
+    requests_mock.get(url, [{"status_code": 401}, {"content": archive.getvalue()}])
+    with get_http_session() as session:
+        client = api.EudaApiClient(session, "test@example.com", "password")
+        client._logged_in = True
+        login = Mock()
+        monkeypatch.setattr(client, "login", login)
+        name, data = client.download_dataset(payload["vin"], "id", "data.zip")
+        assert name == "data.json"
+        assert data == payload
+        login.assert_called_once()