Bump jinja2 from 3.1.3 to 3.1.4 in /app

CHANGELOG.md

@@ -1,6 +1,19 @@
 # Changelog
 Changes to the project will be tracked in this file via the date of change.
 
+## 2024-03-29
+- Adding unencrypted VirusTotal submission support
+- Update dependency (Black)
+
+## 2024-03-20
+- Dependency updates
+
+## 2024-03-04
+- Adding TLSH Card support
+
+## 2024-02-02
+- Adding QR Card support
+
 ## 2024-01-17
 - Bug fix for Visual Basic card where page fails to load if data does not exist for a section.
 

app/blueprints/strelka.py

@@ -4,23 +4,72 @@
 import os
 from collections import defaultdict
 
-from io import BytesIO
-from typing import Dict, Tuple
+from typing import Dict, Tuple, Union
 
 from flask import Blueprint, current_app, jsonify, request, session, Response
-from sqlalchemy import or_, desc, asc, func, case
-from sqlalchemy.dialects.postgresql import JSON
+from sqlalchemy import or_, desc, asc, func, case, cast, String
 from sqlalchemy.orm import joinedload
 
 from database import db
 from models import FileSubmission, User
 from services.auth import auth_required
 from services.strelka import get_db_status, get_frontend_status, submit_data
-from services.virustotal import get_virustotal_positives, create_vt_zip_and_download
+from services.virustotal import (
+    get_virustotal_positives,
+    create_vt_zip_and_download,
+    download_vt_bytes,
+    get_virustotal_widget_url,
+)
 from services.insights import get_insights
 
 strelka = Blueprint("strelka", __name__, url_prefix="/strelka")
 
+# Define the priority of each mimetype (For VirusTotal Scanning Priorization)
+MIMETYPE_PRIORITY = {
+    "application/x-dosexec": 1,  # Executables
+    "application/x-executable": 1,
+    "application/vnd.microsoft.portable-executable": 1,
+    "application/x-elf": 1,
+    "application/zip": 2,  # Archives
+    "application/x-rar-compressed": 2,
+    "application/x-msi": 2,
+    "application/x-7z-compressed": 2,
+    "application/vnd.ms-cab-compressed": 2,
+    "application/x-tar": 2,
+    "application/gzip": 2,
+    "application/octet-stream": 2,  # Unknown streams
+    "text/plain": 3,  # Scripts and source code
+    "text/x-script": 3,
+    "text/javascript": 3,
+    "application/x-bat": 3,
+    "application/x-sh": 3,
+    "application/x-python": 3,
+    "text/x-python": 3,
+    "text/html": 4,  # Web files
+    "application/xhtml+xml": 4,
+    "application/xml": 4,
+    "text/xml": 4,
+    "text/css": 4,
+    "application/pdf": 5,  # Documents
+    "application/msword": 5,
+    "application/vnd.ms-excel": 5,
+    "application/vnd.ms-powerpoint": 5,
+    "application/vnd.openxmlformats-officedocument.wordprocessingml.document": 5,
+    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": 5,
+    "application/vnd.openxmlformats-officedocument.presentationml.presentation": 5,
+    # Images
+    "image/jpeg": 10,
+    "image/png": 10,
+    "image/gif": 10,
+    "image/webp": 10,
+    "image/tiff": 10,
+    "image/bmp": 10,
+    "image/svg+xml": 10,
+    # Other file types
+    "application/json": 20,  # Data formats
+    "application/xml": 20,
+}
+
 
 @strelka.route("/status/strelka", methods=["GET"])
 def get_server_status() -> Tuple[Response, int]:
@@ -60,7 +109,9 @@ def get_database_status() -> Tuple[Response, int]:
 
 @strelka.route("/upload", methods=["POST"])
 @auth_required
-def submit_file(user: User) -> Tuple[Response, int]:
+def submit_file(
+    user: User,
+) -> Union[tuple[Response, int], tuple[dict[str, Union[Response, str]], int]]:
     """
     Submit a file (or hash) to the Strelka file analysis engine and save the analysis results to the database.
 
@@ -75,6 +126,7 @@ def submit_file(user: User) -> Tuple[Response, int]:
         None.
     """
     submitted_hash = ""
