Fix web login: migrate to /api/v2/auth/web/login + push approval

Trade Republic deprecated the legacy /api/v1/auth/web/login endpoint
(all requests now return HTTP 426 CLIENT_VERSION_OUTDATED regardless of
User-Agent). The current web app uses /api/v2/auth/web/login with:

  * x-tr-platform: web
  * x-tr-app-version: 15.7.0  (web track, distinct from the Android version)
  * x-tr-device-info: base64(JSON) browser fingerprint
  * x-aws-waf-token: same value as the aws-waf-token cookie

The v2 flow also no longer issues a numeric code: the user approves the
login via a push notification in the TR mobile app. The web app polls
GET /api/v2/auth/web/login/processes/{processId} until the push is
acknowledged.

Changes:
  * pytr/api.py: add TR_WEB_APP_VERSION / TR_WEB_USER_AGENT / TR_WEB_LOGIN_PATH
    constants (overridable via env), _get_device_id() persisting a stable
    UUID at ~/.pytr/device_id, _build_device_info_header(), _auth_headers()
    helper, and await_web_login_approval() polling loop.
  * initiate_weblogin() now POSTs to v2 with the new headers, polls for
    push approval, and returns 0 to signal the v2 flow.
  * complete_weblogin('') is a no-op for v2 (approval already happened).
  * pytr/account.py: when initiate_weblogin() returns 0, skip the
    code/SMS prompts and call complete_weblogin('') directly.

Refs: #250

Author: twannooitmeer

pytr/account.py

@@ -61,23 +61,28 @@ def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwrigh
         except ValueError as e:
             log.fatal(str(e))
             sys.exit(1)
-        request_time = time.time()
-        print("Enter the code you received to your mobile app as a notification.")
-        print(f"Enter nothing if you want to receive the (same) code as SMS. (Countdown: {countdown})")
-        code = input("Code: ")
-        if code == "":
-            countdown = countdown - (time.time() - request_time)
-            for remaining in range(int(countdown)):
-                print(
-                    f"Need to wait {int(countdown - remaining)} seconds before requesting SMS...",
-                    end="\r",
-                )
-                time.sleep(1)
-            print()
-            tr.resend_weblogin()
-            code = input("SMS requested. Enter the confirmation code:")
-        tr.complete_weblogin(code)
-        log.info("Logged in.")
+        if countdown == 0:
+            # v2 push-approve flow: initiate_weblogin() already polled and got approval.
+            tr.complete_weblogin("")
+            log.info("Logged in.")
+        else:
+            request_time = time.time()
+            print("Enter the code you received to your mobile app as a notification.")
+            print(f"Enter nothing if you want to receive the (same) code as SMS. (Countdown: {countdown})")
+            code = input("Code: ")
+            if code == "":
+                countdown = countdown - (time.time() - request_time)
+                for remaining in range(int(countdown)):
+                    print(
+                        f"Need to wait {int(countdown - remaining)} seconds before requesting SMS...",
+                        end="\r",
+                    )
+                    time.sleep(1)
+                print()
+                tr.resend_weblogin()
+                code = input("SMS requested. Enter the confirmation code:")
+            tr.complete_weblogin(code)
+            log.info("Logged in.")
 
     log.debug(get_settings(tr))
     return tr

pytr/api.py

@@ -21,7 +21,9 @@
 # SOFTWARE.
 
 import asyncio
+import base64
 import json
+import os
 import pathlib
 import re
 import ssl
@@ -45,11 +47,24 @@
 BASE_DIR = home / ".pytr"
 CREDENTIALS_FILE = BASE_DIR / "credentials"
 COOKIES_FILE = BASE_DIR / "cookies.txt"
+DEVICE_ID_FILE = BASE_DIR / "device_id"
+
+# Web app values captured from app.traderepublic.com on 2026-05-29.
+# These can be overridden via environment variables if Trade Republic bumps them.
+TR_WEB_APP_VERSION = os.environ.get("PYTR_TR_APP_VERSION", "15.7.0")
+TR_WEB_USER_AGENT = os.environ.get(
+    "PYTR_TR_USER_AGENT",
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36",
+)
+TR_WEB_LOGIN_PATH = "/api/v2/auth/web/login"
 
 
 class TradeRepublicApi:
     _default_headers = {
-        "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36"
+        "User-Agent": TR_WEB_USER_AGENT,
+        "Origin": "https://app.traderepublic.com",
+        "Referer": "https://app.traderepublic.com/",
     }
     _host = "https://api.traderepublic.com"
     _waf_login_url = "https://app.traderepublic.com/login"
@@ -183,6 +198,46 @@ def _fetch_waf_token_playwright(self, timeout_ms: int = 30000):
             self.log.warning("AWS WAF token not acquired. Value is None.")
         return token
 
