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 Depends, FastAPI, HTTPException, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel
from starlette.middleware.sessions import SessionMiddleware

from .store import (
    MODULE_KEYS,
    SUPER_ADMIN_EMAIL,
    UserStore,
    allowed_modules,
    has_module,
    is_admin,
)

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


load_dotenv()

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


def _data_dir() -> Path:
    """사용자/권한 JSON 저장소 위치. 컨테이너 재배포에도 살아남도록
    DATA_DIR 환경변수로 마운트된 볼륨을 가리킬 수 있다."""
    override = os.getenv("DATA_DIR", "").strip()
    if override:
        return Path(override)
    return BASE_DIR / "data"


DATA_DIR = _data_dir()


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


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()
user_store = UserStore(DATA_DIR / "users.json")


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_session_user(request: Request) -> dict[str, Any] | None:
    """세션에서 로그인 사용자(이메일/이름/사진) 추출. OMS SSO 호환."""
    user = request.session.get("user")
    if isinstance(user, dict):
        return user
    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 get_current_user_record(request: Request) -> dict[str, Any] | None:
    """세션 + 저장소를 합쳐 권한이 포함된 사용자 레코드 반환."""
    sess = get_session_user(request)
    if not sess:
        return None
    rec = user_store.get(sess["email"])
    if rec is None:
        # 세션은 살아있지만 저장소에 없음 — 세션 무효화
        return None
    return rec


def require_admin(request: Request) -> dict[str, Any]:
    rec = get_current_user_record(request)
    if rec is None:
        raise HTTPException(status_code=401, detail="로그인이 필요합니다.")
    if not is_admin(rec):
        raise HTTPException(status_code=403, detail="관리자 권한이 필요합니다.")
    return rec


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 계정만 접속할 수 있습니다."
    return True, ""


# ── ERP 메뉴 정의 ──────────────────────────────────────────────
# 각 항목: key(권한키), title, description, url(env override), status(ready|preparing), category
def _menu_items_for(user_rec: dict[str, Any]) -> list[dict[str, Any]]:
    items = [
        {
            "key": "corm",
            "title": "CORM",
            "subtitle": "CS · 발주 · 반품 · 코드관리",
            "description": "고객 응대와 발주/반품, 코드 관리 업무를 한 곳에서 처리합니다.",
            "url": env("CS_ORDER_URL", "/corm/"),
            "health_url": "/corm/health/db",
            "status": "ready",
            "category": "운영",
        },
        {
            "key": "order",
            "title": "Order",
            "subtitle": "고객 주문 데이터베이스",
            "description": "고객 주문 내역을 조회·검색하고 관련 데이터를 관리합니다.",
            "url": env("CUSTOMER_ORDER_LIST_URL", "/orderlist/"),
            "health_url": "/orderlist/health/db",
            "status": "ready",
            "category": "운영",
        },
        {
            "key": "expense",
            "title": "개인경비",
            "subtitle": "Personal Expense",
            "description": "법인카드/개인경비 사용 내역을 등록·증빙하고 정산을 신청합니다.",
            "url": "#",
            "health_url": None,
            "status": "preparing",
            "category": "관리",
        },
        {
            "key": "vacation",
            "title": "휴가",
            "subtitle": "Vacation",
            "description": "연차/반차/특별휴가 신청과 잔여일수, 결재 현황을 확인합니다.",
            "url": "#",
            "health_url": None,
            "status": "preparing",
            "category": "관리",
        },
    ]
    allowed = allowed_modules(user_rec)
    for item in items:
        item["allowed"] = item["key"] in allowed
    return items


@app.get("/", response_class=HTMLResponse)
async def home(request: Request) -> HTMLResponse:
    sess = get_session_user(request)
    if not sess:
        return render_template(request, "login.html")
    user_rec = user_store.get(sess["email"])
    if user_rec is None:
        # 도메인은 통과했으나 저장소에 없음 — 세션 정리 후 재로그인
        request.session.clear()
        return render_template(request, "login.html")
    menu_items = _menu_items_for(user_rec)
    return render_template(
        request,
        "main.html",
        {
            "user": user_rec,
            "menu_items": menu_items,
            "is_admin": is_admin(user_rec),
        },
    )


@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,
        )

    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 ""

    # 저장소에 사용자 등록/갱신 (신규는 권한 0 — 관리자가 부여)
    user_store.upsert_login(email=email, name=name, picture=picture)

    # OMS 와 공유하는 top-level 키 (SSO 계약)
    request.session["user_email"] = email
    request.session["user_name"] = name
    request.session["user_picture"] = picture
    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"}


# ── 관리자 페이지 ──────────────────────────────────────────────
@app.get("/admin", response_class=HTMLResponse)
async def admin_page(request: Request) -> HTMLResponse:
    rec = get_current_user_record(request)
    if rec is None:
        return RedirectResponse(url="/login", status_code=303)
    if not is_admin(rec):
        return render_template(
            request,
            "denied.html",
            {"reason": "관리자만 접근할 수 있는 페이지입니다."},
            status_code=403,
        )
    users = sorted(user_store.list_all(), key=lambda u: (u["email"] != SUPER_ADMIN_EMAIL, u["email"]))
    return render_template(
        request,
        "admin.html",
        {
            "user": rec,
            "users": users,
            "module_keys": list(MODULE_KEYS),
            "super_admin_email": SUPER_ADMIN_EMAIL,
            "is_admin": True,
        },
    )


# ── 관리자 API ──────────────────────────────────────────────
class UpdatePermissionsBody(BaseModel):
    role: str | None = None
    modules: dict[str, bool] | None = None


@app.get("/api/users")
async def api_list_users(_: dict[str, Any] = Depends(require_admin)) -> JSONResponse:
    return JSONResponse({"users": user_store.list_all()})


@app.put("/api/users/{email}")
async def api_update_user(
    email: str,
    body: UpdatePermissionsBody,
    _: dict[str, Any] = Depends(require_admin),
) -> JSONResponse:
    try:
        rec = user_store.update_permissions(
            email=email, role=body.role, modules=body.modules
        )
    except PermissionError as exc:
        raise HTTPException(status_code=403, detail=str(exc))
    except KeyError as exc:
        raise HTTPException(status_code=404, detail=str(exc))
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    return JSONResponse({"user": rec})


@app.delete("/api/users/{email}")
async def api_delete_user(
    email: str,
    _: dict[str, Any] = Depends(require_admin),
) -> JSONResponse:
    try:
        user_store.delete(email)
    except PermissionError as exc:
        raise HTTPException(status_code=403, detail=str(exc))
    return JSONResponse({"ok": True})