#1082

config/settings/base.py

@@ -251,6 +251,10 @@
 ACCOUNT_EMAIL_VERIFICATION = "optional"
 ACCOUNT_ADAPTER = "allauth.account.adapter.DefaultAccountAdapter"
 
+MFA_SUPPORTED_TYPES = ["totp", "webauthn"]
+MFA_PASSKEY_LOGIN_ENABLED = True
+MFA_WEBAUTHN_ALLOW_INSECURE_ORIGIN = True
+
 # Elasticsearch
 # -------------------------------------------------------------------------------
 ELASTICSEARCH_URL = env("ELASTICSEARCH_URL")

orochi/templates/admin/dump_intermediate.html

@@ -0,0 +1,12 @@
+{% extends "admin/base_site.html" %}
+
+{% block content %}
+<form action="" method="post">
+    {% csrf_token %}
+    <p>Selected dumps:</p>
+    <ul>{{ items|unordered_list }}</ul>
+    {{ form }}
+    <input type="hidden" name="action" value="{{action}}" />
+    <input type="submit" name="apply" value="Confirm" />
+</form>
+{% endblock %}

orochi/utils/volatility_dask_elk.py

@@ -26,6 +26,7 @@
 from clamdpy import ClamdUnixSocket
 from distributed import fire_and_forget, get_client, rejoin, secede
 from django.conf import settings
+from django.core.exceptions import ObjectDoesNotExist
 from elasticsearch import Elasticsearch, helpers
 from elasticsearch_dsl import Search
 from extra_settings.models import Setting
@@ -828,30 +829,35 @@ def unzip_then_run(dump_pk, user_pk, password, restart, move):
 
             # check symbols using banners
             if dump.operating_system in ("Linux", "Mac"):
-                if banner := dump.result_set.get(plugin__name="banners.Banners"):
-                    banner.result = RESULT_STATUS_RUNNING
-                    banner.save()
-                    logging.info(f"[dump {dump_pk}] Running banners plugin")
-                    run_plugin(dump, banner.plugin)
-                    time.sleep(1)
-                    if banner_result := get_banner(banner):
-                        dump.banner = banner_result.strip("\"'")
-                        logging.error(
-                            f"[dump {dump_pk}] guessed banner '{dump.banner}'"
-                        )
-                        dump.save()
+                try:
+                    if banner := dump.result_set.get(plugin__name="banners.Banners"):
+                        banner.result = RESULT_STATUS_RUNNING
+                        banner.save()
+                        logging.info(f"[dump {dump_pk}] Running banners plugin")
+                        run_plugin(dump, banner.plugin)
+                        time.sleep(1)
+                        if banner_result := get_banner(banner):
+                            dump.banner = banner_result.strip("\"'")
+                            logging.error(
+                                f"[dump {dump_pk}] guessed banner '{dump.banner}'"
+                            )
+                            dump.save()
+                except ObjectDoesNotExist:
+                    logging.error(f"[dump {dump_pk}] Banner plugin missing")
             elif dump.operating_system == "Windows":
