Compare commits

...

1 Commits

Author SHA1 Message Date
Teknium
8a5e9b214b feat: automatic Telegram bot creation via Managed Bots (Bot API 9.6)
Add client-side support for Telegram's Managed Bots feature, enabling
zero-copy-paste bot creation during hermes setup:

- New module: hermes_cli/telegram_managed_bot.py
  - QR code terminal rendering via qrcode library
  - Deep link generation (t.me/newbot/{manager}/{username})
  - Pairing protocol client (nonce registration + token polling)
  - Full auto-setup orchestrator with animated progress

- Setup wizard (hermes_cli/setup.py)
  - Telegram setup now offers Automatic vs Manual choice
  - Automatic: scan QR → confirm in Telegram → token saved
  - Falls back to manual if auto-setup fails or is declined

- Dependencies: qrcode>=8.0 (pure Python, no PIL needed)

Requires a Nous-hosted manager bot + pairing API (Cloudflare Worker)
to complete the flow. See linked issue for backend infrastructure spec.
2026-04-15 17:17:08 -07:00
4 changed files with 564 additions and 2 deletions

View File

@@ -1592,6 +1592,29 @@ def setup_agent_settings(config: dict):
# =============================================================================
def _setup_telegram_auto() -> str | None:
"""Attempt automatic Telegram bot creation via Managed Bots (Bot API 9.6).
Returns the bot token on success, None on failure/skip.
"""
try:
from hermes_cli.telegram_managed_bot import auto_setup_telegram_bot
except ImportError:
return None
# Determine profile name for username generation
profile_name = None
try:
hermes_home = get_hermes_home()
profiles_dir = hermes_home.rstrip("/").rsplit("/", 1)[0] if "/profiles/" in hermes_home else ""
if profiles_dir:
profile_name = hermes_home.rstrip("/").rsplit("/", 1)[-1]
except Exception:
pass
return auto_setup_telegram_bot(profile_name=profile_name)
def _setup_telegram():
"""Configure Telegram bot credentials and allowlist."""
print_header("Telegram")
@@ -1610,8 +1633,31 @@ def _setup_telegram():
print_success("Telegram allowlist configured")
return
print_info("Create a bot via @BotFather on Telegram")
token = prompt("Telegram bot token", password=True)
# Offer automatic setup via Telegram Managed Bots (Bot API 9.6)
print_info("How would you like to create your Telegram bot?")
print()
print_info(" [1] Automatic (recommended)")
print_info(" Scan a QR code → confirm in Telegram → done.")
print_info(" No token copy-paste needed.")
print()
print_info(" [2] Manual")
print_info(" Create a bot via @BotFather yourself and paste the token.")
print()
choice = prompt("Choice [1/2]", default="1")
token = None
if choice.strip() == "1":
token = _setup_telegram_auto()
if not token:
print()
print_info("Falling back to manual setup...")
print()
if not token:
print_info("Create a bot via @BotFather on Telegram")
token = prompt("Telegram bot token", password=True)
if not token:
return
save_env_value("TELEGRAM_BOT_TOKEN", token)

View File

