49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
from fastapi import Depends, HTTPException, Request, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from starlette.middleware.sessions import SessionMiddleware
|
|
|
|
from app.config import settings
|
|
from app.database import get_db
|
|
from app.models import AdminUser
|
|
from app.security import verify_password
|
|
|
|
|
|
def install_session_middleware(app: Any) -> None:
|
|
app.add_middleware(
|
|
SessionMiddleware,
|
|
secret_key=settings.secret_key,
|
|
session_cookie=settings.session_cookie,
|
|
max_age=settings.session_max_age,
|
|
same_site="lax",
|
|
https_only=settings.session_https_only,
|
|
)
|
|
|
|
|
|
async def authenticate_admin(session: AsyncSession, username: str, password: str) -> AdminUser | None:
|
|
result = await session.execute(select(AdminUser).where(AdminUser.username == username))
|
|
user = result.scalar_one_or_none()
|
|
if not user or not user.is_active:
|
|
return None
|
|
if not verify_password(password, user.password_hash):
|
|
return None
|
|
return user
|
|
|
|
|
|
async def get_current_admin(request: Request, db: AsyncSession = Depends(get_db)) -> AdminUser:
|
|
admin_id = request.session.get("admin_id")
|
|
if not admin_id:
|
|
raise HTTPException(status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/admin/login"})
|
|
user = await db.get(AdminUser, admin_id)
|
|
if not user or not user.is_active:
|
|
request.session.clear()
|
|
raise HTTPException(status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/admin/login"})
|
|
return user
|
|
|
|
|
|
def login_required(view: Callable):
|
|
return view
|