From b17fe8e2d7b6567c0bc44f2552d74645991f518f Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 5 Sep 2026 15:12:43 +0800 Subject: [PATCH] fix: wire cached-only read-only broker reconciliation Co-Authored-By: Codex --- application/firstrade_client.py | 80 +++++++++++++-- main.py | 21 +++- tests/test_firstrade_client.py | 175 ++++++++++++++++++++++++++++++++ tests/test_request_handling.py | 35 +++++++ 4 files changed, 298 insertions(+), 13 deletions(-) diff --git a/application/firstrade_client.py b/application/firstrade_client.py index 3668130..67d19fe 100644 --- a/application/firstrade_client.py +++ b/application/firstrade_client.py @@ -10,9 +10,11 @@ import os from dataclasses import dataclass from hashlib import sha256 +from math import isfinite from pathlib import Path from time import time from typing import Any, Callable +from urllib.parse import urlsplit from application.account_payload_utils import flatten_values, float_or_none from application.state_persistence import GcsStateStore @@ -52,7 +54,10 @@ class FirstradeCredentials: debug: bool = False @classmethod - def from_env(cls, env: Callable[[str, str | None], str | None] = os.getenv) -> "FirstradeCredentials": + def from_env( + cls, env: Callable[[str, str | None], str | None] = os.getenv, + *, include_login_credentials: bool = True, + ) -> "FirstradeCredentials": def _get_credential(secret_name: str, env_var: str) -> str: try: from quant_platform_kit.cloud import get_secret_store @@ -62,15 +67,15 @@ def _get_credential(secret_name: str, env_var: str) -> str: return env(env_var, "") or "" username = _get_credential("firstrade-username", "FIRSTRADE_USERNAME") - password = _get_credential("firstrade-password", "FIRSTRADE_PASSWORD") + password = _get_credential("firstrade-password", "FIRSTRADE_PASSWORD") if include_login_credentials else "" return cls( username=username.strip(), password=password, - pin=env("FIRSTRADE_PIN", "") or "", - email=env("FIRSTRADE_MFA_EMAIL", "") or "", - phone=env("FIRSTRADE_MFA_PHONE", "") or "", - mfa_secret=_get_credential("firstrade-mfa-secret", "FIRSTRADE_MFA_SECRET"), - mfa_code=env("FIRSTRADE_MFA_CODE", "") or "", + pin=(env("FIRSTRADE_PIN", "") or "") if include_login_credentials else "", + email=(env("FIRSTRADE_MFA_EMAIL", "") or "") if include_login_credentials else "", + phone=(env("FIRSTRADE_MFA_PHONE", "") or "") if include_login_credentials else "", + mfa_secret=_get_credential("firstrade-mfa-secret", "FIRSTRADE_MFA_SECRET") if include_login_credentials else "", + mfa_code=(env("FIRSTRADE_MFA_CODE", "") or "") if include_login_credentials else "", cookie_dir=env("FIRSTRADE_COOKIE_DIR", ".runtime/firstrade-cookies") or ".runtime/firstrade-cookies", reuse_session=(env("FIRSTRADE_REUSE_SESSION", "false") or "").strip().lower() == "true", @@ -264,6 +269,62 @@ def connect(self) -> "FirstradeBrokerClient": self._save_session_cache(cookie_dir) return self + def connect_read_only(self) -> "FirstradeBrokerClient": + """Reuse cached authentication for account reads; never log in or change the cache.""" + if ( + self.live_trading_enabled or self.session is not None or self.account_data is not None + or not self.credentials.username.strip() or not self.credentials.reuse_session + ): + raise FirstradeSafetyError("Firstrade cached-only connection requires a fresh non-trading client.") + payload = self._load_session_cache(Path(self.credentials.cookie_dir)) + if not payload or not isinstance(payload.get("access-token"), str) or not payload["access-token"]: + raise FirstradePlatformError("Firstrade cached session unavailable.") + from firstrade.account import FTAccountData, FTSession + + session_factory = self._session_factory or FTSession + account_data_factory = self._account_data_factory or FTAccountData + session = session_factory(username="", password="", save_session=False, debug=False) + transport = session.session + transport.trust_env = False + original_request = transport.request + + def read_only_request(method, url, **kwargs): + parsed = urlsplit(url) + if str(method).lower() != "get" or ( + parsed.scheme != "https" or parsed.netloc != "api3x.firstrade.com" + or parsed.path not in { + "/private/userinfo", "/private/acct_list", "/private/balances", + "/private/positions", "/private/order_status", + } + ): + raise FirstradeSafetyError("Firstrade read-only request denied.") + kwargs.update(timeout=(5, 15), allow_redirects=False) + response = original_request(method, url, **kwargs) + if not 200 <= response.status_code < 300: + raise FirstradePlatformError("Firstrade account read unavailable.") + body = response.json() + if not isinstance(body, (dict, list)) or (isinstance(body, dict) and body.get("error")): + raise FirstradePlatformError("Firstrade account read unavailable.") + return response + + transport.request = read_only_request + try: + session.build_session_from_tokens(payload) + account_data = account_data_factory(session) + except Exception: + transport.close() + raise FirstradePlatformError("Firstrade cached session unavailable.") from None + self.session, self.account_data = session, account_data + self.session_reused = True + return self + + def close(self) -> None: + session, self.session = self.session, None + self.account_data = None + self.session_reused = False + if session is not None: + session.session.close() + def _build_session(self, session_factory: Callable[..., Any], cookie_dir: Path) -> Any: return session_factory( username=self.credentials.username, @@ -306,9 +367,10 @@ def _is_valid_session_cache_payload(self, payload: Any) -> bool: except (TypeError, ValueError): return False ttl = max(1, int(self.credentials.session_cache_ttl_seconds or 1)) - if saved_at <= 0.0 or (time() - saved_at) > ttl: + age = time() - saved_at + if not isfinite(saved_at) or saved_at <= 0.0 or not 0 <= age <= ttl: return False - return bool(payload.get("ftat") and payload.get("sid")) + return all(isinstance(payload.get(key), str) and payload[key].strip() for key in ("ftat", "sid")) def _try_cached_session( self, diff --git a/main.py b/main.py index 1ca96a8..77170c1 100644 --- a/main.py +++ b/main.py @@ -68,10 +68,14 @@ app = Flask(__name__) register_health_endpoint(app) # GET /health /healthz -# There is intentionally no production builder here: the existing client login -# path can create session artifacts. An explicitly injected ephemeral client is -# required before this private read-only endpoint can contact the provider. -READ_ONLY_BROKER_RECONCILIATION_CLIENT_BUILDER = None +def _build_read_only_reconciliation_client(): + return FirstradeBrokerClient( + FirstradeCredentials.from_env(include_login_credentials=False), + live_trading_enabled=False, + ).connect_read_only() + + +READ_ONLY_BROKER_RECONCILIATION_CLIENT_BUILDER = _build_read_only_reconciliation_client _REDACTED = "" _TELEGRAM_BOT_PATH_RE = re.compile(r"(?i)(/bot)([^/\s]+)") @@ -454,6 +458,7 @@ def _handle_reconciliation(): if not reconciliation_enabled(os.getenv): return jsonify({"status": "blocked", "reason": "broker_reconciliation_disabled"}), 503 + client = None try: settings = _runtime_settings() runtime_target = settings.runtime_target @@ -463,6 +468,8 @@ def _handle_reconciliation(): env_reader=os.getenv, ) requested_account = str(os.getenv("FIRSTRADE_ACCOUNT") or "").strip() + if not requested_account: + raise FirstradeReconciliationUnavailable("Firstrade reconciliation requires an explicit account.") client = READ_ONLY_BROKER_RECONCILIATION_CLIENT_BUILDER() observations = collect_read_only_reconciliation_observations( client, @@ -483,6 +490,12 @@ def _handle_reconciliation(): return jsonify({"status": "blocked", "reason": "broker_reconciliation_unavailable"}), 503 except Exception: return jsonify({"status": "blocked", "reason": "broker_reconciliation_unavailable"}), 503 + finally: + if client is not None: + try: + client.close() + except Exception: + pass # Cleanup must not expose provider details or change the safe response. def _paper_command_consumer_runtime_is_isolated(settings: PlatformRuntimeSettings) -> bool: diff --git a/tests/test_firstrade_client.py b/tests/test_firstrade_client.py index c92eed2..60c65c7 100644 --- a/tests/test_firstrade_client.py +++ b/tests/test_firstrade_client.py @@ -341,3 +341,178 @@ def test_order_reads_preserve_supported_complete_payloads(wrapper, rows): client.account_data = SimpleNamespace(get_orders=lambda *_args, **_kwargs: payload) assert client.get_orders("test-account") == rows + + +def _cached_read_only_client(tmp_path, monkeypatch, *, saved_at=1000.0, account_factory=HeaderCheckingAccountData): + import application.firstrade_client as module + import requests + + monkeypatch.setattr(module, "time", lambda: 1001.0) + calls = [] + + class CachedSession(FakeSession): + def __init__(self, **kwargs): + super().__init__(**kwargs) + self.closed = False + self.session = SimpleNamespace(headers={}, request=self.request, close=self.close) + + def build_session_from_tokens(self, payload): + self.session.headers.update(payload) + + def request(self, method, url, **kwargs): + calls.append((method, url, kwargs)) + response = requests.Response() + response.status_code = 200 + response._content = b'{}' + return response + + def close(self): + self.closed = True + + def login(self): + pytest.fail("read-only connection must never authenticate") + + def login_two(self, _code): + pytest.fail("read-only connection must never send MFA") + + credentials = FirstradeCredentials( + username="synthetic-user", password="", reuse_session=True, + cookie_dir=str(tmp_path / "absent"), debug=True, + ) + store = FakeStateStore() + client = FirstradeBrokerClient( + credentials, session_factory=CachedSession, account_data_factory=account_factory, + session_cache_store=store, + ) + store.payloads[client._session_state_key()] = { + "saved_at": saved_at, "ftat": "synthetic-ftat", "sid": "synthetic-sid", + "access-token": "synthetic-access", + } + return client, store, calls + + +def test_read_only_connection_reuses_cache_without_auth_or_writes(tmp_path, monkeypatch): + client, store, calls = _cached_read_only_client(tmp_path, monkeypatch) + assert client.connect_read_only() is client + assert client.session_reused + assert client.live_trading_enabled is False + assert client.session.kwargs["debug"] is False + assert client.session.kwargs["save_session"] is False + assert not client.session.kwargs.get("password") + assert not client.session.kwargs.get("mfa_secret") + assert not client.session.kwargs.get("profile_path") + assert client.session.session.trust_env is False + assert store.writes == 0 + assert not (tmp_path / "absent").exists() + assert calls == [] + session = client.session + client.close() + assert session.closed + assert client.session is None and client.account_data is None + + +@pytest.mark.parametrize("saved_at", [None, 0, "bad", float("nan"), float("inf"), 1002, -1, 1]) +def test_read_only_connection_rejects_missing_or_expired_cache(tmp_path, monkeypatch, saved_at): + client, store, calls = _cached_read_only_client(tmp_path, monkeypatch, saved_at=saved_at) + from dataclasses import replace + client.credentials = replace(client.credentials, session_cache_ttl_seconds=20) + with pytest.raises(FirstradePlatformError, match="cached session unavailable"): + client.connect_read_only() + assert client.session is None and client.account_data is None + assert not (tmp_path / "absent").exists() + assert store.writes == 0 and calls == [] + + +def test_read_only_failed_account_read_preserves_cache_and_closes(tmp_path, monkeypatch): + seen = [] + def fail(session): + seen.append(session) + raise RuntimeError("synthetic provider error must not escape") + client, store, calls = _cached_read_only_client(tmp_path, monkeypatch, account_factory=fail) + original = dict(store.payloads) + with pytest.raises(FirstradePlatformError, match="^Firstrade cached session unavailable\\.$"): + client.connect_read_only() + assert seen[0].closed + assert client.session is None and client.account_data is None + assert store.payloads == original and store.writes == 0 + assert calls == [] + + +@pytest.mark.parametrize("method,url", [ + ("post", "https://api3x.firstrade.com/private/stock_order"), + ("get", "https://api3x.firstrade.com/private/cancel_order"), + ("get", "https://api3x.firstrade.com/public/quote"), + ("get", "https://other.example/private/userinfo"), + ("get", "http://api3x.firstrade.com/private/userinfo"), +]) +def test_read_only_transport_rejects_other_requests(tmp_path, monkeypatch, method, url): + client, store, calls = _cached_read_only_client(tmp_path, monkeypatch) + client.connect_read_only() + with pytest.raises(FirstradeSafetyError): + client.session.session.request(method, url) + assert calls == [] + client.close() + + +def test_read_only_transport_is_bounded_without_redirects(tmp_path, monkeypatch): + client, store, calls = _cached_read_only_client(tmp_path, monkeypatch) + client.connect_read_only() + client.session.session.request("get", "https://api3x.firstrade.com/private/userinfo") + assert calls[0][2] == {"timeout": (5, 15), "allow_redirects": False} + client.close() + + +@pytest.mark.parametrize("failure", [None, "http", "redirect", "auth", "non-json", "timeout"]) +def test_read_only_real_sdk_stops_on_first_read_failure(tmp_path, monkeypatch, failure): + import json + import requests + from urllib.parse import urlsplit + client, store, _ = _cached_read_only_client(tmp_path, monkeypatch) + client._session_factory = None + client._account_data_factory = None + calls, closed = [], [] + original_close = requests.Session.close + def close(session): + closed.append(True) + original_close(session) + def request(session, method, url, **kwargs): + calls.append(urlsplit(url).path) + assert session.trust_env is False + assert kwargs == {"timeout": (5, 15), "allow_redirects": False} + if failure == "timeout": + raise requests.Timeout("synthetic detail must not escape") + response = requests.Response() + response.status_code = {"http": 401, "redirect": 302}.get(failure, 200) + body = {"error": "synthetic-denial"} if failure == "auth" else {"error": "", "items": []} + response._content = b"not-json" if failure == "non-json" else json.dumps(body).encode() + return response + monkeypatch.setattr(requests.Session, "request", request) + monkeypatch.setattr(requests.Session, "close", close) + if failure: + with pytest.raises(FirstradePlatformError, match="^Firstrade cached session unavailable\\.$"): + client.connect_read_only() + assert calls == ["/private/userinfo"] + else: + client.connect_read_only() + assert calls == ["/private/userinfo", "/private/acct_list"] + client.close() + assert closed == [True] + assert store.writes == 0 + assert not (tmp_path / "absent").exists() + + +def test_read_only_credentials_do_not_read_password_or_mfa(monkeypatch): + import quant_platform_kit.cloud + reads = [] + def get_secret(name, **_kwargs): + reads.append(name) + assert name == "firstrade-username" + return "synthetic-user" + monkeypatch.setattr(quant_platform_kit.cloud, "get_secret_store", lambda: SimpleNamespace(get_secret=get_secret)) + def env(name, default=None): + assert name not in {"FIRSTRADE_PASSWORD", "FIRSTRADE_PIN", "FIRSTRADE_MFA_SECRET", "FIRSTRADE_MFA_CODE", "FIRSTRADE_MFA_EMAIL", "FIRSTRADE_MFA_PHONE"} + return default + credentials = FirstradeCredentials.from_env(env, include_login_credentials=False) + assert credentials.username == "synthetic-user" + assert credentials.password == credentials.mfa_secret == credentials.mfa_code == "" + assert reads == ["firstrade-username"] diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 14fc94b..0fd4a66 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -147,7 +147,12 @@ def test_reconcile_enabled_returns_redacted_blocked_receipt(monkeypatch): ), ) + closed = [] + class FakeClient: + def close(self): + closed.append(True) + def account_numbers(self): return ["account-sensitive-001"] @@ -172,6 +177,7 @@ def get_orders(self, _account, *, per_page=0): assert response.status_code == 200 payload = response.get_json() + assert closed == [True] assert payload["permits_active_lkg"] is False assert payload["expected_digests_configured"] is False assert "account-sensitive-001" not in response.get_data(as_text=True) @@ -528,3 +534,32 @@ def fake_dispatch(targets): assert response.status_code == 200 assert response.get_json()["dispatches_due"] == 0 assert observed["targets"][0]["service_name"] == "firstrade-quant-service" + + +def test_reconciliation_default_builder_uses_cached_only_client(monkeypatch): + from types import SimpleNamespace + calls = [] + credentials = object() + def from_env(**kwargs): + assert kwargs == {"include_login_credentials": False} + return credentials + def client_factory(actual, *, live_trading_enabled): + assert actual is credentials and live_trading_enabled is False + return SimpleNamespace(connect_read_only=lambda: calls.append("cached-only") or "client") + monkeypatch.setattr(main.FirstradeCredentials, "from_env", from_env) + monkeypatch.setattr(main, "FirstradeBrokerClient", client_factory) + assert main.READ_ONLY_BROKER_RECONCILIATION_CLIENT_BUILDER() == "client" + assert calls == ["cached-only"] + + +def test_reconciliation_missing_account_stops_before_client(monkeypatch): + from types import SimpleNamespace + monkeypatch.setenv("FIRSTRADE_BROKER_RECONCILIATION_ENABLED", "true") + monkeypatch.delenv("FIRSTRADE_ACCOUNT", raising=False) + monkeypatch.setattr(main, "_runtime_settings", lambda: SimpleNamespace(runtime_target=object())) + monkeypatch.setattr(main, "validate_reconciliation_preconditions", lambda **_kwargs: None) + calls = [] + monkeypatch.setattr(main, "READ_ONLY_BROKER_RECONCILIATION_CLIENT_BUILDER", lambda: calls.append(True)) + response = main.app.test_client().post("/reconcile") + assert response.status_code == 503 + assert calls == []