guardia-itsm/core/oauth.py

"""
OAuth2 / OpenID Connect 소셜 로그인 — GUARDiA ITSM

지원 제공자:
  google  : Google OAuth2 (GOOGLE_CLIENT_ID / GOOGLE_CLIENT_SECRET)
  github  : GitHub OAuth2  (GITHUB_CLIENT_ID / GITHUB_CLIENT_SECRET)
  kakao   : 카카오 로그인   (KAKAO_CLIENT_ID)
  naver   : 네이버 로그인   (NAVER_CLIENT_ID / NAVER_CLIENT_SECRET)
  keycloak: 온프레미스 Keycloak OIDC (KEYCLOAK_BASE_URL / KEYCLOAK_REALM / ...)

환경변수 미설정 시 해당 제공자는 로그인 화면에서 숨겨진다.
외부망 차단 환경에서는 Keycloak(온프레미스)만 활성화 권장.
"""
from __future__ import annotations

import hashlib
import hmac
import os
import secrets
import time
from typing import Optional
from urllib.parse import urlencode

import httpx

# ── 환경변수 ──────────────────────────────────────────────────────────────────

_BASE = os.getenv("OAUTH_REDIRECT_BASE", "http://localhost:8001")

PROVIDERS: dict[str, dict] = {
    "google": {
        "name":          "Google",
        "icon":          "google",
        "enabled":       bool(os.getenv("GOOGLE_CLIENT_ID")),
        "client_id":     os.getenv("GOOGLE_CLIENT_ID", ""),
        "client_secret": os.getenv("GOOGLE_CLIENT_SECRET", ""),
        "auth_url":      "https://accounts.google.com/o/oauth2/v2/auth",
        "token_url":     "https://oauth2.googleapis.com/token",
        "userinfo_url":  "https://www.googleapis.com/oauth2/v3/userinfo",
        "scope":         "openid email profile",
        "redirect_uri":  f"{_BASE}/api/auth/oauth/google/callback",
    },
    "github": {
        "name":          "GitHub",
        "icon":          "github",
        "enabled":       bool(os.getenv("GITHUB_CLIENT_ID")),
        "client_id":     os.getenv("GITHUB_CLIENT_ID", ""),
        "client_secret": os.getenv("GITHUB_CLIENT_SECRET", ""),
        "auth_url":      "https://github.com/login/oauth/authorize",
        "token_url":     "https://github.com/login/oauth/access_token",
        "userinfo_url":  "https://api.github.com/user",
        "scope":         "read:user user:email",
        "redirect_uri":  f"{_BASE}/api/auth/oauth/github/callback",
    },
    "kakao": {
        "name":          "카카오",
        "icon":          "kakao",
        "enabled":       bool(os.getenv("KAKAO_CLIENT_ID")),
        "client_id":     os.getenv("KAKAO_CLIENT_ID", ""),
        "client_secret": os.getenv("KAKAO_CLIENT_SECRET", ""),
        "auth_url":      "https://kauth.kakao.com/oauth/authorize",
        "token_url":     "https://kauth.kakao.com/oauth/token",
        "userinfo_url":  "https://kapi.kakao.com/v2/user/me",
        "scope":         "profile_nickname account_email",
        "redirect_uri":  f"{_BASE}/api/auth/oauth/kakao/callback",
    },
    "naver": {
        "name":          "네이버",
        "icon":          "naver",
        "enabled":       bool(os.getenv("NAVER_CLIENT_ID")),
        "client_id":     os.getenv("NAVER_CLIENT_ID", ""),
        "client_secret": os.getenv("NAVER_CLIENT_SECRET", ""),
        "auth_url":      "https://nid.naver.com/oauth2.0/authorize",
        "token_url":     "https://nid.naver.com/oauth2.0/token",
        "userinfo_url":  "https://openapi.naver.com/v1/nid/me",
        "scope":         "name email",
        "redirect_uri":  f"{_BASE}/api/auth/oauth/naver/callback",
    },
    "keycloak": {
        "name":          "SSO (Keycloak)",
        "icon":          "sso",
        "enabled":       bool(os.getenv("KEYCLOAK_BASE_URL")),
        "client_id":     os.getenv("KEYCLOAK_CLIENT_ID", "guardia"),
        "client_secret": os.getenv("KEYCLOAK_CLIENT_SECRET", ""),
        "auth_url":      f"{os.getenv('KEYCLOAK_BASE_URL','')}/realms/{os.getenv('KEYCLOAK_REALM','master')}/protocol/openid-connect/auth",
        "token_url":     f"{os.getenv('KEYCLOAK_BASE_URL','')}/realms/{os.getenv('KEYCLOAK_REALM','master')}/protocol/openid-connect/token",
        "userinfo_url":  f"{os.getenv('KEYCLOAK_BASE_URL','')}/realms/{os.getenv('KEYCLOAK_REALM','master')}/protocol/openid-connect/userinfo",
        "scope":         "openid email profile",
        "redirect_uri":  f"{_BASE}/api/auth/oauth/keycloak/callback",
    },
}