+    total_scanned_with_hits = []
 
     if "file" not in request.files and b"hash" not in request.data:
         return (
@@ -111,25 +163,46 @@ def submit_file(user: User) -> Tuple[Response, int]:
         submission = json.loads(request.data)
         submitted_description = submission["description"]
         submitted_hash = submission["hash"]
+        submitted_encrypted = submission.get("encrypted", True)
         submitted_type = "virustotal"
 
         if os.environ.get("VIRUSTOTAL_API_KEY"):
-            file = create_vt_zip_and_download(
-                api_key=os.environ.get("VIRUSTOTAL_API_KEY"),
-                file_hashes=[submitted_hash],
-                password="infected",
-            )
-            if "Error" in str(file):
+            try:
+                if submitted_encrypted:
+                    file = create_vt_zip_and_download(
+                        api_key=os.environ.get("VIRUSTOTAL_API_KEY"),
+                        file_hash=[submitted_hash],
+                        password="infected",
+                    )
+                else:
+                    file = download_vt_bytes(
+                        api_key=os.environ.get("VIRUSTOTAL_API_KEY"),
+                        file_hash=submitted_hash,
+                    )
+
+                # Assume file is a BytesIO object and set a filename attribute
+                file.filename = submitted_hash
+
+            except Exception as e:
                 return (
                     jsonify(
                         {
                             "error": "VirusTotal request was not successful.",
-                            "details": str(file.args[0]),
+                            "details": str(e),
                         }
                     ),
                     400,
                 )
-            file.filename = submitted_hash
+        else:
+            return (
+                jsonify(
+                    {
+                        "error": "VirusTotal request was not successful.",
+                        "details": "No VirusTotal API key has been loaded. Check the README for details.",
+                    }
+                ),
+                400,
+            )
 
     # Submit the file to the Strelka analysis engine.
     if file:
@@ -157,14 +230,25 @@ def submit_file(user: User) -> Tuple[Response, int]:
             # Get the submitted file object from the analysis results.
             submitted_file = response[0]
 
+            def get_mimetype_priority(mime_list):
+                # Return the highest priority of the mimetypes in the list
+                return min(MIMETYPE_PRIORITY.get(mime, 9999) for mime in mime_list)
+
             # If VirusTotal API key provided, get positives (VIRUSTOTAL_API_LIMIT determined Max Scans per Request)
             # -1    = VirusTotal Lookup Error
             # -2    = VirusTotal API Key Not Provided
             # >= 0  = Response Positives from VirusTotal
             if os.environ.get("VIRUSTOTAL_API_KEY"):
                 total_scanned = 0
                 try:
-                    for scanned_file in response:
+                    # Sort files by the priority of their mimetypes
+                    sorted_files = sorted(
+                        response,
+                        key=lambda x: get_mimetype_priority(
+                            x["file"]["flavors"]["mime"]
+                        ),
+                    )
+                    for scanned_file in sorted_files:
                         if total_scanned <= int(os.environ.get("VIRUSTOTAL_API_LIMIT")):
                             scanned_file["enrichment"] = {"virustotal": -2}
                             scanned_file["enrichment"][
@@ -173,7 +257,18 @@ def submit_file(user: User) -> Tuple[Response, int]:
                                 api_key=os.environ.get("VIRUSTOTAL_API_KEY"),
                                 file_hash=scanned_file["scan"]["hash"]["sha256"],
                             )
-                            total_scanned += 1
+                            total_scanned += 1.0
+                            if scanned_file["enrichment"]["virustotal"] > 0:
+                                total_scanned_with_hits.append(
+                                    {
+                                        "file_sha256": scanned_file["scan"]["hash"][
+                                            "sha256"
+                                        ],
+                                        "positives": scanned_file["enrichment"][
+                                            "virustotal"
+                                        ],
+                                    }
+                                )
                         else:
                             scanned_file["enrichment"] = {"virustotal": -3}
                 except Exception as e:
@@ -240,7 +335,20 @@ def submit_file(user: User) -> Tuple[Response, int]:
             db.session.commit()
 
             # Return the analysis results and a 200 status code.
-            return jsonify(response), 200
+            return (
+                jsonify(
+                    {
+                        "file_id": str(new_submission.file_id),
+                        "response": response,
+                        "meta": {
+                            "file_size": int(file_size),
+                            "iocs": list(iocs),
+                            "vt_positives": list(total_scanned_with_hits),
+                        },
+                    }
+                ),
+                200,
+            )
 
         # If an exception occurs, log the error and return an error message.
         except Exception as e:
@@ -498,8 +606,10 @@ def view(user: User) -> Tuple[Dict[str, any], int]:
     # Apply the search filter if there is a search query
     if search_query:
         search_filter = or_(
-            FileSubmission.file_name.like(f"%{search_query}%"),
-            FileSubmission.submitted_description.like(f"%{search_query}%"),
+            FileSubmission.file_name.ilike(f"%{search_query}%"),
+            FileSubmission.submitted_description.ilike(f"%{search_query}%"),
+            cast(FileSubmission.yara_hits, String).ilike(f"%{search_query}%"),
+            User.user_cn.ilike(f"%{search_query}%"),
         )
         base_query = base_query.filter(search_filter)
 
@@ -622,6 +732,51 @@ def get_mime_type_stats(user: User) -> Tuple[Response, int]:
     return jsonify(stats), 200
 
 
+@strelka.route("/virustotal/widget-url", methods=["POST"])
+@auth_required
+def get_vt_widget_url(resource: str) -> Tuple[Response, int]:
+    """
+    Route to get a VirusTotal widget url with customized theme colors.
+
+    Returns:
+        A JSON response containing the VirusTotal widget url or an error message.
+    """
+    data = request.get_json()
+    if not data or "resource" not in data:
+        return jsonify({"error": "Resource identifier is required"}), 400
+
+    # Strelka UI Defaults
+    fg1 = data.get("fg1", "333333")  # Dark text color
+    bg1 = data.get("bg1", "FFFFFF")  # Light background color
+    bg2 = data.get("bg2", "F5F5F5")  # Slightly grey background for differentiation
+    bd1 = data.get("bd1", "E8E8E8")  # Light grey border color
+
+    api_key = os.getenv("VIRUSTOTAL_API_KEY")
+    if not api_key:
+        return jsonify({"error": "VirusTotal API key is not available."}), 500
+
+    try:
+        # Pass the theme colors to the function
+        widget_url = get_virustotal_widget_url(
+            api_key, data["resource"], fg1, bg1, bg2, bd1
+        )
+        return jsonify({"widget_url": widget_url}), 200
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500
+
+
+@strelka.route("/check_vt_api_key", methods=["GET"])
+def check_vt_api_key():
+    """
+    Endpoint to check if the VirusTotal API key is available.
+
+    Returns:
+        A boolean response containing the VirusTotal widget url or an error message.
+    """
+    api_key_exists = bool(os.environ.get("VIRUSTOTAL_API_KEY"))
+    return jsonify({"apiKeyAvailable": api_key_exists}), 200
+
+
 def submissions_to_json(submission: FileSubmission) -> Dict[str, any]:
     """
     Converts the given submission to a dictionary representation that can be