63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
import os
|
|
from datetime import datetime, timedelta
|
|
from fastapi import APIRouter, Depends, HTTPException, 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
|
|
|
|
router = APIRouter()
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
JWT_SECRET = os.getenv("JWT_SECRET", "changeme")
|
|
JWT_ALGO = "HS256"
|
|
JWT_EXP_HOURS = 24
|
|
|
|
|
|
def _make_token(username: str) -> str:
|
|
exp = datetime.utcnow() + 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")
|
|
async def login(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),
|
|
):
|
|
from route.deps import require_admin
|
|
# handled with deps in main
|
|
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}
|