一账通是一款开源的统一身份认证授权管理解决方案，企业级IDaaS/IAM平台系统

from arkid.core.b64_compress import Compress from arkid.core.openapi import get_permissions from arkid.core.models import ( UserPermissionResult, SystemPermission, User, Tenant, App, Permission, UserGroup, ExpiringToken, GroupPermissionResult, AppPermissionResult, OpenPermission, ) from arkid.core.api import api from django.db.models import Q import collections import requests import logging import uuid import jwt import re from oauth2_provider.models import Application class PermissionData(object): ''' 权限数据的统一处理类 ''' def __init__(self): pass def update_system_permission(self): ''' 更新系统权限 ''' # 取得所有的系统权限 return_change_group_info, return_change_sort_ids = self.update_arkid_system_permission() # 更新所有用户的系统权限 tenants = Tenant.valid_objects.all() for tenant in tenants: self.update_arkid_all_user_permission(str(tenant.id), change_sort_ids=return_change_sort_ids) def update_app_permission(self, tenant_id, app_id): ''' 更新应用权限 ''' tenant = Tenant.valid_objects.filter(id=tenant_id).first() app = App.valid_objects.filter(id=app_id).first() if tenant and app: app_config = app.config.config openapi_uris = app_config.get('openapi_uris', None) if openapi_uris: # 取得应用所有的权限 return_change_group_info, return_change_sort_ids = self.update_app_all_permission(tenant, app) # 更新应用所有用户权限 uprs = UserPermissionResult.valid_objects.filter( app=app ).exclude(tenant=tenant) tenants = [] tenants.append(tenant) for upr in uprs: tenant = upr.tenant if tenant not in tenants: tenants.append(tenant) for tenant in tenants: # 需要更新对应租户的所有用户 self.update_app_all_user_permission(tenant, app, change_sort_ids=return_change_sort_ids) def update_only_user_app_permission(self, tenant_id, app_id): ''' 仅仅更新用户的应用权限 ''' tenant = Tenant.valid_objects.filter(id=tenant_id).first() app = App.valid_objects.filter(id=app_id).first() if tenant and app: # 更新应用所有用户权限 uprs = UserPermissionResult.valid_objects.filter( app=app ).exclude(tenant=tenant) tenants = [] tenants.append(tenant) for upr in uprs: tenant = upr.tenant if tenant not in tenants: tenants.append(tenant) for tenant in tenants: # 需要更新对应租户的所有用户 self.update_app_all_user_permission(tenant, app) def update_tenant_permission(self, tenant_id): ''' 更新租户权限 ''' tenant = Tenant.valid_objects.filter(id=tenant_id).first() if tenant and app: # 更新租户的所有用户权限 self.update_tenant_all_user_permission(tenant) def get_platfrom_tenant(self): ''' 获取平台租户 ''' # tenant, _ = Tenant.objects.get_or_create( # slug='', # name="平台租户", # ) tenant = Tenant.objects.filter( slug='' ).first() if tenant is None: tenant = Tenant() tenant.slug = '' tenant.name = '平台租户' tenant.save() return tenant def api_system_permission_check(self, tenant, user, operation_id): ''' 检查api接口权限 ''' systempermission = SystemPermission.valid_objects.filter(tenant=None, is_system=True, operation_id=operation_id, category='api').first() if systempermission: sort_id = systempermission.sort_id permission_result_arr = self.get_permission_result(tenant, user, None) if permission_result_arr and len(permission_result_arr) > sort_id and int(permission_result_arr[sort_id]) == 0: return False return True def api_system_permission_check_app(self, tenant, app, operation_id): ''' 根据应用检查api接口权限 ''' systempermission = SystemPermission.valid_objects.filter(tenant=None, is_system=True, operation_id=operation_id, category='api').first() if systempermission: sort_id = systempermission.sort_id permission_result_arr = self.get_app_system_permission_result(tenant, app, None) if permission_result_arr is None: return False if len(permission_result_arr) <= sort_id: return False if permission_result_arr and len(permission_result_arr) > sort_id and int(permission_result_arr[sort_id]) == 0: return False return True def get_app_system_permission_result(self, tenant, app, select_app): ''' 取得应用解码后的权限数组 ''' apppermissionresult = AppPermissionResult.valid_objects.filter( tenant=tenant, self_app=app, app=select_app, ).first() compress = Compress() permission_result_arr = [] if apppermissionresult: permission_result = compress.decrypt(apppermissionresult.result) permission_result_arr = list(permission_result) return permission_result_arr def get_permission_result(self, tenant, user, app): ''' 取得用户解码后的权限数组 ''' userpermissionresult = UserPermissionResult.valid_objects.filter( user=user, tenant=tenant, app=app, ).first() compress = Compress() permission_result_arr = [] if userpermissionresult: permission_result = compress.decrypt(userpermissionresult.result) permission_result_arr = list(permission_result) return permission_result_arr def add_system_permission_to_user(self, tenant_id, user_id, permission_id): ''' 给某个用户增加系统权限 ''' tenant = Tenant.valid_objects.filter(id=tenant_id).first() user = User.valid_objects.filter(id=user_id).first() permission = SystemPermission.valid_objects.filter(id=permission_id).first() if tenant: self.update_arkid_single_user_permission(tenant, user, permission, 1) else: print('不存在租户或者用户无法更新') def add_user_many_permission(self, permissions_dict): ''' 给用户添加多个权限自动区分类型 ''' user_ids = permissions_dict.get('user_ids', []) data_arr = permissions_dict.get('data_arr', []) tenant_id = permissions_dict.get('tenant_id', None) for user_id in user_ids: if user_id and data_arr and tenant_id: for permission_id in data_arr: permission = SystemPermission.valid_objects.filter(id=permission_id).first() if permission is None: permission = Permission.valid_objects.filter(id=permission_id).first() if isinstance(permission, SystemPermission): # 添加系统权限 self.add_system_permission_to_user(tenant_id, user_id, permission_id) else: # 添加应用权限 self.add_app_permission_to_user(tenant_id, str(permission.app_id), user_id, permission_id