# (c) cavaliba.com - home - views.py import re from django.conf import settings from django.contrib import messages from django.db.models import Q from django.http import Http404, HttpResponse from django.shortcuts import redirect, render from django.utils.translation import gettext as _ from app_data.aaa import start_view from app_data.data import update_bigset from app_data.eav import task_eav_purge, task_eav_refresh from app_data.ipam import task_ipam_refresh from app_data.keyname import rename from app_data.models import DataRevision from app_data.schema import Schema from app_data.user import User from .configuration import get_configuration, get_initial_form, get_post_form, save_form from .dashboard import get_dashboard_cached, invalidate_dashboard_cache from .factory import factory_reset, factory_reset_available, factory_reset_password_ok from .log import DEBUG, INFO, WARNING, log, purge from .models import CavalibaLog from .prototypes import PROTOTYPE_REGISTRY, get_prototype from .sidebar import invalidate_sidebar_cache, list_active_users from .tasks import housekeeping_daily # '' or 'tw4' HTML_TEMPLATE_PATH = settings.CAVALIBA_HTML_FRAMEWORK # ----------------------------------------- # status page # ----------------------------------------- def status(request): return HttpResponse("OK") # ----------------------------------------- # private - private_page # ----------------------------------------- def private(request): context = start_view( request, app="home", view="private", perm="p_home_access", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] log(DEBUG, aaa=aaa, app="home", view="dash", action="view", status="OK") context["dashboard"] = get_dashboard_cached(aaa) return render(request, f"app_home/{HTML_TEMPLATE_PATH}/private.html", context) # ----------------------------------------- # configuration # ----------------------------------------- def configuration(request, appname=None): context = start_view( request, app="home", view="configuration", perm="p_conf_admin", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if not appname: appname = next(iter(settings.CONFIGURATION_TOPLEVEL)) return redirect("app_home:configuration", appname) if appname not in settings.CONFIGURATION_TOPLEVEL: return redirect("app_home:index") if request.method == "POST": form = get_post_form(request, appname=appname) if form.is_valid(): changed_keys = save_form(form, appname=appname) if appname == "home" and "SIDEBAR_CONTENT" in changed_keys: invalidate_sidebar_cache(username=aaa.get("username")) if appname == "home" and "DASHBOARD_CONTENT" in changed_keys: invalidate_dashboard_cache(username=aaa.get("username")) messages.add_message(request, messages.SUCCESS, _("Configuration updated")) for keyname in changed_keys: log( INFO, aaa=aaa, app="home", view="configuration", action="update", status="OK", data=f"{appname}.{keyname}", ) return redirect("app_home:configuration", appname) else: messages.add_message(request, messages.ERROR, _("Invalid configuration")) else: form = get_initial_form(appname) context["appname"] = appname context["appdisplayname"] = settings.CONFIGURATION_TOPLEVEL.get(appname, {}).get( "name", appname ) # app selector pills: template iterates this dict directly (keyname -> {name, icon}), # in insertion order - reorder CONFIGURATION_TOPLEVEL in settings.py to reorder the pills. context["apps"] = settings.CONFIGURATION_TOPLEVEL context["form"] = form return render(request, f"app_home/{HTML_TEMPLATE_PATH}/configuration.html", context) # ----------------------------------------- # log # ----------------------------------------- def logview(request, level="info"): context = start_view( request, app="home", view="log", perm="p_log_view", noauthz="app_home:index", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] # purge ? if request.method == "POST": if request.POST.get("purge"): if "p_log_admin" in aaa["perms"]: count = purge(aaa=aaa) messages.add_message(request, messages.SUCCESS, _("Logs purged: ") + str(count)) log( WARNING, aaa=aaa, app="home", view="log", action="purge", status="OK", data=f"{count} removed", ) return redirect("app_home:log") # bigset or not count = CavalibaLog.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") if request.method == "GET": if request.GET.get("query"): query = request.GET.get("query") if query: m = re.compile(r"[a-zA-Z0-9()_/.-]*$") if not m.match(query): query = "" context["query"] = query # partial query + paginate if bigset try: size = int(request.GET.get("size", 100)) page = int(request.GET.get("page", 1)) except Exception: return redirect("app_home:log") if page < 1 or size > 10000 or size < 1: return redirect("app_home:log") offset = (page - 1) * size limit = offset + size if len(query) > 0: if level not in ["debug", "info", "warning", "error", "critical"]: level = "debug" if level == "info": level_display = _("INFO") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter(level__in=["INFO", "WARNING", "ERROR", "CRITICAL"]) .filter( Q(app__icontains=query) | Q(username__icontains=query) | Q(view__icontains=query) | Q(action__icontains=query) | Q(data__icontains=query) | Q(status__icontains=query) | Q(user_ip__icontains=query) )[offset:limit] ) elif level == "warning": level_display = _("WARNING") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter(level__in=["WARNING", "ERROR", "CRITICAL"]) .filter( Q(app__icontains=query) | Q(username__icontains=query) | Q(view__icontains=query) | Q(action__icontains=query) | Q(data__icontains=query) | Q(status__icontains=query) | Q(user_ip__icontains=query) )[offset:limit] ) elif level == "error": level_display = _("ERROR") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter(level__in=["ERROR", "CRITICAL"]) .filter( Q(app__icontains=query) | Q(username__icontains=query) | Q(view__icontains=query) | Q(action__icontains=query) | Q(data__icontains=query) | Q(status__icontains=query) | Q(user_ip__icontains=query) )[offset:limit] ) elif level == "critical": level_display = _("CRITICAL") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter(level="CRITICAL") .filter( Q(app__icontains=query) | Q(username__icontains=query) | Q(view__icontains=query) | Q(action__icontains=query) | Q(data__icontains=query) | Q(status__icontains=query) | Q(user_ip__icontains=query) )[offset:limit] ) else: level_display = _("DEBUG") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter( Q(app__icontains=query) | Q(username__icontains=query) | Q(view__icontains=query) | Q(action__icontains=query) | Q(data__icontains=query) | Q(status__icontains=query) | Q(user_ip__icontains=query) )[offset:limit] ) else: if level not in ["debug", "info", "warning", "error", "critical"]: level = "info" if level == "info": level_display = _("INFO") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter(level__in=["INFO", "WARNING", "ERROR", "CRITICAL"])[offset:limit] ) elif level == "warning": level_display = _("WARNING") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter(level__in=["WARNING", "ERROR", "CRITICAL"])[offset:limit] ) elif level == "error": level_display = _("ERROR") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter(level__in=["ERROR", "CRITICAL"])[offset:limit] ) elif level == "critical": level_display = _("CRITICAL") logs = CavalibaLog.objects.all().order_by("-id").filter(level="CRITICAL")[offset:limit] else: level_display = _("DEBUG") logs = CavalibaLog.objects.all().order_by("-id")[offset:limit] # PREV | FIRST || CURRENT or ... || LAST | NEXT context["size"] = size context["page"] = page page_last = int(count / size) + 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 context["level"] = level_display context["logs"] = logs return render(request, f"app_home/{HTML_TEMPLATE_PATH}/log.html", context) # ---------------------------------------------------------------------------------------------------- # admin tools MAIN # ----------------------------------------------------------------------------------------------------- def admintools(request): """Admin tools landing page: maintenance jobs and the danger zone. The factory reset row is only offered to an admin account (aaa["is_admin"]), and needs the CAVALIBA_ADMIN_PASSWORD env var to be set to be actionable.""" context = start_view( request, app="admintools", view="main", perm="p_admintools", ) if context["redirect"]: return redirect(context["redirect"]) context["can_factory_reset"] = context["aaa"].get("is_admin", False) context["factory_reset_available"] = factory_reset_available() return render(request, f"app_home/{HTML_TEMPLATE_PATH}/admintools.html", context) # ----------------------------------------- # admin tools - update bigset # ----------------------------------------- def admintools_updatebigset(request): context = start_view( request, app="admintools", view="update_bigset", perm="p_admintools", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if request.method == "POST": messages.add_message(request, messages.SUCCESS, _("Task started...")) update_bigset(aaa=aaa) messages.add_message(request, messages.SUCCESS, _("Task done.")) log( WARNING, aaa=aaa, app="home", view="admintools", action="updatebigset", status="OK", data="done", ) return redirect("app_home:admintools") # ----------------------------------------- # admin tools - EAV refresh # ----------------------------------------- def admintools_eavrefresh(request): context = start_view( request, app="admintools", view="eav_refresh", perm="p_admintools", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if request.method == "POST": messages.add_message(request, messages.SUCCESS, _("EAV Refresh - long task started...")) log( WARNING, aaa=aaa, app="home", view="admintools", action="eav_refresh", status="OK", data="started...", ) task_eav_refresh.delay(dryrun=False, force=False) return redirect("app_home:admintools") # ----------------------------------------- # admin tools - EAV purge # ----------------------------------------- def admintools_eavpurge(request): context = start_view( request, app="admintools", view="eav_purge", perm="p_admintools", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if request.method == "POST": messages.add_message(request, messages.SUCCESS, _("EAV Purge - long task started...")) log( WARNING, aaa=aaa, app="home", view="admintools", action="eav_purge", status="OK", data="started...", ) task_eav_purge.delay(dryrun=False) return redirect("app_home:admintools") # ----------------------------------------- # admin tools - IPAM refresh # ----------------------------------------- def admintools_ipamrefresh(request): context = start_view( request, app="admintools", view="ipam_refresh", perm="p_admintools", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if request.method == "POST": messages.add_message(request, messages.SUCCESS, _("IPAM Refresh - long task started...")) log(INFO, aaa=aaa, app="home", view="admintools", action="ipam_refresh", status="OK", data="ipam_refresh started...") # fmt: skip task_ipam_refresh.delay(dryrun=False) return redirect("app_home:admintools") # ----------------------------------------- # admin tools - Housekeeping # ----------------------------------------- def admintools_housekeeping(request): context = start_view( request, app="admintools", view="housekeeping", perm="p_admintools", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if request.method == "POST": messages.add_message( request, messages.SUCCESS, _("Global housekeeping - long task started...") ) log(WARNING, aaa=aaa, app="home", view="admintools", action="housekeeping", status="OK", data="started...") # fmt: skip housekeeping_daily.delay(username=aaa.get("username")) return redirect("app_home:admintools") # ----------------------------------------- # admin tools - Revision # ----------------------------------------- def admintools_revision(request): context = start_view( request, app="admintools", view="revision", perm="p_admintools", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] # Get pagination parameters from GET request try: page = int(request.GET.get("page", 1)) size = int(request.GET.get("size", 1000)) except Exception: page = 1 size = 1000 # Calculate offset for pagination offset = (page - 1) * size # Retrieve revisions ordered by date descending with pagination revisions = DataRevision.objects.all().order_by("-date")[offset : offset + size] context["revisions"] = revisions context["count"] = DataRevision.objects.count() context["page"] = page context["size"] = size log( DEBUG, aaa=aaa, app="home", view="admintools", action="revision", status="OK", data="view accessed", ) return render(request, f"app_home/{HTML_TEMPLATE_PATH}/revision.html", context) # ----------------------------------------- # admin tools - active users # ----------------------------------------- def admintools_active_user(request): """Admin tools: list users with a currently cached sidebar, a proxy for "active in the last CAVALIBA_CACHE_SIDEBAR_TIMEOUT seconds" (see app_home.sidebar.list_active_users). Only works with a redis cache backend - the sqlite/LocMemCache fallback can't enumerate keys, in which case context["supported"] is False and the template explains why.""" context = start_view( request, app="admintools", view="active_user", perm="p_admintools", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] usernames = list_active_users() users = [] for username in usernames or []: user = User.from_keyname(keyname=username) users.append( { "login": username, "firstname": user.get_attribute_first("firstname") if user else "", "lastname": user.get_attribute_first("lastname") if user else "", "email": user.email if user else "", "last_login": user.last_login if user else "", } ) context["users"] = users context["count"] = len(users) context["supported"] = usernames is not None log(DEBUG, aaa=aaa, app="home", view="admintools", action="active_user", status="OK", data=f"{context['count']} active users") # fmt: skip return render(request, f"app_home/{HTML_TEMPLATE_PATH}/active_user.html", context) # ----------------------------------------- # admin tools - rename a keyname # ----------------------------------------- # above this many referrers, warn that the synchronous real run may exceed the # web request timeout and point at the CLI/API instead KEYNAME_UI_WARN_REFERRERS = 500 def admintools_keyname(request): """Admin tools: rename an instance keyname (and cascade to every referrer) through app_data.keyname.rename - the same engine the CLI and /api/keyname/ use. The page is gated by p_admintools like its siblings; the rename itself also requires p_keyname_rename (context["can_rename"]). The dry run is mandatory and enforced here, not just by the UI flow: a mode="real" POST always re-runs rename(dryrun=True) against the submitted parameters first and writes nothing if that report carries errors. A forged or replayed POST therefore can't skip the check, and the pre-flight scan is read-only.""" context = start_view( request, app="admintools", view="keyname", perm="p_admintools", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] context["can_rename"] = "p_keyname_rename" in aaa["perms"] context["schemas"] = Schema.listall() context["warn_referrers"] = KEYNAME_UI_WARN_REFERRERS context["classname"] = "" context["oldkeyname"] = "" context["newkeyname"] = "" context["force"] = False context["report"] = None context["mode"] = "" if request.method != "POST": return render(request, f"app_home/{HTML_TEMPLATE_PATH}/keyname.html", context) if not context["can_rename"]: log(WARNING, aaa=aaa, app="home", view="admintools", action="keyname", status="KO", data="denied - missing p_keyname_rename") # fmt: skip messages.add_message( request, messages.ERROR, _("The p_keyname_rename permission is required.") ) return render(request, f"app_home/{HTML_TEMPLATE_PATH}/keyname.html", context) classname = request.POST.get("classname", "").strip() oldkeyname = request.POST.get("oldkeyname", "").strip() newkeyname = request.POST.get("newkeyname", "").strip() force = request.POST.get("force", "") in settings.TRUE_LIST mode = "real" if request.POST.get("mode", "") == "real" else "dryrun" context["classname"] = classname context["oldkeyname"] = oldkeyname context["newkeyname"] = newkeyname context["force"] = force report = rename( classname=classname, oldkeyname=oldkeyname, newkeyname=newkeyname, dryrun=True, force=force, ) # mandatory dry run: a real run only proceeds on a clean pre-flight report if mode == "real" and not report["errors"]: report = rename( classname=classname, oldkeyname=oldkeyname, newkeyname=newkeyname, dryrun=False, force=force, ) context["report"] = report context["mode"] = "dryrun" if report["dryrun"] else "real" if report["errors"]: log(WARNING, aaa=aaa, app="home", view="admintools", action="keyname", status="KO", data=f"{classname}/{oldkeyname} -> {newkeyname} - " + "; ".join(report["errors"])) # fmt: skip else: log(WARNING if report["target_renamed"] else DEBUG, aaa=aaa, app="home", view="admintools", action="keyname", status="OK", data=f"{classname}/{oldkeyname} -> {newkeyname} dryrun={report['dryrun']} force={force} referrers_changed={report['referrers_changed']}") # fmt: skip if report["target_renamed"]: messages.add_message( request, messages.SUCCESS, _("Keyname renamed: %(old)s -> %(new)s") % {"old": oldkeyname, "new": newkeyname}, ) # the rename succeeded: offer a clean form, not a re-run of the same one context["classname"] = "" context["oldkeyname"] = "" context["newkeyname"] = "" return render(request, f"app_home/{HTML_TEMPLATE_PATH}/keyname.html", context) # ----------------------------------------- # admin tools - factory reset # ----------------------------------------- def admintools_factory_reset(request): """Admin tools / Danger zone: empty the whole database (and the on-disk state it owns), then re-provision a fresh instance from builtin/init - optionally loading builtin/demo on top. All the work is in app_home.factory. Gated three ways: p_admintools for the page, aaa["is_admin"] (the "admin" account or the role_admin role) for the action, and the admin account password (CAVALIBA_ADMIN_PASSWORD env var) typed in the modal. POST only. The reset clears every cache, which destroys every session including this one, so the report is rendered directly rather than handed to the next request through a message + redirect.""" context = start_view( request, app="admintools", view="factory_reset", perm="p_admintools", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if request.method != "POST": return redirect("app_home:admintools") if not aaa.get("is_admin", False): log(WARNING, aaa=aaa, app="home", view="admintools", action="factory_reset", status="KO", data="denied - not an admin account") # fmt: skip messages.add_message(request, messages.ERROR, _("An admin account is required.")) return redirect("app_home:admintools") if not factory_reset_available(): log(WARNING, aaa=aaa, app="home", view="admintools", action="factory_reset", status="KO", data="denied - CAVALIBA_ADMIN_PASSWORD not set") # fmt: skip messages.add_message( request, messages.ERROR, _( "Factory reset is unavailable: the CAVALIBA_ADMIN_PASSWORD environment variable is not set." ), ) return redirect("app_home:admintools") if not factory_reset_password_ok(request.POST.get("admin_password", "")): log(WARNING, aaa=aaa, app="home", view="admintools", action="factory_reset", status="KO", data="denied - wrong admin account password") # fmt: skip messages.add_message(request, messages.ERROR, _("Wrong admin account password.")) return redirect("app_home:admintools") with_demo = request.POST.get("load_demo", "") in settings.TRUE_LIST report = factory_reset(aaa=aaa, with_demo=with_demo) # log() writes to the log table, which the reset just emptied - so this entry # is written after the wipe, on purpose, and is the first line of the new log log(WARNING, aaa=aaa, app="home", view="admintools", action="factory_reset", status="OK", data=f"database reset - demo={with_demo} db_version={report['db_version']} operator_restored={report['operator']['restored']} errors={len(report['errors'])}") # fmt: skip context["report"] = report context["with_demo"] = with_demo return render(request, f"app_home/{HTML_TEMPLATE_PATH}/factory_reset.html", context) # ---------------------------------------------------------------------------------------------------- # UI prototypes (beta preview) # ----------------------------------------------------------------------------------------------------- def _prototypes_gate(request, context): """Shared beta-preview gate for prototype views: redirects to admintools with a message if BETA_PREVIEW is not enabled. Returns a redirect response or None.""" if get_configuration("home", "BETA_PREVIEW") == "yes": return None messages.add_message( request, messages.WARNING, _("Enable Beta/Preview features in Configuration to access UI prototypes."), ) return redirect("app_home:admintools") def prototypes_index(request): """List available UI prototype pages (beta preview, admin-only).""" context = start_view( request, app="admintools", view="prototypes", perm="p_admintools", ) if context["redirect"]: return redirect(context["redirect"]) gate = _prototypes_gate(request, context) if gate: return gate context["prototypes"] = PROTOTYPE_REGISTRY return render(request, f"app_home/{HTML_TEMPLATE_PATH}/prototypes/index.html", context) def prototypes_detail(request, slug): """Render a single UI prototype page (beta preview, admin-only).""" context = start_view( request, app="admintools", view="prototypes", perm="p_admintools", ) if context["redirect"]: return redirect(context["redirect"]) gate = _prototypes_gate(request, context) if gate: return gate prototype = get_prototype(slug) if not prototype: raise Http404 context["prototype"] = prototype return render(request, prototype["template"], context)