dbx-main/app/main.py

import inspect
import os
from pathlib import Path
from typing import Any

from authlib.integrations.starlette_client import OAuth, OAuthError
from dotenv import load_dotenv
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from starlette.middleware.sessions import SessionMiddleware

# OMS(orderlist) 와 공유하는 세션 키. SessionMiddleware 의 session_cookie 도 동일 이름.
SESSION_COOKIE_DEFAULT = "session"


load_dotenv()

BASE_DIR = Path(__file__).resolve().parent
ALLOWED_DOMAIN = "dbxcorp.co.kr"

# ALLOWED_EMAILS 는 운영 시 .env 의 ALLOWED_EMAILS (쉼표 구분)로 덮어쓸 수 있다.
# env 가 비어 있으면 아래 기본 목록을 그대로 사용 — 기존 운영 호환.
DEFAULT_ALLOWED_EMAILS = (
    "king@dbxcorp.co.kr",
    "julie@dbxcorp.co.kr",
    "ellen@dbxcorp.co.kr",
    "bj@dbxcorp.co.kr",
)


def env(name: str, default: str = "") -> str:
    value = os.getenv(name, "").strip()
    return value if value else default


def _parse_email_list(raw: str) -> set[str]:
    return {e.strip().lower() for e in raw.split(",") if e.strip()}


_env_emails = _parse_email_list(env("ALLOWED_EMAILS", ""))
ALLOWED_EMAILS: set[str] = _env_emails or _parse_email_list(",".join(DEFAULT_ALLOWED_EMAILS))


def _require_session_secret() -> str:
    secret = env("SESSION_SECRET_KEY", "")
    if not secret:
        raise RuntimeError(
            "SESSION_SECRET_KEY 환경변수가 설정되지 않았습니다. "
            "openssl rand -hex 32 로 새 값을 만들어 .env 에 넣고 컨테이너를 재기동하세요. "
            "OMS(orderlist) 와 SSO 하려면 양쪽 .env 에 같은 값이어야 합니다."
        )
    return secret


def build_google_oauth() -> OAuth:
    oauth = OAuth()
    oauth.register(
        name="google",
        client_id=env("GOOGLE_CLIENT_ID"),
        client_secret=env("GOOGLE_CLIENT_SECRET"),
        server_metadata_url="https://accounts.google.com/.well-known/openid-configuration",
        client_kwargs={"scope": "openid email profile"},
    )
    return oauth


app = FastAPI(title="DBX 메인 페이지")
app.add_middleware(
    SessionMiddleware,
    secret_key=_require_session_secret(),
    session_cookie=env("SESSION_COOKIE_NAME", SESSION_COOKIE_DEFAULT),
    https_only=env("SESSION_COOKIE_SECURE", "true").lower() == "true",
    same_site="lax",
    max_age=int(env("SESSION_MAX_AGE", "28800")),
    path="/",
)
app.mount("/static", StaticFiles(directory=str(BASE_DIR / "static")), name="static")

templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
oauth = build_google_oauth()


def public_url_for(request: Request, route_name: str) -> str:
    public_base_url = env("PUBLIC_BASE_URL").rstrip("/")
    if public_base_url:
        return f"{public_base_url}{request.url_for(route_name).path}"
    return str(request.url_for(route_name))


def get_user(request: Request) -> dict[str, Any] | None:
    user = request.session.get("user")
    if isinstance(user, dict):
        return user
    # OMS 가 top-level user_email 만 채워놓은 SSO 세션도 인정
    email = request.session.get("user_email")
    if email:
        return {
            "email": str(email),
            "name": request.session.get("user_name") or str(email),
            "picture": request.session.get("user_picture", "") or "",
        }
    return None


def safe_next(raw: str | None) -> str:
    """Open redirect 방지: 같은 호스트의 절대 경로만 허용."""
    if not raw:
        return "/"
    if not raw.startswith("/") or raw.startswith("//") or raw.startswith("/\\"):
        return "/"
    return raw


