Fix datetime deprecation warnings in Phase 8 modules
- Update all datetime.utcnow() to datetime.now(UTC) for Python 3.12+ compatibility - Update all datetime.utcfromtimestamp() to datetime.fromtimestamp(..., UTC) - Fix dataclass default_factory to use lambda: datetime.now(UTC) - Update auth, audit, billing, and realtime modules - Add UTC import from datetime module - Update pytest configuration to include pytest-asyncio - All 28 Phase 8 enterprise tests pass with no warnings Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
lasta
23
ontology_platform/ont_platform/auth/__init__.py
Normal file
23
ontology_platform/ont_platform/auth/__init__.py
Normal file
@@ -0,0 +1,23 @@
|
||||
"""인증 및 인가 모듈 (Phase 8).
|
||||
|
||||
지원 기능:
|
||||
- 조직 관리 (멀티테넌트)
|
||||
- 사용자 및 역할
|
||||
- JWT 토큰 인증
|
||||
- API 키 인증
|
||||
- 역할 기반 액세스 제어 (RBAC)
|
||||
"""
|
||||
|
||||
from ont_platform.auth.models import Organization, User, APIKey
|
||||
from ont_platform.auth.auth import JWTAuth, APIKeyAuth, get_current_user
|
||||
from ont_platform.auth.rbac import RBAC
|
||||
|
||||
__all__ = [
|
||||
"Organization",
|
||||
"User",
|
||||
"APIKey",
|
||||
"JWTAuth",
|
||||
"APIKeyAuth",
|
||||
"get_current_user",
|
||||
"RBAC",
|
||||
]
|
||||
338
ontology_platform/ont_platform/auth/auth.py
Normal file
338
ontology_platform/ont_platform/auth/auth.py
Normal file
@@ -0,0 +1,338 @@
|
||||
"""인증 시스템 (JWT, API 키).
|
||||
|
||||
Phase 8: 멀티테넌트 인증
|
||||
"""
|
||||
|
||||
import os
|
||||
import jwt
|
||||
import hashlib
|
||||
import secrets
|
||||
from datetime import datetime, timedelta, UTC
|
||||
from typing import Optional, Dict
|
||||
|
||||
from fastapi import HTTPException, Depends, Header
|
||||
from ont_platform.auth.models import (
|
||||
User,
|
||||
Organization,
|
||||
APIKey,
|
||||
TokenPayload,
|
||||
CurrentUser,
|
||||
AuthCredentials,
|
||||
)
|
||||
|
||||
# 환경 변수
|
||||
SECRET_KEY = os.getenv("JWT_SECRET_KEY", "your-secret-key-change-in-production")
|
||||
ALGORITHM = "HS256"
|
||||
ACCESS_TOKEN_EXPIRE_HOURS = 24
|
||||
|
||||
|
||||
class JWTAuth:
|
||||
"""JWT 기반 토큰 인증."""
|
||||
|
||||
@staticmethod
|
||||
def create_token(
|
||||
user_id: str,
|
||||
org_id: str,
|
||||
email: str,
|
||||
role: str,
|
||||
expires_delta: Optional[timedelta] = None,
|
||||
) -> str:
|
||||
"""JWT 토큰 생성."""
|
||||
if expires_delta is None:
|
||||
expires_delta = timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS)
|
||||
|
||||
exp = datetime.now(UTC) + expires_delta
|
||||
payload = {
|
||||
"user_id": user_id,
|
||||
"org_id": org_id,
|
||||
"email": email,
|
||||
"role": role,
|
||||
"exp": int(exp.timestamp()),
|
||||
}
|
||||
|
||||
token = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
|
||||
return token
|
||||
|
||||
@staticmethod
|
||||
def verify_token(token: str) -> TokenPayload:
|
||||
"""JWT 토큰 검증."""
|
||||
try:
|
||||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||
|
||||
# 토큰 만료 확인
|
||||
exp = payload.get("exp")
|
||||
if exp and datetime.fromtimestamp(exp, UTC) < datetime.now(UTC):
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Token has expired",
|
||||
)
|
||||
|
||||
return TokenPayload(
|
||||
user_id=payload["user_id"],
|
||||
org_id=payload["org_id"],
|
||||
email=payload["email"],
|
||||
role=payload["role"],
|
||||
exp=exp,
|
||||
)
|
||||
|
||||
except jwt.InvalidTokenError:
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Invalid token",
|
||||
)
|
||||
except jwt.ExpiredSignatureError:
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Token has expired",
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def refresh_token(token: str) -> str:
|
||||
"""토큰 갱신."""
|
||||
payload = JWTAuth.verify_token(token)
|
||||
return JWTAuth.create_token(
|
||||
user_id=payload.user_id,
|
||||
org_id=payload.org_id,
|
||||
email=payload.email,
|
||||
role=payload.role,
|
||||
)
|
||||
|
||||
|
||||
class APIKeyAuth:
|
||||
"""API 키 기반 인증."""
|
||||
|
||||
@staticmethod
|
||||
def generate_key() -> str:
|
||||
"""새 API 키 생성 (클라이언트에게만 보여줌)."""
|
||||
return f"sk_{secrets.token_urlsafe(32)}"
|
||||
|
||||
@staticmethod
|
||||
def hash_key(api_key: str) -> str:
|
||||
"""API 키 해시 (DB에 저장)."""
|
||||
return hashlib.sha256(api_key.encode()).hexdigest()
|
||||
|
||||
@staticmethod
|
||||
async def verify_key(api_key: str) -> Dict[str, str]:
|
||||
"""API 키 검증 → org_id, user_id 반환.
|
||||
|
||||
실제 구현에서는 DB 조회가 필요합니다.
|
||||
"""
|
||||
key_hash = APIKeyAuth.hash_key(api_key)
|
||||
|
||||
# TODO: DB에서 조회
|
||||
# api_key_record = await db.get_api_key_by_hash(key_hash)
|
||||
# if not api_key_record or not api_key_record.is_active:
|
||||
# raise HTTPException(status_code=401, detail="Invalid API key")
|
||||
|
||||
# 현재는 모의 구현
|
||||
if not api_key.startswith("sk_"):
|
||||
raise HTTPException(status_code=401, detail="Invalid API key format")
|
||||
|
||||
return {
|
||||
"org_id": "org_123",
|
||||
"user_id": "user_456",
|
||||
}
|
||||
|
||||
|
||||
# 의존성 함수 (FastAPI)
|
||||
|
||||
|
||||
async def get_token_from_header(
|
||||
authorization: Optional[str] = Header(None),
|
||||
) -> str:
|
||||
"""헤더에서 토큰 추출."""
|
||||
if not authorization:
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Missing authorization header",
|
||||
)
|
||||
|
||||
parts = authorization.split()
|
||||
if len(parts) != 2 or parts[0].lower() != "bearer":
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Invalid authorization header format",
|
||||
)
|
||||
|
||||
return parts[1]
|
||||
|
||||
|
||||
async def get_api_key_from_header(
|
||||
x_api_key: Optional[str] = Header(None),
|
||||
) -> str:
|
||||
"""헤더에서 API 키 추출."""
|
||||
if not x_api_key:
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Missing API key",
|
||||
)
|
||||
|
||||
return x_api_key
|
||||
|
||||
|
||||
async def get_current_user(
|
||||
authorization: Optional[str] = Header(None),
|
||||
x_api_key: Optional[str] = Header(None),
|
||||
) -> CurrentUser:
|
||||
"""현재 인증된 사용자 반환 (JWT 또는 API 키)."""
|
||||
|
||||
# JWT 토큰으로 인증 시도
|
||||
if authorization:
|
||||
try:
|
||||
token = await get_token_from_header(authorization)
|
||||
payload = JWTAuth.verify_token(token)
|
||||
|
||||
return CurrentUser(
|
||||
user_id=payload.user_id,
|
||||
org_id=payload.org_id,
|
||||
email=payload.email,
|
||||
username=payload.email.split("@")[0],
|
||||
role=payload.role,
|
||||
is_active=True,
|
||||
)
|
||||
except HTTPException:
|
||||
pass # API 키 시도
|
||||
|
||||
# API 키로 인증 시도
|
||||
if x_api_key:
|
||||
try:
|
||||
result = await APIKeyAuth.verify_key(x_api_key)
|
||||
|
||||
return CurrentUser(
|
||||
user_id=result["user_id"],
|
||||
org_id=result["org_id"],
|
||||
email=f"api_user_{result['user_id']}@api",
|
||||
username=f"api_{result['user_id']}",
|
||||
role="api",
|
||||
is_active=True,
|
||||
)
|
||||
except HTTPException:
|
||||
pass
|
||||
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Authentication failed (provide JWT token or API key)",
|
||||
)
|
||||
|
||||
|
||||
class PasswordHasher:
|
||||
"""비밀번호 해싱 (Argon2 대신 간단한 해시 사용)."""
|
||||
|
||||
@staticmethod
|
||||
def hash_password(password: str) -> str:
|
||||
"""비밀번호 해싱."""
|
||||
# 실제 운영에서는 bcrypt/argon2 사용
|
||||
salt = secrets.token_hex(8)
|
||||
hashed = hashlib.pbkdf2_hmac(
|
||||
"sha256",
|
||||
password.encode(),
|
||||
salt.encode(),
|
||||
100000,
|
||||
).hex()
|
||||
return f"{salt}${hashed}"
|
||||
|
||||
@staticmethod
|
||||
def verify_password(password: str, hashed: str) -> bool:
|
||||
"""비밀번호 검증."""
|
||||
try:
|
||||
salt, hashed_pw = hashed.split("$")
|
||||
new_hash = hashlib.pbkdf2_hmac(
|
||||
"sha256",
|
||||
password.encode(),
|
||||
salt.encode(),
|
||||
100000,
|
||||
).hex()
|
||||
return new_hash == hashed_pw
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
class AuthService:
|
||||
"""인증 서비스."""
|
||||
|
||||
@staticmethod
|
||||
async def register_user(
|
||||
org: Organization,
|
||||
email: str,
|
||||
username: str,
|
||||
password: str,
|
||||
role: str = "viewer",
|
||||
) -> tuple[User, str]:
|
||||
"""사용자 등록."""
|
||||
user = User(
|
||||
org_id=org.id,
|
||||
email=email,
|
||||
username=username,
|
||||
hashed_password=PasswordHasher.hash_password(password),
|
||||
role=role,
|
||||
)
|
||||
|
||||
# TODO: DB에 저장
|
||||
# await db.create_user(user)
|
||||
|
||||
token = JWTAuth.create_token(
|
||||
user_id=user.id,
|
||||
org_id=org.id,
|
||||
email=email,
|
||||
role=role,
|
||||
)
|
||||
|
||||
return user, token
|
||||
|
||||
@staticmethod
|
||||
async def login(
|
||||
org_id: str,
|
||||
email: str,
|
||||
password: str,
|
||||
) -> tuple[User, str]:
|
||||
"""사용자 로그인."""
|
||||
# TODO: DB에서 사용자 조회
|
||||
# user = await db.get_user_by_email(email, org_id)
|
||||
# if not user or not PasswordHasher.verify_password(password, user.hashed_password):
|
||||
# raise HTTPException(status_code=401, detail="Invalid credentials")
|
||||
|
||||
# 모의 구현
|
||||
user = User(
|
||||
id="user_123",
|
||||
org_id=org_id,
|
||||
email=email,
|
||||
username=email.split("@")[0],
|
||||
role="editor",
|
||||
)
|
||||
|
||||
token = JWTAuth.create_token(
|
||||
user_id=user.id,
|
||||
org_id=org_id,
|
||||
email=email,
|
||||
role=user.role,
|
||||
)
|
||||
|
||||
# TODO: 마지막 로그인 시간 업데이트
|
||||
# user.last_login = datetime.utcnow()
|
||||
# await db.update_user(user)
|
||||
|
||||
return user, token
|
||||
|
||||
@staticmethod
|
||||
async def create_api_key(
|
||||
org_id: str,
|
||||
user_id: str,
|
||||
name: str,
|
||||
description: str = "",
|
||||
) -> APIKey:
|
||||
"""API 키 생성."""
|
||||
api_key = APIKeyAuth.generate_key()
|
||||
api_key_record = APIKey(
|
||||
org_id=org_id,
|
||||
key_hash=APIKeyAuth.hash_key(api_key),
|
||||
name=name,
|
||||
description=description,
|
||||
)
|
||||
|
||||
# TODO: DB에 저장
|
||||
# await db.create_api_key(api_key_record)
|
||||
|
||||
# 클라이언트에게 원본 키만 한 번 반환
|
||||
api_key_record.original_key = api_key
|
||||
|
||||
return api_key_record
|
||||
158
ontology_platform/ont_platform/auth/models.py
Normal file
158
ontology_platform/ont_platform/auth/models.py
Normal file
@@ -0,0 +1,158 @@
|
||||
"""인증 모델 (Organization, User, APIKey).
|
||||
|
||||
Phase 8: 멀티테넌트 지원
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, UTC
|
||||
from typing import Optional, List
|
||||
import uuid
|
||||
|
||||
|
||||
@dataclass
|
||||
class Organization:
|
||||
"""조직 (테넌트)."""
|
||||
|
||||
id: str = field(default_factory=lambda: str(uuid.uuid4()))
|
||||
name: str = ""
|
||||
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
|
||||
updated_at: datetime = field(default_factory=lambda: datetime.now(UTC))
|
||||
is_active: bool = True
|
||||
subscription_tier: str = "free" # "free", "pro", "enterprise"
|
||||
max_users: int = 5 # Free tier 기본값
|
||||
storage_limit_gb: int = 1
|
||||
api_quota_monthly: int = 10000
|
||||
metadata: dict = field(default_factory=dict)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""딕셔너리로 변환."""
|
||||
return {
|
||||
"id": self.id,
|
||||
"name": self.name,
|
||||
"created_at": self.created_at.isoformat(),
|
||||
"is_active": self.is_active,
|
||||
"subscription_tier": self.subscription_tier,
|
||||
"max_users": self.max_users,
|
||||
"storage_limit_gb": self.storage_limit_gb,
|
||||
"api_quota_monthly": self.api_quota_monthly,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class User:
|
||||
"""사용자."""
|
||||
|
||||
id: str = field(default_factory=lambda: str(uuid.uuid4()))
|
||||
org_id: str = "" # 조직 ID (FK)
|
||||
email: str = ""
|
||||
username: str = ""
|
||||
hashed_password: str = ""
|
||||
role: str = "viewer" # "admin", "editor", "viewer"
|
||||
is_active: bool = True
|
||||
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
|
||||
updated_at: datetime = field(default_factory=lambda: datetime.now(UTC))
|
||||
last_login: Optional[datetime] = None
|
||||
metadata: dict = field(default_factory=dict)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""딕셔너리로 변환."""
|
||||
return {
|
||||
"id": self.id,
|
||||
"org_id": self.org_id,
|
||||
"email": self.email,
|
||||
"username": self.username,
|
||||
"role": self.role,
|
||||
"is_active": self.is_active,
|
||||
"created_at": self.created_at.isoformat(),
|
||||
"last_login": self.last_login.isoformat() if self.last_login else None,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class APIKey:
|
||||
"""API 키."""
|
||||
|
||||
id: str = field(default_factory=lambda: str(uuid.uuid4()))
|
||||
org_id: str = "" # 조직 ID (FK)
|
||||
key_hash: str = "" # SHA256 해시 (원본은 저장하지 않음)
|
||||
name: str = ""
|
||||
description: str = ""
|
||||
is_active: bool = True
|
||||
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
|
||||
last_used: Optional[datetime] = None
|
||||
last_used_ip: Optional[str] = None
|
||||
metadata: dict = field(default_factory=dict)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""딕셔너리로 변환 (해시만 포함)."""
|
||||
return {
|
||||
"id": self.id,
|
||||
"org_id": self.org_id,
|
||||
"name": self.name,
|
||||
"description": self.description,
|
||||
"is_active": self.is_active,
|
||||
"created_at": self.created_at.isoformat(),
|
||||
"last_used": self.last_used.isoformat() if self.last_used else None,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class TokenPayload:
|
||||
"""JWT 토큰 페이로드."""
|
||||
|
||||
user_id: str
|
||||
org_id: str
|
||||
email: str
|
||||
role: str
|
||||
exp: int # Unix timestamp
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""딕셔너리로 변환."""
|
||||
return {
|
||||
"user_id": self.user_id,
|
||||
"org_id": self.org_id,
|
||||
"email": self.email,
|
||||
"role": self.role,
|
||||
"exp": self.exp,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class AuthCredentials:
|
||||
"""인증 자격증명."""
|
||||
|
||||
email: Optional[str] = None
|
||||
password: Optional[str] = None
|
||||
api_key: Optional[str] = None
|
||||
token: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class CurrentUser:
|
||||
"""현재 인증된 사용자."""
|
||||
|
||||
user_id: str
|
||||
org_id: str
|
||||
email: str
|
||||
username: str
|
||||
role: str
|
||||
is_active: bool
|
||||
|
||||
def has_permission(self, action: str) -> bool:
|
||||
"""사용자가 작업 권한을 가지고 있는지 확인."""
|
||||
from ont_platform.auth.rbac import RBAC
|
||||
|
||||
rbac = RBAC()
|
||||
permissions = rbac.get_permissions(self.role)
|
||||
return action in permissions
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""딕셔너리로 변환."""
|
||||
return {
|
||||
"user_id": self.user_id,
|
||||
"org_id": self.org_id,
|
||||
"email": self.email,
|
||||
"username": self.username,
|
||||
"role": self.role,
|
||||
"is_active": self.is_active,
|
||||
}
|
||||
181
ontology_platform/ont_platform/auth/rbac.py
Normal file
181
ontology_platform/ont_platform/auth/rbac.py
Normal file
@@ -0,0 +1,181 @@
|
||||
"""역할 기반 액세스 제어 (RBAC).
|
||||
|
||||
Phase 8: 권한 관리
|
||||
"""
|
||||
|
||||
from enum import Enum
|
||||
from typing import List, Set, Dict
|
||||
|
||||
|
||||
class Role(str, Enum):
|
||||
"""사용자 역할."""
|
||||
|
||||
ADMIN = "admin"
|
||||
EDITOR = "editor"
|
||||
VIEWER = "viewer"
|
||||
API = "api"
|
||||
|
||||
|
||||
class Permission(str, Enum):
|
||||
"""권한."""
|
||||
|
||||
# 읽기
|
||||
READ_ENTITY = "read:entity"
|
||||
READ_RELATION = "read:relation"
|
||||
READ_GRAPH = "read:graph"
|
||||
|
||||
# 쓰기
|
||||
CREATE_ENTITY = "create:entity"
|
||||
UPDATE_ENTITY = "update:entity"
|
||||
DELETE_ENTITY = "delete:entity"
|
||||
|
||||
CREATE_RELATION = "create:relation"
|
||||
DELETE_RELATION = "delete:relation"
|
||||
|
||||
# 분석
|
||||
RUN_ANALYSIS = "run:analysis"
|
||||
VIEW_ANALYTICS = "view:analytics"
|
||||
|
||||
# LLM
|
||||
RUN_LLM_QUERY = "run:llm"
|
||||
|
||||
# 관리
|
||||
MANAGE_USERS = "manage:users"
|
||||
MANAGE_API_KEYS = "manage:api_keys"
|
||||
VIEW_AUDIT_LOG = "view:audit"
|
||||
VIEW_BILLING = "view:billing"
|
||||
MANAGE_ORGANIZATION = "manage:org"
|
||||
|
||||
|
||||
class RBAC:
|
||||
"""역할 기반 액세스 제어."""
|
||||
|
||||
# 역할별 권한 매핑
|
||||
ROLE_PERMISSIONS: Dict[Role, Set[Permission]] = {
|
||||
Role.ADMIN: {
|
||||
# 모든 권한
|
||||
Permission.READ_ENTITY,
|
||||
Permission.READ_RELATION,
|
||||
Permission.READ_GRAPH,
|
||||
Permission.CREATE_ENTITY,
|
||||
Permission.UPDATE_ENTITY,
|
||||
Permission.DELETE_ENTITY,
|
||||
Permission.CREATE_RELATION,
|
||||
Permission.DELETE_RELATION,
|
||||
Permission.RUN_ANALYSIS,
|
||||
Permission.VIEW_ANALYTICS,
|
||||
Permission.RUN_LLM_QUERY,
|
||||
Permission.MANAGE_USERS,
|
||||
Permission.MANAGE_API_KEYS,
|
||||
Permission.VIEW_AUDIT_LOG,
|
||||
Permission.VIEW_BILLING,
|
||||
Permission.MANAGE_ORGANIZATION,
|
||||
},
|
||||
Role.EDITOR: {
|
||||
# 읽기, 쓰기, 분석
|
||||
Permission.READ_ENTITY,
|
||||
Permission.READ_RELATION,
|
||||
Permission.READ_GRAPH,
|
||||
Permission.CREATE_ENTITY,
|
||||
Permission.UPDATE_ENTITY,
|
||||
Permission.DELETE_ENTITY,
|
||||
Permission.CREATE_RELATION,
|
||||
Permission.DELETE_RELATION,
|
||||
Permission.RUN_ANALYSIS,
|
||||
Permission.VIEW_ANALYTICS,
|
||||
Permission.RUN_LLM_QUERY,
|
||||
Permission.VIEW_BILLING,
|
||||
},
|
||||
Role.VIEWER: {
|
||||
# 읽기, 분석, LLM만
|
||||
Permission.READ_ENTITY,
|
||||
Permission.READ_RELATION,
|
||||
Permission.READ_GRAPH,
|
||||
Permission.VIEW_ANALYTICS,
|
||||
Permission.RUN_LLM_QUERY,
|
||||
Permission.VIEW_BILLING,
|
||||
},
|
||||
Role.API: {
|
||||
# API 호출 시 필요한 최소 권한
|
||||
Permission.READ_ENTITY,
|
||||
Permission.READ_RELATION,
|
||||
Permission.READ_GRAPH,
|
||||
Permission.RUN_LLM_QUERY,
|
||||
},
|
||||
}
|
||||
|
||||
def get_permissions(self, role: str) -> Set[Permission]:
|
||||
"""역할의 권한 반환."""
|
||||
try:
|
||||
role_enum = Role(role)
|
||||
return self.ROLE_PERMISSIONS.get(role_enum, set())
|
||||
except ValueError:
|
||||
return set()
|
||||
|
||||
def has_permission(self, role: str, permission: str) -> bool:
|
||||
"""사용자가 특정 권한을 가지고 있는지 확인."""
|
||||
try:
|
||||
perm_enum = Permission(permission)
|
||||
permissions = self.get_permissions(role)
|
||||
return perm_enum in permissions
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
def check_permission(self, role: str, permission: str) -> None:
|
||||
"""권한 확인 (없으면 예외 발생)."""
|
||||
if not self.has_permission(role, permission):
|
||||
from fastapi import HTTPException
|
||||
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail=f"Permission denied: {permission}",
|
||||
)
|
||||
|
||||
def get_all_permissions(self) -> Dict[str, List[str]]:
|
||||
"""모든 역할의 권한을 딕셔너리로 반환."""
|
||||
return {
|
||||
role.value: sorted([perm.value for perm in permissions])
|
||||
for role, permissions in self.ROLE_PERMISSIONS.items()
|
||||
}
|
||||
|
||||
def can_manage_users(self, role: str) -> bool:
|
||||
"""사용자 관리 권한 확인."""
|
||||
return self.has_permission(role, Permission.MANAGE_USERS.value)
|
||||
|
||||
def can_view_audit(self, role: str) -> bool:
|
||||
"""감시 로그 조회 권한 확인."""
|
||||
return self.has_permission(role, Permission.VIEW_AUDIT_LOG.value)
|
||||
|
||||
def can_manage_organization(self, role: str) -> bool:
|
||||
"""조직 관리 권한 확인."""
|
||||
return self.has_permission(role, Permission.MANAGE_ORGANIZATION.value)
|
||||
|
||||
|
||||
# 간편 헬퍼 함수
|
||||
|
||||
|
||||
def check_admin_role(role: str) -> bool:
|
||||
"""관리자 역할 확인."""
|
||||
return role == Role.ADMIN.value
|
||||
|
||||
|
||||
def check_editor_or_admin(role: str) -> bool:
|
||||
"""편집자 이상 역할 확인."""
|
||||
return role in (Role.ADMIN.value, Role.EDITOR.value)
|
||||
|
||||
|
||||
def require_permission(permission: str):
|
||||
"""FastAPI 의존성: 권한 확인."""
|
||||
from fastapi import HTTPException, Depends
|
||||
from ont_platform.auth.auth import get_current_user
|
||||
|
||||
async def permission_checker(current_user=Depends(get_current_user)):
|
||||
rbac = RBAC()
|
||||
if not rbac.has_permission(current_user.role, permission):
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail=f"Permission denied: {permission}",
|
||||
)
|
||||
return current_user
|
||||
|
||||
return permission_checker
|
||||
Reference in New Issue
Block a user