# app_user - views.py # (c) cavaliba.com import json import pprint import re import app_home.cache as cache import jwt from app_data.permissions import has_iam_export_permission from app_data.revision import revision_add_raw, revision_get from app_home.configuration import get_configuration from app_home.log import DEBUG, ERROR, INFO, WARNING, log # Celery tasks from app_sirene.mail import task_send_mail from app_sirene.sms import get_sms_quota, sms_check_valid_number, task_send_sms from django.contrib import messages from django.contrib.auth import logout as django_logout from django.shortcuts import redirect, render from django.utils import timezone from django.utils.translation import gettext as _ from .aaa import get_aaa, get_groups_for_user, start_view from .forms import UserForm, UserPrefForm, UserUploadForm from .models import SireneUser # from .user import import_json # from .user import import_yaml from .user import ( user_create, # or update user_csv_response, user_delete_by_id, user_get_by_id, user_get_by_login, user_get_email, user_get_form, user_get_form_blank, user_get_mobile, user_json_response, user_search, user_update, user_yaml_response, ) # from app_data.permissions import has_user_read_permission # from app_data.permissions import has_user_update_permission # from app_data.permissions import has_user_create_permission # from app_data.permissions import has_user_delete_permission # ----------------------------------------- # private #----------------------------------------- def private(request): context = start_view(request, app="iam", view="private", noauth="app_home:private", perm="p_iam_access", noauthz="app_home:index") if context["redirect"]: return redirect(context["redirect"]) return redirect("app_user:user_list") # ---------------------------------------------------------- # USER list # ---------------------------------------------------------- def list(request): ''' display user list - ?o=csv to export as CSV (yaml, json) ''' context = start_view(request, app="iam", view="list", noauth="app_home:private", perm="p_user_read", noauthz="app_home:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] # p_user_read is enough in start_view : no per-user permission. # bigset or not bigset_size = int(get_configuration("data","DATA_BIGSET_SIZE")) count = SireneUser.objects.count() bigset = True context["bigset"] = bigset context["count"] = count # Filter POST (form) #query = request.GET.get("q", "") query = "" if request.method == "POST": if request.POST.get('query'): query = request.POST.get('query') m = re.compile(r'[a-zA-Z0-9()_/.-]*$') if not m.match(query): #return redirect("app_user:user_list") query = "" context["query"] = query # # partial query + paginate if bigset # if bigset: try: size = int(request.GET.get("size", bigset_size)) page = int(request.GET.get("page",1)) except Exception as e: print(e) return redirect("app_user:user_list") if page < 1 or size > 10000 or size < 1: return redirect("app_user:user_list") users = user_search(query=query, page=page, size=size) # export requested (o=xxxx) output = request.GET.get("o","") if output != "": # permission if not has_iam_export_permission(aaa=aaa): log(WARNING, aaa=aaa, app="iam", view="list", action="export", status="DENY", data=_("Not allowed")) messages.add_message(request, messages.ERROR, _("Not allowed")) return redirect("app_sirene:private") # NEXT : partial export (selection only) # size max for interactive export ? export_max = int(get_configuration("data","EXPORT_INTERACTIVE_MAX_SIZE")) if count > export_max: log(ERROR, aaa=aaa, app="iam", view="list", action="export", status="KO", data="Export too big") messages.add_message(request, messages.ERROR, _("Export too large for interactive export.")) return redirect("app_user:user_list") if output == "csv": log(DEBUG, aaa=aaa, app="iam", view="user_list", action="export_csv", status="OK", data="") response = user_csv_response(users) return response elif output == "json": log(DEBUG, aaa=aaa, app="iam", view="user_list", action="export_json", status="OK", data="") response = user_json_response(users) return response elif output == "yaml": log(DEBUG, aaa=aaa, app="iam", view="user_list", action="export_yaml", status="OK", data="") response = user_yaml_response(users) return response # PREV | FIRST || CURRENT or ... || LAST | NEXT context["size"] = size context["page"] = page page_last = int (count / size) + 1 # if count * size <= page_last: # page_last = page_last + 1 if page > 1: context["page_prev"] = page - 1 else: context["page_prev"] = page if page == 1: context["page_first"] = True else: context["page_first"] = False if page > 1 and page < page_last: context["page_current"] = True else: context["page_current"] = False context["page_last"] = page_last if page < page_last: context["page_last_active"] = False else: context["page_last_active"] = True if page < page_last: context["page_next"] = page + 1 else: context["page_next"] = page # else: # users = SireneUser.objects.all().order_by('login') # # users = SireneUser.objects.all().exclude(login='admin').prefetch_related('roles').order_by('login') upload_form = UserUploadForm() log(DEBUG, aaa=aaa, app="iam", view="list", action="get", status="OK", data=f"{count} users") context["users"] = users context["upload_form"] = upload_form return render(request, 'app_user/user_list.html', context) # ---------------------------------------------------------- # User delete # ---------------------------------------------------------- def delete(request, userid): ''' delete user ''' context = start_view(request, app="iam", view="user_delete", noauth="app_home:index", perm="p_user_delete", noauthz="app_sirene:index") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if request.method == "POST": user, err = user_delete_by_id(userid=userid) if err: messages.add_message(request, messages.ERROR, _("Failed to delete user")) log(ERROR, aaa=aaa, app="iam", view="user_delete", action="delete", status="FAIL", data=err) else: revision_add_raw(aaa=aaa, classname="_user", keyname=user.login, displayname=user.displayname, action="delete") messages.add_message(request, messages.SUCCESS, _("User deleted")) log(INFO, aaa=aaa, app="iam", view="user_delete", action="delete", status="OK", data=f"{user.login}") return redirect("app_user:user_list") # ---------------------------------------------------------- # User detail # ---------------------------------------------------------- def detail(request, userid=None): ''' display role detail ''' context = start_view(request, app="iam", view="user_detail", noauth="app_user:private", perm="p_user_read", noauthz="app_user:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] user = user_get_by_id(userid) if not user: messages.add_message(request, messages.ERROR, _("Not allowed")) log(WARNING, aaa=aaa, app="iam", view="user", action="detail", status="KO", data=_("Not allowed") ) return redirect("app_user:user_list") log(DEBUG, aaa=aaa, app="iam", view="user", action="detail", status="OK", data=user.login ) # get group objects direct, computed, indirect = get_groups_for_user(user, mode="obj") context["direct"] = direct context["computed"] = computed context["indirect"] = indirect permissions = [] for g in direct: for p in g.permissions.all(): if p not in permissions: permissions.append(p) for g in indirect: for p in g.permissions.all(): if p not in permissions: permissions.append(p) context["permissions"] = permissions # revisions - V3.23 context['revisions'] = revision_get(classname="_user", keyname=user.login) # can use YAML Editor ? if 'p_data_import' in aaa['perms']: context['ui_texteditor'] = True context["user"] = user return render(request, 'app_user/user_detail.html', context) # ---------------------------------------------------------- # User edit # ---------------------------------------------------------- def edit(request, userid=None): ''' user edit form''' context = start_view(request, app="iam", view="user.edit", noauth="app_home:index", perm="p_user_update", noauthz="app_user:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] user = None # if user is provided, it's an update if userid: user = user_get_by_id(userid) if not user: messages.add_message(request, messages.ERROR, _("Not allowed")) log(WARNING, aaa=aaa, app="iam", view="user", action="edit", status="KO", data=_("Not allowed")) return redirect("app_user:user_list") # else, it's a create : check p_user_create else: if "p_user_create" not in aaa["perms"]: messages.add_message(request, messages.ERROR, _("Not allowed")) log(WARNING, aaa=aaa, app="iam", view="user", action="create", status="KO", data=_("Not allowed")) return redirect("app_user:user_list") if request.method == "POST": form = UserForm(request.POST) if form.is_valid(): action = "create" # update if user: user, err = user_update(user, form.cleaned_data) action="edit" # create else: user, err = user_create(data=form.cleaned_data) # success ? if user: revision_add_raw(aaa=aaa, classname="_user", keyname=user.login, displayname=user.displayname, action=action) messages.add_message(request, messages.SUCCESS, _("User saved : ") + user.login ) log(INFO, aaa=aaa, app="iam", view="edit", action="save", status="OK", data=f"{user.login}") else: messages.add_message(request, messages.ERROR, _("Failed to save user")) log(WARNING, aaa=aaa, app="iam", view="edit", action="save", status="KO", data="") return redirect("app_user:user_detail", user.id) else: messages.add_message(request, messages.ERROR, _("Invalid User form")) log(DEBUG, aaa=aaa, app="iam", view="edit", action="save", status="KO", data="Invalid User form") else: if user: form = user_get_form(user) log(DEBUG, aaa=aaa, app="iam", view="edit", action="edit", status="OK", data=f"{user.login}") else: form = user_get_form_blank() log(DEBUG, aaa=aaa, app="iam", view="edit", action="new", status="OK", data="") context["form"] = form if user: context["user"] = user return render(request, 'app_user/user_edit.html', context) # # ---------------------------------------------------------- # # User Import # # ---------------------------------------------------------- # def user_import(request): # context = start_view(request, app="user", view="user_import", noauth="app_home:index", # perm="p_iam_access", noauthz="app_user:private") # if context["redirect"]: # return redirect(context["redirect"]) # aaa = get_aaa(request) # if request.method == "POST": # form = UserUploadForm(request.POST, request.FILES) # if form.is_valid(): # file = request.FILES["file"] # if file.multiple_chunks(): # messages.add_message(request, messages.ERROR, _("Import failed - file too big")) # log(ERROR, aaa=aaa, app="user", view="user_import", action="POST", status="FAIL", data="File too big") # return redirect("app_user:user_list") # elif file.name.endswith(".json"): # line_total, line_ok = import_json(file) # elif file.name.endswith(".yaml"): # line_total, line_ok = import_yaml(file) # elif file.name.endswith(".yml"): # line_total, line_ok = import_yaml(file) # else: # messages.add_message(request, messages.ERROR, _("Import failed - unknown file")) # log(ERROR, aaa=aaa, app="user", view="user_import", action="POST", status="FAIL", data=f"Unknown file type {file}") # return redirect("app_user:user_list") # # OK # messages.add_message(request, messages.SUCCESS, _("Imported : ") + f"{line_ok}/{line_total}" ) # log(INFO, aaa=aaa, app="user", view="user_import", action="POST", status="OK", data=f"{file} - {line_ok}/{line_total} entries") # return redirect("app_user:user_list") # # Failed # messages.add_message(request, messages.ERROR, _("Failed to import")) # log(ERROR, aaa=aaa, app="user", view="user_import", action="POST", status="FAIL", data="import failed, not a POST") # return redirect("app_user:user_list") # ---------------------------------------------------------- # User Preferences # ---------------------------------------------------------- def preferences(request): ''' user edit form''' context = start_view(request, app="iam", view="user_preferences", noauth="app_sirene:index", perm="p_user_pref", noauthz="app_home:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] user = user_get_by_login(aaa["username"]) if not user: return redirect("app_home:private") if request.method == "POST": form = UserPrefForm(request.POST) if form.is_valid(): user.displayname = form.cleaned_data["displayname"] user.firstname = form.cleaned_data["firstname"] user.lastname = form.cleaned_data["firstname"] user.want_notifications = form.cleaned_data["want_notifications"] user.want_24 = form.cleaned_data["want_24"] user.want_email = form.cleaned_data["want_email"] user.want_sms = form.cleaned_data["want_sms"] user.secondary_email = form.cleaned_data["secondary_email"] user.secondary_mobile = form.cleaned_data["secondary_mobile"] user.last_update = timezone.now() try: user.save() except Exception as e: messages.add_message(request, messages.ERROR, _("Failed to edit preferences") ) print(e) return redirect("app_home:private") messages.add_message(request, messages.SUCCESS, _("Preferences updated")) return redirect("app_home:private") else: messages.add_message(request, messages.ERROR, _("Invalid preferences")) else: initial = {} # won't be edited initial["is_enabled"] = user.is_enabled initial["email"] = user.email initial["mobile"] = user.mobile initial["firstname"] = user.firstname initial["lastname"] = user.lastname initial["displayname"] = user.displayname initial["want_notifications"] = user.want_notifications initial["want_24"] = user.want_24 initial["want_email"] = user.want_email initial["want_sms"] = user.want_sms initial["secondary_email"] = user.secondary_email initial["secondary_mobile"] = user.secondary_mobile form = UserPrefForm(initial=initial) context["form"] = form context["user"] = user return render(request, 'app_user/user_pref_edit.html', context) # ----------------------------------------- # logout #----------------------------------------- def logout(request): context = start_view(request, app="iam", view="logout", noauth="app_home:index") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] request.session.flush() # Django Auth mode if aaa["auth_mode"] == "local": django_logout(request) # basic / browser mode elif aaa["auth_mode"] == "basic": # HACK redirect with false credential #redirect("http://logout:logout@cavaliba.com") # user must erase his browser history pass # oauth2 else: # already done by flush pass return redirect("app_home:index") # ----------------------------------------- # debug user env #----------------------------------------- def debug_env(request): context = start_view(request, app="iam", view="debug_env", noauth="app_sirene:index", perm="p_user_debug", noauthz="app_sirene:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] #print("DEBUG request.LANGUAGE_CODE=", request.LANGUAGE_CODE) #print("DEBUG get_language()=", get_language()) str_headers = '' for header, value in request.headers.items(): str_headers += f"{header}: {value}\n" # decode JWT Access Token if any # ------------------------------ encoded = request.headers.get('X-Access-Token','') # encoded = request.META.get('X-Access-Token') try: jwt_header = jwt.get_unverified_header(encoded) jwt_payload = jwt.decode(encoded, options={"verify_signature": False}) jwt_header = json.dumps(jwt_header, indent=4) jwt_payload = json.dumps(jwt_payload, indent=4) except Exception as e: log(WARNING, aaa=aaa, app="iam", view="user", action="debug", status="KO", data=e) jwt_header = "n/a" jwt_payload = "n/a" aaa_nice = pprint.pformat(aaa) # OIDC Authorization Bearer Token # ------------------------------- # request.META.get('HTTP_AUTHORIZATION') # request.headers.get('Authorization') try: # Authorization: Bearer {access_token_here} autz_header = request.headers.get('Authorization','') encoded = autz_header.split()[1] tokid_header = jwt.get_unverified_header(encoded) tokid_payload = jwt.decode(encoded, options={"verify_signature": False}) tokid_header = json.dumps(tokid_header, indent=4) tokid_payload = json.dumps(tokid_payload, indent=4) except Exception as e: log(WARNING, aaa=aaa, app="iam", view="user", action="debug", status="KO", data=e) tokid_header = "n/a" tokid_payload = "n/a" context["title"] = "Debug ENV" context["env"] = str_headers context["jwt_header"] = jwt_header context["jwt_payload"] = jwt_payload context["tokid_header"] = tokid_header context["tokid_payload"] = tokid_payload context["aaa_nice"] = aaa_nice return render(request, 'app_user/debug.html', context) #----------------------------------------- # Send test email #----------------------------------------- def email_test(request, userid=None): context = start_view(request, app="iam", view="email_test", noauth="app_sirene:index", perm="p_user_email_test", noauthz="app_sirene:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] subject = get_configuration("sirene","EMAIL_TEST_SUBJECT") text_content = get_configuration("sirene", "EMAIL_TEST_CONTENT") html_content = None dest = "" if userid: user = user_get_by_id(userid) if not user: return redirect("app_user:private") if request.method == "POST": dest = user_get_email(user) task_send_mail.delay(subject, text_content, [dest], html_content=html_content, aaa=aaa) messages.add_message(request, messages.SUCCESS, _("Test Email sent")) return redirect("app_user:private") #----------------------------------------- # SMS Test form User Edit Form #----------------------------------------- def sms_test(request, userid=None): context = start_view(request, app="iam", view="sms_test", noauth="app_sirene:index", perm="p_user_sms_test", noauthz="app_user:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if get_sms_quota(aaa) < 1: messages.add_message(request, messages.ERROR, _("Insufficient SMS Quota")) return redirect("app_user:private") if request.method == "POST": dest = "" if userid: user = user_get_by_id(userid) if not user: return redirect("app_user:private") dest = user_get_mobile(user) if not sms_check_valid_number(dest): messages.add_message(request, messages.ERROR, _("Invalid SMS number")) return redirect("app_user:private") data = get_configuration("sirene", "SMS_TEST") task_send_sms.delay([dest], data, aaa=aaa) messages.add_message(request, messages.SUCCESS, _("Test SMS sent")) return redirect("app_user:private") #----------------------------------------- # Impersonate #----------------------------------------- def impersonate(request, newlogin=None): context = start_view(request, app="iam", view="impersonate", noauth="app_sirene:index", perm="p_iam_impersonate", noauthz="app_user:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] try: impersonate_by = aaa["username"] except Exception: return redirect("app_sirene:private") if newlogin: # logout current user cache.init() request.session.flush() # try relog with newlogin new_aaa = get_aaa(request, auth_mode="impersonate", impersonate_by=impersonate_by, impersonate=newlogin) # check newlogin is not admin if new_aaa["is_admin"]: messages.add_message(request, messages.ERROR, "Can't impersonate ADMIN") cache.init() request.session.flush() # success elif new_aaa["is_authenticated"]: request.session["aaa"] = new_aaa messages.add_message(request, messages.SUCCESS, "Impersonated for " + newlogin) return redirect("app_sirene:private") else: messages.add_message(request, messages.ERROR, "Invalid new login") else: messages.add_message(request, messages.ERROR, "No login provided") #return redirect("app_user:impersonate") return render(request, 'app_user/impersonate.html', context) return redirect("app_sirene:private")