基于AIogram与Pyrogram的Telegram机器人用户授权[400 CODE_EXPIRED]问题

我使用Python 3.12、AIogram 3.x和Pyrogram 2.0.106开发Telegram机器人，目前遇到用户授权相关问题：

• 代码运行状态不稳定，有时提交失败，有时抛出未知错误
• 最致命的问题是触发[400 CODE_EXPIRED]错误——我在收到验证码后10秒内就提交，按逻辑验证码应该仍处于有效期内

以下是我当前的代码：

from aiogram import Router, F, types
from aiogram.filters import CommandStart
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import StatesGroup, State
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup

from pyrogram import Client

class UserAuth(StatesGroup):
    waiting_for_phone = State()
    waiting_for_code = State()


@router.callback_query(lambda cq: cq.data == "add_user")
async def add_user_callback(callback_query: types.CallbackQuery, state: FSMContext):
    await callback_query.answer()
    await callback_query.message.answer("Please enter your phone number:")
    await state.set_state(UserAuth.waiting_for_phone)


@router.message(UserAuth.waiting_for_phone)
async def process_phone(message: types.Message, state: FSMContext):
    phone_number = message.text.strip()
    await state.update_data(phone=phone_number)

    client = Client("user_session", api_id=api_id, api_hash=api_hash)
    await client.connect()

    try:
        sent_code = await client.send_code(phone_number)
        phone_code_hash = sent_code.phone_code_hash

        print(phone_code_hash)
        await message.answer("Code has been sent to your phone. Please enter the code in format: code12345")
        await state.update_data(sent_code=sent_code)
        await state.update_data(phone_code_hash=phone_code_hash)
        await state.set_state(UserAuth.waiting_for_code)
    except Exception as e:
        await message.answer(f"Failed to send code: {e}")
        await state.clear()


@router.message(UserAuth.waiting_for_code)
async def process_code(message: types.Message, state: FSMContext):
    code = message.text.strip()
    data = await state.get_data()
    phone_number = data.get("phone")
    phone_code_hash = data.get("phone_code_hash")

    client = Client("user_session", api_id=api_id, api_hash=api_hash)
    await client.connect()

    try:
        await client.sign_in(phone_number=str(phone_number), phone_code_hash=str(phone_code_hash), phone_code=code)
        await message.answer("Authorization successful! Now you can send messages.")
        await state.clear()

        await client.send_message(chat_id, "You have successfully logged in via the bot!")
    finally:
        await state.clear()
        await client.disconnect()

问题根源分析 #

1. Session实例不共享：process_phone和process_code中分别创建新的Pyrogram Client实例，两个实例的session状态完全独立。Pyrogram的send_code生成的验证码哈希与当前session绑定，后续用新session调用sign_in时，哈希与session不匹配，Telegram会判定验证码已过期。
2. 资源未正确释放：process_phone中创建的Client没有执行断开操作，可能导致session状态混乱、资源泄漏。
3. 错误处理不严谨：process_code的finally块重复调用state.clear()，且未针对Pyrogram特定错误（如两步验证、验证码错误）做针对性处理。

修复后的代码 #

from aiogram import Router, types
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import StatesGroup, State
from aiogram.types import CallbackQuery

from pyrogram import Client
from pyrogram.errors import BadRequest, SessionPasswordNeeded

# 请替换为你的实际配置
api_id = 123456
api_hash = "your_api_hash_here"
chat_id = 123456789

class UserAuth(StatesGroup):
    waiting_for_phone = State()
    waiting_for_code = State()
    waiting_for_password = State()  # 新增：处理两步验证场景

router = Router()

# 全局存储用户专属的Client实例（生产环境建议改用Redis等持久化存储）
user_clients = {}

@router.callback_query(lambda cq: cq.data == "add_user")
async def add_user_callback(callback_query: CallbackQuery, state: FSMContext):
    await callback_query.answer()
    await callback_query.message.answer("请输入你的电话号码：")
    await state.set_state(UserAuth.waiting_for_phone)

