Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 25 additions & 1 deletion .github/workflows/invoke-cloud-run.yml
Original file line number Diff line number Diff line change
Expand Up @@ -141,6 +141,7 @@ jobs:
scheduler_location=""
scheduler_expected_path=""
restore_probe_uri=""
pause_after_run="false"
ensure_invoke_bridge_job() {
local bridge_job="$1"
local bridge_uri="$2"
Expand Down Expand Up @@ -351,6 +352,14 @@ jobs:
scheduler_expected_path="/dry-run"
ensure_invoke_bridge_job "${scheduler_job}" "${service_url}/dry-run" "${scheduler_location}"
;;
/session-renew)
# Manual-only forced read-only session renewal. Dedicated bridge job, kept PAUSED
# between manual invokes so its placeholder schedule never fires on its own.
scheduler_job="${CLOUD_RUN_SERVICE}-session-renew-scheduler"
scheduler_expected_path="/session-renew"
ensure_invoke_bridge_job "${scheduler_job}" "${service_url}/session-renew" "${scheduler_location}"
pause_after_run="true"
;;
/account-facts-sync)
# Dedicated scheduler: OIDC for Cloud Run IAM (X-Serverless-Authorization) plus
# FIRSTRADE_ACCOUNT_FACTS_SYNC_TOKEN in Authorization for app auth.
Expand All @@ -360,7 +369,7 @@ jobs:
;;
*)
echo "Cloud Run service ${CLOUD_RUN_SERVICE} has internal ingress, so GitHub-hosted runners cannot curl ${raw_path} directly." >&2
echo "Use one of the scheduler-backed paths: /run, /probe, /dry-run, /account-facts-sync." >&2
echo "Use one of the scheduler-backed paths: /run, /probe, /dry-run, /account-facts-sync, /session-renew." >&2
exit 1
;;
esac
Expand Down Expand Up @@ -419,6 +428,7 @@ jobs:
echo "scheduler_job=${scheduler_job}"
echo "scheduler_location=${scheduler_location}"
echo "restore_probe_uri=${restore_probe_uri}"
echo "pause_after_run=${pause_after_run}"
} >> "$GITHUB_OUTPUT"

- name: Authenticate for service invocation
Expand Down Expand Up @@ -446,6 +456,7 @@ jobs:
if: steps.service.outputs.invoke_method == 'scheduler'
env:
RESTORE_PROBE_URI: ${{ steps.service.outputs.restore_probe_uri }}
PAUSE_AFTER_RUN: ${{ steps.service.outputs.pause_after_run }}
SERVICE_URL: ${{ steps.service.outputs.url }}
run: |
set -euo pipefail
Expand All @@ -455,6 +466,13 @@ jobs:
scheduler_location="${{ steps.service.outputs.scheduler_location }}"

restore_probe_scheduler() {
if [ "${PAUSE_AFTER_RUN:-false}" = "true" ]; then
echo "Pausing manual-only bridge job ${scheduler_job}."
gcloud scheduler jobs pause "${scheduler_job}" \
--project="${GCP_PROJECT_ID}" \
--location="${scheduler_location}" \
--quiet || echo "::warning::Could not pause ${scheduler_job}; pause it manually."
fi
if [ -z "${RESTORE_PROBE_URI:-}" ]; then
return 0
fi
Expand All @@ -469,6 +487,12 @@ jobs:
--quiet || true
}
trap restore_probe_scheduler EXIT
if [ "${PAUSE_AFTER_RUN:-false}" = "true" ]; then
gcloud scheduler jobs resume "${scheduler_job}" \
--project="${GCP_PROJECT_ID}" \
--location="${scheduler_location}" \
--quiet
fi

