# (c) - cavaliba.com - IAM - role.py import json import re import yaml from app_data.permissions import has_role_create_permission, has_role_delete_permission, has_role_update_permission from app_home.log import WARNING, log from app_user.group import group_get_by_name from app_user.permission import permission_all_keynames, permission_get_by_name from app_user.user import user_get_by_login, user_get_login_by_email from django.forms.models import model_to_dict from django.http import HttpResponse from django.utils import timezone from django.utils.translation import gettext as _ from .forms import RoleForm from .models import SireneGroup, SirenePermission, SireneUser def role_search(query=None, page=1, size=10): from django.db.models import Q m = re.compile(r'[a-zA-Z0-9()_/.-]*$') if query and not m.match(query): return [] offset = (page - 1) * size limit = offset + size qs = SireneGroup.objects\ .filter(is_role=True)\ .prefetch_related("permissions")\ .prefetch_related("users")\ .prefetch_related("subgroups")\ .order_by('keyname') if query: qs = qs.filter( Q(keyname__icontains=query) | Q(displayname__icontains=query) | Q(description__icontains=query) ) return qs[offset:limit] def role_all_filtered(is_enabled=None, filter=None): # prefetch_related(''). if is_enabled: return SireneGroup.objects.filter(is_enabled=True, is_role=True).all() else: return SireneGroup.objects.filter(is_role=True).all() def role_get_by_id(id, enabled_only=False): if enabled_only: return SireneGroup.objects.filter(pk=id, is_role=True, is_enabled=True)\ .prefetch_related('users')\ .prefetch_related('subgroups')\ .prefetch_related('permissions')\ .first() else: return SireneGroup.objects.filter(pk=id, is_role=True)\ .prefetch_related('users')\ .prefetch_related('subgroups')\ .prefetch_related('permissions')\ .first() def role_get_by_name(name, enabled_only=False): if enabled_only: return SireneGroup.objects.filter(keyname=name, is_role=True, is_enabled=True)\ .prefetch_related('users')\ .prefetch_related('subgroups')\ .prefetch_related('permissions')\ .first() else: return SireneGroup.objects.filter(keyname=name, is_role=True)\ .prefetch_related('users')\ .prefetch_related('subgroups')\ .prefetch_related('permissions')\ .first() def role_get_by_data(data, enabled_only=False): keyname = data.get('keyname', None) if not keyname: return return role_get_by_name(keyname, enabled_only=enabled_only) # OBJECTS def role_get_subgroups(role, done=[]): ''' Recurse group Object to get all subgroups Objects''' reply = [] for g in role.subgroups.all(): if g in done: continue done.append(g) reply.append(g) reply += role_get_subgroups(g, done) reply = list(set(reply)) return reply def role_get_form(item=None): # blank if not item: return RoleForm(initial={}) attributs = [ "keyname", "displayname", "description", "is_enabled", "is_builtin", ] initial = {} for attrib in attributs: try: initial[attrib] = getattr(item,attrib, "") except Exception: pass initial["users"] = [i for i in item.users.all()] initial["subgroups"] = [i for i in item.subgroups.all()] initial["permissions"] = [i for i in item.permissions.all()] return RoleForm(initial=initial) def role_get_permissions(roles): # IN roles =["role1", "role2", ...] perms = [] # if builtin role_admin: if "role_admin" in roles: return permission_all_keynames() for rolename in roles: gobj = role_get_by_name(rolename, enabled_only=True) if gobj: for perm in gobj.permissions.all(): perms.append(perm.keyname) perms = list(set(perms)) return perms # ------------------------------------ # expand group(s) objects to users # Recursive # ------------------------------------ # OBJECTS def role_expand_to_users(roles_processed, new_roles): reply = [] if not new_roles: return [] if len(new_roles) == 0: return [] for role in new_roles: if role in roles_processed: # already done continue roles_processed.append(role) # add all users from group g reply += role.users.all() # for user in role.users.all(): # reply.append(user) # recurse to subgroups for sg in role.subgroups.all(): subreply = role_expand_to_users(roles_processed, [sg]) reply += subreply # remove duplicate users reply = list(set(reply)) return reply def role_init(data): keyname = data.get('keyname', None) if not keyname: return None, "role_init: missing keyname" entry = role_get_by_name(keyname) if entry: return entry, None entry = SireneGroup(is_role=True) return role_update(entry, data) def role_create(data): keyname = data.get('keyname', None) if not keyname: return None, "role_create: missing keyname" # check if exists as group (not role) entry0 = SireneGroup.objects.filter(keyname=keyname, is_role=False).first() if entry0: return None, f"role_create: {keyname} already a group" # create or update entry1 = role_get_by_name(keyname) if not entry1: entry1 = SireneGroup(is_role=True) return role_update(entry1, data) def role_update(role, data): ''' update Object with data dict info''' attributs = [ "keyname", "displayname", "is_enabled","description","is_builtin", ] # special_attributs = ["users", "subgroups", "permissions"] if not data: return "role_update: no data" for attrib in attributs: if attrib in data: try: setattr(role, attrib, data[attrib]) except Exception: pass role.last_update = timezone.now() role.save() # users if 'users' in data: # can be a Queryset (form) or list of login (import) role.users.clear() if data["users"]: for user in data["users"]: if type(user) is SireneUser: role.users.add(user) else: if '@' in user: login2 = user_get_login_by_email(user) user2 = user_get_by_login(login2) else: user2 = user_get_by_login(user) if user2: role.users.add(user2) role.last_update = timezone.now() role.save() # subgroups if 'subgroups' in data: # can be a Queryset (form) or list of keynames (import) role.subgroups.clear() if data["subgroups"]: for subgroup in data["subgroups"]: if type(subgroup) is SireneGroup: role.subgroups.add(subgroup) else: # TODO - also get Role ... subgroup2 = group_get_by_name(subgroup) if subgroup2: role.subgroups.add(subgroup2) role.last_update = timezone.now() role.save() # permissions if 'permissions' in data: # can be a Queryset (form) or list of keynames (import) role.permissions.clear() if data["permissions"]: for perm in data["permissions"]: if type(perm) is SirenePermission: role.permissions.add(perm) else: perm2 = permission_get_by_name(perm) if perm2: role.permissions.add(perm2) role.last_update = timezone.now() role.save() return role, None def role_update_by_data(data): role = role_get_by_data(data) if role: return role_update(role, data) else: return None, "role_update: role not found" # 0 def role_delete(robj): if not robj: return None, "role_delete: error" if robj.is_builtin: return None, "role_delete: is builtin" if not robj.is_role: return None, "role_delete: is group" try: robj.delete() return robj, None except Exception as e: return None, f"role_delete: {e}" # 0 def role_delete_by_id(id): robj = role_get_by_id(id) return role_delete(robj) # X def role_delete_by_data(data): role = role_get_by_data(data) if role: return role_delete_by_id(role.id) return None, "role_delete: not found" # 0 def role_enable_by_data(data): role = role_get_by_data(data) if role: role.is_enabled = True role.save() return role, None return None, "role_enable: not found" # 0 def role_disable_by_data(data): role = role_get_by_data(data) if role: role.is_enabled = False role.save() return role, None return None, "role_disable: not found" # ------------------------------------------------------- # LOADER / IMPORTS # ------------------------------------------------------- def load_role(datadict=None, aaa=None): if not datadict: return "load_role: missing data" if not aaa: log(WARNING, aaa=aaa, app="iam", view="role", action="load", status="DENY", data=_("Not allowed")) return "load_role: not allowed" keyname = datadict.get("keyname", None) if not keyname: return "load_role: missing keyname" action = datadict.get("_action", "create") err = None if action == "init": if not has_role_create_permission(aaa=aaa): log(WARNING, aaa=aaa, app="iam", view="role", action="init", status="DENY", data=_("Not allowed")) return "load_role: init not allowed" # if missing, create ; if already exists, don't update role, err = role_init(datadict) elif action == "create": if not has_role_create_permission(aaa=aaa): log(WARNING, aaa=aaa, app="iam", view="role", action="create", status="DENY", data=_("Not allowed")) return f"load_role: create not allowed for {keyname}" # create and/or update with provided data role, err = role_create(datadict) elif action == "update": if not has_role_update_permission(aaa=aaa): log(WARNING, aaa=aaa, app="iam", view="role", action="update", status="DENY", data=_("Not allowed")) return f"load_role: update not allowed for {keyname}" # update only if exists ; don't create role, err = role_update_by_data(datadict) elif action == "delete": if not has_role_delete_permission(aaa=aaa): log(WARNING, aaa=aaa, app="iam", view="role", action="delete", status="DENY", data=_("Not allowed")) return f"load_role: delete not allowed for {keyname}" role, err = role_delete_by_data(datadict) elif action == "enable": if not has_role_update_permission(aaa=aaa): log(WARNING, aaa=aaa, app="iam", view="role", action="enable", status="DENY", data=_("Not allowed")) return f"load_role: enable not allowed for {keyname}" role, err = role_enable_by_data(datadict) elif action == "disable": if not has_role_update_permission(aaa=aaa): log(WARNING, aaa=aaa, app="iam", view="role", action="disable", status="DENY", data=_("Not allowed")) return f"load_role: disable not allowed for {keyname}" role, err = role_disable_by_data(datadict) elif action == "noop": return else: return f"load_role: failed {keyname} , action {action}" if err: return f"load_role: {err}" return # ------------------------------------------------------- # EXPORTs # ------------------------------------------------------- def role_to_dict(role): dict_attributs = ["keyname", "displayname", "is_enabled","description"] m = {} if role.is_role: m["classname"] = "_role" else: m["classname"] = "_group" m.update( model_to_dict(role, fields=dict_attributs) ) # remove null values m2= {} for k,v in m.items(): if not isinstance(v,bool): if v: m2[k] = v else: m2[k] = v # add many-to-many objects m2["users"] = [i.login for i in role.users.all()] m2["subgroups"]= [i.keyname for i in role.subgroups.all()] if role.is_role: m2["permissions"] = [i.keyname for i in role.permissions.all()] return m2 # group def role_to_yaml(role): data = role_to_dict(role) return yaml.dump([data], allow_unicode=True, Dumper=MyYamlDumper, sort_keys=False) def role_listdict_format(roles): ''' list of Models to list of dict ''' dict_attributs = ["keyname", "displayname", "is_enabled","description" ] datalist = [] for role in roles: m = {} m["classname"] = "_role" m.update(model_to_dict(role, fields=dict_attributs)) # remove null values m2= {} for k,v in m.items(): if not isinstance(v,bool): if v: m2[k] = v else: m2[k] = v # special_attributs = ["subgroups", "users", "permissions"] m2["subgroups"]= [i.keyname for i in role.subgroups.all()] m2["users"] = [i.login for i in role.users.all()] m2["permissions"] = [i.keyname for i in role.permissions.all()] m2["id"] = role.id datalist.append(m2) return datalist # JSON def role_json_response(items): datalist = role_listdict_format(items) filedata = json.dumps(datalist, indent=4, ensure_ascii = False) response = HttpResponse(filedata, content_type='text/json') response['Content-Disposition'] = 'attachment; filename="roles.json"' return response # YAML class MyYamlDumper(yaml.SafeDumper): def write_line_break(self, data=None): super().write_line_break(data) if len(self.indents) < 2: super().write_line_break() def role_yaml_response(items): datalist = role_listdict_format(items) filedata = yaml.dump(datalist, allow_unicode=True, Dumper=MyYamlDumper, sort_keys=False) response = HttpResponse(filedata, content_type='text/yaml') response['Content-Disposition'] = 'attachment; filename="roles.yaml"' return response