#1073

Author: Davide Arcuri

compose/local/dask/Dockerfile

@@ -1,4 +1,4 @@
-FROM daskdev/dask:2024.9.1-py3.12
+FROM daskdev/dask:2024.12.1-py3.12
 ENV DEBIAN_FRONTEND noninteractive
 
 ARG local_folder=/uploads
@@ -27,7 +27,7 @@ RUN freshclam
 # Workers should have similar reqs as django
 WORKDIR /
 COPY ./requirements /requirements
-RUN pip install uv==0.5.6 -e git+https://github.com/dadokkio/volatility3.git@e2cdbdc2bf30b8c17ae36b68559ca4ff5c78b461#egg=volatility3 \
+RUN pip install uv==0.5.15 -e git+https://github.com/dadokkio/volatility3.git@a98a23fa41395b9bbd961f6e50d0a0f19201fcc7#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

@@ -1,4 +1,4 @@
-FROM python:3.12.6-slim-bookworm as common-base
+FROM python:3.12.8-slim-bookworm as common-base
 
 ENV DJANGO_SETTINGS_MODULE config.settings.local
 ENV PYTHONUNBUFFERED 1
@@ -44,7 +44,7 @@ RUN /usr/local/go/bin/go build
 FROM common-base
 WORKDIR /
 COPY ./requirements /requirements
-RUN pip install uv==0.5.6 -e git+https://github.com/dadokkio/volatility3.git@e2cdbdc2bf30b8c17ae36b68559ca4ff5c78b461#egg=volatility3 \
+RUN pip install uv==0.5.15 -e git+https://github.com/dadokkio/volatility3.git@a98a23fa41395b9bbd961f6e50d0a0f19201fcc7#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

@@ -1,5 +1,6 @@
 from ninja import NinjaAPI
 
+from orochi.api.routers.admin import router as admin_router
 from orochi.api.routers.auth import router as auth_router
 from orochi.api.routers.bookmarks import router as bookmarks_router
 from orochi.api.routers.customrules import router as customrules_router
@@ -12,6 +13,7 @@
 from orochi.api.routers.utils import router as utils_router
 
 api = NinjaAPI(csrf=True, title="Orochi API", urls_namespace="api")
+api.add_router("/admin/", admin_router, tags=["Admin"])
 api.add_router("/auth/", auth_router, tags=["Auth"])
 api.add_router("/users/", users_router, tags=["Users"])
 api.add_router("/folders/", folders_router, tags=["Folders"])

orochi/api/models.py

@@ -1,6 +1,6 @@
 from enum import Enum
 from pathlib import Path
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Tuple
 
 from django.contrib.auth import get_user_model
 from django.contrib.auth.models import Group
@@ -367,13 +367,16 @@ class RuleOut(Schema):
     headline: Optional[str] = None
 
 
-class RuleFilter(Schema):
+###################################################
+# Datatables
+###################################################
+class TableFilter(Schema):
     search: str = None
     order_column: int = 1
     order_dir: str = Field("asc", pattern="^(asc|desc)$")
 
 
-class CustomPagination(PaginationBase):
+class RulePagination(PaginationBase):
     class Input(Schema):
         start: int
         length: int
@@ -412,8 +415,6 @@ def paginate_queryset(self, queryset, pagination: Input, **params):
 ###################################################
 # Symbols
 ###################################################
-
-
 class SymbolsBannerIn(Schema):
     path: List[str] = []
     index: str
@@ -432,3 +433,37 @@ class UploadFileIn(Schema):
 
 class ISFIn(Schema):
     path: str
+
+
+class SymbolsOut(Schema):
+    id: str
+    path: str
+    action: Tuple[str, str]
+
+
+class CustomSymbolsPagination(PaginationBase):
+    class Input(Schema):
+        start: int
+        length: int
+
+    class Output(Schema):
+        draw: int
+        recordsTotal: int
+        recordsFiltered: int
+        data: List[SymbolsOut]
+
+    items_attribute: str = "data"
+
+    def paginate_queryset(self, queryset, pagination: Input, **params):
+        request = params["request"]
+        return {
+            "draw": request.draw,
+            "recordsTotal": request.total,
+            "recordsFiltered": len(queryset),
+            "data": [
+                SymbolsOut(**{"id": x.id, "path": x.path, "action": x.action})
+                for x in queryset[
+                    pagination.start : pagination.start + pagination.length
+                ]
+            ],
+        }