echo "Triggering ${scheduler_job} at ${started_at}."
gcloud scheduler jobs run "${scheduler_job}" \
Expand Down
5 changes: 3 additions & 2 deletions application/firstrade_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -255,7 +255,8 @@ def __init__(
self.session_reused = False
self.read_only_transport_enabled = False

def connect(self) -> "FirstradeBrokerClient":
def connect(self, *, force_login: bool = False) -> "FirstradeBrokerClient":
"""Log in (or reuse cache). force_login skips cache reuse but still saves the new cache."""
self.read_only_transport_enabled = False
self.credentials.require_login_fields()
session_factory = self._session_factory
Expand All @@ -269,7 +270,7 @@ def connect(self) -> "FirstradeBrokerClient":
cookie_dir = Path(self.credentials.cookie_dir)
cookie_dir.mkdir(parents=True, exist_ok=True)
session = self._build_session(session_factory, cookie_dir)
if self.credentials.reuse_session and self._try_cached_session(
if not force_login and self.credentials.reuse_session and self._try_cached_session(
session,
account_data_factory=account_data_factory,
cookie_dir=cookie_dir,
Expand Down
38 changes: 38 additions & 0 deletions application/session_check_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
from application.firstrade_client import (
FirstradeBrokerClient,
FirstradeCredentials,
FirstradeSafetyError,
is_live_trading_enabled,
mask_account_id,
)
Expand Down Expand Up @@ -457,3 +458,40 @@ def run_session_check(
if maintenance_state_error:
result["session_check_maintenance_state_error"] = maintenance_state_error
return result


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 read-only session renewal: one fresh login, persist the session cache, read balances.

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

resolved_credentials = credentials or FirstradeCredentials.from_env(env_reader)
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)
account = client.select_account(env_reader("FIRSTRADE_ACCOUNT", "") or None)
balances = client.get_balances(account)
print(
"Firstrade forced session renewal "
f"session_reused={bool(getattr(client, 'session_reused', False))} "
f"account={mask_account_id(account)} balances_read={bool(balances)}",
flush=True,
)
return {
"ok": True,
"api_kind": "unofficial-reverse-engineered",
"session_renewed": True,
"session_reused": bool(getattr(client, "session_reused", False)),
"account": mask_account_id(account),
"balances_read": bool(balances),
}
22 changes: 21 additions & 1 deletion main.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
from application.firstrade_client import (
FirstradeBrokerClient,
FirstradeCredentials,
FirstradeMfaRequired,
FirstradePlatformError,
is_live_trading_enabled,
mask_account_id,
Expand Down Expand Up @@ -58,7 +59,7 @@
validate_reconciliation_candidate,
validate_reconciliation_preconditions,
)
from application.session_check_service import run_session_check
from application.session_check_service import run_forced_session_renewal, run_session_check
from notifications.telegram import build_sender
from quant_platform_kit.common.runtime_reports import (
append_runtime_report_error,
Expand Down Expand Up @@ -1014,6 +1015,25 @@ def probe():
return session_check()


@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.
"""
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


@app.post("/monitor-dispatch")
@app.get("/monitor-dispatch")
def monitor_dispatch():
Expand Down
169 changes: 169 additions & 0 deletions tests/test_forced_session_renewal.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,169 @@
from __future__ import annotations

import pytest

from application import session_check_service
from application.firstrade_client import FirstradeCredentials, FirstradeMfaRequired, FirstradeSafetyError
from application.session_check_service import run_forced_session_renewal

ORDER_METHODS = ("place_stock_order", "place_order", "submit_order", "cancel_order", "preview_order")
READ_METHODS = {"connect", "select_account", "get_balances"}


def _credentials(**overrides):
values = dict(
username="u", password="p", reuse_session=True, persist_session_cache=True,
gcs_state_bucket="bucket",
)
values.update(overrides)
return FirstradeCredentials(**values)


class _RecordingClient:
def __init__(self, credentials, *, live_trading_enabled):
self.credentials = credentials
self.live_trading_enabled = live_trading_enabled
self.session_reused = False
self.calls = []

def __getattr__(self, name):
if name in ORDER_METHODS:
raise AssertionError(f"order method reached: {name}")
raise AttributeError(name)

def connect(self, *, force_login=False):
self.calls.append(("connect", force_login))
return self

def select_account(self, requested=None):
self.calls.append(("select_account", requested))
return "12345979"

def get_balances(self, account):
self.calls.append(("get_balances", account))
return {"cash": 1}


def _factory(created):
def make(credentials, *, live_trading_enabled):
client = _RecordingClient(credentials, live_trading_enabled=live_trading_enabled)
created.append(client)
return client

return make


def test_forced_renewal_logs_in_once_read_only():
created = []
result = run_forced_session_renewal(
credentials=_credentials(), client_factory=_factory(created), env_reader=lambda *_: None,
)
assert result["ok"] is True and result["session_renewed"] is True
assert len(created) == 1
client = created[0]
assert client.live_trading_enabled is False
assert client.calls[0] == ("connect", True)
assert {name for name, _ in client.calls} <= READ_METHODS
assert "12345979" not in str(result)


def test_forced_renewal_never_touches_monthly_marker(monkeypatch):
def boom(*_a, **_k):
raise AssertionError("marker path reached")

monkeypatch.setattr(session_check_service, "resolve_session_check_maintenance_decision", boom)
monkeypatch.setattr(session_check_service, "persist_session_check_maintenance", boom)
monkeypatch.setattr(session_check_service, "build_gcs_state_store_from_env", boom)
created = []
run_forced_session_renewal(
credentials=_credentials(), client_factory=_factory(created), env_reader=lambda *_: None,
)


@pytest.mark.parametrize("overrides", [{"reuse_session": False}, {"persist_session_cache": False}])
def test_forced_renewal_requires_cache_persistence(overrides):
with pytest.raises(FirstradeSafetyError):
run_forced_session_renewal(
credentials=_credentials(**overrides), client_factory=_factory([]), env_reader=lambda *_: None,
)


def test_forced_renewal_refuses_live_client():
def make(credentials, *, live_trading_enabled):
return _RecordingClient(credentials, live_trading_enabled=True)

with pytest.raises(FirstradeSafetyError):
run_forced_session_renewal(credentials=_credentials(), client_factory=make, env_reader=lambda *_: None)


def test_connect_force_login_skips_cached_session(tmp_path):
from application.firstrade_client import FirstradeBrokerClient

events = []

class _Session:
def __init__(self, **_kw):
self.session = type("T", (), {"headers": {}, "cookies": None})()

def login(self):
events.append("login")
return False

client = FirstradeBrokerClient(
_credentials(cookie_dir=str(tmp_path), gcs_state_bucket=""),
live_trading_enabled=False,
session_factory=_Session,
account_data_factory=lambda session: object(),
order_factory=lambda *_a: (_ for _ in ()).throw(AssertionError("order factory reached")),
)
client._try_cached_session = lambda *a, **k: events.append("cached") or True
client.connect(force_login=True)
assert events == ["login"]


# --- HTTP route ---------------------------------------------------------------------------


@pytest.fixture
def http_client(monkeypatch):
pytest.importorskip("flask")
import main

def no_orders(*_a, **_k):
raise AssertionError("strategy/order path reached")

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)
return main


def test_route_disabled_without_http_flag(http_client, monkeypatch):
monkeypatch.delenv("FIRSTRADE_RUN_SESSION_CHECK_ON_HTTP", raising=False)
response = http_client.app.test_client().post("/session-renew")
assert response.status_code == 403


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})
response = http_client.app.test_client().post("/session-renew")
assert response.status_code == 200 and calls == [1]


def test_route_reports_mfa_required(http_client, monkeypatch):
monkeypatch.setenv("FIRSTRADE_RUN_SESSION_CHECK_ON_HTTP", "true")

def mfa():
raise FirstradeMfaRequired("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"


def test_route_is_post_only(http_client):
response = http_client.app.test_client().get("/session-renew")
assert response.status_code == 405
10 changes: 10 additions & 0 deletions tests/test_invoke_cloud_run_workflow.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,3 +14,13 @@ def test_invoke_cloud_run_workflow_ensures_probe_and_dry_run_scheduler_bridges()
assert 'ensure_invoke_bridge_job "${scheduler_job}" "${service_url}/probe"' in workflow
assert 'ensure_invoke_bridge_job "${scheduler_job}" "${service_url}/dry-run"' in workflow
assert '--schedule="0 0 1 1 *"' in workflow


def test_invoke_cloud_run_workflow_session_renew_bridge_is_paused_after_run():
workflow_path = Path(__file__).resolve().parents[1] / ".github/workflows/invoke-cloud-run.yml"
workflow = workflow_path.read_text(encoding="utf-8")

assert 'scheduler_job="${CLOUD_RUN_SERVICE}-session-renew-scheduler"' in workflow
assert 'ensure_invoke_bridge_job "${scheduler_job}" "${service_url}/session-renew"' in workflow
assert 'gcloud scheduler jobs pause "${scheduler_job}"' in workflow
assert 'pause_after_run="true"' in workflow
1 change: 1 addition & 0 deletions tests/test_request_handling.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@ def test_cloud_run_route_contracts_are_registered():
"/account-facts-sync": ["POST"],
"/monitor-dispatch": ["GET", "POST"],
"/probe": ["POST"],
"/session-renew": ["POST"],
"/static/<path:filename>": ["GET"],
}

Expand Down
Loading