초기 커밋: Docker/PostgreSQL 이관, 구글 OAuth 제거, Gitea 푸시 스크립트 추가

app/auth.py

@@ -0,0 +1,76 @@
+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}