# 임시 state 저장소 (메모리 — 분산 환경에서는 Redis 사용)
_states: dict[str, float] = {}


def get_enabled_providers() -> list[dict]:
    """활성화된 OAuth 제공자 목록 (민감정보 제외)."""
    return [
        {"id": pid, "name": p["name"], "icon": p["icon"]}
        for pid, p in PROVIDERS.items()
        if p["enabled"]
    ]


def build_auth_url(provider_id: str) -> str:
    """OAuth 인증 URL 생성 + CSRF state 발급."""
    p = PROVIDERS.get(provider_id)
    if not p or not p["enabled"]:
        raise ValueError(f"OAuth 제공자 미설정: {provider_id}")

    state = secrets.token_urlsafe(32)
    _states[state] = time.time()
    # 5분 초과 state 정리
    cutoff = time.time() - 300
    expired = [k for k, v in _states.items() if v < cutoff]
    for k in expired:
        _states.pop(k, None)

    params = {
        "client_id":     p["client_id"],
        "redirect_uri":  p["redirect_uri"],
        "scope":         p["scope"],
        "response_type": "code",
        "state":         state,
    }
    return p["auth_url"] + "?" + urlencode(params)


def verify_state(state: str) -> bool:
    """CSRF state 검증."""
    ts = _states.pop(state, None)
    if ts is None:
        return False
    return (time.time() - ts) < 300  # 5분 이내


async def exchange_code(provider_id: str, code: str) -> Optional[dict]:
    """인가 코드 → 액세스 토큰 교환 → 사용자 정보 조회."""
    p = PROVIDERS.get(provider_id)
    if not p:
        return None

    try:
        async with httpx.AsyncClient(timeout=15.0) as client:
            # 토큰 교환
            token_data = {
                "grant_type":    "authorization_code",
                "code":          code,
                "redirect_uri":  p["redirect_uri"],
                "client_id":     p["client_id"],
                "client_secret": p["client_secret"],
            }
            headers = {"Accept": "application/json"}
            resp = await client.post(p["token_url"], data=token_data, headers=headers)
            resp.raise_for_status()
            tokens = resp.json()

        access_token = tokens.get("access_token")
        if not access_token:
            return None

        async with httpx.AsyncClient(timeout=10.0) as client:
            # 사용자 정보 조회
            userinfo_resp = await client.get(
                p["userinfo_url"],
                headers={"Authorization": f"Bearer {access_token}"},
            )
            userinfo_resp.raise_for_status()
            return userinfo_resp.json()

    except Exception:
        return None


def extract_email(provider_id: str, userinfo: dict) -> Optional[str]:
    """제공자별 이메일 추출."""
    if provider_id in ("google", "keycloak"):
        return userinfo.get("email")
    if provider_id == "github":
        return userinfo.get("email") or f"{userinfo.get('login', '')}@github.com"
    if provider_id == "kakao":
        return userinfo.get("kakao_account", {}).get("email")
    if provider_id == "naver":
        return userinfo.get("response", {}).get("email")
    return userinfo.get("email")


def extract_name(provider_id: str, userinfo: dict) -> str:
    """제공자별 표시 이름 추출."""
    if provider_id in ("google", "keycloak"):
        return userinfo.get("name", "")
    if provider_id == "github":
        return userinfo.get("name") or userinfo.get("login", "")
    if provider_id == "kakao":
        return userinfo.get("kakao_account", {}).get("profile", {}).get("nickname", "")
    if provider_id == "naver":
        return userinfo.get("response", {}).get("name", "")
    return userinfo.get("name", "")