162 lines
5.6 KiB
Python
162 lines
5.6 KiB
Python
import requests
|
|
from datetime import datetime, timezone
|
|
from sqlalchemy.orm import Session
|
|
from fastapi import HTTPException
|
|
from typing import List, Dict, Any
|
|
import hmac
|
|
import hashlib
|
|
from app.models.auth.module_model import Module
|
|
from app.models.auth.module_environment_model import ModuleEnvironment
|
|
from app.models.auth.access_model import Access
|
|
from app.services.auth.trust_service import TrustService
|
|
from app.models.auth.module_access_model import ModuleAccess
|
|
import uuid
|
|
from app.core.redis import sync_redis_client
|
|
|
|
class ModulePermissionService:
|
|
|
|
@staticmethod
|
|
def sync_permissions(db: Session, module_id: str):
|
|
"""
|
|
Connects to the module's default environment and fetches defined permissions.
|
|
Updates the local ModuleAccess table to mirror these permissions.
|
|
|
|
:param module_id: The UUID string of the module
|
|
"""
|
|
try:
|
|
module_uuid = uuid.UUID(module_id)
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail="Invalid module UUID")
|
|
|
|
module = db.query(Module).filter(Module.id == module_uuid).first()
|
|
if not module:
|
|
raise HTTPException(status_code=404, detail="Module not found")
|
|
|
|
env = db.query(ModuleEnvironment).filter(
|
|
ModuleEnvironment.module_id == module.id,
|
|
ModuleEnvironment.is_default == True
|
|
).first()
|
|
|
|
if not env:
|
|
env = db.query(ModuleEnvironment).filter(
|
|
ModuleEnvironment.module_id == module.id,
|
|
ModuleEnvironment.is_active == True
|
|
).first()
|
|
|
|
if not env:
|
|
raise HTTPException(status_code=400, detail="No active environment to sync from")
|
|
|
|
try:
|
|
|
|
payload_body = "{}"
|
|
|
|
secret = env.trust_credentials.get("hmac_secret")
|
|
if not secret:
|
|
pass
|
|
|
|
if secret:
|
|
signature = hmac.new(
|
|
secret.encode("utf-8"),
|
|
payload_body.encode("utf-8"),
|
|
hashlib.sha256
|
|
).hexdigest()
|
|
else:
|
|
signature = ""
|
|
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"X-SaaS-Signature": signature
|
|
}
|
|
|
|
url = f"{env.backend_base_url}{env.permission_sync_endpoint}"
|
|
response = requests.post(url, headers=headers, data=payload_body, timeout=10)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
except Exception as e:
|
|
raise HTTPException(status_code=502, detail=f"Failed to fetch permissions from module: {str(e)}")
|
|
|
|
permissions: List[Dict[str, Any]] = data.get("permissions", [])
|
|
|
|
synced_count = 0
|
|
timestamp = datetime.now(timezone.utc)
|
|
|
|
permission_map = {}
|
|
|
|
for perm in permissions:
|
|
code = perm.get("permission_code")
|
|
if not code:
|
|
continue
|
|
|
|
access = db.query(ModuleAccess).filter(
|
|
ModuleAccess.module_id == module.id,
|
|
ModuleAccess.access_code == code
|
|
).first()
|
|
|
|
if not access:
|
|
access = ModuleAccess(
|
|
access_code=code,
|
|
module_id=module.id,
|
|
name=perm.get("name", code),
|
|
category=perm.get("category", "General"),
|
|
)
|
|
db.add(access)
|
|
else:
|
|
access.name = perm.get("name", access.name)
|
|
access.category = perm.get("category", access.category)
|
|
|
|
access.last_synced_at = timestamp
|
|
access.sync_checksum = perm.get("hash")
|
|
|
|
permission_map[code] = access
|
|
synced_count += 1
|
|
|
|
db.flush()
|
|
|
|
for perm in permissions:
|
|
code = perm.get("permission_code")
|
|
parent_code = perm.get("parent_code")
|
|
|
|
if not code or not parent_code:
|
|
continue
|
|
|
|
access = permission_map.get(code)
|
|
parent_access = permission_map.get(parent_code)
|
|
|
|
if not parent_access:
|
|
parent_access = db.query(ModuleAccess).filter(
|
|
ModuleAccess.module_id == module.id,
|
|
ModuleAccess.access_code == parent_code
|
|
).first()
|
|
|
|
if access and parent_access:
|
|
access.parent_id = parent_access.id
|
|
|
|
db.commit()
|
|
|
|
try:
|
|
if sync_redis_client.client:
|
|
sync_redis_client.client.delete("saas:access:v2:all:full")
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"status": "success",
|
|
"message": "Permissions synced successfully",
|
|
"synced_count": synced_count
|
|
}
|
|
|
|
@staticmethod
|
|
def get_module_permissions(db: Session, module_id: str):
|
|
"""List permissions for a module from local DB."""
|
|
try:
|
|
module_uuid = uuid.UUID(module_id)
|
|
module = db.query(Module).filter(Module.id == module_uuid).first()
|
|
except ValueError:
|
|
module = db.query(Module).filter(Module.module_id == module_id).first()
|
|
if not module:
|
|
raise HTTPException(status_code=404, detail="Module not found")
|
|
|
|
return db.query(ModuleAccess).filter(
|
|
ModuleAccess.module_id == module.id
|
|
).all() |