feat(erp): ERP 대시보드 + 관리자 권한 시스템 도입

- 로그인: dbxcorp.co.kr 도메인 화이트리스트만 사용. ALLOWED_EMAILS 게이트 제거 - king@dbxcorp.co.kr 슈퍼관리자 자동 부여 (강등/삭제 불가) - 신규 로그인 사용자는 JSON 저장소에 자동 등록 (기본 모듈 권한 0) - 관리자 페이지(/admin): 역할/모듈별 토글로 직원 권한 수정 - 메인 대시보드를 ERP 시스템(monochromatic shadcn)으로 재설계 · 모듈: CORM, Order, 개인경비(준비중), 휴가(준비중) · 권한 없는 모듈은 '권한 없음' 배지 + 클릭 불가 - DESIGN.md 토큰 기반 styles → app/static/erp.css - docker-compose: dbx-main-data 볼륨 마운트 (DATA_DIR=/data) 으로 권한 JSON 영구 보관 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-28 23:36:51 +09:00
parent 64c77b36ea
commit 316e3fe8d1
7 changed files with 1351 additions and 165 deletions
@@ -5,12 +5,22 @@ 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 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"

@@ -20,14 +30,17 @@ 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 _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:
@@ -35,14 +48,6 @@ def env(name: str, default: str = "") -> str:
    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:
@@ -80,6 +85,7 @@ app.mount("/static", StaticFiles(directory=str(BASE_DIR / "static")), name="stat

 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:
@@ -89,11 +95,11 @@ def public_url_for(request: Request, route_name: str) -> str:
    return str(request.url_for(route_name))


-def get_user(request: Request) -> dict[str, Any] | None:
+def get_session_user(request: Request) -> dict[str, Any] | None:
+    """세션에서 로그인 사용자(이메일/이름/사진) 추출. OMS SSO 호환."""
    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 {
@@ -104,6 +110,27 @@ def get_user(request: Request) -> dict[str, Any] | None:
    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:
@@ -144,35 +171,79 @@ def is_allowed_google_user(userinfo: dict[str, Any]) -> tuple[bool, str]:
        return False, "Google 계정 이메일 인증이 확인되지 않았습니다."
    if domain != ALLOWED_DOMAIN:
        return False, "회사 Google Workspace 계정만 접속할 수 있습니다."
-    if email not in ALLOWED_EMAILS:
-        return False, "접속 허용 목록에 없는 계정입니다."
    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:
-    user = get_user(request)
-    if not user:
+    sess = get_session_user(request)
+    if not sess:
        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",
-        },
-    ]
+    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, "menu_items": menu_items},
+        {
+            "user": user_rec,
+            "menu_items": menu_items,
+            "is_admin": is_admin(user_rec),
+        },
    )


@@ -186,7 +257,6 @@ async def login(request: Request):
            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")
@@ -226,11 +296,14 @@ async def auth_google(request: Request):
    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
-    # 기존 코드/템플릿 호환용 dict
    request.session["user"] = {
        "email": email,
        "name": name,
@@ -249,3 +322,72 @@ async def logout(request: Request) -> RedirectResponse:
@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})