feat(appliance): initial release of MERICA standalone appliance v1.0.0
This commit is contained in:
@@ -0,0 +1,88 @@
|
||||
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
|
||||
Reference in New Issue
Block a user