orochi/api/routers/admin.py

@@ -0,0 +1,132 @@
+from django.contrib import messages
+from django.core import management
+from ninja import Router
+from ninja.security import django_auth_superuser
+
+from orochi.api.models import ErrorsOut, SuccessResponse
+
+router = Router()
+
+
+@router.get(
+    "/rules/update",
+    auth=django_auth_superuser,
+    response={200: SuccessResponse, 400: ErrorsOut},
+    url_name="update_rules",
+)
+def update_rules(request):
+    """Update rules.
+
+    This endpoint triggers the synchronization of rules using a management command.
+    It returns a success message if the synchronization is successful, or an error message if it fails.
+
+    Args:
+        request: The request object.
+
+    Returns:
+        Tuple[int, dict]: A tuple containing the status code and a dictionary with a message.
+            Returns 200 and a success message if the synchronization is successful.
+            Returns 404 and an error message if the synchronization fails.
+
+    Raises:
+        Exception: If an error occurs during rule synchronization.
+    """
+    try:
+        management.call_command("rules_sync", verbosity=0)
+        messages.add_message(request, messages.INFO, "Sync Rules done")
+        return 200, {"message": "Sync Symbols done"}
+    except Exception as e:
+        messages.add_message(request, messages.ERROR, f"Sync Plugin failed: {e}")
+        return 404, {"errors": "Forbidden"}
+
+
+@router.get(
+    "/rules/generate",
+    auth=django_auth_superuser,
+    response={200: SuccessResponse, 400: ErrorsOut},
+    url_name="generate_default_rule",
+)
+def generate_default_rule(request):
+    """Generate a default rule.
+
+    This endpoint triggers the generation of a default rule using a management command.
+    It returns a success message if the rule creation is successful, or an error message if it fails.
+
+    Args:
+        request: The request object.
+
+    Returns:
+        Tuple[int, dict]: A tuple containing the status code and a dictionary with a message.
+            Returns 200 and a success message if the rule creation is successful.
+            Returns 404 and an error message if the rule creation fails.
+
+    Raises:
+        Exception: If an error occurs during rule generation.
+    """
+    try:
+        management.call_command("generate_default_rule", verbosity=0)
+        messages.add_message(request, messages.INFO, "Default Rule created")
+        return 200, {"message": "Sync Symbols done"}
+    except Exception as e:
+        messages.add_message(request, messages.ERROR, f"Sync Plugin failed: {e}")
+        return 404, {"errors": "Forbidden"}
+
+
+@router.get(
+    "/plugins/update",
+    auth=django_auth_superuser,
+    response={200: SuccessResponse, 400: ErrorsOut},
+    url_name="update_plugins",
+)
+def update_plugins(request):
+    """Update plugins for the application.
+
+    This endpoint triggers a plugin synchronization process. It then redirects to the admin page, displaying a success or error message.
+
+    Args:
+        request: The incoming HTTP request.
+
+    Returns:
+        A redirect to /admin with a success message if the synchronization is successful, or a 404 error with an error message if it fails.
+
+    Raises:
+        Exception: If an error occurs during plugin synchronization.
+    """
+
+    try:
+        management.call_command("plugins_sync", verbosity=0)
+        messages.add_message(request, messages.INFO, "Sync Plugin done")
+        return 200, {"message": "Sync Plugin done"}
+    except Exception as e:
+        messages.add_message(request, messages.ERROR, f"Sync Plugin failed: {e}")
+        return 404, {"errors": "Forbidden"}
+
+
+@router.get(
+    "/symbols/update",
+    auth=django_auth_superuser,
+    response={200: SuccessResponse, 400: ErrorsOut},
+    url_name="update_symbols",
+)
+def update_symbols(request):
+    """Update symbols for the application.
+
+    This endpoint triggers a symbol synchronization process. It then redirects to the admin page, displaying a success or error message.
+
+    Args:
+        request: The incoming HTTP request.
+
+    Returns:
+        A redirect to /admin with a success message if the synchronization is successful, or a 404 error with an error message if it fails.
+
+    Raises:
+        Exception: If an error occurs during symbol synchronization.
+    """
+
+    try:
+        management.call_command("symbols_sync", verbosity=0)
+        messages.add_message(request, messages.INFO, "Sync Symbols done")
+        return 200, {"message": "Sync Symbols done"}
+    except Exception as e:
+        messages.add_message(request, messages.ERROR, f"Sync Symbols failed: {e}")
+        return 404, {"errors": "Forbidden"}

