diff --git a/application/firstrade_client.py b/application/firstrade_client.py index eda1eb2..876364e 100644 --- a/application/firstrade_client.py +++ b/application/firstrade_client.py @@ -144,10 +144,11 @@ def is_live_trading_enabled(env: Callable[[str, str | None], str | None] = os.ge def mask_account_id(account_id: str) -> str: + """Expose at most two trailing characters; fully mask shorter identifiers.""" value = str(account_id or "") - if len(value) <= 4: + if len(value) <= 2: return "*" * len(value) - return f"{'*' * max(0, len(value) - 4)}{value[-4:]}" + return f"{'*' * (len(value) - 2)}{value[-2:]}" def normalize_account_number(value: object) -> str | None: """Coerce Firstrade account identifiers to comparable non-empty strings.""" diff --git a/application/rebalance_service.py b/application/rebalance_service.py index e66e113..cf1bfed 100644 --- a/application/rebalance_service.py +++ b/application/rebalance_service.py @@ -28,7 +28,9 @@ ) from application.runtime_broker_adapters import build_runtime_broker_adapters from application.signal_snapshot import build_signal_snapshot -from application.state_persistence import GcsStateStore, build_gcs_state_store_from_env +from application.state_persistence import ( + GcsStateStore, build_gcs_state_store_from_env, legacy_account_state_id, +) from application.strategy_run_persistence import ( build_strategy_run_state, claim_live_strategy_run, @@ -628,17 +630,18 @@ def log_message(message: str) -> None: evaluation_metadata=getattr(evaluation, "metadata", None), ) masked_account = mask_account_id(account) + state_account = legacy_account_state_id(account) existing_run = None if persist_strategy_runs and not settings.dry_run_only: existing_run = read_latest_strategy_run_state( store=store, - account=masked_account, + account=state_account, strategy_profile=strategy_runtime.profile, run_period=run_period, ) existing_submission_claim = read_live_strategy_run_claim( store=store, - account=masked_account, + account=state_account, strategy_profile=strategy_runtime.profile, run_period=run_period, ) @@ -678,6 +681,7 @@ def log_message(message: str) -> None: strategy_run_persisted = persist_strategy_run_state( store=store, state=duplicate_state, + storage_account=state_account, now=now, ) except Exception as exc: @@ -745,6 +749,7 @@ def log_message(message: str) -> None: strategy_run_persisted = persist_strategy_run_state( store=store, state=planned_state, + storage_account=state_account, now=now, ) except Exception as exc: @@ -763,7 +768,7 @@ def acquire_submission_claim() -> bool: return True submission_claim_acquired = claim_live_strategy_run( store=store, - account=masked_account, + account=state_account, strategy_profile=strategy_runtime.profile, run_period=run_period, now=now, @@ -792,7 +797,7 @@ def acquire_submission_claim() -> bool: if execution_result.idempotency_blocked: existing_run = read_latest_strategy_run_state( store=store, - account=masked_account, + account=state_account, strategy_profile=strategy_runtime.profile, run_period=run_period, ) or { @@ -965,6 +970,7 @@ def acquire_submission_claim() -> bool: result["strategy_run_persisted"] = persist_strategy_run_state( store=store, state=completed_state, + storage_account=state_account, now=now, ) except Exception as exc: diff --git a/application/session_check_service.py b/application/session_check_service.py index 2c7a316..c2b84c2 100644 --- a/application/session_check_service.py +++ b/application/session_check_service.py @@ -22,7 +22,9 @@ is_live_trading_enabled, mask_account_id, ) -from application.state_persistence import GcsStateStore, build_gcs_state_store_from_env +from application.state_persistence import ( + GcsStateStore, build_gcs_state_store_from_env, legacy_account_state_id, +) from strategy_registry import ( FIRSTRADE_PLATFORM, resolve_strategy_definition, @@ -162,7 +164,7 @@ def _session_check_state_key( env: Callable[[str, str | None], str | None] = os.getenv, ) -> str: account_selector = str(env("FIRSTRADE_ACCOUNT", "") or "").strip() - account_key = _safe_key(mask_account_id(account_selector)) if account_selector else "auto" + account_key = _safe_key(legacy_account_state_id(account_selector)) if account_selector else "auto" return ( f"session-checks/{account_key}/{_safe_key(strategy_profile)}/" f"{_safe_key(period)}/latest.json" diff --git a/application/state_persistence.py b/application/state_persistence.py index 35e1f1e..3cae828 100644 --- a/application/state_persistence.py +++ b/application/state_persistence.py @@ -12,6 +12,14 @@ class StatePersistenceError(RuntimeError): """Raised when the configured state store cannot read or write.""" +def legacy_account_state_id(account: str) -> str: + """Keep existing private state/claim locators stable; never use for display.""" + value = str(account or "") + if len(value) <= 4: + return "*" * len(value) + return "*" * (len(value) - 4) + value[-4:] + + def _normalize_prefix(prefix: str | None) -> str: return str(prefix or "").strip("/") @@ -51,15 +59,15 @@ def read_json(self, key: str) -> dict[str, Any] | None: error_msg = str(exc) if "404" in error_msg or "NOT_FOUND" in error_msg.upper(): return None - raise StatePersistenceError(f"GCS read failed for {key}: {exc}") from exc + raise StatePersistenceError("GCS state read failed") from None if text is None: return None try: payload = json.loads(text) except json.JSONDecodeError as exc: - raise StatePersistenceError(f"GCS payload for {key} is not valid JSON: {exc}") from exc + raise StatePersistenceError("GCS state payload is not valid JSON") from None if not isinstance(payload, dict): - raise StatePersistenceError(f"GCS payload for {key} is not a JSON object.") + raise StatePersistenceError("GCS state payload is not a JSON object.") return payload def write_json(self, key: str, payload: dict[str, Any]) -> bool: @@ -68,7 +76,7 @@ def write_json(self, key: str, payload: dict[str, Any]) -> bool: try: _object_store().write_text(uri, data) except Exception as exc: - raise StatePersistenceError(f"GCS write failed for {key}: {exc}") from exc + raise StatePersistenceError("GCS state write failed") from None return True def create_json(self, key: str, payload: dict[str, Any]) -> bool: @@ -77,7 +85,7 @@ def create_json(self, key: str, payload: dict[str, Any]) -> bool: data = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) bucket_name, separator, object_name = uri[5:].partition("/") if not separator or not bucket_name or not object_name: - raise StatePersistenceError(f"Invalid GCS state URI for {key}") + raise StatePersistenceError("Invalid GCS state URI") try: from google.api_core.exceptions import Conflict, PreconditionFailed from google.cloud import storage @@ -97,7 +105,7 @@ def create_json(self, key: str, payload: dict[str, Any]) -> bool: except (Conflict, PreconditionFailed): return False except Exception as exc: - raise StatePersistenceError(f"GCS atomic create failed for {key}: {exc}") from exc + raise StatePersistenceError("GCS state atomic create failed") from None def build_gcs_state_store_from_env( diff --git a/application/strategy_run_persistence.py b/application/strategy_run_persistence.py index ed9dad0..0405425 100644 --- a/application/strategy_run_persistence.py +++ b/application/strategy_run_persistence.py @@ -9,6 +9,7 @@ from typing import Any from application.state_persistence import GcsStateStore +from application.firstrade_client import mask_account_id from quant_platform_kit.common.execution_outcomes import ( DEFAULT_TERMINAL_STRATEGY_RUN_STAGES, STAGE_NO_ACTION, @@ -135,7 +136,7 @@ def claim_live_strategy_run( payload = { "stage": "PENDING_SUBMISSION", "submission_claim_boundary": LIVE_SUBMISSION_CLAIM_BOUNDARY, - "account": account, + "account": mask_account_id(account), "strategy_profile": strategy_profile, "run_period": run_period, "as_of": (now or utcnow()).isoformat(), @@ -233,7 +234,7 @@ def build_strategy_run_state( payload = { "stage": stage, "as_of": as_of.isoformat(), - "account": account, + "account": mask_account_id(account), "strategy_profile": strategy_profile, "strategy_display_name": strategy_display_name, "run_period": run_period, @@ -260,10 +261,11 @@ def persist_strategy_run_state( *, store: GcsStateStore, state: Mapping[str, Any], + storage_account: str | None = None, now: datetime | None = None, ) -> bool: as_of = now or utcnow() - account = str(state.get("account") or "unknown") + account = storage_account if storage_account is not None else str(state.get("account") or "unknown") strategy_profile = str(state.get("strategy_profile") or "unknown") run_period = str(state.get("run_period") or f"{as_of.year:04d}-{as_of.month:02d}") stage = str(state.get("stage") or "UNKNOWN") diff --git a/docs/unified-daily-notification.md b/docs/unified-daily-notification.md index c20a196..0e48ffa 100644 --- a/docs/unified-daily-notification.md +++ b/docs/unified-daily-notification.md @@ -128,3 +128,12 @@ Candidate persistence failures expose only a fixed `candidate_read` or resource paths and credential context stay out of command output. A permission category identifies the failing storage operation; it does not establish that an alerting Secret Manager denial caused the storage failure. + +Account identifiers displayed in logs, notification results and account-selection +errors retain at most two trailing characters. Identifiers of two characters or +fewer are fully masked. Runtime failure alerts omit the exception detail. + +Private state object and submission-claim locators retain their existing identity +format so a display change cannot bypass an existing duplicate-submission lock +or session-maintenance marker. New state payload account fields use the two-character +mask. State-store errors omit object locators and provider exception details. diff --git a/notifications/telegram.py b/notifications/telegram.py index 06141cc..6614243 100644 --- a/notifications/telegram.py +++ b/notifications/telegram.py @@ -2,6 +2,7 @@ from __future__ import annotations +from application.firstrade_client import mask_account_id from quant_platform_kit.common.display_names import display_name from quant_platform_kit.common.operational_notification_localization import resolve_operational_notification_locale @@ -1033,7 +1034,7 @@ def render_cycle_summary(result: Mapping[str, Any], *, lang: str = "en") -> str: if strategy_name: lines.append(translator("strategy_label", name=strategy_name)) if account: - lines.append(translator("account_label", account=account)) + lines.append(translator("account_label", account=mask_account_id(account))) if not submitted: account_snapshot = result.get("heartbeat_account_snapshot") account_snapshot = account_snapshot if isinstance(account_snapshot, Mapping) else {} diff --git a/tests/test_account_alert_redaction.py b/tests/test_account_alert_redaction.py new file mode 100644 index 0000000..9baacda --- /dev/null +++ b/tests/test_account_alert_redaction.py @@ -0,0 +1,114 @@ +"""Exercise real HTTP error handling and Telegram formatting with offline I/O.""" +from copy import deepcopy +import json +from types import SimpleNamespace + +import pytest +import requests + +import main +from application.firstrade_client import FirstradeBrokerClient, FirstradeCredentials +from notifications import unified + + +class MemoryStore: + def __init__(self): + self.data = {} + + def read(self, key): + return deepcopy(self.data.get(key, (None, None))) + + def compare_and_set(self, key, version, value): + current = self.data.get(key, (None, None))[0] + if current != version: + return False + self.data[key] = ((current or 0) + 1, deepcopy(value)) + return True + + +@pytest.mark.parametrize("lang", ["en", "zh"]) +@pytest.mark.parametrize("unified_enabled", [False, True]) +def test_account_selection_failure_through_real_alert_sender( + monkeypatch, capsys, lang, unified_enabled, +): + # Synthetic data only. Keep select_account, HTTP error handling, notification + # routing and build_sender real; replace broker/storage/network boundaries. + account = "87654321" + client = FirstradeBrokerClient(FirstradeCredentials(username="synthetic", password="synthetic")) + monkeypatch.setattr(client, "account_numbers", lambda: []) + monkeypatch.setattr(main, "run_session_check", lambda: client.select_account(account)) + target = dict(platform_id="firstrade", strategy_profile="synthetic", + service_name="synthetic-service", account_scope="synthetic-scope", + account_selector=[account], deployment_selector="synthetic-slot", dry_run_only=False) + env = { + "FIRSTRADE_RUN_SESSION_CHECK_ON_HTTP": "true", + unified.FLAG: str(unified_enabled).lower(), + "QSL_ROUTINE_NOTIFICATION_MODE": "daily_digest", + "RUNTIME_TARGET_ENABLED": "true", "RUNTIME_TARGET_JSON": json.dumps(target), + "DIGEST_STORE_TARGET": "synthetic", "DIGEST_OPAQUE_ACCOUNT_UID": "opaque-synthetic", + "DIGEST_TARGET_ID": "target-synthetic", "DIGEST_CANDIDATES_GCS_URI": "gs://synthetic/candidates", + # No archive source: the immediate incident path must still work. + "EXECUTION_REPORT_GCS_URI": "", "DIGEST_TARGET_BINDINGS_JSON": "", + "QSL_NOTIFICATION_MARKER_GCS_URI": "", "DIGEST_NOTIFICATION_SCHEDULE_JSON": "", + "K_SERVICE": "synthetic-service", "ACCOUNT_REGION": "synthetic-region", + "STRATEGY_DISPLAY_NAME": "synthetic", "QSL_NOTIFY_LANG": lang, + } + for key, value in env.items(): + monkeypatch.setenv(key, value) + monkeypatch.setattr(main, "_get_telegram_token", lambda: "synthetic-token") + monkeypatch.delenv("QSL_GLOBAL_TELEGRAM_CHAT_ID", raising=False) + monkeypatch.setenv("GLOBAL_TELEGRAM_CHAT_ID", "synthetic-chat") + store = MemoryStore() + monkeypatch.setattr(unified, "GcsJsonStore", lambda: store) + sent = [] + + def post(url, *, json, timeout): + assert url == "https://api.telegram.org/botsynthetic-token/sendMessage" + assert timeout == 15 + sent.append(json) + return SimpleNamespace(status_code=200, json=lambda: {"ok": True}) + + monkeypatch.setattr(requests, "post", post) + response = main.app.test_client().post("/probe") + assert response.status_code == 500 + payload = response.get_json() + assert payload["ok"] is False + assert payload["runtime_error_notification_attempted"] is True + assert "******21" in payload["error"] + assert len(sent) == 1 + assert sent[0]["chat_id"] == "synthetic-chat" + assert ("health check failed" if lang == "en" else "健康检查失败") in sent[0]["text"] + assert "Requested account" not in sent[0]["text"] + surfaces = [response.get_data(as_text=True), json.dumps(sent), capsys.readouterr().out] + for surface in surfaces: + assert account not in surface + assert account[-4:] not in surface + if unified_enabled: + assert any("/_notification_markers/" in key for key in store.data) + assert any("/_notification_events/" in key for key in store.data) + + +@pytest.mark.parametrize("account", ["87654321", "****4321"]) +@pytest.mark.parametrize("lang", ["en", "zh"]) +def test_cycle_alert_redacts_raw_and_legacy_accounts_at_log_and_transport(monkeypatch, account, lang): + from application.rebalance_service import _publish_cycle_notification + + sent, logged = [], [] + + def post(url, *, json, timeout): + assert url == "https://api.telegram.org/botsynthetic-token/sendMessage" + sent.append(json) + return SimpleNamespace(status_code=200, json=lambda: {"ok": True}) + + monkeypatch.setattr(requests, "post", post) + result = {"account": account, "strategy_profile": "synthetic", "portfolio": {}, + "execution": {}, "submitted_orders": [], "skipped_orders": [], + "execution_blocked": True, "execution_blocking_skips": [{"reason": "broker_rejected"}]} + settings = SimpleNamespace(tg_token="synthetic-token", tg_chat_id="synthetic-chat", notify_lang=lang) + assert _publish_cycle_notification(result, settings=settings, log_message=logged.append) is True + assert len(sent) == 1 + assert logged and "******21" in "\n".join(logged) + for surface in (json.dumps(sent), "\n".join(logged)): + assert "87654321" not in surface + assert "4321" not in surface + assert result["account"] == account # Broker-facing input remains intact. diff --git a/tests/test_firstrade_client.py b/tests/test_firstrade_client.py index 6417952..e07751a 100644 --- a/tests/test_firstrade_client.py +++ b/tests/test_firstrade_client.py @@ -91,8 +91,24 @@ def build_fake_client(live=False): ).connect() -def test_mask_account_id_keeps_only_last_four_digits(): - assert mask_account_id("12345678") == "****5678" +@pytest.mark.parametrize("value, expected", [ + (None, ""), ("", ""), ("7", "*"), ("78", "**"), + ("678", "*78"), ("5678", "**78"), ("12345678", "******78"), + ("synthetic-78", "**********78"), ("****5678", "******78"), +]) +def test_mask_account_id_keeps_at_most_last_two_characters(value, expected): + assert mask_account_id(value) == expected + assert mask_account_id(expected) == expected + + +def test_account_summaries_and_selection_errors_use_last_two_characters(): + client = build_fake_client() + assert client.list_account_summaries()[0]["account"] == "******78" + with pytest.raises(FirstradePlatformError) as caught: + client.select_account("87654321") + assert "******21" in str(caught.value) + assert "4321" not in str(caught.value) + assert "87654321" not in str(caught.value) def test_notional_live_order_rejects_size_above_local_cap(): diff --git a/tests/test_forced_session_renewal.py b/tests/test_forced_session_renewal.py index 55910ea..e37a887 100644 --- a/tests/test_forced_session_renewal.py +++ b/tests/test_forced_session_renewal.py @@ -56,7 +56,7 @@ def make(credentials, *, live_trading_enabled): return make -def test_forced_renewal_logs_in_once_read_only(): +def test_forced_renewal_logs_in_once_read_only(capsys): created = [] result = run_forced_session_renewal( credentials=_credentials(), client_factory=_factory(created), env_reader=lambda *_: None, @@ -68,7 +68,12 @@ def test_forced_renewal_logs_in_once_read_only(): assert client.calls[0] == ("connect", (True, True)) assert [call[0] for call in client.calls] == ["connect", "select_account", "get_balances", "save_verified_session_cache"] assert {name for name, _ in client.calls} <= READ_METHODS - assert "12345979" not in str(result) + assert result["account"] == "******79" + log = capsys.readouterr().out + assert "account=******79" in log + for value in ("12345979", "5979"): + assert value not in str(result) + assert value not in log def test_forced_renewal_never_touches_monthly_marker(monkeypatch): diff --git a/tests/test_notification_off_equivalence.py b/tests/test_notification_off_equivalence.py index cc70897..e4e2fa2 100644 --- a/tests/test_notification_off_equivalence.py +++ b/tests/test_notification_off_equivalence.py @@ -175,6 +175,14 @@ def test_flags_off_entrypoints_renderers_match_main_golden(monkeypatch, flag, mo assert actual[key] == expected[key] actual_broker = actual["broker_cycle"] expected_broker = expected["broker_cycle"] + # The historical golden retains the old display mask. Only persisted account + # payloads change; storage locators and every broker call must stay exact. + for call in expected_broker["calls"]: + if call[0] == "state" and call[1] in {"write_json", "create_json"}: + payload = call[2][0][1] + if "account" in payload: + assert payload["account"] == "**************ctor" + payload["account"] = "****************or" assert [call for call in actual_broker["calls"] if call[0] in {"broker", "state"}] == [ call for call in expected_broker["calls"] if call[0] in {"broker", "state"}] for key in ("ok", "execution_status", "orders_pending_count", "submitted_orders"): diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index bd588f2..ea8855c 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -1421,7 +1421,7 @@ def test_render_cycle_summary_formats_skipped_orders_in_unified_chinese_template ) assert "🔔 【调仓指令】" in message - assert "🆔 账户: ****1234" in message + assert "🆔 账户: ******34" in message assert "📌 策略账户概览" in message assert "⏱ 执行时点:" not in message assert "🎯 信号:" not in message @@ -1595,7 +1595,7 @@ def test_render_cycle_summary_formats_skipped_orders_in_unified_english_template ) assert "🔔 【Rebalance Instruction】" in message - assert "🆔 Account: ****1234" in message + assert "🆔 Account: ******34" in message assert "📌 Strategy Account" in message assert "⏱ Timing:" not in message assert "🎯 Signal:" not in message @@ -1789,10 +1789,10 @@ def test_native_repeat_funding_result_keeps_late_persistence_error(live_cycle, m action_done=False, ), ) - def fail_final_funding_persistence(*, store, state, now): + def fail_final_funding_persistence(*, store, state, storage_account, now): if state["stage"] == "FUNDING_BLOCKED": raise RuntimeError("synthetic_final_persistence_failure") - return persist_strategy_run_state(store=store, state=state, now=now) + return persist_strategy_run_state(store=store, state=state, storage_account=storage_account, now=now) monkeypatch.setattr("application.rebalance_service.persist_strategy_run_state", fail_final_funding_persistence) result = run(store=store) @@ -1801,3 +1801,18 @@ def fail_final_funding_persistence(*, store, state, now): assert result["strategy_run_persisted"] is False assert "synthetic_final_persistence_failure" in result["strategy_run_persistence_error"] assert _should_publish_cycle_notification(result) + + +def test_preexisting_submission_claim_still_blocks_after_display_redaction(live_cycle): + from application.strategy_run_persistence import LIVE_SUBMISSION_CLAIM_BOUNDARY + + run, client = live_cycle + key = "strategy-runs/claims/____5678/tqqq_growth_income/2026_09.json" + claim = {"stage": "PENDING_SUBMISSION", "submission_claim_boundary": LIVE_SUBMISSION_CLAIM_BOUNDARY} + store = FakeStateStore() + store.payloads[key] = dict(claim) + result = run(store=store) + assert result["submission_claim_blocks_repeat"] is True + assert client.orders == [] + assert store.payloads[key] == claim + assert not store.writes diff --git a/tests/test_session_check_service.py b/tests/test_session_check_service.py index 1b3a0df..c028d0c 100644 --- a/tests/test_session_check_service.py +++ b/tests/test_session_check_service.py @@ -232,3 +232,16 @@ def test_session_check_policy_skip_does_not_require_credentials_or_client(): assert result["session_check_skipped"] is True assert result["session_check_policy"] == "skip" assert result["session_check_policy_reason"] == "policy_skip" + + +def test_preexisting_account_maintenance_marker_still_skips_after_display_redaction(): + key = "session-checks/____5678/russell_top50_leader_rotation/2026_06/latest.json" + store = FakeStateStore({key: {"checked_at": "2026-06-01T01:02:03+00:00"}}) + result = run_session_check( + client_factory=ExplodingClient, state_store=store, + env_reader=_env({"STRATEGY_PROFILE": "russell_top50_leader_rotation", "FIRSTRADE_ACCOUNT": "12345678"}), + now=datetime(2026, 6, 15, tzinfo=timezone.utc), + ) + assert result["session_check_skipped"] is True + assert store.reads == [key] + assert store.writes == [] diff --git a/tests/test_state_error_redaction.py b/tests/test_state_error_redaction.py new file mode 100644 index 0000000..c47f45d --- /dev/null +++ b/tests/test_state_error_redaction.py @@ -0,0 +1,33 @@ +"""Private legacy state locators never escape through diagnostic exceptions.""" +import traceback +from types import SimpleNamespace + +import pytest + +from application import state_persistence as state + + +@pytest.mark.parametrize("operation", ["read_json", "write_json", "create_json"]) +def test_state_errors_omit_private_locator_and_provider_details(monkeypatch, operation): + key = "strategy-runs/____4321/synthetic/period/latest.json" + + def fail(*args, **kwargs): + raise RuntimeError(f"synthetic provider failure: {key} account=87654321 token=synthetic-secret") + + monkeypatch.setattr(state, "_object_store", lambda: SimpleNamespace(read_text=fail, write_text=fail)) + store = state.GcsStateStore(bucket="synthetic", client_factory=fail) + with pytest.raises(state.StatePersistenceError) as caught: + method = getattr(store, operation) + method(key) if operation == "read_json" else method(key, {}) + rendered = "".join(traceback.format_exception(caught.type, caught.value, caught.tb)) + for private in (key, "4321", "87654321", "synthetic-secret"): + assert private not in str(caught.value) + assert private not in rendered + + +@pytest.mark.parametrize("payload", ["invalid-json", "[]"]) +def test_invalid_state_payload_error_omits_private_locator(monkeypatch, payload): + monkeypatch.setattr(state, "_object_store", lambda: SimpleNamespace(read_text=lambda _: payload)) + with pytest.raises(state.StatePersistenceError) as caught: + state.GcsStateStore(bucket="synthetic").read_json("strategy-runs/____4321/latest.json") + assert "4321" not in str(caught.value)