fix(longbridge): send expired_at and fail-closed on refresh errors

src/quant_platform_kit/longbridge/auth.py

@@ -5,11 +5,16 @@
 import hmac
 import json
 import time
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from typing import Any, Callable
+from urllib.parse import urlencode
 
 from quant_platform_kit.cloud import get_secret_store, get_secret_store_rw
 
+# Vendor default for Legacy Access Token lifetime when requesting a refresh.
+# https://open.longportapp.com/docs/refresh-token-api.md
+DEFAULT_REFRESHED_TOKEN_LIFETIME_DAYS = 90
+
 
 def fetch_token_from_secret(
     project_id: str,
@@ -54,6 +59,25 @@ def _format_expiry(expiry_timestamp: float) -> str:
     return datetime.fromtimestamp(expiry_timestamp, timezone.utc).isoformat()
 
 
+def _format_refresh_expired_at(
+    *,
+    now: datetime | None = None,
+    lifetime_days: int = DEFAULT_REFRESHED_TOKEN_LIFETIME_DAYS,
+) -> str:
+    """ISO-8601/RFC3339 timestamp for the *new* token's expiry (vendor-required query param)."""
+    base = now if now is not None else datetime.now(timezone.utc)
+    if base.tzinfo is None:
+        base = base.replace(tzinfo=timezone.utc)
+    expiry = base + timedelta(days=int(lifetime_days))
+    # Match vendor docs example (...Z) and official SDK Rfc3339 output.
+    return expiry.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+
+
+def _refresh_query_params(*, expired_at: str) -> str:
+    """Query string used both in the request URL and in the HMAC params slot."""
+    return urlencode({"expired_at": expired_at})
+
+
 def refresh_token_if_needed(
     current_token: str,
     *,
@@ -62,6 +86,7 @@ def refresh_token_if_needed(
     app_key: str | None,
     app_secret: str | None,
     refresh_threshold_days: int = 30,
+    refreshed_token_lifetime_days: int = DEFAULT_REFRESHED_TOKEN_LIFETIME_DAYS,
     requests_module: Any | None = None,
     secret_client_factory: Callable[[], Any] | None = None,
 ) -> str:
@@ -83,29 +108,41 @@ def refresh_token_if_needed(
     if requests_module is None:
         import requests as requests_module
 
+    expired_at = _format_refresh_expired_at(lifetime_days=refreshed_token_lifetime_days)
+    params = _refresh_query_params(expired_at=expired_at)
     headers = {
         "X-Api-Key": app_key,
         "Authorization": current_token,
         "X-Timestamp": str(int(time.time())),
         "Content-Type": "application/json; charset=utf-8",
     }
-    headers["X-Api-Signature"] = _longport_sign("GET", "/v1/token/refresh", headers, "", "", app_secret)
+    headers["X-Api-Signature"] = _longport_sign("GET", "/v1/token/refresh", headers, params, "", app_secret)
     response = requests_module.get(
-        "https://openapi.longportapp.com/v1/token/refresh",
+        f"https://openapi.longportapp.com/v1/token/refresh?{params}",
         headers=headers,
         timeout=15,
     ).json()
     if response.get("code") != 0:
+        code = response.get("code")
+        message = response.get("message") or "unknown error"
         if expiry_timestamp is not None and expiry_timestamp <= now:
-            code = response.get("code")
-            message = response.get("message") or "unknown error"
             raise RuntimeError(
                 f"LongPort token in secret '{secret_name}' expired at {_format_expiry(expiry_timestamp)}; "
                 f"refresh failed with code {code}: {message}"
             )
-        return current_token
-
-    new_token = response["data"]["token"]
+        # Fail closed while still pre-expiry so broken refresh cannot sit silent until 401003.
+        raise RuntimeError(
+            f"LongPort token refresh failed for secret '{secret_name}' "
+            f"(token still within expiry window); code {code}: {message}"
+        )
+
+    data = response.get("data") or {}
+    new_token = data.get("token")
+    if not isinstance(new_token, str) or not new_token.strip():
+        raise RuntimeError(
+            f"LongPort token refresh for secret '{secret_name}' returned an empty token payload"
+        )
+    new_token = new_token.strip()
     get_secret_store_rw().update_secret(secret_name, new_token, project_id=project_id)
 
     return new_token

tests/test_longbridge_auth.py

@@ -7,8 +7,14 @@
 import types
 import unittest
 from unittest.mock import patch
+from urllib.parse import parse_qs, urlparse
 
-from quant_platform_kit.longbridge.auth import build_contexts, fetch_token_from_secret, refresh_token_if_needed
+from quant_platform_kit.longbridge.auth import (
+    DEFAULT_REFRESHED_TOKEN_LIFETIME_DAYS,
+    build_contexts,
+    fetch_token_from_secret,
+    refresh_token_if_needed,
+)
 
 
 class FakeSecretStore:
@@ -50,9 +56,22 @@ def destroy_latest_secret(self, secret_name: str, *, project_id: str | None = No
         self.destroyed.append(f"projects/{project_id}/secrets/{secret_name}/versions/3")
 
 
-class FakeRequests:
-    @staticmethod
-    def get(url, headers, timeout):
+class RecordingRequests:
+    """Captures the refresh GET URL/headers and returns a successful body."""
+
+    last_url: str | None = None
+    last_headers: dict | None = None
+
+    @classmethod
+    def reset(cls) -> None:
+        cls.last_url = None
+        cls.last_headers = None
+
+    @classmethod
+    def get(cls, url, headers, timeout):
+        cls.last_url = url
+        cls.last_headers = dict(headers)
+
         class Response:
             @staticmethod
             def json():
@@ -103,24 +122,36 @@ def test_refresh_token_if_needed_returns_same_token_when_far_from_expiry(self) -
         self.assertEqual(refreshed, token)
 
     @patch("quant_platform_kit.longbridge.auth.get_secret_store_rw")
-    def test_refresh_token_if_needed_persists_new_token(self, mock_get_store_rw) -> None:
+    def test_refresh_token_if_needed_persists_new_token_with_expired_at(self, mock_get_store_rw) -> None:
         payload = {"exp": 1}
         encoded = base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")).decode("utf-8").rstrip("=")
         token = f"aaa.{encoded}.bbb"
         fake_store = FakeSecretStoreReadWrite(token)
         mock_get_store_rw.return_value = fake_store
+        RecordingRequests.reset()
 
         refreshed = refresh_token_if_needed(
             token,
             project_id="demo",
             secret_name="token",
             app_key="key",
             app_secret="secret",
-            requests_module=FakeRequests,
+            requests_module=RecordingRequests,
+            refreshed_token_lifetime_days=DEFAULT_REFRESHED_TOKEN_LIFETIME_DAYS,
         )
 
         self.assertEqual(refreshed, "new-token")
         self.assertEqual(fake_store.created_parent, "projects/demo/secrets/token")
+        self.assertIsNotNone(RecordingRequests.last_url)
+        parsed = urlparse(RecordingRequests.last_url)
+        self.assertEqual(parsed.path, "/v1/token/refresh")
+        query = parse_qs(parsed.query)
+        self.assertIn("expired_at", query)
+        expired_at = query["expired_at"][0]
+        self.assertTrue(expired_at.endswith("Z"))
+        self.assertRegex(expired_at, r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$")
+        # Signature must cover the same encoded query string used in the URL.
+        self.assertIn("Signature=", RecordingRequests.last_headers["X-Api-Signature"])
 
     def test_refresh_token_if_needed_raises_clear_error_when_expired_and_refresh_fails(self) -> None:
         payload = {"exp": 1}
@@ -140,22 +171,26 @@ def test_refresh_token_if_needed_raises_clear_error_when_expired_and_refresh_fai
         self.assertIn("longport_token_sg", str(context.exception))
         self.assertIn("refresh failed with code 401003", str(context.exception))
 
-    def test_refresh_token_if_needed_returns_same_token_when_refresh_fails_but_token_not_expired(self) -> None:
+    def test_refresh_token_if_needed_fail_closed_when_refresh_fails_but_token_not_expired(self) -> None:
         payload = {"exp": int(time.time()) + 86400}
         encoded = base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")).decode("utf-8").rstrip("=")
         token = f"aaa.{encoded}.bbb"
 
-        refreshed = refresh_token_if_needed(
-            token,
-            project_id="demo",
-            secret_name="token",
-            app_key="key",
-            app_secret="secret",
-            refresh_threshold_days=30,
-            requests_module=FakeFailedRequests,
-        )
+        with self.assertRaises(RuntimeError) as context:
+            refresh_token_if_needed(
+                token,
+                project_id="demo",
+                secret_name="longport_token_paper",
+                app_key="key",
+                app_secret="secret",
+                refresh_threshold_days=30,
+                requests_module=FakeFailedRequests,
+            )
 
-        self.assertEqual(refreshed, token)
+        message = str(context.exception)
+        self.assertIn("longport_token_paper", message)
+        self.assertIn("still within expiry window", message)
+        self.assertIn("401003", message)
 
     def test_refresh_token_if_needed_raises_clear_error_when_expired_and_app_credentials_missing(self) -> None:
         payload = {"exp": 1}