bot.py - master - Amino/LeakBot

bot.py
432 строки15 KB
Amino
upload files
20 дек 2025, 13:38
20 дек 2025, 13:38c41443e
import asyncio
from typing import Optional

from aiogram import Bot, Dispatcher, F
from aiogram.enums import ParseMode
from aiogram.client.default import DefaultBotProperties
from aiogram.filters import CommandStart
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import (
    Message,
    CallbackQuery,
    InlineKeyboardMarkup,
    InlineKeyboardButton,
    WebAppInfo,
)
from sqlalchemy import select, update, desc
from sqlalchemy.ext.asyncio import AsyncSession
from datetime import datetime

from config import settings
from db import User, PasswordCheck, LoginLeak, get_session, init_db
from external import check_password_pwned, check_login_leak


class CheckStates(StatesGroup):
    waiting_for_password = State()
    waiting_for_login = State()


def main_menu_kb() -> InlineKeyboardMarkup:
    return InlineKeyboardMarkup(
        inline_keyboard=[
            [
                InlineKeyboardButton(
                    text="🔐 Проверить данные", callback_data="check_data"
                )
            ],
            [
                InlineKeyboardButton(
                    text="📜 История проверок", callback_data="history"
                )
            ],
            [
                InlineKeyboardButton(
                    text="📱 Открыть мини‑приложение",
                    web_app=WebAppInfo(url=f"{settings.miniapp_host.rstrip('/')}/app"),
                )
            ],
        ]
    )


def check_menu_kb() -> InlineKeyboardMarkup:
    return InlineKeyboardMarkup(
        inline_keyboard=[
            [InlineKeyboardButton(text="👤 Проверить логин", callback_data="check_login")],
            [
                InlineKeyboardButton(
                    text="🔑 Проверить пароль", callback_data="check_password"
                )
            ],
        ]
    )


def history_menu_kb() -> InlineKeyboardMarkup:
    return InlineKeyboardMarkup(
        inline_keyboard=[
            [
                InlineKeyboardButton(
                    text="🔑 История паролей", callback_data="history_passwords"
                )
            ],
            [
                InlineKeyboardButton(
                    text="👤 История логинов", callback_data="history_logins"
                )
            ],
            [
                InlineKeyboardButton(
                    text="⚠️ Только утечки", callback_data="history_compromised"
                )
            ],
            [
                InlineKeyboardButton(
                    text="◀️ Назад", callback_data="back_to_main"
                )
            ],
        ]
    )


async def get_or_create_user(
    session: AsyncSession, tg_id: int, username: Optional[str]
) -> User:
    result = await session.execute(select(User).where(User.telegram_id == tg_id))
    user = result.scalar_one_or_none()
    if user is None:
        user = User(telegram_id=tg_id, username=username, checks_count=0)
        session.add(user)
        await session.commit()
        await session.refresh(user)
    return user


dp = Dispatcher()


@dp.message(CommandStart())
async def cmd_start(message: Message, state: FSMContext, session: AsyncSession):
    user = await get_or_create_user(
        session, message.from_user.id, message.from_user.username
    )

    text = (
        "👋 Привет! Это бот для проверки логинов и паролей на предмет утечек.\n\n"
        f"Твой Telegram ID: <code>{user.telegram_id}</code>\n"
        f"Имя: <b>{user.username or '—'}</b>\n"
        f"Всего проверок: <b>{user.checks_count}</b>\n\n"
        "Нажми кнопку ниже, чтобы начать проверку."
    )
    await message.answer(text, reply_markup=main_menu_kb())


@dp.callback_query(F.data == "check_data")
async def cb_check_data(callback: CallbackQuery):
    await callback.message.edit_text(
        "Выбери тип проверки:", reply_markup=check_menu_kb()
    )
    await callback.answer()


@dp.callback_query(F.data == "check_password")
async def cb_check_password(callback: CallbackQuery, state: FSMContext):
    await callback.message.answer(
        "Отправь мне пароль, который нужно проверить.\n"
        "<b>Важно:</b> пароль будет захеширован перед отправкой в сервис утечек.",
        parse_mode=ParseMode.HTML,
    )
    await state.set_state(CheckStates.waiting_for_password)
    await callback.answer()


@dp.callback_query(F.data == "check_login")
async def cb_check_login(callback: CallbackQuery, state: FSMContext):
    await callback.message.answer("Отправь логин или email для проверки.")
    await state.set_state(CheckStates.waiting_for_login)
    await callback.answer()