-                regipy = dump.result_set.get(
-                    plugin__name="windows.registry.hivelist.HiveList"
-                )
-                logging.info(f"[dump {dump_pk}] Running regipy plugins")
-                # run_plugin(dump, regipy.plugin, regipy_plugins=True)
-                dask_client = get_client()
-                fire_and_forget(
-                    dask_client.submit(
-                        run_plugin, dump, regipy.plugin, regipy_plugins=True
+                try:
+                    regipy = dump.result_set.get(
+                        plugin__name="windows.registry.hivelist.HiveList"
                     )
-                )
+                    logging.info(f"[dump {dump_pk}] Running regipy plugins")
+                    dask_client = get_client()
+                    fire_and_forget(
+                        dask_client.submit(
+                            run_plugin, dump, regipy.plugin, regipy_plugins=True
+                        )
+                    )
+                except ObjectDoesNotExist:
+                    logging.error(f"[dump {dump_pk}] HiveList plugin missing")
 
         if restart or check_runnable(dump.pk, dump.operating_system, dump.banner):
             dask_client = get_client()

orochi/website/admin.py

@@ -1,15 +1,23 @@
 from django.contrib import admin
+from django.contrib.auth import get_user_model
 from django.contrib.auth.models import Group
+from django.http import HttpResponseRedirect
+from django.shortcuts import render
 from django_admin_listfilter_dropdown.filters import RelatedDropdownFilter
 from django_admin_multiple_choice_list_filter.list_filters import (
     MultipleChoiceListFilter,
 )
 from django_file_form.model_admin import FileFormAdmin
 from django_file_form.models import TemporaryUploadedFile
 from guardian.admin import GuardedModelAdmin
+from guardian.shortcuts import assign_perm, get_objects_for_user, get_perms, remove_perm
 
 from orochi.website.defaults import RESULT
-from orochi.website.forms import PluginCreateAdminForm, PluginEditAdminForm
+from orochi.website.forms import (
+    PluginCreateAdminForm,
+    PluginEditAdminForm,
+    UserListForm,
+)
 from orochi.website.models import (
     Bookmark,
     CustomRule,
@@ -64,11 +72,65 @@ class ResultAdmin(admin.ModelAdmin):
 
 @admin.register(Dump)
 class DumpAdmin(GuardedModelAdmin):
+    actions = ["assign_to_users", "remove_from_users"]
     list_display = ("name", "author", "index", "status")
     search_fields = ["author__name", "name", "index"]
     list_filter = ("author", "status", "created_at")
     exclude = ("suggested_symbols_path", "regipy_plugins", "banner")
 
+    def assign_to_users(self, request, queryset):
+        if "apply" in request.POST:
+            users = request.POST.getlist("authorized_users")
+            for item in queryset:
+                for user_pk in users:
+                    user = get_user_model().objects.get(pk=user_pk)
+                    assign_perm("can_see", user, item)
+            self.message_user(
+                request, f"{len(queryset)} dumps added to {len(users)} users"
+            )
+            return HttpResponseRedirect(request.get_full_path())
+        form = UserListForm(
+            initial={"_selected_action": queryset.values_list("id", flat=True)}
+        )
+        return render(
+            request,
+            "admin/dump_intermediate.html",
+            context={
+                "items": queryset,
+                "form": form,
+                "title": "Assign dumps to users",
+                "action": "assign_to_users",
+            },
+        )
+
+    def remove_from_users(self, request, queryset):
+        if "apply" in request.POST:
+            users = request.POST.getlist("authorized_users")
+            for item in queryset:
+                for user_pk in users:
+                    user = get_user_model().objects.get(pk=user_pk)
+                    remove_perm("can_see", user, item)
+            self.message_user(
+                request, f"{len(queryset)} dumps removed from {len(users)} users"
+            )
+            return HttpResponseRedirect(request.get_full_path())
+        form = UserListForm(
+            initial={"_selected_action": queryset.values_list("id", flat=True)}
+        )
+        return render(
+            request,
+            "admin/dump_intermediate.html",
+            context={
+                "items": queryset,
+                "form": form,
+                "title": "Remove dumps from users",
+                "action": "remove_from_users",
+            },
+        )
+
+    assign_to_users.short_description = "Assign dump to users"
+    remove_from_users.short_description = "Remove dumps from users"
+
     def get_queryset(self, request):
         return super(DumpAdmin, self).get_queryset(request).prefetch_related("plugins")
 

orochi/website/forms.py

@@ -206,6 +206,22 @@ class Meta:
         }
 
 
+######################################
+# ADMIN USERLIST
+######################################
+class UserListForm(forms.Form):
+    _selected_action = forms.CharField(widget=forms.MultipleHiddenInput)
+    authorized_users = forms.TypedMultipleChoiceField(
+        required=False,
+    )
+
+    def __init__(self, *args, **kwargs):
+        super(UserListForm, self).__init__(*args, **kwargs)
+        self.fields["authorized_users"].choices = [
+            (x.pk, x.username) for x in get_user_model().objects.all()
+        ]
+
+
 ######################################
 # CREATE PLUGIN FROM ADMIN
 ######################################