external.py - master - Amino/LeakBot

external.py
117 строк4 KB
Amino
upload files
20 дек 2025, 13:38
20 дек 2025, 13:38c41443e
import hashlib
from typing import Tuple, Optional

import httpx

from config import settings


HIBP_PW_API_URL = "https://api.pwnedpasswords.com/range/{prefix}"
HIBP_ACCOUNT_API_URL = "https://haveibeenpwned.com/api/v3/breachedaccount/{account}"
LEAKCHECK_PUBLIC_API_URL = "https://leakcheck.io/api/public"


#Реализация k-Anonymity для HIBP: sha1(password -> первые 5 символов отправляем в HIBP -> в ответе ищем оставшуюся часть.

async def check_password_pwned(password: str) -> Tuple[bool, str, str]:

    sha1 = hashlib.sha1(password.encode("utf-8")).hexdigest().upper()
    prefix, suffix = sha1[:5], sha1[5:]

    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.get(
            HIBP_PW_API_URL.format(prefix=prefix),
            headers={"Add-Padding": "true"},
        )
        resp.raise_for_status()
        lines = resp.text.splitlines()

    count = 0
    for line in lines:
        parts = line.split(":")
        if len(parts) != 2:
            continue
        hash_tail, found_count = parts
        if hash_tail.strip().upper() == suffix:
            count = int(found_count.strip())
            break

    compromised = count > 0
    details = (
        f"Этот пароль встречается в базах утечек <b>{count}</b> раз(а)."
        if compromised
        else "По этому паролю записей в базе HIBP не найдено."
    )
    return compromised, prefix, details


async def check_login_leak(
    login: str,
) -> Tuple[bool, Optional[str], Optional[str]]:
    """
    Проверка логина/емейла через публичный API LeakCheck.

    Документация: https://wiki.leakcheck.io/en/api/public
    Примеры ответа:
    {
        "success": true,
        "found": 3,
        "fields": ["username", "first_name", "address"],
        "sources": [
            {"name": "Evony.com", "date": "2016-07"},
            ...
        ]
    }
    """
    try:
        async with httpx.AsyncClient(timeout=15) as client:
            resp = await client.get(
                LEAKCHECK_PUBLIC_API_URL,
                params={"check": login},
            )
            resp.raise_for_status()
            data = resp.json()
    except httpx.HTTPError as e:
        return (
            False,
            None,
            f"Не удалось запросить LeakCheck (public API): {e.__class__.__name__}",
        )

    success = data.get("success", False)
    if not success:
        return (
            False,
            "LeakCheck",
            "По этому логину/email утечек в LeakCheck не найдено.",
        )

    found = int(data.get("found", 0) or 0)
    if found <= 0:
        return (
            False,
            "LeakCheck",
            "По этому логину/email утечек в LeakCheck не найдено.",
        )

    fields = data.get("fields") or []
    sources = data.get("sources") or []

    sources_str = ", ".join(
        f"{s.get('name', 'unknown')} ({s.get('date', 'n/a')})"
        for s in sources
        if isinstance(s, dict)
    )
    if not sources_str:
        sources_str = "источники не указаны"

    fields_str = ", ".join(fields) if fields else "поля не указаны"

    details = (
        f"⚠️ Найдено утечек: <b>{found}</b> (по данным LeakCheck).\n"
        f"Поля в записях: <code>{fields_str}</code>\n"
        f"Источники: <b>{sources_str}</b>\n\n"
        "Рекомендуем сменить пароли и включить 2FA.\n"
        "<i>Powered by LeakCheck</i>"
    )
    return True, "LeakCheck", details