feat: SSO via shared session with dbx-main, sub-path hosting at /orderlist

2026-05-24 05:35:19 +09:00
parent 75ab736e8f
commit c966d41628
4 changed files with 159 additions and 51 deletions
@@ -1,4 +1,24 @@
+"""
+SSO 모드 인증 모듈.
+
+dbx-main(메인 페이지) 이 Google OAuth 를 수행하여 같은 도메인의 세션 쿠키에
+사용자 이메일을 저장한다고 가정한다. OMS 는 그 세션을 읽기만 한다.
+
+호환 방식 (둘 중 하나 이상이 충족되면 인증 성공):
+  1. Starlette SessionMiddleware 공유 — dbx-main 과 OMS 가 동일한
+     SESSION_SECRET_KEY / SESSION_COOKIE_NAME 을 사용하고, 세션에
+     'user_email' 키가 저장돼 있다.
+  2. 신뢰된 업스트림 헤더 — Nginx/NPM 의 auth_request 같은 메커니즘으로
+     dbx-main 이 인증한 사용자의 이메일을 헤더로 전달한다.
+     ALLOW_TRUSTED_USER_HEADER=1 일 때만 활성화. 헤더 이름은
+     TRUSTED_USER_HEADER (기본 X-Forwarded-User).
+
+세션에 사용자 이메일이 있고 OMS DB 에 그 사용자가 없으면, 정책에 따라
+자동 프로비저닝한다 (AUTO_PROVISION_USERS=1 일 때, 기본 1).
+"""
+
 import os
+from typing import Optional

 from fastapi import APIRouter, Depends, HTTPException, Request, status
 from fastapi.responses import RedirectResponse
@@ -10,8 +30,22 @@ 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")
+# ---- 환경설정 ----------------------------------------------------------
+MAIN_LOGIN_URL = os.getenv("MAIN_LOGIN_URL", "https://dbx.no1king.freeddns.org/login")
+MAIN_LOGOUT_URL = os.getenv("MAIN_LOGOUT_URL", "https://dbx.no1king.freeddns.org/logout")
+SESSION_USER_KEY = os.getenv("SESSION_USER_KEY", "user_email")
+SESSION_NAME_KEY = os.getenv("SESSION_NAME_KEY", "user_name")
+
+ALLOW_TRUSTED_USER_HEADER = os.getenv("ALLOW_TRUSTED_USER_HEADER", "0") == "1"
+TRUSTED_USER_HEADER = os.getenv("TRUSTED_USER_HEADER", "X-Forwarded-User")
+TRUSTED_NAME_HEADER = os.getenv("TRUSTED_NAME_HEADER", "X-Forwarded-User-Name")
+
+AUTO_PROVISION_USERS = os.getenv("AUTO_PROVISION_USERS", "1") == "1"
+ADMIN_EMAILS = {
+    e.strip().lower()
+    for e in os.getenv("ADMIN_EMAILS", "king@dbxcorp.co.kr").split(",")
+    if e.strip()
+}


 def app_path(path: str = "/") -> str:
@@ -21,17 +55,59 @@ def app_path(path: str = "/") -> str:
    return f"{root_path}{path}"


+def _resolve_authenticated_email(request: Request) -> Optional[tuple[str, Optional[str]]]:
+    """현재 요청에서 인증된 사용자의 (email, name) 을 추출. 없으면 None."""
+    # 1) 신뢰된 업스트림 헤더 우선 (명시적으로 켜진 경우만)
+    if ALLOW_TRUSTED_USER_HEADER:
+        header_email = request.headers.get(TRUSTED_USER_HEADER)
+        if header_email:
+            header_name = request.headers.get(TRUSTED_NAME_HEADER)
+            return header_email.strip().lower(), header_name
+
+    # 2) 공유 세션 쿠키 (Starlette SessionMiddleware)
+    session_email = request.session.get(SESSION_USER_KEY)
+    if session_email:
+        session_name = request.session.get(SESSION_NAME_KEY)
+        return session_email.strip().lower(), session_name
+
+    return None
+
+
+def _provision_user(db: Session, email: str, name: Optional[str]) -> User:
+    """OMS DB 에 사용자를 새로 생성."""
+    is_admin = email.lower() in ADMIN_EMAILS
+    user = User(
+        email=email,
+        name=name or email.split("@")[0],
+        is_admin=is_admin,
+        is_approved=True,  # dbx-main 에서 인증된 사용자는 신뢰
+    )
+    db.add(user)
+    db.commit()
+    db.refresh(user)
+    return user
+
+
 def get_current_user(request: Request, db: Session = Depends(get_db)):
-    user_email = request.session.get("user_email")
-    if not user_email:
+    resolved = _resolve_authenticated_email(request)
+    if not resolved:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")

-    user = db.query(User).filter(User.email == user_email).first()
+    email, name = resolved
+    user = db.query(User).filter(User.email == email).first()
    if not user:
-        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
+        if not AUTO_PROVISION_USERS:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="User is not registered in OMS",
+            )
+        user = _provision_user(db, email, name)

    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")
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="User is not approved by administrator",
+        )

    return user

@@ -42,35 +118,32 @@ def get_admin_user(current_user: User = Depends(get_current_user)):
    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("/"))
+def _next_param(request: Request) -> str:
+    """로그인 후 돌아올 절대 경로(서브경로 prefix 포함)."""
+    root = os.getenv("APP_ROOT_PATH", "").rstrip("/")
+    return f"{root}/" if root else "/"


@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)
+async def login(request: Request):
+    """OMS 자체 로그인 없음. dbx-main 의 로그인 페이지로 위임."""
+    sep = "&" if "?" in MAIN_LOGIN_URL else "?"
+    target = f"{MAIN_LOGIN_URL}{sep}next={_next_param(request)}"
+    return RedirectResponse(url=target, status_code=302)


@router.get("/logout")
 async def logout(request: Request):
-    request.session.pop("user_email", None)
-    return RedirectResponse(url=app_path("/"))
+    """OMS 세션 흔적 제거 후 dbx-main 로그아웃으로 위임."""
+    request.session.pop(SESSION_USER_KEY, None)
+    request.session.pop(SESSION_NAME_KEY, None)
+    return RedirectResponse(url=MAIN_LOGOUT_URL, status_code=302)


@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}
+    return {
+        "email": current_user.email,
+        "name": current_user.name,
+        "is_admin": current_user.is_admin,
+    }