#1073

.vscode/settings.json

@@ -20,12 +20,9 @@
     "commandCenter.border": "#e7e7e799"
   },
   "peacock.color": "#922744",
-  "python.linting.pylintEnabled": false,
-  "python.linting.mypyEnabled": false,
-  "python.linting.enabled": true,
-  "python.formatting.provider": "none",
-  "python.linting.flake8Enabled": false,
   "[python]": {
     "editor.defaultFormatter": "ms-python.black-formatter"
-  }
+  },
+  "python.testing.unittestEnabled": false,
+  "python.testing.pytestEnabled": true
 }

compose/local/dask/Dockerfile

@@ -27,7 +27,7 @@ RUN freshclam
 # Workers should have similar reqs as django
 WORKDIR /
 COPY ./requirements /requirements
-RUN pip install uv==0.4.25 -e git+https://github.com/dadokkio/volatility3.git@7b0cb4facd1e1714a36793a27c0570461a3f02a1#egg=volatility3 \
+RUN pip install uv==0.5.6 -e git+https://github.com/dadokkio/volatility3.git@e2cdbdc2bf30b8c17ae36b68559ca4ff5c78b461#egg=volatility3 \
   && uv pip install --no-cache --system -r /requirements/base.txt
 
 COPY ./compose/local/dask/prepare.sh /usr/bin/prepare.sh

compose/local/django/Dockerfile

@@ -44,7 +44,7 @@ RUN /usr/local/go/bin/go build
 FROM common-base
 WORKDIR /
 COPY ./requirements /requirements
-RUN pip install uv==0.4.25 -e git+https://github.com/dadokkio/volatility3.git@7b0cb4facd1e1714a36793a27c0570461a3f02a1#egg=volatility3 \
+RUN pip install uv==0.5.6 -e git+https://github.com/dadokkio/volatility3.git@e2cdbdc2bf30b8c17ae36b68559ca4ff5c78b461#egg=volatility3 \
   && uv pip install --no-cache --system -r /requirements/base.txt
 
 COPY ./compose/local/__init__.py /src/volatility3/volatility3/framework/constants/__init__.py

orochi/api/api.py

@@ -7,6 +7,7 @@
 from orochi.api.routers.folders import router as folders_router
 from orochi.api.routers.plugins import router as plugins_router
 from orochi.api.routers.rules import router as rules_router
+from orochi.api.routers.symbols import router as symbols_router
 from orochi.api.routers.users import router as users_router
 from orochi.api.routers.utils import router as utils_router
 
@@ -20,3 +21,4 @@
 api.add_router("/bookmarks/", bookmarks_router, tags=["Bookmarks"])
 api.add_router("/rules/", rules_router, tags=["Rules"])
 api.add_router("/customrules/", customrules_router, tags=["Custom Rules"])
+api.add_router("/symbols/", symbols_router, tags=["Symbols"])

orochi/api/models.py

@@ -407,3 +407,24 @@ def paginate_queryset(self, queryset, pagination: Input, **params):
                 ]
             ],
         }
+
+
+###################################################
+# Symbols
+###################################################
+
+
+class SymbolsBannerIn(Schema):
+    path: List[str] = []
+    index: str
+    operating_system: OSEnum
+    banner: str = None
+
+
+class SymbolsInfo(Schema):
+    original_name: Optional[str] = None
+    local_folder: Optional[str] = None
+
+
+class SymbolsIn(Schema):
+    info: Optional[List[SymbolsInfo]] = []

orochi/api/routers/dumps.py

@@ -6,7 +6,6 @@
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.db import transaction
-from django.http import HttpResponse
 from django.shortcuts import get_object_or_404
 from guardian.shortcuts import assign_perm, get_objects_for_user, get_perms, remove_perm
 from ninja import File, PatchDict, Query, Router, UploadedFile
@@ -22,7 +21,12 @@
     ResultSmallOutSchema,
     SuccessResponse,
 )