@dp.message(CheckStates.waiting_for_password)
async def process_password(
    message: Message, state: FSMContext, session: AsyncSession
):
    raw_password = message.text.strip()
    if not raw_password:
        await message.answer("Пароль не должен быть пустым, попробуй ещё раз.")
        return

    user = await get_or_create_user(
        session, message.from_user.id, message.from_user.username
    )

    compromised, hash_prefix, details = await check_password_pwned(raw_password)

    session.add(
        PasswordCheck(
            user_id=user.id,
            hash_prefix=hash_prefix,
            compromised=compromised,
        )
    )
    await session.execute(
        update(User)
        .where(User.id == user.id)
        .values(checks_count=User.checks_count + 1)
    )
    await session.commit()

    if compromised:
        text = (
            "⚠️ <b>Пароль найден в утечках!</b>\n\n"
            f"{details}\n\n"
            "Рекомендуем срочно сменить этот пароль во всех сервисах."
        )
    else:
        text = (
            "✅ <b>Пароль не найден в известных утечках по данным сервиса.</b>\n\n"
            "Тем не менее, используй уникальные и сложные пароли."
        )

    await message.answer(text, parse_mode=ParseMode.HTML, reply_markup=main_menu_kb())
    await state.clear()


@dp.message(CheckStates.waiting_for_login)
async def process_login(message: Message, state: FSMContext, session: AsyncSession):
    login = message.text.strip()
    if not login:
        await message.answer("Логин не должен быть пустым, попробуй ещё раз.")
        return

    user = await get_or_create_user(
        session, message.from_user.id, message.from_user.username
    )

    compromised, source, details = await check_login_leak(login)

    session.add(
        LoginLeak(
            user_id=user.id,
            login=login,
            source=source or "unknown",
            compromised=compromised,
            details=details,
        )
    )
    await session.execute(
        update(User)
        .where(User.id == user.id)
        .values(checks_count=User.checks_count + 1)
    )
    await session.commit()

    if compromised:
        text = (
            "⚠️ <b>Логин/почта найдены в утечках!</b>\n\n"
            f"{details or 'Детали утечки доступны в источнике.'}\n\n"
            "Рекомендуем сменить пароли и включить 2FA."
        )
    else:
        text = "✅ <b>По этому логину утечек не найдено в доступных источниках.</b>"

    await message.answer(text, parse_mode=ParseMode.HTML, reply_markup=main_menu_kb())
    await state.clear()


@dp.callback_query(F.data == "history")
async def cb_history(callback: CallbackQuery):
    await callback.message.edit_text(
        "📜 <b>История проверок</b>\n\n"
        "Выбери, какую историю хочешь посмотреть:",
        reply_markup=history_menu_kb(),
        parse_mode=ParseMode.HTML,
    )
    await callback.answer()


@dp.callback_query(F.data == "back_to_main")
async def cb_back_to_main(callback: CallbackQuery, session: AsyncSession):
    user = await get_or_create_user(
        session, callback.from_user.id, callback.from_user.username
    )
    text = (
        "👋 Привет! Это бот для проверки логинов и паролей на предмет утечек.\n\n"
        f"Твой Telegram ID: <code>{user.telegram_id}</code>\n"
        f"Имя: <b>{user.username or '—'}</b>\n"
        f"Всего проверок: <b>{user.checks_count}</b>\n\n"
        "Нажми кнопку ниже, чтобы начать проверку."
    )
    await callback.message.edit_text(text, reply_markup=main_menu_kb())
    await callback.answer()


@dp.callback_query(F.data == "history_passwords")
async def cb_history_passwords(callback: CallbackQuery, session: AsyncSession):
    user = await get_or_create_user(
        session, callback.from_user.id, callback.from_user.username
    )

    result = await session.execute(
        select(PasswordCheck)
        .where(PasswordCheck.user_id == user.id)
        .order_by(desc(PasswordCheck.created_at))
        .limit(10)
    )
    checks = result.scalars().all()

    if not checks:
        text = "🔑 <b>История проверок паролей</b>\n\n" "У тебя пока нет проверенных паролей."
        kb = InlineKeyboardMarkup(
            inline_keyboard=[
                [InlineKeyboardButton(text="◀️ Назад", callback_data="history")]
            ]
        )
    else:
        compromised_count = sum(1 for c in checks if c.compromised)
        text = (
            f"🔑 <b>История проверок паролей</b>\n\n"
            f"Всего записей: <b>{len(checks)}</b>\n"
            f"Скомпрометированных: <b>{compromised_count}</b>\n\n"
        )
        for i, check in enumerate(checks, 1):
            status = "⚠️" if check.compromised else "✅"
            date_str = check.created_at.strftime("%d.%m.%Y %H:%M") if check.created_at else "—"
            text += (
                f"{i}. {status} <code>{check.hash_prefix}...</code>\n"
                f"   Дата: {date_str}\n\n"
            )
        kb = InlineKeyboardMarkup(
            inline_keyboard=[
                [InlineKeyboardButton(text="◀️ Назад", callback_data="history")]
            ]
        )

    await callback.message.edit_text(text, reply_markup=kb, parse_mode=ParseMode.HTML)
    await callback.answer()


