fix: wire cached-only read-only broker reconciliation

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,

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 = "<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:

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"]