66 lines
2.5 KiB
Python
66 lines
2.5 KiB
Python
import os
|
|
from datetime import datetime, timedelta, timezone
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from jose import jwt
|
|
from passlib.context import CryptContext
|
|
|
|
from bd.database import get_db, AsyncSessionLocal
|
|
from bd.models import Admin
|
|
from rate_limiter import limiter
|
|
|
|
router = APIRouter()
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
_raw_secret = os.getenv("JWT_SECRET", "")
|
|
if not _raw_secret:
|
|
import secrets as _s
|
|
_raw_secret = _s.token_hex(32)
|
|
print("[WARN] JWT_SECRET не задан — сгенерирован временный секрет. Установи JWT_SECRET в .env!")
|
|
|
|
JWT_SECRET = _raw_secret
|
|
JWT_ALGO = "HS256"
|
|
JWT_EXP_HOURS = 8
|
|
|
|
|
|
def _make_token(username: str) -> str:
|
|
exp = datetime.now(timezone.utc) + timedelta(hours=JWT_EXP_HOURS)
|
|
return jwt.encode({"sub": username, "exp": exp}, JWT_SECRET, algorithm=JWT_ALGO)
|
|
|
|
|
|
async def ensure_default_admin():
|
|
username = os.getenv("ADMIN_USERNAME", "admin")
|
|
password = os.getenv("ADMIN_PASSWORD", "changeme")
|
|
async with AsyncSessionLocal() as db:
|
|
existing = (await db.execute(select(Admin))).first()
|
|
if not existing:
|
|
admin = Admin(username=username, password_hash=pwd_context.hash(password))
|
|
db.add(admin)
|
|
await db.commit()
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
|
|
@router.post("/login")
|
|
@limiter.limit("5/minute")
|
|
async def login(request: Request, data: LoginRequest, db: AsyncSession = Depends(get_db)):
|
|
admin = (await db.execute(select(Admin).where(Admin.username == data.username))).scalar_one_or_none()
|
|
if not admin or not pwd_context.verify(data.password, admin.password_hash):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Неверный логин или пароль")
|
|
return {"access_token": _make_token(admin.username), "token_type": "bearer"}
|
|
|
|
|
|
@router.post("/change-password")
|
|
async def change_password(data: dict, db: AsyncSession = Depends(get_db)):
|
|
admin = (await db.execute(select(Admin).where(Admin.username == data["username"]))).scalar_one_or_none()
|
|
if not admin or not pwd_context.verify(data["current_password"], admin.password_hash):
|
|
raise HTTPException(status_code=400, detail="Неверный текущий пароль")
|
|
admin.password_hash = pwd_context.hash(data["new_password"])
|
|
await db.commit()
|
|
return {"ok": True}
|