From 469977682200e86c5e1ca311cdd04ec23b1ea4cb Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:18:11 +0800 Subject: [PATCH 1/4] feat: guard daily renewal and alert on missed deadlines Co-Authored-By: Codex --- README.md | 10 +- application/firstrade_client.py | 12 +- application/session_check_service.py | 51 ++++- application/session_renewal_guard.py | 277 ++++++++++++++++++++++++++ docs/session-renewal.md | 145 ++++++++++++++ main.py | 29 +-- tests/test_forced_session_renewal.py | 108 +++++++++- tests/test_session_renewal_guard.py | 286 +++++++++++++++++++++++++++ 8 files changed, 889 insertions(+), 29 deletions(-) create mode 100644 application/session_renewal_guard.py create mode 100644 docs/session-renewal.md create mode 100644 tests/test_session_renewal_guard.py diff --git a/README.md b/README.md index 8c8259b..6a7f955 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,14 @@ Direct runtime profiles can usually run from market history or portfolio state. 4. Confirm rollback steps and artifact versions. 5. Enable scheduled or live execution only after the above checks are clear. +## Daily session renewal + +The existing Scheduler-backed renewal endpoint now claims a dedicated GCS cycle +before login and blocks replay after failures or unknown outcomes. MFA and device +verification stop for owner recovery. A broker-free deadline endpoint checks +completion by 21:15 UTC. See [renewal and alert setup](docs/session-renewal.md) for +notification delivery semantics and the separately approved watchdog rollout. + ## Read-only account facts The existing **Diagnose cached-balance stage failure** manual workflow can inspect saved @@ -81,7 +89,7 @@ The existing Runtime Target Lifecycle manual entry with `metadata_only=true` ins `POST /account-facts-sync` is a cached-session-only balance snapshot path. It is disabled unless `FIRSTRADE_ACCOUNT_FACTS_SYNC_ENABLED=true`; it does not log in, refresh credentials, schedule itself, submit orders, or send notifications. It publishes only provider-reported equity and, when present, provider `cash_balance`; it does not read positions, so a positions read failure or incomplete positions cannot block the balance snapshot. Buying power and positions are not used to calculate assets or cash. -The existing daily Execution Report Heartbeat also invokes this snapshot path once through the already-configured Scheduler bridge. Its independent snapshot job checks the exact service URL, POST route and OIDC audience, never creates or updates the bridge, and requires a new successful Scheduler completion plus the matching Cloud Run request HTTP 200. The endpoint returns 200 only after the publisher accepts its acknowledgement; its INFO log may be suppressed and is not required. Dispatch acceptance alone is not success, and independent website latest/identity/history readback remains a separate acceptance check. The existing daily schedule is unchanged; report and notification jobs remain independent. A manual `snapshot_only=true` run checks this path without reports or notifications. Expired/missing cached authentication fails the snapshot job; `/session-renew` remains manual and is never invoked automatically. No Cloud Run image update is needed for this workflow-only integration. +The existing daily Execution Report Heartbeat also invokes this snapshot path once through the already-configured Scheduler bridge. Its independent snapshot job checks the exact service URL, POST route and OIDC audience, never creates or updates the bridge, and requires a new successful Scheduler completion plus the matching Cloud Run request HTTP 200. The endpoint returns 200 only after the publisher accepts its acknowledgement; its INFO log may be suppressed and is not required. Dispatch acceptance alone is not success, and independent website latest/identity/history readback remains a separate acceptance check. The existing daily schedule is unchanged; report and notification jobs remain independent. A manual `snapshot_only=true` run checks this path without reports or notifications. Expired/missing cached authentication fails the snapshot job; `/session-renew` is never invoked by this snapshot workflow. No Cloud Run image update is needed for this workflow-only integration. The Cloud Run service must remain IAM-protected. The caller needs `roles/run.invoker` and must send its Google-signed ID token in `X-Serverless-Authorization`; the separate application token goes in `Authorization: Bearer …`. See [Cloud Run service-to-service authentication](https://cloud.google.com/run/docs/authenticating/service-to-service). Neither token is a Firstrade credential. diff --git a/application/firstrade_client.py b/application/firstrade_client.py index 876364e..bc27699 100644 --- a/application/firstrade_client.py +++ b/application/firstrade_client.py @@ -75,7 +75,7 @@ class FirstradeCredentials: @classmethod def from_env( cls, env: Callable[[str, str | None], str | None] = os.getenv, - *, include_login_credentials: bool = True, + *, include_login_credentials: bool = True, include_mfa_credentials: bool = True, ) -> "FirstradeCredentials": def _get_credential(secret_name: str, env_var: str) -> str: try: @@ -90,11 +90,11 @@ def _get_credential(secret_name: str, env_var: str) -> str: return cls( username=username.strip(), password=password, - 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 "", + pin=(env("FIRSTRADE_PIN", "") or "") if include_login_credentials and include_mfa_credentials else "", + email=(env("FIRSTRADE_MFA_EMAIL", "") or "") if include_login_credentials and include_mfa_credentials else "", + phone=(env("FIRSTRADE_MFA_PHONE", "") or "") if include_login_credentials and include_mfa_credentials else "", + mfa_secret=_get_credential("firstrade-mfa-secret", "FIRSTRADE_MFA_SECRET") if include_login_credentials and include_mfa_credentials else "", + mfa_code=(env("FIRSTRADE_MFA_CODE", "") or "") if include_login_credentials and include_mfa_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", diff --git a/application/session_check_service.py b/application/session_check_service.py index c2b84c2..c0e6279 100644 --- a/application/session_check_service.py +++ b/application/session_check_service.py @@ -5,7 +5,7 @@ import json import os from collections.abc import Callable, Mapping -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import datetime, timezone from typing import Any @@ -19,6 +19,7 @@ FirstradeBrokerClient, FirstradeCredentials, FirstradeSafetyError, + FirstradeMfaRequired, is_live_trading_enabled, mask_account_id, ) @@ -462,29 +463,65 @@ def run_session_check( return result +class SessionRenewalFailure(FirstradeSafetyError): + """Fixed, non-sensitive failure classification for unattended renewal.""" + + def __init__(self, reason: str, stage: str): + super().__init__(reason) + self.reason = reason + self.stage = stage + + def run_forced_session_renewal( *, credentials: FirstradeCredentials | None = None, client_factory: Callable[..., FirstradeBrokerClient] = FirstradeBrokerClient, env_reader: Callable[[str, str | None], str | None] = os.getenv, ) -> dict[str, Any]: - """Manual session renewal: fresh login, select account, read balances, then persist cache. + """One renewal attempt: fresh login, validate account, then persist shared cache. Bypasses (never reads, writes, or deletes) the monthly session-check marker. The client is always built with live trading disabled and only account-read methods are called. A successful result requires the shared cache write to complete successfully. """ - resolved_credentials = credentials or FirstradeCredentials.from_env(env_reader) + resolved_credentials = credentials or FirstradeCredentials.from_env( + env_reader, include_mfa_credentials=False, + ) + # Never consume a seed, code, PIN or OTP destination in unattended renewal. + resolved_credentials = replace( + resolved_credentials, pin="", email="", phone="", mfa_secret="", mfa_code="", debug=False, + ) if not resolved_credentials.reuse_session or not resolved_credentials.persist_session_cache: raise FirstradeSafetyError("Forced session renewal requires session reuse and cache persistence.") client = client_factory(resolved_credentials, live_trading_enabled=False) if getattr(client, "live_trading_enabled", False): raise FirstradeSafetyError("Forced session renewal requires a non-trading client.") - client.connect(force_login=True, defer_cache_save=True) - account = client.select_account(env_reader("FIRSTRADE_ACCOUNT", "") or None) - balances = client.get_balances(account) - client.save_verified_session_cache() + from firstrade.exceptions import LoginError, LoginRequestError, LoginResponseError + + try: + client.connect(force_login=True, defer_cache_save=True) + except FirstradeMfaRequired: + raise + except LoginRequestError as exc: + reason = "auth_rejected" if exc.status_code in {401, 403} else "auth_outcome_unknown" + raise SessionRenewalFailure(reason, "auth") from None + except LoginResponseError: + raise SessionRenewalFailure("auth_rejected", "auth") from None + except LoginError: + # SDK 0.0.39 raises LoginError for device/OTP challenges without a method. + raise FirstradeMfaRequired("Manual authentication required.") from None + except Exception: + raise SessionRenewalFailure("auth_outcome_unknown", "auth") from None + try: + account = client.select_account(env_reader("FIRSTRADE_ACCOUNT", "") or None) + balances = client.get_balances(account) + except Exception: + raise SessionRenewalFailure("account_read_failed", "account_read") from None + try: + client.save_verified_session_cache() + except Exception: + raise SessionRenewalFailure("cache_write_outcome_unknown", "cache_write") from None print( "Firstrade forced session renewal " f"session_reused={bool(getattr(client, 'session_reused', False))} " diff --git a/application/session_renewal_guard.py b/application/session_renewal_guard.py new file mode 100644 index 0000000..1a0e808 --- /dev/null +++ b/application/session_renewal_guard.py @@ -0,0 +1,277 @@ +"""Single-writer renewal state and a broker-free daily deadline check.""" + +from __future__ import annotations + +import json +import os +import uuid +from datetime import datetime, timedelta, timezone +from typing import Callable + +from application.firstrade_client import FirstradeMfaRequired +from application.session_check_service import SessionRenewalFailure +from application.state_persistence import GcsStateStore, StatePersistenceError + +ATTEMPT_KEY = "maintenance/session-renew-attempt.json" +ALERT_PREFIX = "maintenance/session-renew-alerts" + + +def utcnow() -> datetime: + return datetime.now(timezone.utc) + + +def cycle_at(now: datetime) -> datetime: + now = now.astimezone(timezone.utc) + start = now.replace(hour=21, minute=0, second=0, microsecond=0) + return start if now >= start else start - timedelta(days=1) + + +def parse_time(value: str) -> datetime: + result = datetime.fromisoformat(value.replace("Z", "+00:00")) + if result.tzinfo is None: + raise ValueError("Timezone required") + return result.astimezone(timezone.utc) + + +class RenewalStore: + """Generation-pinned reads and compare-and-swap; no lease expiry or takeover.""" + + def __init__(self, state: GcsStateStore): + self.state = state + + @classmethod + def from_env(cls): + bucket = os.getenv("FIRSTRADE_GCS_STATE_BUCKET", "").strip() + if not bucket: + raise StatePersistenceError("Renewal state unavailable") + return cls(GcsStateStore(bucket, os.getenv("FIRSTRADE_STATE_PREFIX", "firstrade-platform"))) + + def _blob(self): + from google.cloud import storage + + uri = self.state._object_uri(ATTEMPT_KEY) + bucket, _, name = uri[5:].partition("/") + return storage.Client().bucket(bucket).blob(name) + + def read(self): + from google.api_core.exceptions import NotFound + + blob = self._blob() + try: + blob.reload(timeout=15, retry=None) + except NotFound: + return None, 0 + generation = int(blob.generation) + # A concurrent replacement is a failed read, never interpreted as absence. + value = json.loads(blob.download_as_text( + if_generation_match=generation, timeout=15, retry=None, + )) + if not isinstance(value, dict) or value.get("schema") != 1: + raise StatePersistenceError("Invalid renewal state") + if value.get("status") not in {"running", "completed", "manual_required", "ready"}: + raise StatePersistenceError("Invalid renewal state") + cycle = parse_time(value["cycle"]) + if cycle != cycle_at(cycle): + raise StatePersistenceError("Invalid renewal cycle") + if value["status"] == "completed" and parse_time(value["completed_at"]) < cycle: + raise StatePersistenceError("Invalid renewal completion") + return value, generation + + def write(self, value, generation): + from google.api_core.exceptions import PreconditionFailed + + blob = self._blob() + try: + blob.upload_from_string(json.dumps(value), content_type="application/json", + if_generation_match=generation, timeout=15, retry=None) + except PreconditionFailed: + return None + # The SDK populates generation from the upload response. Never do a fresh reload: + # it might observe another writer and grant authority over that writer's record. + if not blob.generation: + raise StatePersistenceError("Renewal write outcome unknown") + return int(blob.generation) + + def claim_alert(self, cycle: str, kind: str): + date = parse_time(cycle).strftime("%Y-%m-%d") + key = f"{ALERT_PREFIX}/{date}/{kind}.json" + if self.state.create_json(key, { + "schema": 1, "cycle": cycle, "kind": kind, "attempted_at": utcnow().isoformat(), + "delivery": "unknown", + }): + return True + receipt = self.state.read_json(key) or {} + return "already_acknowledged" if receipt.get("delivery") == "acknowledged" else "unconfirmed" + + def finish_alert(self, cycle, kind, delivery): + date = parse_time(cycle).strftime("%Y-%m-%d") + self.state.write_json(f"{ALERT_PREFIX}/{date}/{kind}.json", { + "schema": 1, "cycle": cycle, "kind": kind, "delivery": delivery, + "finished_at": utcnow().isoformat(), + }) + + +def emit_event(kind: str, cycle: str, reason: str, stage: str): + # Deliberately exclude exceptions, broker responses, accounts and auth data. + print(json.dumps({"severity": "INFO" if kind == "recovered" else "ERROR", + "event": "session_renewal", "kind": kind, "cycle": cycle, + "reason": reason, "stage": stage}), flush=True) + + +def send_alert(kind: str, cycle: str, reason: str, stage: str) -> bool: + """Use only the existing injected notification credentials, never broker secrets.""" + from notifications.telegram import build_sender + + token = os.getenv("TELEGRAM_TOKEN") + chat = os.getenv("QSL_GLOBAL_TELEGRAM_CHAT_ID") or os.getenv("GLOBAL_TELEGRAM_CHAT_ID") + if not token or not chat: + return False + titles = {"failure": "Firstrade 会话续期失败", "deadline": "Firstrade 会话续期未在 21:15 UTC 前完成", + "recovered": "Firstrade 会话续期已恢复"} + action = "请检查人工认证或存储状态；系统不会自动重试登录。" + if kind == "deadline": + action = "请检查调度、续期记录和存储；不要重复触发登录。" + elif kind == "recovered": + action = "账户验证与共享缓存保存已完成。" + return build_sender(token, chat)( + f"{titles[kind]}\n周期：{cycle}\n原因：{reason}\n阶段：{stage}\n{action}" + ) is True + + +def alert(store, *, kind, cycle, reason, stage, sender=send_alert): + emit_event(kind, cycle, reason, stage) + try: + claimed = store.claim_alert(cycle, kind) if store is not None else None + if claimed in {"already_acknowledged", "unconfirmed"}: + if claimed == "unconfirmed": + print(json.dumps({"severity": "ERROR", "event": "session_renewal_notification", + "kind": kind, "cycle": cycle, "delivery": "unconfirmed", + "dedup_available": True}), flush=True) + return claimed + except Exception: + # Storage failure must not also silence the notification. In this degraded + # case each incoming request attempts one send; it still never logs in. + claimed = None + try: + acknowledged = sender(kind, cycle, reason, stage) is True + except Exception: + acknowledged = False + delivery = "acknowledged" if acknowledged else "unconfirmed" + if claimed is True: + try: + store.finish_alert(cycle, kind, delivery) + except Exception: + pass # Never resend a notification whose delivery may have succeeded. + print(json.dumps({"severity": "INFO" if acknowledged else "ERROR", + "event": "session_renewal_notification", "kind": kind, "cycle": cycle, + "delivery": delivery, "dedup_available": claimed is not None}), flush=True) + return delivery + + +def run_guarded_renewal( + renew: Callable, *, store=None, schedule_time=None, now=None, clock=utcnow, sender=send_alert, +): + """At most one login for a daily cycle; any unknown outcome needs owner recovery.""" + now = now or clock() + cycle = cycle_at(now).isoformat() + if schedule_time: + try: + scheduled = parse_time(schedule_time) + if scheduled != cycle_at(scheduled) or scheduled > now: + raise ValueError("Invalid renewal schedule time") + if scheduled != cycle_at(now): + return {"ok": False, "status": "stale_delivery", "session_renewed": False}, 200 + except (ValueError, TypeError): + return {"ok": False, "error": "invalid_schedule_time"}, 400 + try: + store = store or RenewalStore.from_env() + previous, generation = store.read() + if previous: + old_cycle = parse_time(previous["cycle"]) + status = previous["status"] + if status not in {"completed", "ready"}: + return {"ok": False, "status": "not_replayed", "session_renewed": False, + "manual_recovery_required": True}, 200 + if old_cycle >= parse_time(cycle): + return {"ok": status == "completed", "status": "duplicate_delivery", + "session_renewed": False}, 200 + attempt = {"schema": 1, "cycle": cycle, "status": "running", + "execution_id": uuid.uuid4().hex, "started_at": now.isoformat(), + "manual_recovery_required": True} + claimed_generation = store.write(attempt, generation) + if claimed_generation is None: + return {"ok": False, "status": "concurrent_delivery", "session_renewed": False}, 200 + except Exception: + delivery = alert(store, kind="failure", cycle=cycle, + reason="guard_state_unavailable", stage="guard", sender=sender) + return {"ok": False, "error": "guard_state_unavailable", "alert_delivery": delivery}, 503 + + try: + result = renew() + if result.get("ok") is not True or result.get("session_renewed") is not True: + raise SessionRenewalFailure("renewal_result_invalid", "validation") + except Exception as exc: + reason, stage = "renewal_outcome_unknown", "auth" + if isinstance(exc, FirstradeMfaRequired): + reason = "mfa_or_device_verification_required" + elif isinstance(exc, SessionRenewalFailure): + reason, stage = exc.reason, exc.stage + failed = {**attempt, "status": "manual_required", "reason": reason, "stage": stage, + "finished_at": clock().isoformat()} + try: + store.write(failed, claimed_generation) + except Exception: + pass # The durable running record still prevents replay, including on future days. + delivery = alert(store, kind="failure", cycle=cycle, reason=reason, stage=stage, sender=sender) + return {"ok": False, "error": reason, "stage": stage, "manual_recovery_required": True, + "alert_delivery": delivery}, 409 if isinstance(exc, FirstradeMfaRequired) else 503 + + completed = {**attempt, "status": "completed", "manual_recovery_required": False, + "completed_at": clock().isoformat()} + try: + if store.write(completed, claimed_generation) is None: + raise StatePersistenceError("Completion conflict") + except Exception: + delivery = alert(store, kind="failure", cycle=cycle, reason="completion_outcome_unknown", + stage="completion", sender=sender) + return {"ok": False, "error": "completion_outcome_unknown", + "manual_recovery_required": True, "alert_delivery": delivery}, 503 + print(json.dumps({"severity": "INFO", "event": "session_renewal", "kind": "completed", + "cycle": cycle, "execution_id": attempt["execution_id"]}), flush=True) + if previous and previous["status"] == "ready": + alert(store, kind="recovered", cycle=cycle, reason="renewal_completed", stage="completion", sender=sender) + return {**result, "cycle": cycle, "status": "completed"}, 200 + + +def check_renewal_deadline(*, store=None, schedule_time=None, now=None, sender=send_alert): + """Check the 21:15 UTC cutoff without loading credentials or invoking renewal.""" + now = now or utcnow() + cycle = cycle_at(now) + if schedule_time: + try: + scheduled = parse_time(schedule_time) + if scheduled != cycle_at(scheduled) + timedelta(minutes=15) or scheduled > now: + raise ValueError("Invalid deadline schedule time") + cycle = cycle_at(scheduled) + if cycle != cycle_at(now): + return {"ok": False, "status": "stale_delivery"}, 200 + except (ValueError, TypeError): + return {"ok": False, "error": "invalid_schedule_time"}, 400 + cutoff = cycle + timedelta(minutes=15) + if now < cutoff: + return {"ok": True, "status": "not_due"}, 200 + reason = "not_completed_by_2115_utc" + try: + store = store or RenewalStore.from_env() + record, _ = store.read() + if (record and record["cycle"] == cycle.isoformat() and record["status"] == "completed" + and cycle <= parse_time(record["completed_at"]) <= cutoff): + return {"ok": True, "status": "completed_by_deadline", "cycle": cycle.isoformat()}, 200 + except Exception: + reason = "deadline_state_unavailable" + delivery = alert(store, kind="deadline", cycle=cycle.isoformat(), reason=reason, + stage="deadline", sender=sender) + # HTTP 200 acknowledges an observed incident, not successful renewal. Failed + # notification transport remains non-2xx so Scheduler failure monitoring can see it. + return {"ok": False, "error": reason, "cycle": cycle.isoformat(), + "alert_delivery": delivery}, 200 if delivery in {"acknowledged", "already_acknowledged"} else 503 diff --git a/docs/session-renewal.md b/docs/session-renewal.md new file mode 100644 index 0000000..13a8797 --- /dev/null +++ b/docs/session-renewal.md @@ -0,0 +1,145 @@ +# Daily session renewal + +`POST /session-renew` uses the existing IAM-protected Cloud Run service and the +existing daily 21:00 UTC Cloud Scheduler job. It remains disabled unless +`FIRSTRADE_RUN_SESSION_CHECK_ON_HTTP=true`. It never runs a strategy or submits +orders. Successful renewal means account selection, balance read and shared +cache persistence completed; it does not establish trading eligibility or prove +an independent process can use the cache. + +## Duplicate delivery and manual recovery + +The runtime uses the existing `FIRSTRADE_GCS_STATE_BUCKET` and +`FIRSTRADE_STATE_PREFIX`. The dedicated object +`maintenance/session-renew-attempt.json` contains only a schema version, UTC +cycle, random execution ID, timestamps, status and fixed failure classification. +It does not contain authentication material or account data. Deployment must +first confirm that this path is unused, and that no other implementation writes +it. Do not reuse any trading or monthly session-check marker. + +A generation-pinned read followed by a generation-matched write admits exactly +one attempt. A missing object is created with generation match zero. No login +starts if state cannot be read or the claim outcome is unknown. A completed +cycle suppresses duplicates; only a later daily cycle can start a new attempt. +`X-CloudScheduler-ScheduleTime` identifies the original scheduled cycle across +retries and midnight. It must be 21:00 UTC; old deliveries are acknowledged +without login, and future/malformed values are rejected. The header is not an +authentication mechanism: retain Cloud Run IAM and protected ingress. Requests +without the header use the most recent 21:00 UTC cycle, not a new manual attempt. + +`running` has **no timeout takeover**. A crash, authentication failure, MFA or +new-device challenge, account-read failure, or uncertain cache/completion write +blocks subsequent automatic logins, including on later days. The route returns +`ok=false` on failure. A duplicate delivery can return HTTP 200 to acknowledge +that no new login was attempted; this is not a renewal success signal. + +The renewal credential loader never reads MFA seed/code/PIN/email/phone fields. +Even caller-supplied MFA material is removed before constructing the client, and +SDK debug output is disabled. No challenge response, OTP request, seed-based +code generation, or login retry is attempted. Other independently authorized +client uses keep their existing behavior. + +Recovery belongs to the owner. After inspecting the original attempt and +resolving authentication or storage outside this endpoint, the owner may +separately authorize a generation-matched transition of this object to `ready`. +Keep its existing `schema`, `cycle` and execution history; never delete the +object or backdate the cycle. Confirm there is no in-flight attempt first. +The next **later** daily cycle can then try once. No reset endpoint or automatic +reset is provided. A successful first cycle after `ready` sends one recovery +notification. Code rollback does not undo authentication side effects or clear +this maintenance state. + +## Failure and deadline notifications + +The runtime uses only the existing injected `TELEGRAM_TOKEN` and +`QSL_GLOBAL_TELEGRAM_CHAT_ID` (or `GLOBAL_TELEGRAM_CHAT_ID`). It does not resolve +broker secrets to send alerts. Messages contain fixed reason/stage, UTC cycle +and an action hint; they omit account, balance, exception text and credentials. +No new bot or token is required when this injection is already present. + +The first failure attempts one notification. Atomic claims in +`maintenance/session-renew-alerts/YYYY-MM-DD/{failure,deadline,recovered}.json` +limit each event to one send attempt across instances. An acknowledged transport +response is recorded separately from an unknown/failed delivery; it is not +proof that a person received the notification. A crash after the claim, or a +failed/unknown delivery, is not automatically resent. If storage itself cannot +claim an alert, each incoming request makes one notification attempt and emits +`dedup_available=false`; cross-instance notification deduplication is unavailable +in that degraded state. Login remains blocked. Do not configure automatic +Scheduler retries as a substitute for notification recovery. + +`POST /session-renew-deadline` reads only maintenance state and notification +configuration. It never loads broker credentials, reads session cookies, logs +in, renews, or publishes account facts. Schedule it at **21:15 UTC daily** using +the existing approved service root as OIDC audience. It requires a completed +record for that day's 21:00 cycle with `completed_at <= 21:15 UTC`. Missing, +older, running, failed and unreadable records alert; a completion after the +cutoff does not erase the missed deadline. Early calls do nothing and stale +Scheduler deliveries do not check a different cycle. The cutoff is exact; +Scheduler dispatch, logging and notification delivery may be delayed. + +Structured stdout events use `event=session_renewal` and kinds `failure`, +`deadline`, `completed`, `recovered`. Notification events use +`event=session_renewal_notification` and `delivery=acknowledged|unconfirmed`. +Observe these fixed fields instead of treating HTTP 200 as renewal completion. + +## Rollout requires separate owner approval + +A merged PR does not deploy or activate monitoring. Prepare the following +concrete items in protected deployment material and obtain owner sign-off: + +1. The reviewed image digest, current serving revision, rollback target and + completion of any concurrent cache/deployment task. One deployment owner + performs the rollout; do not deploy this branch alongside another release. +2. The exact existing bucket/prefix, unused maintenance paths, runtime identity + and object read/create/update permissions. Overwriting GCS objects also needs + `storage.objects.delete`. Reuse existing grants if sufficient; do not add + project-wide roles. Keep these paths out of automated lifecycle deletion. +3. Existing Telegram target ID and token Secret reference, with receiver testing + separately authorized. Never place either value in Git. Do not read the MFA + seed or modify any authentication secret. +4. A distinct watchdog Scheduler job: POST to `/session-renew-deadline`, schedule + `15 21 * * *`, timezone `UTC`, deadline `180s`, retries zero, existing approved + Scheduler identity, OIDC audience equal to the service root. Do not change + the existing renewal job's `0 21 * * *`, 600-second deadline or zero retries. +5. An independent Cloud Monitoring policy for failed watchdog delivery and + unconfirmed notification events. A watchdog in the same service cannot send + its own Telegram alert while the service is down. Activation is incomplete + until this separate failure path has a verified notification channel. An + empty channel inventory is a missing configuration, not a working alert. + +Example Cloud Monitoring log-based policy input (replace placeholders only in +protected deployment material; this is not an apply command): + +```json +{ + "displayName": "Session renewal watchdog delivery failure", + "combiner": "OR", + "conditions": [{ + "displayName": "Watchdog or notification transport failed", + "conditionMatchedLog": { + "filter": "(resource.type=\"cloud_scheduler_job\" AND resource.labels.job_id=\"\" AND jsonPayload.@type=\"type.googleapis.com/google.cloud.scheduler.logging.AttemptFinished\" AND severity>=ERROR) OR (resource.type=\"cloud_run_revision\" AND resource.labels.service_name=\"\" AND jsonPayload.event=\"session_renewal_notification\" AND jsonPayload.delivery=\"unconfirmed\")" + } + }], + "alertStrategy": {"notificationRateLimit": {"period": "3600s"}, "autoClose": "86400s"}, + "notificationChannels": ["projects//notificationChannels/"], + "enabled": true +} +``` + +This policy reports failed dispatch/transport, not a successful business cycle. +Also verify that the watchdog job stays enabled; a deleted or paused monitor has +no failed execution to alert on. Keep the existing external runtime guard as an +additional observer. Do not send test messages, trigger either endpoint, deploy, +change IAM or secrets merely to validate a PR. + +Offline acceptance covers concurrent duplicate delivery, stale delivery across +midnight, state-write uncertainty, crash persistence, MFA/auth rejection, +future-day blocking, no MFA secret access, the deadline cutoff and notification +acknowledgement. After an approved rollout, separately authorize any test +notification and observe natural cycles. Do not trigger login for acceptance +without explicit authorization. + +References: [Scheduler deduplication](https://docs.cloud.google.com/scheduler/docs/overview), +[GCS preconditions](https://docs.cloud.google.com/storage/docs/request-preconditions), +[log-based alerting](https://docs.cloud.google.com/logging/docs/alerting/log-based-alerts). diff --git a/main.py b/main.py index e0b5104..455f5f8 100644 --- a/main.py +++ b/main.py @@ -67,6 +67,7 @@ validate_reconciliation_preconditions, ) from application.session_check_service import run_forced_session_renewal, run_session_check +from application.session_renewal_guard import run_guarded_renewal, check_renewal_deadline from notifications.telegram import build_sender from quant_platform_kit.common.runtime_reports import ( append_runtime_report_error, @@ -1152,21 +1153,23 @@ def probe(): @app.post("/session-renew") def session_renew(): - """Manual-only read-only session renewal (Cloud Run IAM / internal ingress like /probe). - - Logs in once, persists the session cache, reads balances. Never runs strategy or orders, - and never reads or changes the monthly session-check marker. - """ + """IAM-protected, duplicate-safe daily renewal. Never runs strategy or orders.""" if not _flag("FIRSTRADE_RUN_SESSION_CHECK_ON_HTTP"): return jsonify({"ok": False, "error": "session_check_on_http_disabled"}), 403 - try: - return jsonify(run_forced_session_renewal()) - except FirstradeMfaRequired: - app.logger.warning("session_renew mfa_required") - return jsonify({"ok": False, "error": "mfa_required"}), 409 - except FirstradePlatformError as exc: - app.logger.warning("session_renew failed exc_type=%s", type(exc).__name__) - return jsonify({"ok": False, "error": "session_renew_failed"}), 503 + result, status = run_guarded_renewal( + run_forced_session_renewal, + schedule_time=request.headers.get("X-CloudScheduler-ScheduleTime"), + ) + return jsonify(result), status + + +@app.post("/session-renew-deadline") +def session_renew_deadline(): + """IAM-protected watchdog; reads maintenance state and alerts, never logs in.""" + result, status = check_renewal_deadline( + schedule_time=request.headers.get("X-CloudScheduler-ScheduleTime"), + ) + return jsonify(result), status @app.post("/monitor-dispatch") diff --git a/tests/test_forced_session_renewal.py b/tests/test_forced_session_renewal.py index e37a887..6a2a159 100644 --- a/tests/test_forced_session_renewal.py +++ b/tests/test_forced_session_renewal.py @@ -206,6 +206,15 @@ def no_orders(*_a, **_k): monkeypatch.setattr(main, "run_strategy_cycle", no_orders) monkeypatch.setattr(main, "run_session_check", no_orders) monkeypatch.setattr(main, "_run_strategy_cycle_with_report", no_orders) + from unittest.mock import Mock + from application.session_renewal_guard import run_guarded_renewal + store = Mock() + store.read.return_value = (None, 0) + store.write.return_value = 1 + store.claim_alert.return_value = True + monkeypatch.setattr(main, "run_guarded_renewal", lambda renew, **kwargs: run_guarded_renewal( + renew, store=store, sender=lambda *_: True, **kwargs, + )) return main @@ -218,7 +227,7 @@ def test_route_disabled_without_http_flag(http_client, monkeypatch): def test_route_runs_forced_renewal_only(http_client, monkeypatch): monkeypatch.setenv("FIRSTRADE_RUN_SESSION_CHECK_ON_HTTP", "true") calls = [] - monkeypatch.setattr(http_client, "run_forced_session_renewal", lambda: calls.append(1) or {"ok": True}) + monkeypatch.setattr(http_client, "run_forced_session_renewal", lambda: calls.append(1) or {"ok": True, "session_renewed": True}) response = http_client.app.test_client().post("/session-renew") assert response.status_code == 200 and calls == [1] @@ -232,9 +241,104 @@ def mfa(): monkeypatch.setattr(http_client, "run_forced_session_renewal", mfa) response = http_client.app.test_client().post("/session-renew") assert response.status_code == 409 - assert response.get_json()["error"] == "mfa_required" + assert response.get_json()["error"] == "mfa_or_device_verification_required" def test_route_is_post_only(http_client): response = http_client.app.test_client().get("/session-renew") assert response.status_code == 405 + + +def test_renewal_does_not_read_mfa_secret_or_configuration(monkeypatch): + from quant_platform_kit import cloud + requested = [] + + class SecretStore: + def get_secret(self, name, **kwargs): + assert name in {"firstrade-username", "firstrade-password"} + requested.append(name) + return "synthetic" + + def env(name, default=None): + assert not any(part in name for part in ("MFA", "PIN")) + return {"FIRSTRADE_REUSE_SESSION": "true", "FIRSTRADE_PERSIST_SESSION_CACHE": "true"}.get(name, default) + + monkeypatch.setattr(cloud, "get_secret_store", lambda: SecretStore()) + created = [] + run_forced_session_renewal(client_factory=_factory(created), env_reader=env) + assert requested == ["firstrade-username", "firstrade-password"] + assert not created[0].credentials.mfa_secret + + +def test_renewal_strips_injected_mfa_material_and_debug(): + created = [] + run_forced_session_renewal(credentials=_credentials(pin="synthetic", email="x@example.invalid", + phone="synthetic", mfa_secret="synthetic", mfa_code="synthetic", debug=True), + client_factory=_factory(created), env_reader=lambda *_: None) + creds = created[0].credentials + assert not any((creds.pin, creds.email, creds.phone, creds.mfa_secret, creds.mfa_code, creds.debug)) + + +@pytest.mark.parametrize("sdk_error,reason", [ + ("LoginError", "mfa"), ("LoginResponseError", "auth_rejected"), + ("LoginRequestError", "auth_rejected"), +]) +def test_sdk_challenges_never_retry_or_read_accounts(sdk_error, reason): + from firstrade import exceptions + from application.session_check_service import SessionRenewalFailure + calls = [] + class Client(_RecordingClient): + def connect(self, **kwargs): + calls.append("login") + error = getattr(exceptions, sdk_error) + raise error(403 if sdk_error == "LoginRequestError" else "private-synthetic") + def select_account(self, requested=None): + pytest.fail("account read after auth challenge") + with pytest.raises(FirstradeMfaRequired if reason == "mfa" else SessionRenewalFailure) as caught: + run_forced_session_renewal(credentials=_credentials(), client_factory=Client) + assert calls == ["login"] and "private-synthetic" not in str(caught.value) + + +def test_deadline_http_route_never_reaches_broker(http_client, monkeypatch): + from application import session_renewal_guard as guard + from datetime import datetime, timezone + from unittest.mock import Mock + def no_broker(*args, **kwargs): + pytest.fail("broker or credentials reached") + monkeypatch.setattr(http_client, "run_forced_session_renewal", no_broker) + monkeypatch.setattr(FirstradeCredentials, "from_env", no_broker) + store = Mock() + store.read.return_value = (None, 0) + store.claim_alert.return_value = True + monkeypatch.setattr(http_client, "check_renewal_deadline", lambda **kwargs: guard.check_renewal_deadline( + store=store, sender=lambda *_: True, now=datetime(2026, 1, 2, 21, 15, tzinfo=timezone.utc), **kwargs)) + response = http_client.app.test_client().post("/session-renew-deadline") + assert response.status_code == 200 + assert response.get_json()["error"] == "not_completed_by_2115_utc" + + +@pytest.mark.parametrize("challenge", ["mfa", "new_device"]) +def test_real_sdk_challenge_boundary_never_submits_second_factor(tmp_path, challenge): + from firstrade.account import FTSession + from application.firstrade_client import FirstradeBrokerClient + from types import SimpleNamespace + calls = [] + class Session(FTSession): + def _load_cookies(self): + return "" + def _request(self, method, url, **kwargs): + calls.append(method) + assert len(calls) <= 2, "SDK attempted MFA, OTP or login retry" + payload = {"error": "", "mfa": challenge == "mfa", "t_token": "synthetic", + "otp": [{"channel": "sms", "recipientMask": "synthetic"}]} + return SimpleNamespace(status_code=200, json=lambda: payload) + def no_accounts(_): + pytest.fail("account data reached after challenge") + def client(credentials, *, live_trading_enabled): + return FirstradeBrokerClient(credentials, live_trading_enabled=False, + session_factory=Session, account_data_factory=no_accounts) + with pytest.raises(FirstradeMfaRequired): + run_forced_session_renewal(credentials=_credentials(cookie_dir=str(tmp_path), + mfa_secret="synthetic", mfa_code="synthetic", phone="synthetic", pin="synthetic"), + client_factory=client) + assert calls == ["get", "post"] diff --git a/tests/test_session_renewal_guard.py b/tests/test_session_renewal_guard.py new file mode 100644 index 0000000..884beb7 --- /dev/null +++ b/tests/test_session_renewal_guard.py @@ -0,0 +1,286 @@ +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from copy import deepcopy +from datetime import datetime, timedelta, timezone +from threading import Lock, Event + +import pytest + +from application import session_renewal_guard as guard +from application.firstrade_client import FirstradeMfaRequired +from application.session_check_service import SessionRenewalFailure + +NOW = datetime(2026, 1, 2, 21, 0, tzinfo=timezone.utc) + + +class Store: + def __init__(self): + self.record, self.generation = None, 0 + self.alerts = {} + self.lock = Lock() + self.fail_read = False + self.fail_write = None + + def read(self): + with self.lock: + if self.fail_read: + raise RuntimeError("synthetic-private-data") + return deepcopy(self.record), self.generation + + def write(self, value, generation): + with self.lock: + if value["status"] == self.fail_write: + raise TimeoutError("synthetic-private-data") + if generation != self.generation: + return None + self.record = deepcopy(value) + self.generation += 1 + return self.generation + + def claim_alert(self, cycle, kind): + with self.lock: + key = (cycle, kind) + if key in self.alerts: + return "already_acknowledged" if self.alerts[key] == "acknowledged" else "unconfirmed" + self.alerts[key] = "unknown" + return True + + def finish_alert(self, cycle, kind, delivery): + self.alerts[cycle, kind] = delivery + + +def run(store, renew, **kwargs): + return guard.run_guarded_renewal(renew, store=store, now=NOW, clock=lambda: NOW, **kwargs) + + +def success(): + return {"ok": True, "session_renewed": True} + + +def test_duplicate_and_concurrent_delivery_only_log_in_once(): + store, entered, release = Store(), Event(), Event() + calls, alerts = [], [] + + def renew(): + calls.append(1) + entered.set() + assert release.wait(5) + return success() + + with ThreadPoolExecutor(max_workers=2) as pool: + first = pool.submit(run, store, renew, sender=lambda *x: alerts.append(x) or True) + assert entered.wait(5) + second = run(store, renew, sender=lambda *x: alerts.append(x) or True) + assert second[0]["status"] == "not_replayed" + release.set() + assert first.result()[0]["status"] == "completed" + assert run(store, renew)[0]["status"] == "duplicate_delivery" + assert calls == [1] and alerts == [] + + +def test_two_readers_of_absent_state_use_atomic_claim(): + from threading import Barrier + store, barrier = Store(), Barrier(2) + original_read = store.read + + def read(): + value = original_read() + barrier.wait(5) + return value + + store.read = read + calls = [] + with ThreadPoolExecutor(max_workers=2) as pool: + results = list(pool.map(lambda _: run(store, lambda: calls.append(1) or success()), range(2))) + assert calls == [1] + assert {r[0]["status"] for r in results} == {"completed", "concurrent_delivery"} + + +@pytest.mark.parametrize("error", [FirstradeMfaRequired("private"), + SessionRenewalFailure("auth_rejected", "auth"), TimeoutError("private")]) +def test_failures_stop_and_block_future_cycles(error, capsys): + store, calls, sent = Store(), [], [] + + def renew(): + calls.append(1) + raise error + + result, status = run(store, renew, sender=lambda *x: sent.append(x) or True) + assert status in {409, 503} + assert result["manual_recovery_required"] is True + assert store.record["status"] == "manual_required" + for offset in [0, 1, 3]: + response, _ = guard.run_guarded_renewal(renew, store=store, now=NOW + timedelta(days=offset)) + assert response["status"] == "not_replayed" + assert len(calls) == len(sent) == 1 + assert "private" not in capsys.readouterr().out + + +def test_unrecorded_failure_and_completion_timeout_keep_durable_running_claim(): + for fail_write in ["manual_required", "completed"]: + store = Store() + store.fail_write = fail_write + def renew(): + if fail_write == "manual_required": + raise TimeoutError() + return success() + assert run(store, renew, sender=lambda *_: True)[1] == 503 + assert store.record["status"] == "running" + result, _ = guard.run_guarded_renewal(lambda: pytest.fail("replayed"), store=store, + now=NOW + timedelta(days=1)) + assert result["status"] == "not_replayed" + + +@pytest.mark.parametrize("failure", ["read", "claim"]) +def test_state_failure_prevents_credential_loading_or_login(failure): + store, alerts = Store(), [] + store.fail_read = failure == "read" + store.fail_write = "running" if failure == "claim" else None + result, status = run(store, lambda: pytest.fail("login reached"), sender=lambda *x: alerts.append(x) or True) + assert status == 503 and result["error"] == "guard_state_unavailable" + assert len(alerts) == 1 + + +def test_crash_does_not_release_claim(): + store = Store() + def crash(): + raise SystemExit() + with pytest.raises(SystemExit): + run(store, crash) + assert store.record["status"] == "running" + assert run(store, lambda: pytest.fail("replayed"))[0]["status"] == "not_replayed" + + +def test_next_cycle_after_success_and_owner_recovery(): + for previous_status in ["completed", "ready"]: + store, sent = Store(), [] + store.record = {"cycle": (NOW - timedelta(days=1)).isoformat(), "status": previous_status} + result, status = run(store, success, sender=lambda *x: sent.append(x) or True) + assert status == 200 and result["status"] == "completed" + assert len(sent) == (previous_status == "ready") + if sent: + assert sent[0][0] == "recovered" + + +@pytest.mark.parametrize("header,status", [ + ("2026-01-01T21:00:00Z", 200), ("2026-01-03T21:00:00Z", 400), + ("2026-01-02T20:00:00Z", 400), ("2026-01-02T21:00:00", 400), ("bad", 400), +]) +def test_stale_future_and_invalid_schedule_never_login(header, status): + result, actual = run(Store(), lambda: pytest.fail("login reached"), schedule_time=header) + assert actual == status + assert result["ok"] is False + + +def test_schedule_header_is_stable_across_midnight(): + store = Store() + run(store, success) + result, _ = guard.run_guarded_renewal(lambda: pytest.fail("login reached"), store=store, + now=NOW + timedelta(hours=4), schedule_time=NOW.isoformat()) + assert result["status"] == "duplicate_delivery" + + +@pytest.mark.parametrize("record_kind", ["missing", "running", "failed", "yesterday", "late", "read_error", "ontime"]) +def test_deadline_reads_only_exact_cycle_and_cutoff(record_kind): + store, sent = Store(), [] + if record_kind != "missing": + store.record = {"cycle": NOW.isoformat(), "status": "completed", "completed_at": (NOW + timedelta(minutes=10)).isoformat()} + if record_kind in {"running", "failed"}: + store.record["status"] = record_kind + if record_kind == "yesterday": + store.record["cycle"] = (NOW - timedelta(days=1)).isoformat() + if record_kind == "late": + store.record["completed_at"] = (NOW + timedelta(minutes=16)).isoformat() + store.fail_read = record_kind == "read_error" + args = dict(store=store, now=NOW + timedelta(minutes=20), + schedule_time=(NOW + timedelta(minutes=15)).isoformat(), sender=lambda *x: sent.append(x) or True) + result, status = guard.check_renewal_deadline(**args) + assert status == 200 + assert result["ok"] is (record_kind == "ontime") + guard.check_renewal_deadline(**args) + assert len(sent) == (0 if record_kind == "ontime" else 1) + + +def test_deadline_not_due_and_stale_delivery(): + store = Store() + store.fail_read = True + for kwargs, expected in [ + ({"now": NOW}, "not_due"), + ({"now": NOW + timedelta(days=1, minutes=15), "schedule_time": (NOW + timedelta(minutes=15)).isoformat()}, "stale_delivery"), + ]: + result, _ = guard.check_renewal_deadline(store=store, sender=lambda *_: pytest.fail("alert reached"), **kwargs) + assert result["status"] == expected + + +def test_unconfirmed_notification_is_not_claimed_delivered_or_retried(): + store, sent = Store(), [] + args = dict(store=store, now=NOW + timedelta(minutes=15), sender=lambda *x: sent.append(x) or False) + for _ in range(2): + result, status = guard.check_renewal_deadline(**args) + assert status == 503 and result["alert_delivery"] == "unconfirmed" + assert len(sent) == 1 + + +def test_alert_storage_failure_still_attempts_notification(): + store, sent = Store(), [] + def fail(*_): + raise TimeoutError() + store.claim_alert = fail + result, _ = guard.check_renewal_deadline(store=store, now=NOW + timedelta(minutes=15), + sender=lambda *x: sent.append(x) or True) + assert result["alert_delivery"] == "acknowledged" and len(sent) == 1 + + +def test_injected_notification_only(monkeypatch): + from notifications import telegram + calls = [] + monkeypatch.setenv("TELEGRAM_TOKEN", "synthetic-token") + monkeypatch.setenv("GLOBAL_TELEGRAM_CHAT_ID", "synthetic-target") + monkeypatch.delenv("QSL_GLOBAL_TELEGRAM_CHAT_ID", raising=False) + monkeypatch.setattr(telegram, "build_sender", lambda token, chat: lambda message: calls.append((token, chat, message)) or True) + assert guard.send_alert("deadline", NOW.isoformat(), "not_completed_by_2115_utc", "deadline") + assert "21:15 UTC" in calls[0][2] + monkeypatch.delenv("TELEGRAM_TOKEN") + assert guard.send_alert("failure", NOW.isoformat(), "auth_rejected", "auth") is False + + +def test_generation_pinned_read_and_write(monkeypatch): + from google.cloud import storage + from google.api_core.exceptions import PreconditionFailed, NotFound + from application.state_persistence import GcsStateStore + calls = [] + class Blob: + generation = 7 + missing = False + conflict = False + def reload(self, **kwargs): + calls.append(("reload", kwargs)) + if self.missing: + raise NotFound("missing") + def download_as_text(self, **kwargs): + calls.append(("read", kwargs)) + return '{"schema":1,"status":"running","cycle":"2026-01-02T21:00:00+00:00"}' + def upload_from_string(self, *args, **kwargs): + calls.append(("write", kwargs)) + if self.conflict: + raise PreconditionFailed("conflict") + self.generation = 8 + blob = Blob() + class Client: + def bucket(self, name): + return self + def blob(self, name): + return blob + monkeypatch.setattr(storage, "Client", Client) + store = guard.RenewalStore(GcsStateStore("synthetic-bucket")) + value, generation = store.read() + assert generation == 7 + assert store.write(value, generation) == 8 + assert calls[1][1]["if_generation_match"] == 7 + assert calls[2][1]["if_generation_match"] == 7 + assert all(c[1]["retry"] is None for c in calls) + blob.conflict = True + assert store.write(value, generation) is None + blob.missing = True + assert store.read() == (None, 0) From 232e02af230a2575796ac7d1aa3786dcf85d055c Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:27:59 +0800 Subject: [PATCH 2/4] test: include the renewal deadline route in the contract Co-Authored-By: Codex --- tests/test_request_handling.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index a091446..0ddd688 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -88,6 +88,7 @@ def test_cloud_run_route_contracts_are_registered(): "/monitor-dispatch": ["GET", "POST"], "/probe": ["POST"], "/session-renew": ["POST"], + "/session-renew-deadline": ["POST"], "/static/": ["GET"], } From f4e99e216ce3c538424fbfc8255062d24025c7e3 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:30:24 +0800 Subject: [PATCH 3/4] fix: match renewal guard and session cache prefix defaults Co-Authored-By: Codex --- application/session_renewal_guard.py | 2 +- tests/test_session_renewal_guard.py | 11 +++++++++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/application/session_renewal_guard.py b/application/session_renewal_guard.py index 1a0e808..40a0a81 100644 --- a/application/session_renewal_guard.py +++ b/application/session_renewal_guard.py @@ -44,7 +44,7 @@ def from_env(cls): bucket = os.getenv("FIRSTRADE_GCS_STATE_BUCKET", "").strip() if not bucket: raise StatePersistenceError("Renewal state unavailable") - return cls(GcsStateStore(bucket, os.getenv("FIRSTRADE_STATE_PREFIX", "firstrade-platform"))) + return cls(GcsStateStore(bucket, os.getenv("FIRSTRADE_STATE_PREFIX", "firstrade-platform") or "firstrade-platform")) def _blob(self): from google.cloud import storage diff --git a/tests/test_session_renewal_guard.py b/tests/test_session_renewal_guard.py index 884beb7..7a94c82 100644 --- a/tests/test_session_renewal_guard.py +++ b/tests/test_session_renewal_guard.py @@ -284,3 +284,14 @@ def blob(self, name): assert store.write(value, generation) is None blob.missing = True assert store.read() == (None, 0) + + +@pytest.mark.parametrize("prefix", [None, "", "synthetic-prefix"]) +def test_guard_uses_the_same_prefix_default_as_session_cache(monkeypatch, prefix): + monkeypatch.setenv("FIRSTRADE_GCS_STATE_BUCKET", "synthetic-bucket") + if prefix is None: + monkeypatch.delenv("FIRSTRADE_STATE_PREFIX", raising=False) + else: + monkeypatch.setenv("FIRSTRADE_STATE_PREFIX", prefix) + store = guard.RenewalStore.from_env() + assert store.state.prefix == (prefix or "firstrade-platform") From 6103023c1f4c83432b4825587519ac6b2a7c6e2f Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:34:21 +0800 Subject: [PATCH 4/4] fix: classify unavailable renewal credentials before login Co-Authored-By: Codex --- application/session_check_service.py | 5 +++++ tests/test_forced_session_renewal.py | 24 ++++++++++++++++++++++++ 2 files changed, 29 insertions(+) diff --git a/application/session_check_service.py b/application/session_check_service.py index 409a719..ffe51df 100644 --- a/application/session_check_service.py +++ b/application/session_check_service.py @@ -18,6 +18,7 @@ from application.firstrade_client import ( FirstradeBrokerClient, FirstradeCredentials, + FirstradeCredentialsError, FirstradeSafetyError, FirstradeMfaRequired, is_live_trading_enabled, @@ -527,6 +528,10 @@ def run_forced_session_renewal( resolved_credentials = credentials or FirstradeCredentials.from_env( env_reader, include_mfa_credentials=False, ) + try: + resolved_credentials.require_login_fields() + except FirstradeCredentialsError: + raise SessionRenewalFailure("credentials_unavailable", "credentials") from None # Never consume a seed, code, PIN or OTP destination in unattended renewal. resolved_credentials = replace( resolved_credentials, pin="", email="", phone="", mfa_secret="", mfa_code="", debug=False, diff --git a/tests/test_forced_session_renewal.py b/tests/test_forced_session_renewal.py index 6a2a159..e98db20 100644 --- a/tests/test_forced_session_renewal.py +++ b/tests/test_forced_session_renewal.py @@ -342,3 +342,27 @@ def client(credentials, *, live_trading_enabled): mfa_secret="synthetic", mfa_code="synthetic", phone="synthetic", pin="synthetic"), client_factory=client) assert calls == ["get", "post"] + + +@pytest.mark.parametrize("missing_field", ["username", "password"]) +def test_missing_credentials_alert_without_constructing_broker(missing_field): + from unittest.mock import Mock + from application.session_renewal_guard import run_guarded_renewal + store, sender = Mock(), Mock(return_value=True) + store.read.return_value = (None, 0) + store.write.return_value = 1 + store.claim_alert.return_value = True + credentials = _credentials(**{missing_field: ""}) + def factory(*args, **kwargs): + pytest.fail("broker constructed without credentials") + result, status = run_guarded_renewal( + lambda: run_forced_session_renewal(credentials=credentials, client_factory=factory), + store=store, sender=sender, + ) + assert status == 503 + assert result["error"] == "credentials_unavailable" + assert result["stage"] == "credentials" + assert result["manual_recovery_required"] is True + failed = store.write.call_args_list[-1].args[0] + assert failed["status"] == "manual_required" + assert sender.call_args.args[2:] == ("credentials_unavailable", "credentials")