import os from contextvars import ContextVar import casbin import redis from casbin_sqlalchemy_adapter import Adapter from app.config import settings from app.db.session import SessionLocal from common_logging import get_logger logger = get_logger(__name__) _request_context: ContextVar[dict] = ContextVar("request_context", default=None) _enforcer: casbin.Enforcer | None = None _redis_client: redis.Redis | None = None def get_redis_client() -> redis.Redis: global _redis_client if _redis_client is None: if settings.REDIS_URL: _redis_client = redis.from_url(settings.REDIS_URL, decode_responses=True) else: _redis_client = redis.Redis( host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=settings.REDIS_DB, password=settings.REDIS_PASSWORD, decode_responses=True, ) return _redis_client def get_enforcer() -> casbin.Enforcer: global _enforcer if _enforcer is None: base_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) model_path = os.path.join(base_dir, "casbin", "model.conf") adapter = Adapter(settings.DATABASE_URL) _enforcer = casbin.Enforcer(model_path, adapter) def domain_match(domain1: str, domain2: str) -> bool: return domain1 == domain2 or domain2 == "*" def resource_match(resource1: str, resource2: str) -> bool: return resource1 == resource2 or resource2 == "*" def action_match(action1: str, action2: str) -> bool: return action1 == action2 or action2 == "*" _enforcer.add_function("domainMatch", domain_match) _enforcer.add_function("resourceMatch", resource_match) _enforcer.add_function("actionMatch", action_match) logger.info("✅ Casbin enforcer initialized with database adapter") logger.info(f" Model: {model_path}") logger.info(f" Database: {settings.DATABASE_URL.split('@')[-1]}") return _enforcer def _log_permission_check( user_id: int, tenant_id: int, resource: str, action: str, result: bool, role_id: int | None = None, ): try: from app.tasks.audit_tasks import log_permission_check ctx = _request_context.get() log_permission_check.delay( user_id=user_id, tenant_id=tenant_id, resource=resource, action=action, result=result, role_id=role_id, ip_address=ctx.get("ip_address"), user_agent=ctx.get("user_agent"), request_path=ctx.get("path"), ) except Exception as e: logger.error(f"Failed to queue permission check log: {e}") def _log_permission_change(action_type: str, tenant_id: int, resource: str, action: str, role: str): try: from app.tasks.audit_tasks import log_permission_change ctx = _request_context.get() log_permission_change.delay( action_type=action_type, tenant_id=tenant_id, resource=resource, action=action, role=role, user_id=ctx.get("user_id", 0), ip_address=ctx.get("ip_address"), user_agent=ctx.get("user_agent"), request_path=ctx.get("path"), ) except Exception as e: logger.error(f"Failed to queue permission change log: {e}") def set_request_context(ip_address: str = None, user_agent: str = None, path: str = None): _request_context.set({"ip_address": ip_address, "user_agent": user_agent, "path": path}) def check_permission(user_id: int, tenant_id: int, resource: str, action: str) -> bool: cache_key = f"perm:{user_id}:{tenant_id}:{resource}:{action}" try: redis_client = get_redis_client() cached = redis_client.get(cache_key) if cached is not None: logger.info(f"Permission check cached: user={user_id}, resource={resource}, action={action}, result={cached == '1'}", user_id=user_id, tenant_id=tenant_id) return cached == "1" except Exception as e: logger.warning(f"Redis cache error: {e}") enforcer = get_enforcer() subject = f"user:{user_id}" domain = str(tenant_id) result = enforcer.enforce(subject, domain, resource, action) logger.info(f"Permission check: resource={resource}, action={action}, result={result}", user_id=user_id, tenant_id=tenant_id) try: redis_client.setex(cache_key, 300, "1" if result else "0") except Exception as e: logger.warning(f"Redis cache write error: {e}") return result def add_role_for_user(user_id: int, role: str, tenant_id: int) -> bool: from app.models.role import Role from app.models.tenant import Tenant from app.models.user import User db = SessionLocal() try: user = db.query(User).filter(User.id == user_id).first() if not user: raise ValueError(f"User {user_id} does not exist") tenant = db.query(Tenant).filter(Tenant.id == tenant_id).first() if not tenant: raise ValueError(f"Tenant {tenant_id} does not exist") if user.tenant_id is not None and user.tenant_id != tenant_id: raise ValueError(f"User {user_id} does not belong to tenant {tenant_id}") role_code = role.split(":")[-1] if ":" in role else role role_obj = db.query(Role).filter(Role.code == role_code).first() if not role_obj: raise ValueError(f"Role {role_code} does not exist") if ( role_obj.is_system and user.tenant_id is not None and (role_code in {"platform_admin", "platform_user"}) ): raise ValueError(f"Cannot assign platform role {role_code} to tenant user") if role_obj.tenant_id is not None and role_obj.tenant_id != tenant_id: raise ValueError(f"Role {role_code} does not belong to tenant {tenant_id}") enforcer = get_enforcer() subject = f"user:{user_id}" domain = str(tenant_id) success = enforcer.add_grouping_policy(subject, role, domain) if success: enforcer.load_policy() logger.info(f"✅ Added role {role} for user {user_id} in tenant {tenant_id}") invalidate_user_permissions(user_id, tenant_id) else: logger.warning(f"⚠️ Failed to add role {role} for user {user_id} in tenant {tenant_id}") return success finally: db.close() def remove_role_for_user(user_id: int, role: str, tenant_id: int) -> bool: enforcer = get_enforcer() subject = f"user:{user_id}" domain = str(tenant_id) success = enforcer.remove_grouping_policy(subject, role, domain) if success: enforcer.load_policy() logger.info(f"✅ Removed role {role} from user {user_id} in tenant {tenant_id}") invalidate_user_permissions(user_id, tenant_id) return success def get_roles_for_user(user_id: int, tenant_id: int) -> list[str]: enforcer = get_enforcer() subject = f"user:{user_id}" domain = str(tenant_id) roles = enforcer.get_roles_for_user_in_domain(subject, domain) return roles def get_user_roles(user_id: int, tenant_id: int) -> list[dict]: roles = get_roles_for_user(user_id, tenant_id) return [{"role_code": role.replace("role:", "")} for role in roles] def get_users_for_role(role: str, tenant_id: int) -> list[str]: enforcer = get_enforcer() domain = str(tenant_id) users = enforcer.get_users_for_role_in_domain(role, domain) return users def add_permission_for_role(role: str, tenant_id: int, resource: str, action: str) -> bool: enforcer = get_enforcer() domain = str(tenant_id) success = enforcer.add_policy(role, domain, resource, action) if success: enforcer.load_policy() logger.info( f"✅ Added permission: role={role}, tenant={tenant_id}, resource={resource}, action={action}" ) _log_permission_change("grant", tenant_id, resource, action, role) _invalidate_role_permissions(role, tenant_id) return success def remove_permission_for_role(role: str, tenant_id: int, resource: str, action: str) -> bool: enforcer = get_enforcer() domain = str(tenant_id) success = enforcer.remove_policy(role, domain, resource, action) if success: enforcer.load_policy() logger.info( f"✅ Removed permission: role={role}, tenant={tenant_id}, resource={resource}, action={action}" ) _log_permission_change("revoke", tenant_id, resource, action, role) _invalidate_role_permissions(role, tenant_id) return success def get_permissions_for_role(role: str, tenant_id: int) -> list[list[str]]: enforcer = get_enforcer() domain = str(tenant_id) permissions = enforcer.get_filtered_policy(0, role, domain) return permissions def save_policy() -> bool: enforcer = get_enforcer() return enforcer.save_policy() def load_policy() -> bool: enforcer = get_enforcer() return enforcer.load_policy() def invalidate_user_permissions(user_id: int, tenant_id: int) -> None: try: redis_client = get_redis_client() pattern = f"perm:{user_id}:{tenant_id}:*" keys = redis_client.keys(pattern) if keys: redis_client.delete(*keys) logger.info(f"Invalidated {len(keys)} cached permissions for user {user_id}") except Exception as e: logger.warning(f"Failed to invalidate permissions cache: {e}") def _invalidate_role_permissions(role: str, tenant_id: int) -> None: try: users = get_users_for_role(role, tenant_id) for user_subject in users: if user_subject.startswith("user:"): user_id = int(user_subject.split(":")[1]) invalidate_user_permissions(user_id, tenant_id) except Exception as e: logger.warning(f"Failed to invalidate role permissions: {e}") def can_access_knowledge_base(user, kb) -> bool: if user.role in ["platform_admin", "platform_user"]: return True if user.role == "customer_admin": return kb.tenant_id == user.tenant_id if user.role == "customer_user": return kb.created_by == user.id or kb.tenant_id == user.tenant_id return False def can_access_document(user, doc) -> bool: if user.role in ["platform_admin", "platform_user"]: return True if user.role == "customer_admin": return doc.tenant_id == user.tenant_id if user.role == "customer_user": return doc.tenant_id == user.tenant_id return False def get_accessible_knowledge_bases(db, user) -> list[int]: from sqlalchemy import or_ from app.models.knowledge_base import KnowledgeBase if user.role in ["platform_admin", "platform_user"]: return [kb.id for kb in db.query(KnowledgeBase.id).all()] if user.role == "customer_admin": return [ kb.id for kb in db.query(KnowledgeBase.id) .filter(KnowledgeBase.tenant_id == user.tenant_id) .all() ] if user.role == "customer_user": return [ kb.id for kb in db.query(KnowledgeBase.id) .filter( or_(KnowledgeBase.created_by == user.id, KnowledgeBase.tenant_id == user.tenant_id) ) .all() ] return []