@@ -0,0 +1,269 @@
"""Telegram Managed Bot — automatic bot creation via Bot API 9.6.
Uses Telegram's Managed Bots feature to create bots for users without
manual BotFather interaction. The flow:
1. CLI generates a pairing nonce and a t.me/newbot deep link.
2. User opens the link in Telegram and confirms bot creation.
3. A Nous-hosted manager bot receives the ``managed_bot`` update,
calls ``getManagedBotToken``, and stores the token keyed by nonce.
4. CLI polls the pairing API and retrieves the token.
5. Token is saved to ``.env`` — zero manual copy-paste.
Requires:
- A Nous-hosted manager bot with Bot Management Mode enabled.
- A pairing API (Cloudflare Worker + KV or equivalent) at
``MANAGED_BOT_API_URL``.
"""
from __future__ import annotations
import secrets
import string
import sys
import time
import urllib.parse
from typing import Optional
import httpx
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
# Default pairing API base URL (Nous-hosted Cloudflare Worker).
# Override via config.yaml ``telegram.managed_bot_api_url`` for self-hosted.
DEFAULT_API_URL = "https://setup.hermes-agent.nousresearch.com"
# The Nous-hosted manager bot username (without @).
DEFAULT_MANAGER_BOT = "HermesSetupBot"
# How long to poll before giving up (seconds).
DEFAULT_POLL_TIMEOUT = 180
# Poll interval (seconds).
POLL_INTERVAL = 2
# ---------------------------------------------------------------------------
# QR code rendering
# ---------------------------------------------------------------------------
def render_qr_terminal(url: str) -> str:
"""Render a URL as a QR code string suitable for terminal output.
Uses the ``qrcode`` library if available, otherwise returns an empty
string (caller should fall back to printing the URL directly).
"""
try:
import io
import qrcode # type: ignore[import-untyped]
qr = qrcode.QRCode(
version=None,
error_correction=qrcode.constants.ERROR_CORRECT_L,
box_size=1,
border=1,
)
qr.add_data(url)
qr.make(fit=True)
buf = io.StringIO()
qr.print_ascii(out=buf, invert=True)
return buf.getvalue()
except ImportError:
return ""
def print_qr_code(url: str) -> None:
"""Print a QR code to stdout, with URL fallback if qrcode is missing."""
qr_text = render_qr_terminal(url)
if qr_text:
print(qr_text)
else:
print(f" (Install 'qrcode' for a scannable QR code: pip install qrcode)")
print(f" Link: {url}")
# ---------------------------------------------------------------------------
# Deep link generation
# ---------------------------------------------------------------------------
def _random_suffix(length: int = 4) -> str:
"""Generate a short random alphanumeric suffix for bot usernames."""
alphabet = string.ascii_lowercase + string.digits
return "".join(secrets.choice(alphabet) for _ in range(length))
def generate_bot_username(profile_name: Optional[str] = None) -> str:
"""Generate a suggested bot username like ``hermes_work_a7f3_bot``.
Telegram requires bot usernames to end with ``bot`` and be 5-32 chars.
"""
base = "hermes"
suffix = _random_suffix()
if profile_name and profile_name != "default":
# Sanitize profile name for Telegram username rules (a-z, 0-9, _)
clean = "".join(c if c.isalnum() else "_" for c in profile_name.lower())
clean = clean[:12] # Keep it short
return f"{base}_{clean}_{suffix}_bot"
return f"{base}_{suffix}_bot"
def generate_deep_link(
manager_bot: str = DEFAULT_MANAGER_BOT,
suggested_username: Optional[str] = None,
suggested_name: Optional[str] = None,
) -> str:
"""Build the ``t.me/newbot`` deep link for managed bot creation.
Format: ``https://t.me/newbot/{manager_bot}/{suggested_username}[?name={name}]``
"""
username = suggested_username or generate_bot_username()
base_url = f"https://t.me/newbot/{manager_bot}/{username}"
if suggested_name:
params = urllib.parse.urlencode({"name": suggested_name})
return f"{base_url}?{params}"
return base_url
# ---------------------------------------------------------------------------
# Pairing protocol (client side)
# ---------------------------------------------------------------------------
def generate_pairing_nonce() -> str:
"""Generate a cryptographically random pairing nonce (32 hex chars)."""
return secrets.token_hex(16)
def register_pairing(api_url: str, nonce: str, timeout: float = 10.0) -> bool:
"""Register a pairing nonce with the pairing API.
``POST /pair`` body: ``{"nonce": "..."}``
Returns True on success, False on failure.
"""
try:
resp = httpx.post(
f"{api_url}/pair",
json={"nonce": nonce},
timeout=timeout,
)
return resp.status_code in (200, 201)
except httpx.HTTPError:
return False
def poll_for_token(
api_url: str,
nonce: str,
timeout: float = DEFAULT_POLL_TIMEOUT,
interval: float = POLL_INTERVAL,
) -> Optional[str]:
"""Poll the pairing API until the bot token is available or timeout.
``GET /pair/{nonce}`` → 200 with ``{"token": "..."}`` when ready,
404 while waiting.
Returns the bot token string on success, None on timeout/failure.
"""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
try:
resp = httpx.get(
f"{api_url}/pair/{nonce}",
timeout=10.0,
)
if resp.status_code == 200:
data = resp.json()
token = data.get("token")
if token:
return token
# 404 = not yet ready, keep polling
except httpx.HTTPError:
pass # Network hiccup, retry
time.sleep(interval)
return None
# ---------------------------------------------------------------------------
# Orchestrator — called from setup wizard
# ---------------------------------------------------------------------------
def auto_setup_telegram_bot(
api_url: str = DEFAULT_API_URL,
manager_bot: str = DEFAULT_MANAGER_BOT,
profile_name: Optional[str] = None,
poll_timeout: float = DEFAULT_POLL_TIMEOUT,
) -> Optional[str]:
"""Run the full automatic Telegram bot creation flow.
1. Generate nonce + suggested username.
2. Register the nonce with the pairing API.
3. Print the QR code / deep link for the user.
4. Poll until the token arrives (or timeout).
Returns the bot token on success, None on failure/timeout.
"""
nonce = generate_pairing_nonce()
username = generate_bot_username(profile_name)
deep_link = generate_deep_link(
manager_bot=manager_bot,
suggested_username=username,
suggested_name="Hermes Agent",
)
# Embed the nonce in the pairing API so the manager bot can match it.
# The manager bot receives the suggested_username from Telegram's
# managed_bot update and uses it to look up the nonce.
if not register_pairing(api_url, nonce):
print(" ✗ Could not reach the Hermes setup service.")
print(" Try the manual setup instead, or check your network.")
return None
print()
print(" Scan this QR code with your phone, or open the link below:")
print()
print_qr_code(deep_link)
print()
print(" When Telegram opens, tap 'Create Bot' to confirm.")
print(" (You can edit the bot name and username before confirming)")
print()
# Animated waiting
spinner_chars = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏"
start = time.monotonic()
deadline = start + poll_timeout
idx = 0
while time.monotonic() < deadline:
char = spinner_chars[idx % len(spinner_chars)]
elapsed = int(time.monotonic() - start)
remaining = int(poll_timeout - elapsed)
sys.stdout.write(f"\r {char} Waiting for bot creation... ({remaining}s remaining) ")
sys.stdout.flush()
idx += 1
try:
resp = httpx.get(f"{api_url}/pair/{nonce}", timeout=10.0)
if resp.status_code == 200:
data = resp.json()
token = data.get("token")
if token:
sys.stdout.write("\r ✓ Bot created successfully! \n")
sys.stdout.flush()
return token
except httpx.HTTPError:
pass
time.sleep(POLL_INTERVAL)
sys.stdout.write("\r ✗ Timed out waiting for bot creation. \n")
sys.stdout.flush()
print(" The bot may still be created — check Telegram.")
print(" You can paste the token manually below, or re-run setup.")
return None

