From f59aa1a758a5efb57036559592b543e8d2f420dc Mon Sep 17 00:00:00 2001 From: curo1305 Date: Tue, 19 May 2026 23:15:22 +0200 Subject: [PATCH] feat(plugin/telegram_bot): replace bare prompts with 5-step guided setup wizard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 1 — create bot via @BotFather (instructions + press-any-key pause) Step 2 — find Telegram user ID via @userinfobot (instructions + pause) Step 3 — set session passphrase with security explanation Step 4 — save all three vault keys, print ✓ confirmations Step 5 — configuration complete marker Adds setup cancellation on empty token, updated tests: happy path, mismatch, and cancel all covered; press_any_key_to_continue calls properly patched. Co-Authored-By: Claude Sonnet 4.6 --- .../bundled_plugins/telegram_bot/plugin.py | 95 ++++++++++++++++--- tests/unit/test_telegram_bot.py | 57 +++++++++-- 2 files changed, 135 insertions(+), 17 deletions(-) diff --git a/src/pyra/bundled_plugins/telegram_bot/plugin.py b/src/pyra/bundled_plugins/telegram_bot/plugin.py index fd93573..03e49a4 100644 --- a/src/pyra/bundled_plugins/telegram_bot/plugin.py +++ b/src/pyra/bundled_plugins/telegram_bot/plugin.py @@ -129,35 +129,108 @@ class TelegramBotPlugin(BasePlugin): def setup(self, console: Any, vault_writer: Callable[[str, str], None]) -> None: import questionary + from rich.panel import Panel + from rich.rule import Rule - console.print("[bold]Telegram Bot Setup[/bold]") + console.print() + console.print(Panel( + "[bold]Telegram Bot Setup Wizard[/bold]\n\n" + "This wizard connects Pyra to Telegram so you can chat with your\n" + "assistant from anywhere. You will need Telegram open on your phone\n" + "or desktop to complete the next steps.", + border_style="cyan", + )) + + # ── Step 1: Create bot ──────────────────────────────────────────────── + console.print() + console.print(Rule("[bold cyan]Step 1 / 5[/bold cyan] Create your Telegram bot")) + console.print() console.print( - "1. Create a bot via @BotFather on Telegram to get your bot token.\n" - "2. Find your Telegram user ID by messaging @userinfobot.\n" + " 1. Open Telegram and search for [bold]@BotFather[/bold]\n" + " 2. Send [bold]/newbot[/bold] and follow the prompts\n" + " 3. Choose a display name (e.g. [dim]My Pyra Assistant[/dim])\n" + " 4. Choose a username ending in [bold]bot[/bold] " + "(e.g. [dim]my_pyra_bot[/dim])\n" + " 5. BotFather replies with a token that looks like:\n" + " [dim]123456789:AABBccDDeeFFggHHiiJJkkLL[/dim]" ) + console.print() + questionary.press_any_key_to_continue( + " Press any key when you have your token ready ..." + ).ask() + console.print() - token = questionary.password("Bot token (from @BotFather):").ask() - if not token: + token = questionary.password(" Bot token:").ask() + if not token or not token.strip(): + console.print("[dim]Setup cancelled.[/dim]") return + token = token.strip() + + # ── Step 2: Find user ID ────────────────────────────────────────────── + console.print() + console.print(Rule("[bold cyan]Step 2 / 5[/bold cyan] Find your Telegram user ID")) + console.print() + console.print( + " Your user ID is a permanent number that identifies your account.\n" + " It never changes, even if you change your username.\n\n" + " 1. Search for [bold]@userinfobot[/bold] in Telegram\n" + " 2. Send any message (e.g. [dim]/start[/dim])\n" + " 3. Copy the [bold]Id:[/bold] number from the reply " + "(e.g. [dim]123456789[/dim])" + ) + console.print() + questionary.press_any_key_to_continue( + " Press any key when you have your user ID ready ..." + ).ask() + console.print() allowed = questionary.text( - "Allowed Telegram user IDs (comma-separated, e.g. 123456789):" + " Allowed Telegram user IDs (comma-separated, leave blank to allow anyone):" ).ask() + if allowed is None: + console.print("[dim]Setup cancelled.[/dim]") + return - passphrase = questionary.password("Session passphrase:").ask() + # ── Step 3: Session passphrase ──────────────────────────────────────── + console.print() + console.print(Rule("[bold cyan]Step 3 / 5[/bold cyan] Set a session passphrase")) + console.print() + console.print( + " The passphrase is an extra layer of security. Every new chat\n" + " session must pass this challenge before Pyra responds — even\n" + " if someone else gains access to your Telegram account." + ) + console.print() + + passphrase = questionary.password(" Session passphrase:").ask() if not passphrase: + console.print("[dim]Setup cancelled.[/dim]") return - confirm = questionary.password("Confirm passphrase:").ask() + confirm = questionary.password(" Confirm passphrase:").ask() if passphrase != confirm: - console.print("[red]Passphrases do not match.[/red]") + console.print("[red]Passphrases do not match. Run setup again to retry.[/red]") return + # ── Step 4: Save to vault ───────────────────────────────────────────── + console.print() + console.print(Rule("[bold cyan]Step 4 / 5[/bold cyan] Saving configuration")) + console.print() + pw_hash = bcrypt.hashpw(passphrase.encode(), bcrypt.gensalt()).decode() - vault_writer("plugin:telegram_bot:token", token.strip()) + vault_writer("plugin:telegram_bot:token", token) vault_writer("plugin:telegram_bot:allowed_users", (allowed or "").strip()) vault_writer("plugin:telegram_bot:passphrase_hash", pw_hash) - console.print("[green]Telegram bot configured. Enable it and start the daemon.[/green]") + + allowed_display = (allowed or "").strip() or "[dim](any user — consider restricting)[/dim]" + console.print(f" [green]✓[/green] Bot token stored in vault") + console.print(f" [green]✓[/green] Allowed users: {allowed_display}") + console.print(f" [green]✓[/green] Passphrase stored as bcrypt hash") + + # ── Step 5: Done ────────────────────────────────────────────────────── + console.print() + console.print(Rule("[bold cyan]Step 5 / 5[/bold cyan] Configuration complete")) + console.print() def config_fields(self) -> list[ConfigField]: return [ diff --git a/tests/unit/test_telegram_bot.py b/tests/unit/test_telegram_bot.py index 247e010..f835b17 100644 --- a/tests/unit/test_telegram_bot.py +++ b/tests/unit/test_telegram_bot.py @@ -166,19 +166,64 @@ class TestPluginLifecycle: fields = plugin.config_fields() assert any(f.key == "rate_limit" for f in fields) - def test_setup_mismatched_passphrase(self, capsys): - """setup() exits cleanly when passphrases don't match.""" + def _patch_setup(self, token, allowed, pass1, pass2): + """Return a context manager that patches all questionary calls used by setup().""" + pw_answers = iter([token, pass1, pass2]) + return ( + patch("questionary.password", + side_effect=lambda *a, **kw: MagicMock(ask=lambda: next(pw_answers))), + patch("questionary.text", + return_value=MagicMock(ask=lambda: allowed)), + patch("questionary.press_any_key_to_continue", + return_value=MagicMock(ask=lambda: None)), + ) + + def test_setup_mismatched_passphrase(self): + """setup() writes nothing to the vault when passphrases don't match.""" plugin = TelegramBotPlugin() console = MagicMock() vault_writer = MagicMock() - answers = iter(["fake-token", "user123", "pass1", "pass2"]) - with patch("questionary.password", side_effect=lambda *a, **kw: MagicMock(ask=lambda: next(answers))), \ - patch("questionary.text", return_value=MagicMock(ask=lambda: next(answers))): + pw_patch, text_patch, pakc_patch = self._patch_setup( + "fake-token", "123456789", "pass1", "pass2" + ) + with pw_patch, text_patch, pakc_patch: + plugin.setup(console, vault_writer) + + vault_writer.assert_not_called() + console.print.assert_called_with( + "[red]Passphrases do not match. Run setup again to retry.[/red]" + ) + + def test_setup_happy_path(self): + """setup() writes all three vault keys when credentials are valid.""" + plugin = TelegramBotPlugin() + console = MagicMock() + vault_writer = MagicMock() + + pw_patch, text_patch, pakc_patch = self._patch_setup( + "real-token", "111222333", "s3cr3t", "s3cr3t" + ) + with pw_patch, text_patch, pakc_patch: + plugin.setup(console, vault_writer) + + calls = {call[0][0]: call[0][1] for call in vault_writer.call_args_list} + assert calls.get("plugin:telegram_bot:token") == "real-token" + assert calls.get("plugin:telegram_bot:allowed_users") == "111222333" + assert "plugin:telegram_bot:passphrase_hash" in calls + + def test_setup_cancelled_on_empty_token(self): + """setup() exits without writing if the token prompt is cancelled.""" + plugin = TelegramBotPlugin() + console = MagicMock() + vault_writer = MagicMock() + + with patch("questionary.password", return_value=MagicMock(ask=lambda: None)), \ + patch("questionary.press_any_key_to_continue", + return_value=MagicMock(ask=lambda: None)): plugin.setup(console, vault_writer) vault_writer.assert_not_called() - console.print.assert_called_with("[red]Passphrases do not match.[/red]") # ── Auth session state ────────────────────────────────────────────────────────