401 lines
15 KiB
Python
401 lines
15 KiB
Python
"""
|
|
SSO 통합 인증 — SAML 2.0 / OIDC / OAuth2
|
|
|
|
지원 IdP:
|
|
- 행정안전부 공통로그인 (GPKI / SAML 2.0)
|
|
- Google Workspace (OIDC)
|
|
- Microsoft Entra ID (OIDC / OAuth2)
|
|
- 범용 SAML 2.0 / OIDC
|
|
|
|
엔드포인트:
|
|
GET /api/sso/config — SSO 설정 목록
|
|
POST /api/sso/config — SSO IdP 설정 등록
|
|
DELETE /api/sso/config/{id} — 설정 삭제
|
|
GET /api/sso/login/{provider} — SSO 로그인 리다이렉트
|
|
GET /api/sso/callback/saml — SAML ACS (Assertion Consumer Service)
|
|
GET /api/sso/callback/oidc — OIDC 콜백
|
|
POST /api/sso/test/{id} — 설정 테스트
|
|
GET /api/sso/metadata — SP Metadata XML (SAML)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import logging
|
|
import secrets
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional
|
|
from urllib.parse import urlencode, urlparse
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
|
|
from fastapi.responses import RedirectResponse
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from core.auth import create_access_token, get_current_user, require_admin_role
|
|
from database import get_db
|
|
from models import User, UserRole, SSOConfig, SSOSession # 신규 모델
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/sso", tags=["SSO 통합 인증"])
|
|
|
|
BASE_URL = "https://zioinfo.co.kr:8443" # SP(서비스 제공자) 베이스 URL
|
|
|
|
|
|
# ── Pydantic 스키마 ──────────────────────────────────────────────────────────
|
|
|
|
class SSOConfigCreate(BaseModel):
|
|
name: str = Field(..., max_length=100, description="표시명 (예: 행안부 공통로그인)")
|
|
provider_type: str = Field(..., pattern="^(SAML|OIDC|OAUTH2)$")
|
|
# SAML 설정
|
|
idp_metadata_url: Optional[str] = None
|
|
idp_sso_url: Optional[str] = None
|
|
idp_cert: Optional[str] = None
|
|
# OIDC 설정
|
|
client_id: Optional[str] = None
|
|
client_secret: Optional[str] = None
|
|
discovery_url: Optional[str] = None # .well-known/openid-configuration
|
|
scopes: str = Field("openid email profile", description="요청할 scope 목록")
|
|
# 공통
|
|
attribute_mapping: Dict[str, str] = Field(
|
|
default_factory=lambda: {"email": "email", "name": "name"},
|
|
description="IdP 속성 → GUARDiA 속성 매핑"
|
|
)
|
|
default_role: UserRole = UserRole.ENGINEER
|
|
is_active: bool = True
|
|
|
|
|
|
class SSOConfigOut(BaseModel):
|
|
id: int
|
|
name: str
|
|
provider_type: str
|
|
is_active: bool
|
|
created_at: datetime
|
|
|
|
|
|
# ── SAML 헬퍼 ────────────────────────────────────────────────────────────────
|
|
|
|
def _build_saml_authn_request(idp_sso_url: str, sp_entity_id: str, acs_url: str) -> str:
|
|
"""SAML AuthnRequest 생성 (간소화 버전, 서명 없음)."""
|
|
req_id = f"_req_{secrets.token_hex(16)}"
|
|
now = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
xml = f"""<?xml version="1.0"?>
|
|
<samlp:AuthnRequest xmlns:samlp="urn:oasis:names:tc:SAML:2.0:protocol"
|
|
ID="{req_id}" Version="2.0" IssueInstant="{now}"
|
|
Destination="{idp_sso_url}"
|
|
AssertionConsumerServiceURL="{acs_url}"
|
|
ProtocolBinding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST">
|
|
<saml:Issuer xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion">{sp_entity_id}</saml:Issuer>
|
|
<samlp:NameIDPolicy Format="urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress"/>
|
|
</samlp:AuthnRequest>"""
|
|
return base64.b64encode(xml.encode()).decode()
|
|
|
|
|
|
def _parse_saml_response(saml_response_b64: str) -> dict:
|
|
"""SAML Response에서 속성 추출 (실제 운영 시 xmlsec1 검증 필요)."""
|
|
import re
|
|
try:
|
|
decoded = base64.b64decode(saml_response_b64).decode('utf-8', 'replace')
|
|
# NameID (이메일)
|
|
email_m = re.search(r'<(?:[^:]+:)?NameID[^>]*>([^<]+)</(?:[^:]+:)?NameID>', decoded)
|
|
email = email_m.group(1).strip() if email_m else None
|
|
# 속성값 추출
|
|
attrs = {}
|
|
for m in re.finditer(
|
|
r'<(?:[^:]+:)?AttributeValue[^>]*>([^<]+)</(?:[^:]+:)?AttributeValue>',
|
|
decoded
|
|
):
|
|
attrs[m.group(1).strip()] = m.group(1).strip()
|
|
return {"email": email, "name": attrs.get("name", email), "raw_attrs": attrs}
|
|
except Exception as e:
|
|
logger.error(f"SAML 파싱 실패: {e}")
|
|
return {}
|
|
|
|
|
|
# ── OIDC 헬퍼 ────────────────────────────────────────────────────────────────
|
|
|
|
async def _oidc_discovery(discovery_url: str) -> Optional[dict]:
|
|
"""OIDC 디스커버리 문서 조회."""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10, verify=False) as client:
|
|
r = await client.get(discovery_url)
|
|
return r.json() if r.status_code == 200 else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
async def _exchange_code(config: SSOConfig, code: str, redirect_uri: str) -> Optional[dict]:
|
|
"""OIDC 인가코드 → 토큰 교환."""
|
|
discovery = await _oidc_discovery(config.discovery_url)
|
|
if not discovery:
|
|
return None
|
|
token_endpoint = discovery.get("token_endpoint")
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15, verify=False) as client:
|
|
r = await client.post(
|
|
token_endpoint,
|
|
data={
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": redirect_uri,
|
|
"client_id": config.client_id,
|
|
"client_secret": config.client_secret_enc, # 복호화 필요
|
|
},
|
|
)
|
|
return r.json() if r.status_code == 200 else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
async def _get_userinfo(discovery_url: str, access_token: str) -> Optional[dict]:
|
|
"""OIDC userinfo 엔드포인트 호출."""
|
|
discovery = await _oidc_discovery(discovery_url)
|
|
if not discovery:
|
|
return None
|
|
userinfo_ep = discovery.get("userinfo_endpoint")
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10, verify=False) as client:
|
|
r = await client.get(userinfo_ep, headers={"Authorization": f"Bearer {access_token}"})
|
|
return r.json() if r.status_code == 200 else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
# ── 유저 동기화 ──────────────────────────────────────────────────────────────
|
|
|
|
async def _upsert_sso_user(
|
|
attributes: dict, config: SSOConfig, db: AsyncSession
|
|
) -> Optional[User]:
|
|
"""SSO 속성으로 GUARDiA 사용자 생성/업데이트 (Just-In-Time 프로비저닝)."""
|
|
mapping = json.loads(config.attribute_mapping) if config.attribute_mapping else {}
|
|
email = attributes.get(mapping.get("email", "email")) or attributes.get("email")
|
|
name = attributes.get(mapping.get("name", "name")) or attributes.get("name")
|
|
|
|
if not email:
|
|
logger.warning("SSO 속성에서 이메일을 찾을 수 없습니다")
|
|
return None
|
|
|
|
# 기존 유저 조회
|
|
row = await db.execute(select(User).where(User.email == email))
|
|
user = row.scalar_one_or_none()
|
|
|
|
if user:
|
|
# 기존 유저 업데이트
|
|
if name and not user.name:
|
|
user.name = name
|
|
else:
|
|
# 신규 유저 생성 (비밀번호 없음 — SSO 전용)
|
|
user = User(
|
|
email=email,
|
|
name=name or email.split("@")[0],
|
|
hashed_password="!sso_login_only!", # 직접 로그인 불가
|
|
role=config.default_role,
|
|
tenant_id=config.tenant_id,
|
|
is_active=True,
|
|
sso_provider=config.id,
|
|
created_at=datetime.utcnow(),
|
|
)
|
|
db.add(user)
|
|
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
return user
|
|
|
|
|
|
# ── 엔드포인트 ───────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/config", response_model=List[SSOConfigOut])
|
|
async def list_sso_configs(
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(get_current_user),
|
|
):
|
|
rows = await db.execute(
|
|
select(SSOConfig).where(SSOConfig.tenant_id == user.tenant_id)
|
|
)
|
|
configs = rows.scalars().all()
|
|
return [
|
|
SSOConfigOut(id=c.id, name=c.name, provider_type=c.provider_type,
|
|
is_active=c.is_active, created_at=c.created_at)
|
|
for c in configs
|
|
]
|
|
|
|
|
|
@router.post("/config")
|
|
async def create_sso_config(
|
|
req: SSOConfigCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(require_admin_role),
|
|
):
|
|
"""SSO IdP 설정 등록 (client_secret는 암호화 저장)."""
|
|
config = SSOConfig(
|
|
tenant_id=user.tenant_id,
|
|
name=req.name,
|
|
provider_type=req.provider_type,
|
|
idp_metadata_url=req.idp_metadata_url,
|
|
idp_sso_url=req.idp_sso_url,
|
|
idp_cert=req.idp_cert,
|
|
client_id=req.client_id,
|
|
client_secret_enc=req.client_secret, # TODO: AES-256-GCM 암호화
|
|
discovery_url=req.discovery_url,
|
|
scopes=req.scopes,
|
|
attribute_mapping=json.dumps(req.attribute_mapping),
|
|
default_role=req.default_role,
|
|
is_active=req.is_active,
|
|
created_at=datetime.utcnow(),
|
|
)
|
|
db.add(config)
|
|
await db.commit()
|
|
await db.refresh(config)
|
|
return {"ok": True, "id": config.id}
|
|
|
|
|
|
@router.delete("/config/{config_id}")
|
|
async def delete_sso_config(
|
|
config_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(require_admin_role),
|
|
):
|
|
row = await db.execute(
|
|
select(SSOConfig).where(SSOConfig.id == config_id, SSOConfig.tenant_id == user.tenant_id)
|
|
)
|
|
config = row.scalar_one_or_none()
|
|
if not config:
|
|
raise HTTPException(404, "SSO 설정을 찾을 수 없습니다")
|
|
await db.delete(config)
|
|
await db.commit()
|
|
return {"ok": True}
|
|
|
|
|
|
@router.get("/login/{config_id}")
|
|
async def sso_login_redirect(
|
|
config_id: int,
|
|
request: Request,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""SSO 로그인 페이지로 리다이렉트."""
|
|
row = await db.execute(
|
|
select(SSOConfig).where(SSOConfig.id == config_id, SSOConfig.is_active == True)
|
|
)
|
|
config = row.scalar_one_or_none()
|
|
if not config:
|
|
raise HTTPException(404, "SSO 설정을 찾을 수 없습니다")
|
|
|
|
state = secrets.token_urlsafe(32)
|
|
|
|
if config.provider_type == "SAML":
|
|
acs_url = f"{BASE_URL}/api/sso/callback/saml?config_id={config_id}"
|
|
sp_entity_id = f"{BASE_URL}/api/sso/metadata"
|
|
authn_request = _build_saml_authn_request(
|
|
config.idp_sso_url or "", sp_entity_id, acs_url
|
|
)
|
|
redirect_url = f"{config.idp_sso_url}?" + urlencode({
|
|
"SAMLRequest": authn_request,
|
|
"RelayState": state,
|
|
})
|
|
return RedirectResponse(redirect_url)
|
|
|
|
elif config.provider_type in ("OIDC", "OAUTH2"):
|
|
discovery = await _oidc_discovery(config.discovery_url or "")
|
|
if not discovery:
|
|
raise HTTPException(400, "OIDC 디스커버리 실패")
|
|
auth_endpoint = discovery.get("authorization_endpoint", "")
|
|
redirect_uri = f"{BASE_URL}/api/sso/callback/oidc"
|
|
params = {
|
|
"response_type": "code",
|
|
"client_id": config.client_id,
|
|
"redirect_uri": redirect_uri,
|
|
"scope": config.scopes,
|
|
"state": f"{config_id}:{state}",
|
|
}
|
|
return RedirectResponse(f"{auth_endpoint}?{urlencode(params)}")
|
|
|
|
raise HTTPException(400, "지원하지 않는 IdP 타입")
|
|
|
|
|
|
@router.post("/callback/saml")
|
|
async def saml_acs(
|
|
request: Request,
|
|
config_id: int = Query(...),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""SAML ACS — IdP에서 전달한 Assertion 처리 → JWT 발급."""
|
|
form = await request.form()
|
|
saml_response = form.get("SAMLResponse", "")
|
|
if not saml_response:
|
|
raise HTTPException(400, "SAMLResponse 없음")
|
|
|
|
attrs = _parse_saml_response(str(saml_response))
|
|
if not attrs.get("email"):
|
|
raise HTTPException(401, "SAML Assertion에서 이메일 추출 실패")
|
|
|
|
row = await db.execute(select(SSOConfig).where(SSOConfig.id == config_id))
|
|
config = row.scalar_one_or_none()
|
|
if not config:
|
|
raise HTTPException(404, "SSO 설정 없음")
|
|
|
|
user = await _upsert_sso_user(attrs, config, db)
|
|
if not user:
|
|
raise HTTPException(401, "사용자 동기화 실패")
|
|
|
|
token = create_access_token({"sub": user.email, "user_id": user.id})
|
|
# SPA로 토큰 전달 (실제: secure cookie 또는 post message)
|
|
return RedirectResponse(f"/?sso_token={token}&provider=saml")
|
|
|
|
|
|
@router.get("/callback/oidc")
|
|
async def oidc_callback(
|
|
code: str = Query(...),
|
|
state: str = Query(""),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""OIDC 콜백 — 인가코드 → 토큰 → 사용자 정보 → JWT 발급."""
|
|
config_id_str = state.split(":")[0] if ":" in state else "0"
|
|
try:
|
|
config_id = int(config_id_str)
|
|
except ValueError:
|
|
raise HTTPException(400, "잘못된 state")
|
|
|
|
row = await db.execute(select(SSOConfig).where(SSOConfig.id == config_id))
|
|
config = row.scalar_one_or_none()
|
|
if not config:
|
|
raise HTTPException(404, "SSO 설정 없음")
|
|
|
|
redirect_uri = f"{BASE_URL}/api/sso/callback/oidc"
|
|
tokens = await _exchange_code(config, code, redirect_uri)
|
|
if not tokens:
|
|
raise HTTPException(401, "토큰 교환 실패")
|
|
|
|
userinfo = await _get_userinfo(config.discovery_url or "", tokens.get("access_token", ""))
|
|
if not userinfo:
|
|
raise HTTPException(401, "사용자 정보 조회 실패")
|
|
|
|
user = await _upsert_sso_user(userinfo, config, db)
|
|
if not user:
|
|
raise HTTPException(401, "사용자 동기화 실패")
|
|
|
|
jwt_token = create_access_token({"sub": user.email, "user_id": user.id})
|
|
return RedirectResponse(f"/?sso_token={jwt_token}&provider=oidc")
|
|
|
|
|
|
@router.get("/metadata")
|
|
async def sp_metadata():
|
|
"""SP (Service Provider) SAML Metadata XML."""
|
|
acs_url = f"{BASE_URL}/api/sso/callback/saml"
|
|
entity_id = f"{BASE_URL}/api/sso/metadata"
|
|
xml = f"""<?xml version="1.0"?>
|
|
<md:EntityDescriptor xmlns:md="urn:oasis:names:tc:SAML:2.0:metadata"
|
|
entityID="{entity_id}">
|
|
<md:SPSSODescriptor
|
|
AuthnRequestsSigned="false"
|
|
WantAssertionsSigned="false"
|
|
protocolSupportEnumeration="urn:oasis:names:tc:SAML:2.0:protocol">
|
|
<md:AssertionConsumerService
|
|
Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST"
|
|
Location="{acs_url}"
|
|
index="1"/>
|
|
</md:SPSSODescriptor>
|
|
</md:EntityDescriptor>"""
|
|
return Response(content=xml, media_type="application/xml")
|