172 lines
6.3 KiB
Python
172 lines
6.3 KiB
Python
"""
|
|
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
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.models import User
|
|
|
|
router = APIRouter(tags=["auth"])
|
|
|
|
|
|
# ---- 환경설정 ----------------------------------------------------------
|
|
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:
|
|
root_path = os.getenv("APP_ROOT_PATH", "").rstrip("/")
|
|
if path == "/":
|
|
return root_path or "/"
|
|
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)):
|
|
resolved = _resolve_authenticated_email(request)
|
|
if not resolved:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
|
|
|
|
email, name = resolved
|
|
user = db.query(User).filter(User.email == email).first()
|
|
if not user:
|
|
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 and email.lower() not in ADMIN_EMAILS:
|
|
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 and current_user.email.lower() not in ADMIN_EMAILS:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
|
|
return current_user
|
|
|
|
|
|
|
|
def _safe_next(raw: Optional[str]) -> str:
|
|
"""클라이언트가 넘긴 next 를 검증. 같은 호스트의 절대경로만 허용.
|
|
|
|
dbx-main 측에서도 동일 규칙으로 한 번 더 검증하지만, OMS 단계에서
|
|
먼저 거른다. 외부 URL / 프로토콜 상대 URL / //, /\\ 등 차단.
|
|
"""
|
|
root = os.getenv("APP_ROOT_PATH", "").rstrip("/")
|
|
fallback = f"{root}/" if root else "/"
|
|
if not raw:
|
|
return fallback
|
|
candidate = raw.strip()
|
|
if not candidate.startswith("/"):
|
|
return fallback
|
|
if candidate.startswith("//") or candidate.startswith("/\\"):
|
|
return fallback
|
|
return candidate
|
|
|
|
|
|
@router.get("/login")
|
|
async def login(request: Request, next: Optional[str] = None):
|
|
"""OMS 자체 로그인 없음. dbx-main 의 로그인 페이지로 위임.
|
|
|
|
클라이언트가 ?next= 로 원래 가려던 경로를 넘기면 그대로 전달한다.
|
|
없으면 OMS 의 root_path 로 폴백.
|
|
"""
|
|
safe = _safe_next(next)
|
|
sep = "&" if "?" in MAIN_LOGIN_URL else "?"
|
|
target = f"{MAIN_LOGIN_URL}{sep}next={safe}"
|
|
return RedirectResponse(url=target, status_code=302)
|
|
|
|
|
|
@router.get("/logout")
|
|
async def logout(request: Request):
|
|
"""전적으로 dbx-main 에 위임.
|
|
|
|
중요: OMS 는 절대로 request.session 을 modify 하지 않는다.
|
|
NPM 이 proxy_cookie_path / /orderlist/; 로 Set-Cookie 의 path 를
|
|
재작성하므로, OMS 가 세션 쿠키를 쓰면 dbx-main 의 path=/ 쿠키와
|
|
분리된 path=/orderlist/ 쿠키가 만들어져 SSO 가 깨진다.
|
|
"""
|
|
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,
|
|
}
|