89 lines
3.5 KiB
Python
89 lines
3.5 KiB
Python
import os
|
|
import json
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import List, Dict, Any, Tuple
|
|
from app.core.config import settings
|
|
from app.core.auth.models import RBACSettingsSchema, CapabilityTier
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Cache for loaded settings
|
|
_SETTINGS_CACHE: Dict[str, Any] = {}
|
|
|
|
|
|
def get_rbac_storage_path() -> str:
|
|
"""Resolve storage path for dynamic RBAC settings file."""
|
|
base_dir = settings.METADATA_DIR
|
|
if not os.path.exists(base_dir):
|
|
base_dir = settings.CONFIGS_DIR
|
|
return os.path.join(base_dir, "rbac_mappings.json")
|
|
|
|
|
|
def load_rbac_settings() -> RBACSettingsSchema:
|
|
"""Load dynamic RBAC settings from persistent storage, fallback to config.py defaults."""
|
|
filepath = get_rbac_storage_path()
|
|
|
|
if os.path.exists(filepath):
|
|
try:
|
|
with open(filepath, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
return RBACSettingsSchema(**data)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to read custom RBAC settings from {filepath}: {e}. Using defaults.")
|
|
|
|
# Fallback to defaults defined in config.py
|
|
return RBACSettingsSchema(
|
|
editor_roles=list(settings.AUTH_EDITOR_ROLES),
|
|
editor_groups=list(settings.AUTH_EDITOR_GROUPS),
|
|
viewer_roles=list(settings.AUTH_VIEWER_ROLES),
|
|
viewer_groups=list(settings.AUTH_VIEWER_GROUPS),
|
|
admin_roles=list(settings.AUTH_ADMIN_ROLES),
|
|
admin_groups=list(settings.AUTH_ADMIN_GROUPS),
|
|
updated_at=datetime.now(timezone.utc).isoformat(),
|
|
updated_by="system_default"
|
|
)
|
|
|
|
|
|
def save_rbac_settings(schema: RBACSettingsSchema, author: str = "admin") -> RBACSettingsSchema:
|
|
"""Save updated dynamic RBAC settings to persistent storage."""
|
|
filepath = get_rbac_storage_path()
|
|
os.makedirs(os.path.dirname(filepath), exist_ok=True)
|
|
|
|
schema.updated_at = datetime.now(timezone.utc).isoformat()
|
|
schema.updated_by = author
|
|
|
|
with open(filepath, "w", encoding="utf-8") as f:
|
|
json.dump(schema.model_dump(), f, indent=2)
|
|
|
|
logger.info(f"Dynamic RBAC settings successfully updated by {author} at {filepath}")
|
|
return schema
|
|
|
|
|
|
def resolve_user_capabilities(user_roles: List[str], user_groups: List[str]) -> Tuple[str, bool, bool, bool, bool]:
|
|
"""
|
|
Resolve raw user roles & groups into internal capabilities:
|
|
Returns (capability, can_edit, can_deploy, is_admin, is_readonly)
|
|
"""
|
|
rbac = load_rbac_settings()
|
|
|
|
roles_set = {str(r).strip().lower() for r in user_roles if r}
|
|
groups_set = {str(g).strip().lower() for g in user_groups if g}
|
|
|
|
admin_roles_set = {str(r).strip().lower() for r in rbac.admin_roles}
|
|
admin_groups_set = {str(g).strip().lower() for g in rbac.admin_groups}
|
|
|
|
editor_roles_set = {str(r).strip().lower() for r in rbac.editor_roles}
|
|
editor_groups_set = {str(g).strip().lower() for g in rbac.editor_groups}
|
|
|
|
# 1. Check Admin Capability (Highest Priority)
|
|
if roles_set.intersection(admin_roles_set) or groups_set.intersection(admin_groups_set):
|
|
return CapabilityTier.ADMIN.value, True, True, True, False
|
|
|
|
# 2. Check Editor Capability (Data Engineer - Full Canvas Access)
|
|
if roles_set.intersection(editor_roles_set) or groups_set.intersection(editor_groups_set):
|
|
return CapabilityTier.EDITOR.value, True, True, False, False
|
|
|
|
# 3. Default Safe Fallback: Viewer (Pimpinan / Auditor / Read-Only)
|
|
return CapabilityTier.VIEWER.value, False, False, False, True
|