-from orochi.website.defaults import RESULT_STATUS_NOT_STARTED, RESULT_STATUS_RUNNING
+from orochi.utils.volatility_dask_elk import check_runnable, get_banner
+from orochi.website.defaults import (
+    DUMP_STATUS_COMPLETED,
+    RESULT_STATUS_NOT_STARTED,
+    RESULT_STATUS_RUNNING,
+)
 from orochi.website.models import Dump, Folder, Result, UserPlugin
 from orochi.website.views import index_f_and_f
 
@@ -93,7 +97,7 @@ def delete_dump(request, pk: UUID):
         }
 
 
-@router.get("/{pk}", response=DumpInfoSchema, auth=django_auth)
+@router.get("/{pk}", response={200: DumpInfoSchema, 400: ErrorsOut}, auth=django_auth)
 def get_dump_info(request, pk: UUID):
     """
     Summary:
@@ -111,11 +115,16 @@ def get_dump_info(request, pk: UUID):
     """
     dump = get_object_or_404(Dump, index=pk)
     if dump not in get_objects_for_user(request.user, "website.can_see"):
-        return HttpResponse("Forbidden", status=403)
-    return dump
+        return 400, {"errors": "Forbidden"}
+    return 200, dump
 
 
-@router.post("/", url_name="create_index", response=DumpSchema, auth=django_auth)
+@router.post(
+    "/",
+    url_name="create_index",
+    response={200: DumpSchema, 400: ErrorsOut},
+    auth=django_auth,
+)
 def create_dump(request, payload: DumpIn, upload: Optional[UploadedFile] = File(None)):
     """
     Creates a new dump index and handles the associated file uploads. This function processes the provided payload to create a dump entry in the database and manages file storage based on the input parameters.
@@ -161,7 +170,7 @@ def create_dump(request, payload: DumpIn, upload: Optional[UploadedFile] = File(
             dump.upload.save(Path(upload.name).name, upload)
             move = True
         else:
-            return HttpResponse("Bad Request", status=400)
+            return 400, {"errors": "Bad Request"}
         dump.save()
         Result.objects.bulk_create(
             [
@@ -196,10 +205,15 @@ def create_dump(request, payload: DumpIn, upload: Optional[UploadedFile] = File(
         )
         return dump
     except Exception as excp:
-        return HttpResponse(f"Bad Request ({excp})", status=400)
+        return 400, {"errors": f"Bad Request ({excp})"}
 
 
-@router.patch("/{pk}", url_name="edit_index", response=DumpSchema, auth=django_auth)
+@router.patch(
+    "/{pk}",
+    url_name="edit_index",
+    response={200: DumpSchema, 400: ErrorsOut},
+    auth=django_auth,
+)
 def edit_dump(request, pk: UUID, payload: PatchDict[DumpEditIn]):
     """
     Edits an existing dump based on the provided payload. This function updates the dump's attributes and manages user permissions for accessing the dump.
@@ -230,7 +244,7 @@ def edit_dump(request, pk: UUID, payload: PatchDict[DumpEditIn]):
             if "can_see" in get_perms(user, dump) and user != request.user
         ]
 
-        if payload["folder"]:
+        if payload.get("folder"):
             folder, _ = Folder.objects.get_or_create(
                 name=payload["folder"]["name"], user=request.user
             )
@@ -243,19 +257,15 @@ def edit_dump(request, pk: UUID, payload: PatchDict[DumpEditIn]):
                 for user_pk in payload.get("authorized_users", []):
                     user = get_user_model().objects.get(pk=user_pk)
                     if user.pk not in auth_users:
-                        assign_perm(
-                            "can_see",
-                            user,
-                            dump,
-                        )
+                        assign_perm("can_see", user, dump)
                 for user_pk in auth_users:
                     if user_pk not in payload.get("authorized_users", []):
                         user = get_user_model().objects.get(pk=user_pk)
                         remove_perm("can_see", user, dump)
         dump.save()
         return dump
     except Exception as excp:
-        return HttpResponse(f"Bad Request ({excp})", status=400)
+        return 400, {"errors": f"Bad Request ({excp})"}
 
 
 @router.get(
@@ -302,10 +312,64 @@ def get_dump_plugins(request, pks: List[UUID], filters: Query[DumpFilters] = Non
     auth=django_auth,
 )
 def get_dump_plugin_status(request, pks: List[UUID], plugin_name: int):
+    """
+    Retrieve the status of a specific plugin for a list of dumps. This function checks the user's permissions and returns the relevant results based on the provided dump indices and plugin name.
+
+    Args:
+        request: The HTTP request object.
+        pks (List[UUID]): A list of UUIDs representing the dump indices.
+        plugin_name (int): The name of the plugin to filter results by.
+
+    Returns:
+        QuerySet: A queryset containing the results related to the specified dumps and plugin.
+
+    Raises:
+        PermissionDenied: If the user does not have permission to view the dumps.
+    """
     dumps_ok = get_objects_for_user(request.user, "website.can_see")
     dumps = [
         dump.index for dump in Dump.objects.filter(index__in=pks) if dump in dumps_ok
     ]
     return Result.objects.select_related("dump", "plugin").filter(
         dump__index__in=dumps, plugin__name=plugin_name
     )
+
+
+@router.get(
+    "/{pk}/reload_symbols",
+    url_name="reload_symbols",
+    auth=django_auth,
+    response={200: SuccessResponse, 400: ErrorsOut},
+)
+def reload_symbols(request, pk: UUID):
+    """
+    Reload the symbols for a specific dump identified by its primary key. This function checks user permissions, attempts to reload the banner if necessary, and updates the dump's status accordingly.
+
+    Args:
+        request: The HTTP request object.
+        pk (UUID): The primary key of the dump to reload symbols for.
+
+    Returns:
+        Tuple[int, dict]: A tuple containing the HTTP status code and a message indicating the result of the operation.
+
+    Raises:
+        Http404: If the dump with the specified primary key does not exist.
+    """
+    try:
+        dump = get_object_or_404(Dump, index=pk)
+        if dump not in get_objects_for_user(request.user, "website.can_see"):
+            return 403, {"message": "Unauthorized"}
+
+        # Try to reload banner from elastic if first time was not successful
+        if not dump.banner:
+            banner = dump.result_set.get(plugin__name="banners.Banners")
+            if banner_result := get_banner(banner):
+                dump.banner = banner_result.strip("\"'")
+                dump.save()
+
+        if check_runnable(dump.pk, dump.operating_system, dump.banner):
+            dump.status = DUMP_STATUS_COMPLETED
+            dump.save()
+        return 200, {"message": f"Symbol for index {dump.name} has been reloaded."}
+    except Exception as excp:
+        return 400, {"errors": f"Bad Request ({excp})"}

orochi/api/routers/symbols.py

@@ -0,0 +1,151 @@
+import os
+import shutil
+import subprocess
+from pathlib import Path
+from typing import List, Optional
+
+import magic
+from django.shortcuts import get_object_or_404
+from extra_settings.models import Setting
+from ninja import File, Router
+from ninja.files import UploadedFile
+from ninja.security import django_auth
+
+from orochi.api.models import ErrorsOut, SuccessResponse, SymbolsBannerIn, SymbolsIn
+from orochi.utils.download_symbols import Downloader
+from orochi.utils.volatility_dask_elk import check_runnable, refresh_symbols
+from orochi.website.defaults import DUMP_STATUS_COMPLETED
+from orochi.website.models import Dump
+
+router = Router()
+
+
+@router.post(
+    "/banner",
+    auth=django_auth,
+    response={200: SuccessResponse, 400: ErrorsOut},
+)
+def banner_symbols(request, payload: SymbolsBannerIn):
+    """
+    Handles the POST request to download banner symbols based on the provided payload.
+    It checks the status of the download and updates the corresponding dump object accordingly.
+
+    Args:
+        request: The HTTP request object.
+        payload (SymbolsBannerIn): The input data containing the index and path for the symbols.
+
+    Returns:
+        tuple: A tuple containing the HTTP status code and a message or error details.
+
+    Raises:
+        Exception: If an error occurs during the download process or while updating the dump status.
+    """
+    try:
+        dump = get_object_or_404(Dump, index=payload.index)
+
+        d = Downloader(url_list=payload.path)
+        d.download_list()
+
+        if check_runnable(dump.pk, dump.operating_system, dump.banner):
+            dump.status = DUMP_STATUS_COMPLETED
+            dump.save()
+            return 200, {"message": "Symbol downloaded successfully"}
+        return 400, {"errors": "Downloaded symbols not properly installed"}
+    except Exception as excp:
+        return 400, {"errors": str(excp)}
+
+
+@router.post(
+    "/upload",
+    url_name="upload_symbols",
+    auth=django_auth,
+    response={200: SuccessResponse, 400: ErrorsOut},
+)
+def upload_symbols(
+    request, payload: SymbolsIn, symbols: Optional[List[UploadedFile]] = File(None)
+):
+    """
+    Uploads a list of symbol files to a specified directory and extracts them if they are in a compressed format. This function handles file writing and type checking to ensure proper processing of the uploaded symbols.
+
+    Args:
+        request: The HTTP request object.
+        symbols (List[UploadedFile]): A list of uploaded files representing the symbols to be processed.
+
+    Returns:
+        tuple: A tuple containing the HTTP status code and a message indicating the result of the upload.
+
+    Raises:
+        HttpResponse: Returns a 400 Bad Request response if an error occurs during the upload process.
+    """
+    try:
+        path = Path(Setting.get("VOLATILITY_SYMBOL_PATH")) / "added"
+        path.mkdir(parents=True, exist_ok=True)
+        if payload.info:
+            for item in payload.info:
+                start = item.local_folder
+                original_name = item.original_name
+                start = start.replace("/upload/upload", "/media/uploads")
+                filename = original_name
+                filepath = f"{path}/{filename}"
+                shutil.move(start, filepath)
+                filetype = magic.from_file(filepath, mime=True)
+                if filetype in [
+                    "application/zip",
+                    "application/x-7z-compressed",
+                    "application/x-rar",
+                    "application/gzip",
+                    "application/x-tar",
+                ]:
+                    subprocess.call(["7z", "e", filepath, f"-o{path}", "-y"])
+        elif symbols:
+            for symbol in symbols:
+                filepath = f"{path}/{Path(symbol.name).name}"
+                with open(filepath, "wb") as f:
+                    f.write(symbol.read())
+                filetype = magic.from_file(filepath, mime=True)
+                if filetype in [
+                    "application/zip",
+                    "application/x-7z-compressed",
+                    "application/x-rar",
+                    "application/gzip",
+                    "application/x-tar",
+                ]:
+                    subprocess.call(["7z", "e", filepath, f"-o{path}", "-y"])
+        refresh_symbols()
+        return 200, {"message": "Symbols uploaded."}
+
+    except Exception as excp:
+        return 400, {"errors": str(excp)}
+
+
+@router.delete(
+    "/delete",
+    url_name="delete_symbol",
+    auth=django_auth,
+    response={200: SuccessResponse, 405: ErrorsOut},
+)
+def delete_symbol(request, path):
+    """Delete a specific symbol file from the symbols directory.
+
+    Attempts to delete a symbol file located in the added symbols path. If the file exists and is within the added directory,
+    it will be removed and symbols will be refreshed.
+
+    Args:
+        request: The incoming HTTP request.
+        path: The relative path of the symbol file to delete.
+
+    Returns:
+        200: A success message if the symbol is deleted successfully.
+        400: An error response with exception details if deletion fails.
+
+    Raises:
+        Exception: If there are any issues during the deletion process.
+    """
+    try:
+        symbol_path = f"{Setting.get('VOLATILITY_SYMBOL_PATH')}{path}"
+        if Path(symbol_path).exists() and symbol_path.find("/added/") != -1:
+            os.unlink(symbol_path)
+            refresh_symbols()
+            return 200, {"message": "Symbols deleted."}
+    except Exception as excp:
+        return 400, {"errors": str(excp)}