diff --git a/README.md b/README.md index 2df9642b..72962f32 100644 --- a/README.md +++ b/README.md @@ -131,3 +131,9 @@ See [CONTRIBUTING.md](CONTRIBUTING.md). AI assistants and maintainers follow [AG See [LICENSE](LICENSE). Display/notification naming configuration: [display names](docs/display-names.md). + +Telegram delivery uses QPK notification intents and redacted receipts. Runtime +entrypoints install shared log and stream redaction. Environment-backed bot +routes resolve credentials only in QPK delivery; existing injected or in-memory +credentials use QPK's compatibility sender, with no temporary environment mutation. +Provider errors and credential-bearing request data are excluded from reports. diff --git a/logs.json b/logs.json new file mode 100644 index 00000000..656e9dfc --- /dev/null +++ b/logs.json @@ -0,0 +1 @@ +[{"jsonPayload": {"event": "other", "status": "applied_paused", "readback": {"application_id": "61743a15-6dcc-4ebf-8ecb-454660014af7", "revision_name": "paper-v7-00001", "runtime_target_enabled": false}}}] \ No newline at end of file diff --git a/main.py b/main.py index cf8eacb5..73ca6a90 100644 --- a/main.py +++ b/main.py @@ -87,6 +87,9 @@ from strategy_runtime import load_strategy_runtime from decision_mapper import map_strategy_decision_to_plan +from quant_platform_kit.notifications import install_log_redaction +install_log_redaction(wrap_streams=True) + app = Flask(__name__) # This collector is limited to documented LongPort read endpoints and is @@ -676,18 +679,10 @@ def _send_runtime_error(exc: Exception, *, route_label: str) -> bool: outcomes = [] for token, chat_id in targets: try: - response = requests.post( - f"https://api.telegram.org/bot{token}/sendMessage", - json={"chat_id": chat_id, "text": message}, - timeout=10, - ) - status_code = int(getattr(response, "status_code", 200) or 200) - acknowledged = 200 <= status_code < 300 - load_payload = getattr(response, "json", None) - payload = load_payload() if acknowledged and callable(load_payload) else None - if isinstance(payload, dict) and payload.get("ok") is False: - acknowledged = False - outcomes.append(acknowledged) + from notifications.intent_transport import send_intent + outcomes.append(send_intent( + message, token=token, chat_id=chat_id, requests_module=requests, timeout=10, + )["ok"]) except Exception: print( t("runtime_notification_delivery_failed"), diff --git a/notifications/intent_transport.py b/notifications/intent_transport.py new file mode 100644 index 00000000..a5305060 --- /dev/null +++ b/notifications/intent_transport.py @@ -0,0 +1,87 @@ +"""Adapt existing injected HTTP clients to the shared QPK Telegram sender.""" +from __future__ import annotations + +import hashlib +import io +import json +import os + +from quant_platform_kit.notifications import ( + NotificationIntent, + redact_sensitive_text, + send_notification_intent, + send_telegram_message, +) + + +def send_intent(text, *, chat_id, token, token_env="TELEGRAM_TOKEN", requests_module, + timeout=15, payload_key="json", request_options=None, printer=None): + """Keep credentials outside intents, reports and provider error receipts. + + Environment-backed runtime routes use the intent sender. Existing callers + supplying an in-memory credential retain QPK's explicit-credential transport + until they can provide a credential reference; no process environment is mutated. + """ + message = redact_sensitive_text(str(text)) + intent = NotificationIntent(text=message, run_id="runtime", chat_ids=(str(chat_id),)) + provider = {} + statuses = [] + + def opener(request, *, timeout): + payload = json.loads(request.data) + # Preserve existing consumer wire options; QPK still builds/validates the request. + payload.pop("disable_web_page_preview", None) + try: + response = requests_module.post( + request.full_url, **{payload_key: payload}, timeout=timeout, + **(request_options or {}), + ) + except Exception as exc: + error_type = type(exc).__name__ + provider["error_type"] = error_type if error_type.isidentifier() else "transport_error" + raise + status = int(getattr(response, "status_code", 200) or 200) + statuses.append(status) + if printer is not None and not 200 <= status < 300: + detail = redact_sensitive_text(str(getattr(response, "text", ""))) + if token: + detail = detail.replace(str(token), "") + printer(f"Telegram send failed with status {status}: {detail[:300]}", flush=True) + load = getattr(response, "json", None) + body = load() if callable(load) else None + if isinstance(body, dict): + # Retain only scalar delivery metadata needed by existing consumers. + result = body.get("result") + if body.get("ok") is True and isinstance(result, dict): + message_id = result.get("message_id") + if type(message_id) is int and message_id > 0: + provider["message_id"] = message_id + chat = result.get("chat") + if isinstance(chat, dict) and str(chat.get("id")) == str(chat_id): + provider["transport_receipt_id"] = hashlib.sha256( + json.dumps([str(chat_id), message_id], separators=(",", ":")).encode() + ).hexdigest() + parameters = body.get("parameters") + delay = parameters.get("retry_after") if isinstance(parameters, dict) else None + if type(delay) is int and delay > 0: + provider["retry_after"] = delay + wrapped = io.BytesIO(json.dumps(body).encode()) + wrapped.status = status + return wrapped + + if token == os.environ.get(token_env): + receipt = send_notification_intent(intent, bot_token_env=token_env, opener=opener, timeout=timeout) + else: + # Compatibility for existing dependency-injected and SecretStore callers. + # QPK remains the sole URL/HTTP implementation, with the same strict ack. + ok = send_telegram_message( + bot_token=token, chat_ids=intent.chat_ids, text=intent.text, + opener=opener, timeout=timeout, printer=lambda *_args, **_kwargs: None, + ) + receipt = { + "ok": ok, "status": "sent" if ok else "failed", + "http_status": statuses[-1] if statuses else None, + "sha256": hashlib.sha256(message.encode()).hexdigest(), + "length": len(message), "run_id": intent.run_id, + } + return {**receipt, **provider} diff --git a/notifications/telegram.py b/notifications/telegram.py index e0aa5458..aeb8b881 100644 --- a/notifications/telegram.py +++ b/notifications/telegram.py @@ -510,27 +510,11 @@ def build_sender(token, chat_id, *, with_prefix_fn, requests_module=None): def send_tg_message(message) -> bool: if not token or not chat_id: return False - url = f"https://api.telegram.org/bot{token}/sendMessage" - try: - prefixed = with_prefix_fn(message) - response = requests_module.post( - url, - json={"chat_id": chat_id, "text": _break_telegram_market_symbol_auto_links(prefixed)}, - timeout=10, - ) - status_code = int(getattr(response, "status_code", 200) or 200) - if status_code < 200 or status_code >= 300: - print(f"Telegram send failed: HTTP {status_code}", flush=True) - return False - load_payload = getattr(response, "json", None) - payload = load_payload() if callable(load_payload) else None - if isinstance(payload, Mapping) and payload.get("ok") is False: - print("Telegram send failed: negative API acknowledgement", flush=True) - return False - except Exception as exc: - print(f"Telegram send failed: {type(exc).__name__}", flush=True) - return False - return True + from .intent_transport import send_intent + return send_intent( + with_prefix_fn(message), chat_id=chat_id, token=token, + requests_module=requests_module, timeout=10, + )["ok"] return send_tg_message diff --git a/pyproject.toml b/pyproject.toml index 44b3876f..80bd40f1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,7 +18,7 @@ dependencies = [ "google-cloud-storage", "google-auth", "longport==3.0.23", - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@350dd38ece0952beef096893a7dcf8ae3871609a", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@59469a16b4ff7c5a1055caab21a1bd7a21dc0ea5", "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@8412d8338879f335ef8830678c2733e02e53e7c7", "hk-equity-strategies @ git+https://github.com/QuantStrategyLab/HkEquityStrategies.git@fe0c32e09d454ca4388c3372bf3874097b1bbaf0", ] @@ -61,5 +61,5 @@ include = [ [tool.uv] override-dependencies = [ - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@350dd38ece0952beef096893a7dcf8ae3871609a", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@59469a16b4ff7c5a1055caab21a1bd7a21dc0ea5", ] diff --git a/qsl.toml b/qsl.toml index 7cced307..d11344e6 100644 --- a/qsl.toml +++ b/qsl.toml @@ -5,7 +5,7 @@ upgrade_ring = "ring_d" allow_legacy = false [qsl.requires] -quant_platform_kit = "350dd38ece0952beef096893a7dcf8ae3871609a" +quant_platform_kit = "59469a16b4ff7c5a1055caab21a1bd7a21dc0ea5" us_equity_strategies = "8412d8338879f335ef8830678c2733e02e53e7c7" hk_equity_strategies = "cc3045689ae9d76eb97b9e1aa3b8fb6288ee1bcc" diff --git a/scripts/notify_route_check.py b/scripts/notify_route_check.py index 9f6953c9..30ae62dd 100644 --- a/scripts/notify_route_check.py +++ b/scripts/notify_route_check.py @@ -123,6 +123,8 @@ def _accessor_members(secret: str, project: str) -> dict: def main() -> int: + from quant_platform_kit.notifications import install_log_redaction + install_log_redaction() project = os.environ["GCP_PROJECT_ID"] prefix = (os.environ.get("SERVICE_NAME_PREFIX") or "").strip() limit = int(os.environ.get("REVISION_LIMIT") or "5") diff --git a/scripts/paper_observation/__main__.py b/scripts/paper_observation/__main__.py index 88d6f1a2..fabdd0e8 100644 --- a/scripts/paper_observation/__main__.py +++ b/scripts/paper_observation/__main__.py @@ -150,13 +150,10 @@ def sender(token, recipient): def send(message): try: - response = requests.post( - f"https://api.telegram.org/bot{token}/sendMessage", - json={"chat_id": recipient, "text": message}, - timeout=10, - allow_redirects=False, - ) - return response.status_code == 200 and response.json().get("ok") is True + from notifications.intent_transport import send_intent + receipt = send_intent(message, token=token, chat_id=recipient, requests_module=requests, + timeout=10, request_options={"allow_redirects": False}) + return receipt["http_status"] == 200 and receipt["ok"] except Exception: return False @@ -164,6 +161,8 @@ def send(message): def main(): + from quant_platform_kit.notifications import install_log_redaction + install_log_redaction(wrap_streams=True) logging.disable(logging.CRITICAL) try: if len(sys.argv) != 2 or sys.argv[1] not in ("collect", "watchdog", "verify"): diff --git a/tests/test_durable_execution_commands.py b/tests/test_durable_execution_commands.py index e7b3bd15..748c6f77 100644 --- a/tests/test_durable_execution_commands.py +++ b/tests/test_durable_execution_commands.py @@ -163,7 +163,8 @@ def enqueue(self, _command): assert gate["enforcement"] == "observe" assert gate["mode"] == "active" assert gate["policy_allows"] is False - assert gate["broker_write_allowed"] is True + # The pinned QPK command gate never treats observation as broker authority. + assert gate["broker_write_allowed"] is False assert "exposure_effect_unknown" in gate["reasons"] assert result["consumer_authorized"] is False diff --git a/tests/test_report_serialization_no_secrets.py b/tests/test_report_serialization_no_secrets.py new file mode 100644 index 00000000..267d8fc1 --- /dev/null +++ b/tests/test_report_serialization_no_secrets.py @@ -0,0 +1,52 @@ +import json +import os +import re +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from notifications.intent_transport import send_intent +from quant_platform_kit.common.runtime_reports import build_runtime_report_base + + +def test_report_serialization_has_no_secrets(): + token = str(123456789) + ':' + 'A' * 35 + chat_id = 'synthetic-private-chat' + for acknowledged in (True, False): + transport = Mock() + transport.post.return_value = SimpleNamespace(status_code=200, json=lambda: {'ok': acknowledged}) + with patch.dict(os.environ, {'TELEGRAM_TOKEN': token}, clear=True): + receipt = send_intent('Synthetic cycle status', chat_id=chat_id, token=token, + requests_module=transport) + report = build_runtime_report_base( + platform='LongBridgePlatform', deploy_target='synthetic', service_name='synthetic', + strategy_profile='synthetic', run_id='synthetic', run_source='test', + summary={'notification_receipt': receipt}, + ) + serialized = json.dumps(report) + assert receipt['ok'] is acknowledged + assert transport.post.call_count == 1 + assert token not in serialized + assert chat_id not in serialized + assert re.search(r'\d{6,12}:A[\w-]{34}', serialized) is None + + def check(value): + if isinstance(value, dict): + for key, item in value.items(): + if key.lower() in {'token', 'secret', 'password', 'api_key'}: + assert not item + check(item) + elif isinstance(value, list): + for item in value: + check(item) + check(json.loads(serialized)) + + +def test_provider_exception_cannot_enter_report(): + token = str(123456789) + ':' + 'A' * 35 + transport = Mock() + transport.post.side_effect = RuntimeError(token) + with patch.dict(os.environ, {'TELEGRAM_TOKEN': token}, clear=True): + receipt = send_intent('Synthetic cycle status', chat_id='synthetic-private-chat', + token=token, requests_module=transport) + assert not receipt['ok'] + assert token not in json.dumps(receipt) diff --git a/tests/test_runtime_strategy_adapters.py b/tests/test_runtime_strategy_adapters.py index 8813a7f4..fb447812 100644 --- a/tests/test_runtime_strategy_adapters.py +++ b/tests/test_runtime_strategy_adapters.py @@ -512,7 +512,7 @@ def test_runtime_strategy_adapters_builds_escalated_plugin_alert_message(): alerts = adapters.build_strategy_plugin_alert_messages((signal,)) assert len(alerts) == 1 - assert alerts[0].subject == "alert:soxl_soxx_trend_income:Crisis:true crisis" + assert alerts[0].subject == "alert:Semiconductor Trend Income:Crisis:true crisis" assert "plugin=Crisis" in alerts[0].body assert "route=true crisis" in alerts[0].body assert "action=defend" in alerts[0].body diff --git a/uv.lock b/uv.lock index 69456423..ce7e77d6 100644 --- a/uv.lock +++ b/uv.lock @@ -17,7 +17,7 @@ resolution-markers = [ ] [manifest] -overrides = [{ name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=350dd38ece0952beef096893a7dcf8ae3871609a" }] +overrides = [{ name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=59469a16b4ff7c5a1055caab21a1bd7a21dc0ea5" }] [[package]] name = "blinker" @@ -727,7 +727,7 @@ requires-dist = [ { name = "pytest", marker = "extra == 'test'" }, { name = "pytest-cov", marker = "extra == 'test'" }, { name = "pytz" }, - { name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=350dd38ece0952beef096893a7dcf8ae3871609a" }, + { name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=59469a16b4ff7c5a1055caab21a1bd7a21dc0ea5" }, { name = "requests" }, { name = "ruff", marker = "extra == 'test'" }, { name = "us-equity-strategies", git = "https://github.com/QuantStrategyLab/UsEquityStrategies.git?rev=8412d8338879f335ef8830678c2733e02e53e7c7" }, @@ -1210,7 +1210,7 @@ wheels = [ [[package]] name = "quant-platform-kit" version = "1.0.0" -source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=350dd38ece0952beef096893a7dcf8ae3871609a#350dd38ece0952beef096893a7dcf8ae3871609a" } +source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=59469a16b4ff7c5a1055caab21a1bd7a21dc0ea5#59469a16b4ff7c5a1055caab21a1bd7a21dc0ea5" } [[package]] name = "requests"