orochi/api/routers/customrules.py

@@ -16,8 +16,8 @@
     ListStr,
     ListStrAction,
     RuleData,
-    RuleFilter,
     SuccessResponse,
+    TableFilter,
 )
 from orochi.website.models import CustomRule
 
@@ -32,7 +32,7 @@
 )
 @paginate(CustomRulePagination)
 def list_custom_rules(
-    request: HttpRequest, draw: Optional[int], filters: RuleFilter = Query(...)
+    request: HttpRequest, draw: Optional[int], filters: TableFilter = Query(...)
 ):
     rules = CustomRule.objects.filter(Q(public=True) | Q(user=request.user))
     request.draw = draw

orochi/api/routers/rules.py

@@ -14,15 +14,15 @@
 from ninja.security import django_auth
 
 from orochi.api.models import (
-    CustomPagination,
     ErrorsOut,
     ListStr,
     RuleBuildSchema,
     RuleEditInSchena,
-    RuleFilter,
     RuleOut,
+    RulePagination,
     RulesOutSchema,
     SuccessResponse,
+    TableFilter,
 )
 from orochi.website.models import CustomRule
 from orochi.ya.models import Rule, Ruleset
@@ -31,9 +31,9 @@
 
 
 @router.get("/", auth=django_auth, url_name="list_rules", response=List[RuleOut])
-@paginate(CustomPagination)
+@paginate(RulePagination)
 def list_rules(
-    request: HttpRequest, draw: Optional[int], filters: RuleFilter = Query(...)
+    request: HttpRequest, draw: Optional[int], filters: TableFilter = Query(...)
 ):
     """Retrieve a list of rules based on the provided filters and pagination.
 
@@ -43,7 +43,7 @@ def list_rules(
     Args:
         request (HttpRequest): The HTTP request object containing user and query information.
         draw (int, optional): A draw counter for the DataTables plugin to ensure proper response handling.
-        filters (RuleFilter, optional): An object containing search and order criteria. Defaults to Query(...).
+        filters (TableFilter, optional): An object containing search and order criteria. Defaults to Query(...).
 
     Returns:
         List[RuleOut]: A list of rules that match the specified filters and pagination settings.

orochi/api/routers/symbols.py

@@ -9,18 +9,24 @@
 
 import magic
 import requests
+from django.http import HttpRequest
 from django.shortcuts import get_object_or_404
 from django.utils.text import slugify
 from extra_settings.models import Setting
-from ninja import File, Router
+from ninja import File, Query, Router
 from ninja.files import UploadedFile
+from ninja.pagination import paginate
 from ninja.security import django_auth
+from volatility3.framework import automagic, contexts
 
 from orochi.api.models import (
+    CustomSymbolsPagination,
     ErrorsOut,
     ISFIn,
     SuccessResponse,
     SymbolsBannerIn,
+    SymbolsOut,
+    TableFilter,
     UploadFileIn,
 )
 from orochi.utils.download_symbols import Downloader
@@ -31,6 +37,46 @@
 router = Router()
 
 
+@router.get("/", auth=django_auth, url_name="list_symbols", response=List[SymbolsOut])
+@paginate(CustomSymbolsPagination)
+def list_symbols(
+    request: HttpRequest, draw: Optional[int], filters: TableFilter = Query(...)
+):
+    symbols = []
+
+    ctx = contexts.Context()
+    automagics = automagic.available(ctx)
+    if banners := [x for x in automagics if x._config_path == "automagic.SymbolFinder"]:
+        banner = banners[0].banners
+    else:
+        banner = []
+
+    request.draw = draw
+    request.total = len(banner)
+    request.search = filters.search or None
+
+    for k, v in banner.items():
+        try:
+            k = k.decode("utf-8")
+            v = str(v)
+        except AttributeError:
+            k = str(k)
+
+        if filters.search and (filters.search not in k and filters.search not in v):
+            continue
+
+        if "file://" in v:
+            path = v.replace("file://", "").replace(
+                Setting.get("VOLATILITY_SYMBOL_PATH"), ""
+            )
+            action = ("list", "-") if "/added/" not in v else ("delete", path)
+        else:
+            action = ("down", v)
+
+        symbols.append(SymbolsOut(id=k, path=path, action=action))
+    return symbols
+
+
 @router.post(
     "/banner",
     auth=django_auth,