@dp.callback_query(F.data == "history_logins")
async def cb_history_logins(callback: CallbackQuery, session: AsyncSession):
    user = await get_or_create_user(
        session, callback.from_user.id, callback.from_user.username
    )

    result = await session.execute(
        select(LoginLeak)
        .where(LoginLeak.user_id == user.id)
        .order_by(desc(LoginLeak.created_at))
        .limit(10)
    )
    checks = result.scalars().all()

    if not checks:
        text = "👤 <b>История проверок логинов</b>\n\n" "У тебя пока нет проверенных логинов."
        kb = InlineKeyboardMarkup(
            inline_keyboard=[
                [InlineKeyboardButton(text="◀️ Назад", callback_data="history")]
            ]
        )
    else:
        compromised_count = sum(1 for c in checks if c.compromised)
        text = (
            f"👤 <b>История проверок логинов</b>\n\n"
            f"Всего записей: <b>{len(checks)}</b>\n"
            f"Скомпрометированных: <b>{compromised_count}</b>\n\n"
        )
        for i, check in enumerate(checks, 1):
            status = "⚠️" if check.compromised else "✅"
            login_display = check.login[:20] + "..." if len(check.login) > 20 else check.login
            date_str = check.created_at.strftime("%d.%m.%Y %H:%M") if check.created_at else "—"
            text += (
                f"{i}. {status} <code>{login_display}</code>\n"
                f"   Источник: {check.source}\n"
                f"   Дата: {date_str}\n\n"
            )
        kb = InlineKeyboardMarkup(
            inline_keyboard=[
                [InlineKeyboardButton(text="◀️ Назад", callback_data="history")]
            ]
        )

    await callback.message.edit_text(text, reply_markup=kb, parse_mode=ParseMode.HTML)
    await callback.answer()


@dp.callback_query(F.data == "history_compromised")
async def cb_history_compromised(callback: CallbackQuery, session: AsyncSession):
    user = await get_or_create_user(
        session, callback.from_user.id, callback.from_user.username
    )

    # Пароли
    pass_result = await session.execute(
        select(PasswordCheck)
        .where(PasswordCheck.user_id == user.id, PasswordCheck.compromised == True)
        .order_by(desc(PasswordCheck.created_at))
        .limit(5)
    )
    pass_checks = pass_result.scalars().all()

    # Логины
    login_result = await session.execute(
        select(LoginLeak)
        .where(LoginLeak.user_id == user.id, LoginLeak.compromised == True)
        .order_by(desc(LoginLeak.created_at))
        .limit(5)
    )
    login_checks = login_result.scalars().all()

    if not pass_checks and not login_checks:
        text = (
            "⚠️ <b>Найденные утечки</b>\n\n"
            "У тебя пока нет скомпрометированных данных. Это хорошо! 🎉"
        )
    else:
        text = "⚠️ <b>Найденные утечки</b>\n\n"
        if pass_checks:
            text += "🔑 <b>Пароли:</b>\n"
            for i, check in enumerate(pass_checks, 1):
                date_str = check.created_at.strftime("%d.%m.%Y") if check.created_at else "—"
                text += f"{i}. <code>{check.hash_prefix}...</code> ({date_str})\n"
            text += "\n"
        if login_checks:
            text += "👤 <b>Логины:</b>\n"
            for i, check in enumerate(login_checks, 1):
                login_display = check.login[:25] + "..." if len(check.login) > 25 else check.login
                date_str = check.created_at.strftime("%d.%m.%Y") if check.created_at else "—"
                text += f"{i}. <code>{login_display}</code> ({date_str})\n"
                if check.details:
                    details_short = check.details[:100] + "..." if len(check.details) > 100 else check.details
                    text += f"   {details_short}\n"

    kb = InlineKeyboardMarkup(
        inline_keyboard=[
            [InlineKeyboardButton(text="◀️ Назад", callback_data="history")]
        ]
    )

    await callback.message.edit_text(text, reply_markup=kb, parse_mode=ParseMode.HTML)
    await callback.answer()


async def session_middleware(handler, event, data):
    async for session in get_session():
        data["session"] = session
        return await handler(event, data)


async def main() -> None:
    await init_db()
    bot = Bot(
        token=settings.bot_token,
        default=DefaultBotProperties(parse_mode=ParseMode.HTML),
    )
    dp.update.middleware.register(session_middleware)
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())