def render_template(
    request: Request,
    name: str,
    context: dict[str, Any] | None = None,
    status_code: int = 200,
) -> HTMLResponse:
    template_context = {"request": request, **(context or {})}
    first_param = next(iter(inspect.signature(templates.TemplateResponse).parameters))
    if first_param == "request":
        return templates.TemplateResponse(
            request,
            name,
            template_context,
            status_code=status_code,
        )
    return templates.TemplateResponse(
        name,
        template_context,
        status_code=status_code,
    )


def is_allowed_google_user(userinfo: dict[str, Any]) -> tuple[bool, str]:
    email = str(userinfo.get("email", "")).lower().strip()
    email_verified = bool(userinfo.get("email_verified"))
    domain = email.rsplit("@", 1)[-1] if "@" in email else ""

    if not email_verified:
        return False, "Google 계정 이메일 인증이 확인되지 않았습니다."
    if domain != ALLOWED_DOMAIN:
        return False, "회사 Google Workspace 계정만 접속할 수 있습니다."
    if email not in ALLOWED_EMAILS:
        return False, "접속 허용 목록에 없는 계정입니다."
    return True, ""


@app.get("/", response_class=HTMLResponse)
async def home(request: Request) -> HTMLResponse:
    user = get_user(request)
    if not user:
        return render_template(request, "login.html")

    menu_items = [
        {
            "title": "CS 발주 업무",
            "description": "CS 발주 업무 페이지로 이동",
            "url": env("CS_ORDER_URL", "/corm/"),
            "health_url": "/corm/health/db",
        },
        {
            "title": "고객 주문리스트 프로그램",
            "description": "고객 주문리스트 프로그램으로 이동",
            "url": env("CUSTOMER_ORDER_LIST_URL", "/orderlist/"),
            "health_url": "/orderlist/health/db",
        },
    ]
    return render_template(
        request,
        "main.html",
        {"user": user, "menu_items": menu_items},
    )


@app.get("/login")
async def login(request: Request):
    if not env("GOOGLE_CLIENT_ID") or not env("GOOGLE_CLIENT_SECRET"):
        return render_template(
            request,
            "denied.html",
            {"reason": "Google OAuth 환경 변수가 아직 설정되지 않았습니다."},
            status_code=500,
        )

    # SSO: ?next= 로 들어온 목적지를 OAuth 콜백 뒤에 사용하기 위해 세션에 임시 보관
    request.session["_post_login_next"] = safe_next(request.query_params.get("next"))

    redirect_uri = public_url_for(request, "auth_google")
    return await oauth.google.authorize_redirect(
        request,
        redirect_uri,
        hd=ALLOWED_DOMAIN,
        prompt="select_account",
    )


@app.get("/auth/google")
async def auth_google(request: Request):
    try:
        token = await oauth.google.authorize_access_token(request)
        userinfo = token.get("userinfo")
        if userinfo is None:
            userinfo = await oauth.google.userinfo(token=token)
    except OAuthError as exc:
        return render_template(
            request,
            "denied.html",
            {"reason": f"Google 로그인 실패: {exc.error}"},
            status_code=401,
        )

    allowed, reason = is_allowed_google_user(dict(userinfo))
    if not allowed:
        request.session.clear()
        return render_template(
            request,
            "denied.html",
            {"reason": reason},
            status_code=403,
        )

    email = str(userinfo.get("email", "")).lower().strip()
    name = userinfo.get("name") or email
    picture = userinfo.get("picture", "") or ""
    # OMS 와 공유하는 top-level 키 (SSO 계약)
    request.session["user_email"] = email
    request.session["user_name"] = name
    request.session["user_picture"] = picture
    # 기존 코드/템플릿 호환용 dict
    request.session["user"] = {
        "email": email,
        "name": name,
        "picture": picture,
    }
    next_url = safe_next(request.session.pop("_post_login_next", "/"))
    return RedirectResponse(url=next_url, status_code=303)


@app.get("/logout")
async def logout(request: Request) -> RedirectResponse:
    request.session.clear()
    return RedirectResponse(url="/", status_code=303)


@app.get("/healthz")
async def healthz() -> dict[str, str]:
    return {"status": "ok"}