+    def _get_device_id(self) -> str:
+        """Return a stable per-install device UUID, creating it on first use."""
+        try:
+            return DEVICE_ID_FILE.read_text().strip()
+        except FileNotFoundError:
+            BASE_DIR.mkdir(parents=True, exist_ok=True)
+            device_id = uuid.uuid4().hex + uuid.uuid4().hex  # 64 hex chars, matches web app
+            DEVICE_ID_FILE.write_text(device_id)
+            return device_id
+
+    def _build_device_info_header(self) -> str:
+        """Build the base64(JSON) x-tr-device-info payload the web app sends."""
+        payload = {
+            "stableDeviceId": self._get_device_id(),
+            "model": "Apple Macintosh",
+            "browser": "Chrome",
+            "browserVersion": "148.0.0.0",
+            "os": "Mac OS",
+            "osVersion": "10.15.7",
+            "timezone": "Europe/Amsterdam",
+            "timezoneOffset": -120,
+            "screen": "1800x1169x30",
+            "preferredLanguages": ["en", "en-US"],
+            "numberOfCores": 12,
+            "deviceMemory": 16,
+        }
+        raw = json.dumps(payload, separators=(",", ":")).encode("utf-8")
+        return base64.b64encode(raw).decode("ascii")
+
+    def _auth_headers(self) -> Dict[str, str]:
+        """Headers required by /api/v2/auth/web/login."""
+        headers = {
+            "x-tr-platform": "web",
+            "x-tr-app-version": TR_WEB_APP_VERSION,
+            "x-tr-device-info": self._build_device_info_header(),
+        }
+        if self._waf_token:
+            headers["x-aws-waf-token"] = self._waf_token
+        return headers
+
     def _set_waf_cookie(self, token: str):
         """Set the aws-waf-token cookie on the web session."""
         cookie = Cookie(
@@ -222,8 +277,9 @@ def initiate_weblogin(self):
             self.log.warning("No WAF token available.")
 
         r = self._websession.post(
-            f"{self._host}/api/v1/auth/web/login",
+            f"{self._host}{TR_WEB_LOGIN_PATH}",
             json={"phoneNumber": self.phone_no, "pin": self.pin},
+            headers=self._auth_headers(),
         )
         self.log.debug(f"Web login returned: {r.status_code}")
         r.raise_for_status()
@@ -237,20 +293,78 @@ def initiate_weblogin(self):
                 raise ValueError(str(err))
             else:
                 raise ValueError("processId not in reponse")
-        return int(j["countdownInSeconds"]) + 1
+        self.log.info("Web login keys: %s", sorted(j.keys()))
+
+        # v2 web login uses push-to-approve in the TR mobile app: no SMS/code.
+        # Poll the process endpoint until the push is approved (or rejected/expired).
+        self.await_web_login_approval()
+        return 0
+
+    def await_web_login_approval(self, timeout_s: int = 180, interval_s: float = 2.0):
+        """Block until the TR mobile-app push for this login process is approved."""
+        if not self._process_id:
+            raise ValueError("Initiate web login first.")
+        url = f"{self._host}/api/v2/auth/web/login/processes/{self._process_id}"
+        deadline = time.time() + timeout_s
+        print(
+            f"Approve the login in your Trade Republic mobile app "
+            f"(waiting up to {timeout_s}s)...",
+            flush=True,
+        )
+        first_body_logged = False
+        while time.time() < deadline:
+            r = self._websession.get(url, headers=self._auth_headers())
+            if r.status_code == 200:
+                try:
+                    j = r.json()
+                except ValueError:
+                    j = {}
+                if not first_body_logged:
+                    self.log.info("Login poll keys: %s", sorted(j.keys()) if j else "<empty>")
+                    first_body_logged = True
+                state = str(j.get("state") or j.get("status") or "").upper()
+                if state in ("APPROVED", "COMPLETED", "SUCCESS", "OK", "DONE"):
+                    self.save_websession()
+                    self.log.info("Push approved.")
+                    return
+                if state in ("REJECTED", "DECLINED", "FAILED", "EXPIRED"):
+                    raise ValueError(f"Login process {state.lower()}: {j}")
+                # State unknown but session cookie present? Treat as approved.
+                for c in self._websession.cookies:
+                    if c.name in ("tr_session",) and c.value:
+                        self.save_websession()
+                        self.log.info("Session cookie detected, login complete.")
+                        return
+            elif r.status_code in (401, 403, 404, 410):
+                raise ValueError(
+                    f"Login process rejected or expired ({r.status_code}): {r.text[:200]}"
+                )
+            else:
+                self.log.warning("Login poll %s: %s", r.status_code, r.text[:200])
+            time.sleep(interval_s)
+        raise TimeoutError("Push approval not received within timeout.")
 
     def resend_weblogin(self):
         r = self._websession.post(
-            f"{self._host}/api/v1/auth/web/login/{self._process_id}/resend",
-            headers=self._default_headers,
+            f"{self._host}{TR_WEB_LOGIN_PATH}/{self._process_id}/resend",
+            headers=self._auth_headers(),
         )
         r.raise_for_status()
 
     def complete_weblogin(self, verify_code):
         if not self._process_id and not self._websession:
             raise ValueError("Initiate web login first.")
 
-        r = self._websession.post(f"{self._host}/api/v1/auth/web/login/{self._process_id}/{verify_code}")
+        # v2 push flow: approval already happened during initiate_weblogin().
+        # An empty verify_code signals nothing left to do.
+        if not verify_code:
+            self.save_websession()
+            return
+
+        r = self._websession.post(
+            f"{self._host}{TR_WEB_LOGIN_PATH}/{self._process_id}/{verify_code}",
+            headers=self._auth_headers(),
+        )
         r.raise_for_status()
         self.save_websession()