@router.message(UserAuth.waiting_for_phone)
async def process_phone(message: types.Message, state: FSMContext):
    phone_number = message.text.strip()
    await state.update_data(phone=phone_number)
    user_id = message.from_user.id

    # 创建用户专属的Client实例，避免session冲突
    client = Client(f"user_session_{user_id}", api_id=api_id, api_hash=api_hash)
    user_clients[user_id] = client
    
    try:
        await client.connect()
        sent_code = await client.send_code(phone_number)
        await state.update_data(phone_code_hash=sent_code.phone_code_hash)
        await message.answer("验证码已发送到你的手机，请输入验证码（如：12345）：")
        await state.set_state(UserAuth.waiting_for_code)
    except BadRequest as e:
        await message.answer(f"发送验证码失败：{str(e)}")
        await client.disconnect()
        del user_clients[user_id]
        await state.clear()
    except Exception as e:
        await message.answer(f"未知错误：{str(e)}")
        await client.disconnect()
        del user_clients[user_id]
        await state.clear()

@router.message(UserAuth.waiting_for_code)
async def process_code(message: types.Message, state: FSMContext):
    code = message.text.strip()
    data = await state.get_data()
    phone_number = data.get("phone")
    phone_code_hash = data.get("phone_code_hash")
    user_id = message.from_user.id

    if user_id not in user_clients:
        await message.answer("会话已过期，请重新开始授权流程")
        await state.clear()
        return

    client = user_clients[user_id]

    try:
        await client.sign_in(phone_number=phone_number, phone_code_hash=phone_code_hash, phone_code=code)
        await message.answer("授权成功！现在你可以发送消息了。")
        await client.send_message(chat_id, "你已通过机器人成功登录！")
        
        # 按需选择是否保留Client实例或断开
        # await client.disconnect()
        # del user_clients[user_id]
        await state.clear()
    except SessionPasswordNeeded:
        await message.answer("你的账号开启了两步验证，请输入密码：")
        await state.set_state(UserAuth.waiting_for_password)
    except BadRequest as e:
        error_msg = str(e)
        if "CODE_EXPIRED" in error_msg:
            await message.answer("验证码已过期，请重新获取")
        elif "PHONE_CODE_INVALID" in error_msg:
            await message.answer("验证码错误，请重新输入")
        else:
            await message.answer(f"登录失败：{error_msg}")
        await state.clear()
        await client.disconnect()
        del user_clients[user_id]
    except Exception as e:
        await message.answer(f"未知错误：{str(e)}")
        await state.clear()
        await client.disconnect()
        del user_clients[user_id]

@router.message(UserAuth.waiting_for_password)
async def process_password(message: types.Message, state: FSMContext):
    password = message.text.strip()
    data = await state.get_data()
    user_id = message.from_user.id

    if user_id not in user_clients:
        await message.answer("会话已过期，请重新开始授权流程")
        await state.clear()
        return

    client = user_clients[user_id]

    try:
        await client.check_password(password=password)
        await message.answer("授权成功！现在你可以发送消息了。")
        await client.send_message(chat_id, "你已通过机器人成功登录！")
        
        # 按需选择是否保留Client实例或断开
        # await client.disconnect()
        # del user_clients[user_id]
        await state.clear()
    except BadRequest as e:
        if "PASSWORD_HASH_INVALID" in str(e):
            await message.answer("密码错误，请重新输入")
        else:
            await message.answer(f"验证失败：{str(e)}")
            await state.clear()
            await client.disconnect()
            del user_clients[user_id]
    except Exception as e:
        await message.answer(f"未知错误：{str(e)}")
        await state.clear()
        await client.disconnect()
        del user_clients[user_id]

关键修复点 #

• 共享Client实例：为每个用户创建专属的Client实例并存储，确保send_code和sign_in使用同一个session，解决哈希不匹配导致的验证码过期问题。
• 完善错误处理：针对Pyrogram的特定错误类型做针对性提示，提升用户体验，同时避免无意义的报错信息。
• 资源管理：在错误发生时及时断开Client连接并删除实例，避免资源泄漏和session混乱。
• 支持两步验证：新增两步验证的处理逻辑，覆盖更多用户场景。

内容的提问来源于stack exchange，提问作者Junior-ila