king
77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
import os
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from fastapi.responses import RedirectResponse
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.models import User
|
|
|
|
router = APIRouter(tags=["auth"])
|
|
|
|
|
|
DEFAULT_ADMIN_EMAIL = os.getenv("DEFAULT_ADMIN_EMAIL", "king@dbxcorp.co.kr")
|
|
DEFAULT_ADMIN_NAME = os.getenv("DEFAULT_ADMIN_NAME", "Administrator")
|
|
|
|
|
|
def app_path(path: str = "/") -> str:
|
|
root_path = os.getenv("APP_ROOT_PATH", "").rstrip("/")
|
|
if path == "/":
|
|
return root_path or "/"
|
|
return f"{root_path}{path}"
|
|
|
|
|
|
def get_current_user(request: Request, db: Session = Depends(get_db)):
|
|
user_email = request.session.get("user_email")
|
|
if not user_email:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
|
|
|
|
user = db.query(User).filter(User.email == user_email).first()
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
|
|
|
|
if not user.is_approved and not user.is_admin:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="User is not approved by administrator")
|
|
|
|
return user
|
|
|
|
|
|
def get_admin_user(current_user: User = Depends(get_current_user)):
|
|
if not current_user.is_admin:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
|
|
return current_user
|
|
|
|
|
|
def _login_as_default_admin(request: Request, db: Session) -> RedirectResponse:
|
|
email = DEFAULT_ADMIN_EMAIL
|
|
user = db.query(User).filter(User.email == email).first()
|
|
if not user:
|
|
user = User(email=email, name=DEFAULT_ADMIN_NAME, is_admin=True, is_approved=True)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
|
|
request.session["user_email"] = email
|
|
return RedirectResponse(url=app_path("/"))
|
|
|
|
|
|
@router.get("/login")
|
|
async def login(request: Request, db: Session = Depends(get_db)):
|
|
return _login_as_default_admin(request, db)
|
|
|
|
|
|
@router.get("/login/test")
|
|
async def login_test(request: Request, db: Session = Depends(get_db)):
|
|
return _login_as_default_admin(request, db)
|
|
|
|
|
|
@router.get("/logout")
|
|
async def logout(request: Request):
|
|
request.session.pop("user_email", None)
|
|
return RedirectResponse(url=app_path("/"))
|
|
|
|
|
|
@router.get("/api/users/me")
|
|
async def read_users_me(current_user: User = Depends(get_current_user)):
|
|
return {"email": current_user.email, "name": current_user.name, "is_admin": current_user.is_admin}
|