From fd6132e6ce7cf36411831355b8e12b19c39a103f Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Thu, 2 Jul 2026 01:19:17 +0200 Subject: [PATCH] Allow Leo paid research gate by chat title (#32) --- .../install_telegram_smart_research_gates.py | 23 +++++++-- telegram/bot.py | 16 +++++-- telegram/http_chat_proxy.py | 30 ++++++++++++ ...t_install_telegram_smart_research_gates.py | 48 +++++++++++++++++++ tests/test_telegram_leo_x402_bridge.py | 28 +++++++++++ 5 files changed, 139 insertions(+), 6 deletions(-) diff --git a/scripts/install_telegram_smart_research_gates.py b/scripts/install_telegram_smart_research_gates.py index 10c891f..b48b5c4 100644 --- a/scripts/install_telegram_smart_research_gates.py +++ b/scripts/install_telegram_smart_research_gates.py @@ -17,6 +17,7 @@ from pathlib import Path APPROVAL_REF_RE = re.compile(r"^[A-Za-z0-9._:@/-]{8,256}$") CHAT_ID_RE = re.compile(r"^-?\d+$") +CHAT_TITLE_RE = re.compile(r"^[^\r\n=]{1,128}$") MAX_SMART_RESEARCH_USD = 0.06 @@ -29,6 +30,7 @@ def parse_args(argv: list[str]) -> argparse.Namespace: parser.add_argument("--secrets-dir", default="/opt/teleo-eval/secrets") parser.add_argument("--allow-paid", action="store_true", help="Enable paid smart research for one allowed chat") parser.add_argument("--allowed-chat-id", help="Telegram chat id allowed to trigger paid smart research") + parser.add_argument("--allowed-chat-title", help="Telegram chat title allowed to trigger paid smart research") parser.add_argument("--max-usd", default="0.01", help="Maximum spend per Telegram smart-research call") parser.add_argument( "--approval-ref-from-stdin", @@ -75,8 +77,12 @@ def validate_max_usd(value: str) -> str: def validate_paid_inputs(args: argparse.Namespace) -> str | None: if not args.allow_paid: return None - if not args.allowed_chat_id or not CHAT_ID_RE.match(args.allowed_chat_id): - raise ValueError("--allowed-chat-id is required for --allow-paid and must be an integer") + if args.allowed_chat_id and not CHAT_ID_RE.match(args.allowed_chat_id): + raise ValueError("--allowed-chat-id must be an integer") + if args.allowed_chat_title and not CHAT_TITLE_RE.match(args.allowed_chat_title.strip()): + raise ValueError("--allowed-chat-title shape is invalid") + if not args.allowed_chat_id and not args.allowed_chat_title: + raise ValueError("--allowed-chat-id or --allowed-chat-title is required for --allow-paid") approval_ref = read_approval_ref(from_stdin=args.approval_ref_from_stdin) if not APPROVAL_REF_RE.match(approval_ref): raise ValueError("approval ref shape is invalid") @@ -125,7 +131,14 @@ def run_command(command: list[str], *, dry_run: bool) -> dict: } -def build_env_content(*, allow_paid: bool, allowed_chat_id: str | None, max_usd: str, approval_ref_path: Path) -> str: +def build_env_content( + *, + allow_paid: bool, + allowed_chat_id: str | None, + allowed_chat_title: str | None, + max_usd: str, + approval_ref_path: Path, +) -> str: lines = [ f"LIVINGIP_LEO_TELEGRAM_SMART_RESEARCH_ALLOW_PAID={'1' if allow_paid else '0'}", f"LIVINGIP_LEO_TELEGRAM_SMART_RESEARCH_MAX_USD={max_usd}", @@ -133,6 +146,8 @@ def build_env_content(*, allow_paid: bool, allowed_chat_id: str | None, max_usd: ] if allow_paid and allowed_chat_id: lines.insert(1, f"LIVINGIP_LEO_TELEGRAM_SMART_RESEARCH_ALLOWED_CHAT_ID={allowed_chat_id}") + if allow_paid and allowed_chat_title: + lines.insert(2, f"LIVINGIP_LEO_TELEGRAM_SMART_RESEARCH_ALLOWED_CHAT_TITLE={allowed_chat_title.strip()}") return "\n".join(lines) + "\n" @@ -149,6 +164,7 @@ def main(argv: list[str] | None = None) -> int: env_content = build_env_content( allow_paid=args.allow_paid, allowed_chat_id=args.allowed_chat_id, + allowed_chat_title=args.allowed_chat_title, max_usd=max_usd, approval_ref_path=approval_ref_path, ) @@ -175,6 +191,7 @@ def main(argv: list[str] | None = None) -> int: "approvalRefWritten": approval_ref_written, "approvalRefPresent": bool(args.allow_paid), "allowedChatIdPresent": bool(args.allowed_chat_id), + "allowedChatTitlePresent": bool(args.allowed_chat_title), "maxUsd": max_usd, "paidEnabled": bool(args.allow_paid), "dryRun": args.dry_run, diff --git a/telegram/bot.py b/telegram/bot.py index 475d4a8..d8a6da0 100644 --- a/telegram/bot.py +++ b/telegram/bot.py @@ -63,6 +63,7 @@ from http_chat_proxy import ( extract_smart_research_goal, post_chat_proxy, should_attach_structured_market_context, + smart_research_chat_matches_gate, smart_research_payment_fields_for_message, smart_research_command_names, split_smart_research_auto_resume_reply, @@ -740,14 +741,20 @@ def sanitize_message(text: str) -> str: return text[:2000] -def _smart_research_payment_gate(chat_id: int) -> dict: +def _smart_research_payment_gate(chat_id: int, chat_title: str | None = None) -> dict: """Return paid smart-research fields only when all server-side gates pass.""" max_allowed_usd = 0.06 if os.getenv("LIVINGIP_LEO_TELEGRAM_SMART_RESEARCH_ALLOW_PAID") != "1": return {"allow_paid_execution": False} allowed_chat_id = os.getenv("LIVINGIP_LEO_TELEGRAM_SMART_RESEARCH_ALLOWED_CHAT_ID", "").strip() - if not allowed_chat_id or allowed_chat_id != str(chat_id): + allowed_chat_title = os.getenv("LIVINGIP_LEO_TELEGRAM_SMART_RESEARCH_ALLOWED_CHAT_TITLE", "").strip() + if not smart_research_chat_matches_gate( + chat_id=chat_id, + chat_title=chat_title, + allowed_chat_id=allowed_chat_id, + allowed_chat_title=allowed_chat_title, + ): return {"allow_paid_execution": False} try: @@ -1391,7 +1398,10 @@ async def handle_tagged(update: Update, context: ContextTypes.DEFAULT_TYPE): if AGENT_HTTP_RESEARCH_PROXY_URL and smart_research_goal: payment_gate = smart_research_payment_fields_for_message( paid_work_order_id=paid_work_order_id, - configured_payment_gate=_smart_research_payment_gate(msg.chat_id), + configured_payment_gate=_smart_research_payment_gate( + msg.chat_id, + msg.chat.title if getattr(msg, "chat", None) else None, + ), ) proxy_research_goal = smart_research_goal if should_attach_structured_market_context(smart_research_goal): diff --git a/telegram/http_chat_proxy.py b/telegram/http_chat_proxy.py index 9dcfce6..7b50e48 100644 --- a/telegram/http_chat_proxy.py +++ b/telegram/http_chat_proxy.py @@ -77,6 +77,36 @@ _SOCIAL_DISCUSSION_RE = re.compile( ) +def _normalize_chat_title(value: str | None) -> str: + return re.sub(r"\s+", " ", str(value or "")).strip().casefold() + + +def smart_research_chat_matches_gate( + *, + chat_id: int, + chat_title: str | None = None, + allowed_chat_id: str | None = None, + allowed_chat_title: str | None = None, +) -> bool: + """Return whether a Telegram chat matches the configured paid-research gate.""" + if allowed_chat_id and allowed_chat_id.strip() == str(chat_id): + return True + + if not allowed_chat_title: + return False + + normalized_title = _normalize_chat_title(chat_title) + if not normalized_title: + return False + + allowed_titles = { + _normalize_chat_title(part) + for part in re.split(r"[,;\n]", allowed_chat_title) + if _normalize_chat_title(part) + } + return normalized_title in allowed_titles + + def smart_research_command_names( command_prefixes: tuple[str, ...] | list[str] = DEFAULT_SMART_RESEARCH_COMMAND_PREFIXES, ) -> list[str]: diff --git a/tests/test_install_telegram_smart_research_gates.py b/tests/test_install_telegram_smart_research_gates.py index 1fe317b..f79a418 100644 --- a/tests/test_install_telegram_smart_research_gates.py +++ b/tests/test_install_telegram_smart_research_gates.py @@ -61,6 +61,7 @@ def test_installs_paid_gate_from_stdin_without_echoing_secret_or_chat_id(tmp_pat assert proof["paidEnabled"] is True assert proof["approvalRefPresent"] is True assert proof["allowedChatIdPresent"] is True + assert proof["allowedChatTitlePresent"] is False assert proof["maxUsd"] == "0.06" assert proof["secretValuesIncluded"] is False @@ -75,6 +76,53 @@ def test_installs_paid_gate_from_stdin_without_echoing_secret_or_chat_id(tmp_pat assert stat.S_IMODE(os.stat(approval_path).st_mode) == 0o600 +def test_installs_paid_gate_with_title_fallback_without_echoing_secret(tmp_path): + approval_ref = "approval_ref_livingip_x402_20260622" + chat_title = "Leo" + proof_path = tmp_path / "proof.json" + proc = run_installer( + [ + "--agent", + "leo", + "--secrets-dir", + str(tmp_path / "secrets"), + "--allow-paid", + "--allowed-chat-title", + chat_title, + "--max-usd", + "0.01", + "--approval-ref-from-stdin", + "--no-chown", + "--output", + str(proof_path), + ], + approval_ref=approval_ref, + ) + + assert proc.returncode == 0, proc.stderr + combined_output = proc.stdout + proc.stderr + proof_path.read_text() + assert approval_ref not in combined_output + + proof = json.loads(proof_path.read_text()) + env_path = Path(proof["envPath"]) + approval_path = Path(proof["approvalRefPath"]) + assert proof["ok"] is True + assert proof["agent"] == "leo" + assert proof["paidEnabled"] is True + assert proof["approvalRefPresent"] is True + assert proof["allowedChatIdPresent"] is False + assert proof["allowedChatTitlePresent"] is True + assert proof["maxUsd"] == "0.01" + assert proof["secretValuesIncluded"] is False + + env_content = env_path.read_text() + assert "LIVINGIP_LEO_TELEGRAM_SMART_RESEARCH_ALLOW_PAID=1" in env_content + assert "LIVINGIP_LEO_TELEGRAM_SMART_RESEARCH_ALLOWED_CHAT_TITLE=Leo" in env_content + assert "LIVINGIP_LEO_TELEGRAM_SMART_RESEARCH_ALLOWED_CHAT_ID=" not in env_content + assert f"LIVINGIP_LEO_TELEGRAM_SMART_RESEARCH_APPROVAL_REF_FILE={approval_path}" in env_content + assert approval_path.read_text().strip() == approval_ref + + def test_installs_disabled_gate_without_approval_ref(tmp_path): proof_path = tmp_path / "proof.json" proc = run_installer( diff --git a/tests/test_telegram_leo_x402_bridge.py b/tests/test_telegram_leo_x402_bridge.py index 6b170be..1591d28 100644 --- a/tests/test_telegram_leo_x402_bridge.py +++ b/tests/test_telegram_leo_x402_bridge.py @@ -23,6 +23,7 @@ from http_chat_proxy import ( # noqa: E402 extract_chat_proxy_reply, extract_smart_research_goal, should_attach_structured_market_context, + smart_research_chat_matches_gate, split_smart_research_auto_resume_reply, smart_research_payment_fields_for_message, smart_research_command_names, @@ -192,6 +193,33 @@ def test_smart_research_payment_fields_resume_uses_capped_gate(): assert payment_fields == configured_gate +def test_smart_research_chat_gate_allows_exact_chat_id(): + assert smart_research_chat_matches_gate( + chat_id=-1001234567890, + chat_title="Different title", + allowed_chat_id="-1001234567890", + allowed_chat_title=None, + ) + + +def test_smart_research_chat_gate_allows_explicit_title_fallback(): + assert smart_research_chat_matches_gate( + chat_id=-1001111111111, + chat_title=" Leo ", + allowed_chat_id="8331406290", + allowed_chat_title="Leo", + ) + + +def test_smart_research_chat_gate_rejects_unmatched_chat(): + assert not smart_research_chat_matches_gate( + chat_id=-1001111111111, + chat_title="Other group", + allowed_chat_id="8331406290", + allowed_chat_title="Leo", + ) + + @pytest.mark.parametrize( ("http_status", "payload", "expected"), [