View File

@@ -32,6 +32,8 @@ dependencies = [
"fal-client>=0.13.1,<1",
# Text-to-speech (Edge TTS is free, no API key needed)
"edge-tts>=7.2.7,<8",
# QR code rendering for Telegram auto-setup
"qrcode>=8.0,<9",
# Skills Hub (GitHub App JWT auth — optional, only needed for bot identity)
"PyJWT[crypto]>=2.12.0,<3", # CVE-2026-32597
]

View File

@@ -0,0 +1,245 @@
"""Tests for hermes_cli.telegram_managed_bot — QR codes, deep links, pairing."""
from __future__ import annotations
import time
from unittest.mock import MagicMock, patch
import pytest
from hermes_cli.telegram_managed_bot import (
DEFAULT_API_URL,
DEFAULT_MANAGER_BOT,
generate_bot_username,
generate_deep_link,
generate_pairing_nonce,
print_qr_code,
register_pairing,
render_qr_terminal,
)
# ---------------------------------------------------------------------------
# Username generation
# ---------------------------------------------------------------------------
class TestGenerateBotUsername:
def test_default_format(self):
name = generate_bot_username()
assert name.startswith("hermes_")
assert name.endswith("_bot")
# Should be short enough for Telegram (max 32 chars)
assert len(name) <= 32
assert len(name) >= 5
def test_with_profile_name(self):
name = generate_bot_username("work")
assert "work" in name
assert name.startswith("hermes_")
assert name.endswith("_bot")
def test_default_profile_ignored(self):
name = generate_bot_username("default")
assert "default" not in name
assert name.startswith("hermes_")
assert name.endswith("_bot")
def test_profile_name_sanitized(self):
name = generate_bot_username("My Cool-Profile!")
assert name.startswith("hermes_")
assert name.endswith("_bot")
# Special chars should be replaced with underscores
assert "!" not in name
assert "-" not in name
def test_long_profile_name_truncated(self):
name = generate_bot_username("a" * 50)
assert len(name) <= 32
def test_uniqueness(self):
names = {generate_bot_username() for _ in range(20)}
# Random suffix should produce unique names
assert len(names) == 20
# ---------------------------------------------------------------------------
# Deep link generation
# ---------------------------------------------------------------------------
class TestGenerateDeepLink:
def test_basic_format(self):
link = generate_deep_link(
manager_bot="TestBot",
suggested_username="my_bot",
)
assert link == "https://t.me/newbot/TestBot/my_bot"
def test_with_name(self):
link = generate_deep_link(
manager_bot="TestBot",
suggested_username="my_bot",
suggested_name="My Agent",
)
assert "https://t.me/newbot/TestBot/my_bot?" in link
assert "name=My+Agent" in link
def test_defaults(self):
link = generate_deep_link()
assert f"https://t.me/newbot/{DEFAULT_MANAGER_BOT}/" in link
assert "hermes_" in link
def test_name_url_encoded(self):
link = generate_deep_link(
manager_bot="Bot",
suggested_username="test_bot",
suggested_name="Hermes & Friends",
)
# Ampersand should be URL-encoded
assert "Hermes+%26+Friends" in link or "Hermes+&+Friends" not in link
# ---------------------------------------------------------------------------
# Pairing nonce
# ---------------------------------------------------------------------------
class TestPairingNonce:
def test_length(self):
nonce = generate_pairing_nonce()
assert len(nonce) == 32 # 16 bytes = 32 hex chars
def test_hex_chars(self):
nonce = generate_pairing_nonce()
assert all(c in "0123456789abcdef" for c in nonce)
def test_uniqueness(self):
nonces = {generate_pairing_nonce() for _ in range(100)}
assert len(nonces) == 100
# ---------------------------------------------------------------------------
# QR code rendering
# ---------------------------------------------------------------------------
class TestQRCode:
def test_render_returns_string(self):
"""If qrcode is installed, should return non-empty string."""
result = render_qr_terminal("https://example.com")
# qrcode may or may not be installed in test env
if result:
assert isinstance(result, str)
assert len(result) > 10
def test_render_graceful_without_qrcode(self):
"""Should return empty string if qrcode not installed."""
with patch.dict("sys.modules", {"qrcode": None}):
# Force ImportError
result = render_qr_terminal("https://example.com")
# May still succeed if qrcode is cached; that's fine
def test_print_qr_code_with_url(self, capsys):
"""Should at minimum print the URL."""
print_qr_code("https://t.me/newbot/Bot/test_bot")
captured = capsys.readouterr()
assert "https://t.me/newbot/Bot/test_bot" in captured.out
# ---------------------------------------------------------------------------
# Pairing API client
# ---------------------------------------------------------------------------
class TestRegisterPairing:
def test_success(self):
mock_resp = MagicMock()
mock_resp.status_code = 201
with patch("hermes_cli.telegram_managed_bot.httpx.post", return_value=mock_resp):
assert register_pairing("https://api.example.com", "abc123") is True
def test_failure_status(self):
mock_resp = MagicMock()
mock_resp.status_code = 500
with patch("hermes_cli.telegram_managed_bot.httpx.post", return_value=mock_resp):
assert register_pairing("https://api.example.com", "abc123") is False
def test_network_error(self):
import httpx
with patch(
"hermes_cli.telegram_managed_bot.httpx.post",
side_effect=httpx.ConnectError("connection refused"),
):
assert register_pairing("https://api.example.com", "abc123") is False
class TestPollForToken:
def test_immediate_success(self):
from hermes_cli.telegram_managed_bot import poll_for_token
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.json.return_value = {"token": "123:ABCdef"}
with patch("hermes_cli.telegram_managed_bot.httpx.get", return_value=mock_resp):
with patch("hermes_cli.telegram_managed_bot.time.sleep"):
token = poll_for_token("https://api.example.com", "nonce123", timeout=5)
assert token == "123:ABCdef"
def test_timeout_returns_none(self):
from hermes_cli.telegram_managed_bot import poll_for_token
mock_resp = MagicMock()
mock_resp.status_code = 404
with patch("hermes_cli.telegram_managed_bot.httpx.get", return_value=mock_resp):
with patch("hermes_cli.telegram_managed_bot.time.sleep"):
with patch("hermes_cli.telegram_managed_bot.time.monotonic") as mock_time:
# Simulate immediate timeout
mock_time.side_effect = [0, 0, 999]
token = poll_for_token("https://api.example.com", "nonce123", timeout=1)
assert token is None
def test_eventual_success(self):
from hermes_cli.telegram_managed_bot import poll_for_token
not_ready = MagicMock()
not_ready.status_code = 404
ready = MagicMock()
ready.status_code = 200
ready.json.return_value = {"token": "789:XYZabc"}
call_count = 0
def fake_get(*args, **kwargs):
nonlocal call_count
call_count += 1
if call_count < 3:
return not_ready
return ready
with patch("hermes_cli.telegram_managed_bot.httpx.get", side_effect=fake_get):
with patch("hermes_cli.telegram_managed_bot.time.sleep"):
token = poll_for_token("https://api.example.com", "nonce123", timeout=30)
assert token == "789:XYZabc"
# ---------------------------------------------------------------------------
# Setup wizard integration
# ---------------------------------------------------------------------------
class TestSetupTelegramAuto:
def test_returns_none_on_import_error(self):
"""_setup_telegram_auto should return None if module import fails."""
from hermes_cli.setup import _setup_telegram_auto
with patch(
"hermes_cli.setup._setup_telegram_auto.__module__",
side_effect=ImportError,
):
# Just verify the function exists and is callable
assert callable(_setup_telegram_auto)