feat: add headless auth

2026-08-26 10:17:18 +00:00 · 2026-04-18 09:53:00 +02:00
parent 1fb3beeeea
commit b74bad33e2
14 changed files with 244 additions and 32 deletions
@@ -33,7 +33,6 @@ jobs:
      - name: check quality
        run: |
          source .venv/bin/activate
-          prospector
          pre-commit run -a
      - name: test app
        env:
@@ -9,4 +9,4 @@ repos:
        hooks:
        -   id: prospector
            language: system
-
+            args: ["--without-tool", "dodgy"]
@@ -21,5 +21,10 @@ COPY --from=builder /usr/local/bin/  /usr/local/bin/
 RUN  apt-get install -y --no-install-recommends $PYTHON_DEP curl && \
     apt-get clean ; \
     rm -rf /var/lib/apt/lists/*
+
+# Install Playwright and WebKit dependencies
+RUN pip3 install --break-system-packages playwright && \
+    playwright install --with-deps webkit
+
 COPY /docker_files/init.sh /init.sh
 CMD /init.sh
@@ -76,3 +76,5 @@ If you want to thank me for my work :smile:

 [![donate](https://www.paypalobjects.com/en_US/i/btn/btn_donate_LG.gif)](https://www.paypal.com/donate?hosted_button_id=SM652WPXFNCXS)

+## Acknowledgements
+- thanks to @tamcore for https://github.com/tamcore/stelloauth 
@@ -14,8 +14,17 @@
    
 - For everyone :
      ```pip3 install psa-car-controller```
-  
-            
+
+- Install [Playwright](https://playwright.dev/python/) for automatic OAuth login:
+
+  ```bash
+  pip3 install playwright
+  playwright install --with-deps webkit
+  ```
+
+  > If Playwright is not installed or the browser fails to start, automatic login falls back to the manual flow.
+
+
 1.3 start the app:
        
 Start the app with charge control enabled :
@@ -125,6 +125,7 @@ class RemoteClient:

    def stop(self):
        if self.mqtt_client:
+            logger.info("stop mqtt...")
            self.mqtt_client.on_disconnect = None
            self.mqtt_client.disconnect()
        if self.update_thread:
@@ -135,7 +136,7 @@ class RemoteClient:
        if len(self.vehicles_list) > 0:
            try:
                self.wakeup(self.vehicles_list[0].vin)
-            except RateLimitException:
+            except Exception:
                logger.exception("__keep_mqtt")
        self.update_thread = threading.Timer(timeout, self.__keep_mqtt)
        self.update_thread.daemon = True
@@ -210,8 +210,7 @@ class Otp:
        params.update(R)
        xml = self.request(params)
        if xml["err"] != "OK":
-            logger.error("Error during activation: %s", xml)
-            return Otp.NOK
+            raise ConfigException(f"Error during activation: {xml}")
        self.data.synchro(xml, self.generate_kma(self.codepin))

        if self.mode == Otp.OTP_MODE:
@@ -269,14 +268,13 @@ class Otp:
        try:
            if self.activation_start():
                res = self.activation_finalyze()
-                if res != Otp.NOK:
-                    if res == Otp.OTP_TWICE:
-                        self.mode = Otp.OTP_MODE
-                        self.activation_start()
-                        assert self.activation_finalyze() == Otp.OK
-                    otp_code = self._get_otp_code()
-                    assert otp_code is not None
-                    logger.debug("otp code: %s", otp_code)
+                if res == Otp.OTP_TWICE:
+                    self.mode = Otp.OTP_MODE
+                    self.activation_start()
+                    assert self.activation_finalyze() == Otp.OK
+                otp_code = self._get_otp_code()
+                assert otp_code is not None
+                logger.debug("otp code: %s", otp_code)
        except AssertionError as e:
            raise ConfigException("Can't get otp code") from e
        return otp_code
@@ -23,6 +23,7 @@ class ApkParser:
        self.culture = None
        self.client_id = None
        self.client_secret = None
+        self.version = None

    @staticmethod
    def __get_cultures_code(file, country_code):
@@ -45,6 +46,7 @@ class ApkParser:

        # Get Customer id
        self.site_code = BRAND[package_name]["brand_code"] + "_" + self.country_code + "_ESP"
+        self.version = a.androidversion.get("Name")
        pfx_cert = a.get_file("assets/MWPMYMA1.pfx")
        save_key_to_pem(pfx_cert, b"y5Y2my5B")

@@ -14,10 +14,9 @@ from psa_car_controller.psacc.application.charge_control import ChargeControl, C

 logger = logging.getLogger(__name__)

-APP_VERSION = "1.51.1"
 GITHUB_USER = "flobz"
 GITHUB_REPO = "psa_apk"
-TIMEOUT_IN_S = 10
+TIMEOUT_IN_S = 30
 app = PSACarController()


@@ -37,6 +36,7 @@ class InitialSetup:
        self.site_code = apk_parser.site_code
        self.client_id = apk_parser.client_id
        self.client_secret = apk_parser.client_secret
+        self.version = apk_parser.version
        self.country_code = country_code
        self.user_info = None
        self.customer_id = None
@@ -64,7 +64,7 @@ class InitialSetup:
            raise e
        except Exception as ex:
            msg = traceback.format_exc() + f"\nHOST_BRANDID : {apk_parser.host_brandid_prod} " \
-                                           f"sitecode: {apk_parser.site_code}"
+                f"sitecode: {apk_parser.site_code}"
            try:
                msg += res.text
            except BaseException:
@@ -87,7 +87,7 @@ class InitialSetup:
                params={
                    "culture": self.culture,
                    "width": 1080,
-                    "version": APP_VERSION
+                    "version": self.version
                },
                data=json.dumps({"site_code": self.site_code, "ticket": self.token}),
                headers={
@@ -96,7 +96,7 @@ class InitialSetup:
                    "Source-Agent": "App-Android",
                    "Token": self.token,
                    "User-Agent": "okhttp/4.8.0",
-                    "Version": APP_VERSION
+                    "Version": self.version
                },
                cert=("certs/public.pem", "certs/private.pem"),
                timeout=TIMEOUT_IN_S
@@ -0,0 +1,148 @@
+import logging
+import os
+import time
+from urllib.parse import urlparse, parse_qs
+from playwright import sync_api as playwright_sync
+from playwright.sync_api import Page
+
+logger = logging.getLogger(__name__)
+
+# Selectors used by the Gigya login form
+EMAIL_SELECTOR = '#gigya-login-form input[name="username"]'
+PASSWORD_SELECTOR = '#gigya-login-form input[name="password"]'  # nosec dodgy: password
+SUBMIT_SELECTOR = '#gigya-login-form input[type="submit"]'
+REMEMBER_ME_SELECTOR = 'label[for="gigya-checkbox-remember"]'
+
+# ForgeRock AM consent page selectors
+AUTHORIZE_SELECTORS = [
+    'input[name="decision"][value="allow"]',
+    'button[name="decision"][value="allow"]',
+    '#allow',
+    'input[name="allow"]',
+    'button[name="allow"]',
+    'input[type="submit"][value="Allow"]',
+    'input[type="submit"][value="Erlauben"]',
+    'input[type="submit"][value="Autoriser"]',
+    '#cvs_from input[type="submit"]',
+]
+
+TIMEOUT_MS = 60_000
+
+
+class HeadlessOAuthError(Exception):
+    """Exception raised when headless OAuth fails, carrying debug info."""
+
+    def __init__(self, message, url, html, logs):
+        super().__init__(message)
+        self.url = url
+        self.html = html
+        self.logs = logs
+
+
+class PlaywrightNotInstalled(Exception):
+    """Exception raised when Playwright is not installed."""
+
+
+class FormException(Exception):
+    """Exception raised when form submit fails."""
+
+
+def _fill_credentials(page: Page, email, password):
+    """Wait for and fill the login form if visible."""
+    try:
+        page.wait_for_selector(EMAIL_SELECTOR, timeout=30_000)
+    except Exception:  # pylint: disable=broad-except
+        pass
+
+    if page.is_visible(EMAIL_SELECTOR):
+        logger.info("Filling credentials")
+        # Use type instead of fill to be more human-like and avoid some bot detection
+        page.click(EMAIL_SELECTOR)
+        page.type(EMAIL_SELECTOR, email, delay=50)
+        page.click(PASSWORD_SELECTOR)
+        page.type(PASSWORD_SELECTOR, password, delay=50)
+
+        try:
+            page.click(REMEMBER_ME_SELECTOR)
+        except playwright_sync.TimeoutError:
+            logger.warning("Remember me checkbox not found")
+        submit_button = page.locator(SUBMIT_SELECTOR)
+        submit_button.click()
+        submit_button.wait_for(state="detached", timeout=10_000)
+    else:
+        logger.info(
+            "Login form not visible, checking if already authenticated or on error page")
+
+
+def check_for_error(page):
+    if errors := page.locator('div.gigya-error-msg.gigya-form-error-msg.gigya-error-msg-active').all_inner_texts():
+        raise FormException(f"Authentication failed: {errors}")
+
+
+def get_code(page: Page, scheme: str) -> str:
+    """get the OAuth code in the URL or consent page."""
+    code = [None]
+
+    def find_code_in_url(url):
+        logger.info("Checking for oauth2 code in %s", url)
+        if url.startswith(scheme + "://") and (code_found := parse_qs(urlparse(url).query).get("code", [None])[0]):
+            code[0] = code_found
+    page.on('request', lambda req: find_code_in_url(req.url))
+    deadline = time.time() + 30
+    while time.time() < deadline:
+        for selector in AUTHORIZE_SELECTORS:
+            if page.is_visible(selector):
+                logger.info("Clicking authorization consent: %s", selector)
+                page.click(selector)
+                break
+            time.sleep(1)
+            if code[0]:
+                return code[0]
+
+    raise RuntimeError("Can't find oauth2 code")
+
+
+def get_oauth_code_headless(auth_url: str, email: str, password: str,
+                            scheme: str) -> str:
+    """Automate PSA/Stellantis OAuth login using a native Playwright WebKit browser."""
+
+    console_logs = []
+
+    def on_console(msg):
+        console_logs.append(f"[{msg.type}] {msg.text}")
+
+    if not playwright_sync:
+        raise PlaywrightNotInstalled("Playwright is not installed, run 'pip install playwright'")
+
+    with playwright_sync.sync_playwright() as p:
+        # Launch native browser (prefer webkit fallback to chromium)
+        browser = None
+        for browser_type in [p.webkit, p.chromium]:
+            try:
+                logger.info("Launching headless %s", browser_type.name)
+                browser = browser_type.launch(
+                    headless=os.environ.get("NO_HEADLESS") is None)
+                break
+            except Exception as exc:  # pylint: disable=broad-except
+                logger.debug("Failed to launch %s: %s", browser_type.name, exc)
+
+        if not browser:
+            raise PlaywrightNotInstalled("Could not launch any Playwright browser. "
+                                         "Please run 'playwright install --with-deps webkit'")
+
+        context = browser.new_context()
+        try:
+            page = context.new_page()
+            page.on("console", on_console)
+            page.goto(auth_url, wait_until="networkidle", timeout=TIMEOUT_MS)
+            _fill_credentials(page, email, password)
+            return get_code(page, scheme)
+        except (playwright_sync.TimeoutError, RuntimeError) as e:
+            logger.exception("Headless OAuth failed: %s", e)
+            check_for_error(page)
+            raise HeadlessOAuthError(
+                "Headless OAuth failed: could not capture authorization code.",
+                url=page.url, html=page.content(), logs=console_logs
+            ) from e
+        finally:
+            browser.close()
@@ -71,8 +71,8 @@ class PSACarController(metaclass=Singleton):
                logger.exception(
                    "Can't connect to mqtt broker your are not connected to internet or PSA MQTT server is "
                    "down !")
-            except ConfigException:
-                logger.error("start_remote_control failed redo otp config")
+            except ConfigException as e:
+                logger.error("start_remote_control failed redo otp config: %s", e)

    def load_app(self) -> bool:
        my_logger(handler_level=int(self.args.debug))
@@ -15,9 +15,10 @@ logger = logging.getLogger(__name__)

 def get_oauth_config_layout(redirect_url):
    return dbc.Row(dbc.Col(md=12, lg=2, className="m-3", children=[
-        dbc.Row(html.H2('Connection to PSA')),
+        dbc.Row(html.H2('Connection to PSA (manual fallback)')),
        dbc.Row(className="ms-2", children=[
            html.Div(html.P([
+                "Automatic login failed. Complete the OAuth flow manually:", html.Br(),
                html.A("1. Click here", href=redirect_url, target="_blank"), html.Br(),
                "2. Complete the login procedure there too until you see 'LOGIN SUCCESSFUL'", html.Br(),
                "3. Open your browser's DevTools (F12) and then the click on 'Network' tab", html.Br(),
@@ -6,6 +6,7 @@ from dash.exceptions import PreventUpdate
 from flask import request

 from psa_car_controller.psa.otp.otp import new_otp_session
+from psa_car_controller.psa.setup.headless_oauth import HeadlessOAuthError, get_oauth_code_headless
 from psa_car_controller.psacc.application.car_controller import PSACarController
 from psa_car_controller.psa.setup.app_decoder import InitialSetup
 from psa_car_controller.common.mylogger import LOG_FILE
@@ -135,16 +136,61 @@ def connectPSA(n_clicks, app_name, email, password, countrycode):  # pylint: dis
        try:
            global INITIAL_SETUP
            INITIAL_SETUP = InitialSetup(app_name, email, password, countrycode)
-            redirect_uri = parse.quote(INITIAL_SETUP.psacc.manager.generate_redirect_url())
-            return dbc.Alert(["Success !", html.A(" Go to login",
-                                                  href=f"{request.url_root}config_connect?url={redirect_uri}")],
-                             color="success")
+            auth_url = INITIAL_SETUP.psacc.manager.generate_redirect_url()
+            scheme = INITIAL_SETUP.psacc.manager.redirect_uri.split("://")[0]
        except Exception as e:
-            res = str(e)
            logger.exception(e)
-            return dbc.Alert(res, color="danger")
-    else:
-        return ""
+            return dbc.Alert(str(e), color="danger")
+
+        # Attempt automatic headless OAuth
+        try:
+            code = get_oauth_code_headless(auth_url, email, password, scheme)
+            INITIAL_SETUP.connect(code)
+            return dbc.Alert(
+                ["Login successful! ", html.A("Go to OTP config", href=request.url_root + "config_otp")],
+                color="success"
+            )
+        except HeadlessOAuthError as e:
+            redirect_uri = parse.quote(auth_url)
+            return dbc.Alert(
+                [
+                    html.P("Automatic login failed. Please complete manually: "),
+                    html.A("Go to login", href=f"{request.url_root}config_connect?url={redirect_uri}"),
+                    html.Hr(),
+                    html.P("Debug information (please include in GitHub issue):"),
+                    dbc.Label("Last URL:"),
+                    dbc.Input(value=e.url, readonly=True, style={"margin-bottom": "10px"}),
+                    dbc.Label("Console Logs:"),
+                    dbc.Textarea(
+                        value="\n".join(e.logs),
+                        style={
+                            "height": "100px",
+                            "font-family": "monospace",
+                            "font-size": "12px",
+                            "margin-bottom": "10px"
+                        },
+                        readonly=True,
+                    ),
+                    dbc.Label("HTML Content:"),
+                    dbc.Textarea(
+                        value=e.html,
+                        style={"height": "200px", "font-family": "monospace", "font-size": "12px"},
+                        readonly=True,
+                    ),
+                ],
+                color="warning"
+            )
+        except Exception as e:  # pylint: disable=broad-except
+            logger.warning("Headless OAuth failed (%s), falling back to manual flow", e)
+
+        # Manual fallback
+        redirect_uri = parse.quote(auth_url)
+        return dbc.Alert(
+            ["Automatic login failed. Please complete manually: ",
+             html.A("Go to login", href=f"{request.url_root}config_connect?url={redirect_uri}")],
+            color="warning"
+        )
+    return ""


@dash_app.callback(
@@ -28,6 +28,7 @@ argparse = "^1.4.0"
 geojson = "^2.5.0"
 reverse-geocode = "^1.4.1"
 androguard = "^4.1.2"
+playwright = ">=1.40.0"
 pycryptodomex = "3.23.0"  # Pin version as they introduce breaking changes in minor versions
 pydantic = "^1.9.0"
 "ruamel.yaml" = ">=0.15.0"
@@ -54,7 +55,7 @@ pre-commit = "^2.17.0"
 coverage = "^6.3.2"
 deepdiff = "^5.7.0"
 greenery = "^3.3.5"
-autopep8 = "2.0.4"
+autopep8 = "2.3.2"
 pylint = "3.3.6"

 [build-system]