webapp.py - master - Amino/LeakBot

webapp.py
245 строк7 KB
Amino
upload files
20 дек 2025, 13:38
20 дек 2025, 13:38c41443e
from fastapi import FastAPI, Depends, HTTPException
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update, desc
from pathlib import Path

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


app = FastAPI(title="LeakBot Mini App")

BASE_DIR = Path(__file__).parent
STATIC_DIR = BASE_DIR / "static"

app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")


class CheckPasswordRequest(BaseModel):
    telegram_id: int
    password: str


class CheckLoginRequest(BaseModel):
    telegram_id: int
    login: str


async def get_or_create_user(
    session: AsyncSession, tg_id: int, username: str | None = None
) -> 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


@app.on_event("startup")
async def on_startup():
    await init_db()


@app.get("/", response_class=HTMLResponse)
async def root():
    """Корневой маршрут для проверки работы сервера"""
    return HTMLResponse("<h1>LeakBot API работает!</h1><p><a href='/app'>Открыть Mini App</a></p>")


@app.get("/app", response_class=HTMLResponse)
async def mini_app_page():
    """Главная страница Mini App"""
    html_file = STATIC_DIR / "index.html"
    if not html_file.exists():
        raise HTTPException(status_code=500, detail=f"HTML файл не найден: {html_file}")
    
    with open(html_file, "r", encoding="utf-8") as f:
        return HTMLResponse(f.read())


@app.post("/api/check-password")
async def api_check_password(
    payload: CheckPasswordRequest, session: AsyncSession = Depends(get_session)
):
    user = await get_or_create_user(session, payload.telegram_id)
    compromised, hash_prefix, details = await check_password_pwned(payload.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()

    return {
        "compromised": compromised,
        "details": details,
    }


@app.post("/api/check-login")
async def api_check_login(
    payload: CheckLoginRequest, session: AsyncSession = Depends(get_session)
):
    user = await get_or_create_user(session, payload.telegram_id)
    compromised, source, details = await check_login_leak(payload.login)

    session.add(
        LoginLeak(
            user_id=user.id,
            login=payload.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()

    return {
        "compromised": compromised,
        "source": source,
        "details": details,
    }


@app.get("/api/profile/{telegram_id}")
async def api_profile(
    telegram_id: int, session: AsyncSession = Depends(get_session)
):
    result = await session.execute(select(User).where(User.telegram_id == telegram_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    return {
        "telegram_id": user.telegram_id,
        "username": user.username,
        "checks_count": user.checks_count,
    }


@app.get("/api/history/passwords/{telegram_id}")
async def api_history_passwords(
    telegram_id: int,
    limit: int = 20,
    compromised_only: bool = False,
    session: AsyncSession = Depends(get_session),
):
    user = await get_or_create_user(session, telegram_id)
    query = select(PasswordCheck).where(PasswordCheck.user_id == user.id)
    if compromised_only:
        query = query.where(PasswordCheck.compromised == True)
    query = query.order_by(desc(PasswordCheck.created_at)).limit(limit)

    result = await session.execute(query)
    checks = result.scalars().all()

    return {
        "checks": [
            {
                "id": c.id,
                "hash_prefix": c.hash_prefix,
                "compromised": c.compromised,
                "created_at": c.created_at.isoformat() if c.created_at else None,
            }
            for c in checks
        ],
        "total": len(checks),
    }


@app.get("/api/history/logins/{telegram_id}")
async def api_history_logins(
    telegram_id: int,
    limit: int = 20,
    compromised_only: bool = False,
    session: AsyncSession = Depends(get_session),
):
    user = await get_or_create_user(session, telegram_id)
    query = select(LoginLeak).where(LoginLeak.user_id == user.id)
    if compromised_only:
        query = query.where(LoginLeak.compromised == True)
    query = query.order_by(desc(LoginLeak.created_at)).limit(limit)

    result = await session.execute(query)
    checks = result.scalars().all()

    return {
        "checks": [
            {
                "id": c.id,
                "login": c.login,
                "source": c.source,
                "compromised": c.compromised,
                "details": c.details,
                "created_at": c.created_at.isoformat() if c.created_at else None,
            }
            for c in checks
        ],
        "total": len(checks),
    }


@app.get("/api/history/compromised/{telegram_id}")
async def api_history_compromised(
    telegram_id: int,
    limit: int = 10,
    session: AsyncSession = Depends(get_session),
):
    user = await get_or_create_user(session, telegram_id)

    pass_result = await session.execute(
        select(PasswordCheck)
        .where(PasswordCheck.user_id == user.id, PasswordCheck.compromised == True)
        .order_by(desc(PasswordCheck.created_at))
        .limit(limit)
    )
    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(limit)
    )
    login_checks = login_result.scalars().all()

    return {
        "passwords": [
            {
                "id": c.id,
                "hash_prefix": c.hash_prefix,
                "created_at": c.created_at.isoformat() if c.created_at else None,
            }
            for c in pass_checks
        ],
        "logins": [
            {
                "id": c.id,
                "login": c.login,
                "source": c.source,
                "details": c.details,
                "created_at": c.created_at.isoformat() if c.created_at else None,
            }
            for c in login_checks
        ],
    }