From 43e7d12779659365faf92becd744af9f5cc12d76 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 9 Oct 2026 14:15:45 +0800 Subject: [PATCH] fix(longbridge): send expired_at and fail-closed on refresh errors Legacy /v1/token/refresh requires expired_at; omit it and soft-return the stale token let pre-expiry failures sit silent until 401003. Include the query in the HMAC params slot, default new-token lifetime to 90 days, and raise on any non-zero refresh response while the token is still valid. --- src/quant_platform_kit/longbridge/auth.py | 53 ++++++++++++++--- tests/test_longbridge_auth.py | 69 +++++++++++++++++------ 2 files changed, 97 insertions(+), 25 deletions(-) diff --git a/src/quant_platform_kit/longbridge/auth.py b/src/quant_platform_kit/longbridge/auth.py index 6f810080..16ac9538 100644 --- a/src/quant_platform_kit/longbridge/auth.py +++ b/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 diff --git a/tests/test_longbridge_auth.py b/tests/test_longbridge_auth.py index 6ac2c1c4..26a97931 100644 --- a/tests/test_longbridge_auth.py +++ b/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,12 +122,13 @@ 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, @@ -116,11 +136,22 @@ def test_refresh_token_if_needed_persists_new_token(self, mock_get_store_rw) -> 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}