# (c) cavaliba.com - data - views_iam.py # IAM-specific actions accessible from generic instance_detail import json import pprint import jwt from django.contrib import messages from django.contrib.auth import logout as django_logout from django.shortcuts import redirect, render from django.utils.translation import gettext as _ import app_home.cache as cache from app_data.aaa import get_aaa, start_view from app_data.forms_iam import UserPrefForm from app_data.user import User from app_home.log import ERROR, WARNING, log # ------------------------------------------------------- def preferences(request): 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.from_keyname(keyname=aaa["username"], expand=True) if not user: messages.add_message(request, messages.ERROR, _("User not found")) return redirect("app_home:private") if request.method == "POST": form = UserPrefForm(request.POST) if form.is_valid(): user.displayname = form.cleaned_data["displayname"] user.set_field_value_single("firstname", form.cleaned_data["firstname"]) user.set_field_value_single("lastname", form.cleaned_data["lastname"]) user.set_field_value_single( "want_notifications", form.cleaned_data["want_notifications"] ) user.set_field_value_single("want_24", form.cleaned_data["want_24"]) user.set_field_value_single("want_email", form.cleaned_data["want_email"]) user.set_field_value_single("want_sms", form.cleaned_data["want_sms"]) user.set_field_value_single("secondary_email", form.cleaned_data["secondary_email"]) user.set_field_value_single("secondary_mobile", form.cleaned_data["secondary_mobile"]) try: user.update() messages.add_message(request, messages.SUCCESS, _("Preferences updated")) return redirect("app_home:private") except Exception: messages.add_message(request, messages.ERROR, _("Failed to edit preferences")) return redirect("app_home:private") else: messages.add_message(request, messages.ERROR, _("Invalid preferences")) else: initial = { "is_enabled": user.is_enabled, "email": user.get_attribute_first("email"), "mobile": user.get_attribute_first("mobile"), "firstname": user.get_attribute_first("firstname"), "lastname": user.get_attribute_first("lastname"), "displayname": user.displayname, "want_notifications": user.want_notifications, "want_24": user.want_24, "want_email": user.is_field_true("want_email"), "want_sms": user.is_field_true("want_sms"), "secondary_email": user.get_attribute_first("secondary_email"), "secondary_mobile": user.get_attribute_first("secondary_mobile"), } form = UserPrefForm(initial=initial) context["form"] = form context["user"] = user return render(request, "app_data/iam/user_pref_edit.html", context) # ------------------------------------------------------- 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"] str_headers = "" for header, value in request.headers.items(): str_headers += f"{header}: {value}\n" encoded = request.headers.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) try: 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_data/iam/debug.html", context) # ------------------------------------------------------- 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() if aaa["auth_mode"] == "local": django_logout(request) return redirect("app_home:index") # ------------------------------------------------------- def impersonate(request, id=None): context = start_view( request, app="iam", view="impersonate", noauth="app_sirene:index", perm="p_iam_impersonate", noauthz="app_data:private", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] impersonate_by = aaa["username"] if not id: messages.add_message(request, messages.ERROR, "No user provided") return redirect("app_data:private") user = User.from_id(id) if not user: log( ERROR, aaa=aaa, app="data", view="impersonate", action="post", status="KO", data=f"User not found id={id}", ) return redirect("app_data:private") newlogin = user.login cache.init() request.session.flush() new_aaa = get_aaa( request, auth_mode="impersonate", impersonate_by=impersonate_by, impersonate=newlogin ) if new_aaa["is_admin"]: messages.add_message(request, messages.ERROR, "Can't impersonate ADMIN") cache.init() request.session.flush() 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: request.session.flush() messages.add_message(request, messages.ERROR, "Invalid new login") return redirect("app_data:private") # ------------------------------------------------------- def email_test(request, id=None): context = start_view( request, app="data", view="email_test", noauth="app_sirene:index", perm="p_user_email_test", noauthz="app_data:private", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if not id: return redirect("app_data:private") user = User.from_id(id) if not user: log( ERROR, aaa=aaa, app="data", view="email_test", action="post", status="KO", data=f"User not found id={id}", ) return redirect("app_data:private") if request.method == "POST": user.send_test_email(aaa=aaa) messages.add_message(request, messages.SUCCESS, _("Test Email sent")) return redirect("app_data:instance_detail", id) # ------------------------------------------------------- def sms_test(request, id=None): context = start_view( request, app="data", view="sms_test", noauth="app_sirene:index", perm="p_user_sms_test", noauthz="app_data:private", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if not id: return redirect("app_data:private") user = User.from_id(id) if not user: log( ERROR, aaa=aaa, app="data", view="sms_test", action="post", status="KO", data=f"User not found id={id}", ) return redirect("app_data:private") if request.method == "POST": sent = user.send_test_sms(aaa=aaa) if sent: messages.add_message(request, messages.SUCCESS, _("Test SMS sent")) else: messages.add_message( request, messages.ERROR, _("SMS test failed (quota or invalid number)") ) return redirect("app_data:instance_detail", id)