Merge pull request #480 from intelowlproject/develop

1.6.0

Author: mlodic

.env_template

@@ -13,4 +13,4 @@ COMPOSE_FILE=docker/default.yml:docker/local.override.yml
 #COMPOSE_FILE=docker/default.yml:docker/local.override.yml:docker/elasticsearch.yml
 
 # If you want to run a specific version, populate this
-# REACT_APP_INTELOWL_VERSION="1.5.2"
+# REACT_APP_INTELOWL_VERSION="1.6.0"

api/urls.py

@@ -1,6 +1,6 @@
 # This file is a part of GreedyBear https://github.com/honeynet/GreedyBear
 # See the file 'LICENSE' for copying permission.
-from api.views import StatisticsViewSet, enrichment_view, feeds, feeds_advanced, feeds_pagination, general_honeypot_list
+from api.views import StatisticsViewSet, command_sequence_view, enrichment_view, feeds, feeds_advanced, feeds_pagination, general_honeypot_list
 from django.urls import include, path
 from rest_framework import routers
 
@@ -14,6 +14,7 @@
     path("feeds/advanced/", feeds_advanced),
     path("feeds/<str:feed_type>/<str:attack_type>/<str:prioritize>.<str:format_>", feeds),
     path("enrichment", enrichment_view),
+    path("command_sequence", command_sequence_view),
     path("general_honeypot", general_honeypot_list),
     # router viewsets
     path("", include(router.urls)),

api/views/__init__.py

@@ -1,3 +1,4 @@
+from api.views.command_sequence import *
 from api.views.enrichment import *
 from api.views.feeds import *
 from api.views.general_honeypot import *

api/views/command_sequence.py

@@ -0,0 +1,97 @@
+# This file is a part of GreedyBear https://github.com/honeynet/GreedyBear
+# See the file 'LICENSE' for copying permission.
+import logging
+
+from api.views.utils import is_ip_address, is_sha256hash
+from certego_saas.apps.auth.backend import CookieTokenAuthentication
+from django.http import Http404, HttpResponseBadRequest
+from greedybear.consts import FEEDS_LICENSE, GET
+from greedybear.models import IOC, CommandSequence, CowrieSession, Statistics, viewType
+from rest_framework import status
+from rest_framework.decorators import api_view, authentication_classes, permission_classes
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.response import Response
+
+logger = logging.getLogger(__name__)
+
+
+@api_view([GET])
+@authentication_classes([CookieTokenAuthentication])
+@permission_classes([IsAuthenticated])
+def command_sequence_view(request):
+    """
+    View function that handles command sequence queries based on IP addresses or SHA-256 hashes.
+
+    Retrieves and returns command sequences and related IOCs based on the query parameter.
+    If IP address is given, returns all command sequences executed from this IP.
+    If SHA-256 hash is given, returns details about the specific command sequence.
+    Can include similar command sequences if requested.
+
+    Args:
+        request: The HTTP request object containing query parameters
+        query (str): The search term, can be either an IP address or a SHA-256 hash.
+        include_similar (bool): When parameter is present, returns related command sequences based on clustering.
+
+    Returns:
+        Response object with command sequence data or an error response
+
+    Raises:
+        Http404: If the requested resource is not found
+    """
+    observable = request.query_params.get("query")
+    include_similar = request.query_params.get("include_similar") is not None
+    logger.info(f"Command Sequence view requested by {request.user} for {observable}")
+    source_ip = str(request.META["REMOTE_ADDR"])
+    request_source = Statistics(source=source_ip, view=viewType.COMMAND_SEQUENCE_VIEW.value)
+    request_source.save()
+
+    if not observable:
+        return HttpResponseBadRequest("Missing required 'query' parameter")
+
+    if is_ip_address(observable):
+        sessions = CowrieSession.objects.filter(source__name=observable, start_time__isnull=False, commands__isnull=False)
+        sequences = set(s.commands for s in sessions)
+        seqs = [
+            {
+                "time": s.start_time,
+                "command_sequence": "\n".join(s.commands.commands),
+                "command_sequence_hash": s.commands.commands_hash,
+            }
+            for s in sessions
+        ]
+        related_iocs = IOC.objects.filter(cowriesession__commands__in=sequences).distinct().only("name")
+        if include_similar:
+            related_clusters = set(s.cluster for s in sequences if s.cluster is not None)
+            related_iocs = IOC.objects.filter(cowriesession__commands__cluster__in=related_clusters).distinct().only("name")
+        if not seqs:
+            raise Http404(f"No command sequences found for IP: {observable}")
+        data = {
+            "license": FEEDS_LICENSE,
+            "executed_commands": seqs,
+            "executed_by": sorted([ioc.name for ioc in related_iocs]),
+        }
+        return Response(data, status=status.HTTP_200_OK)
+
+    if is_sha256hash(observable):
+        try:
+            seq = CommandSequence.objects.get(commands_hash=observable)
+            seqs = CommandSequence.objects.filter(cluster=seq.cluster) if include_similar and seq.cluster is not None else [seq]
+            commands = ["\n".join(seq.commands) for seq in seqs]
+            sessions = CowrieSession.objects.filter(commands__in=seqs, start_time__isnull=False)
+            iocs = [
+                {
+                    "time": s.start_time,
+                    "ip": s.source.name,
+                }
+                for s in sessions
+            ]
+            data = {
+                "license": FEEDS_LICENSE,
+                "commands": commands,
+                "iocs": sorted(iocs, key=lambda d: d["time"], reverse=True),
+            }
+            return Response(data, status=status.HTTP_200_OK)
+        except CommandSequence.DoesNotExist as exc:
+            raise Http404(f"No command sequences found with hash: {observable}") from exc
+
+    return HttpResponseBadRequest("Query must be a valid IP address or SHA-256 hash")

api/views/utils.py

@@ -2,7 +2,9 @@
 # See the file 'LICENSE' for copying permission.
 import csv
 import logging
+import re
 from datetime import datetime, timedelta
+from ipaddress import ip_address
 
 from api.serializers import FeedsRequestSerializer, FeedsResponseSerializer
 from django.contrib.postgres.aggregates import ArrayAgg
@@ -287,3 +289,40 @@ def feeds_response(iocs, feed_params, valid_feed_types, dict_only=False, verbose
                 return Response(resp_data, status=status.HTTP_200_OK)
         case _:
             return HttpResponseBadRequest()
+
+
+def is_ip_address(string: str) -> bool:
+    """
+    Validate if a string is a valid IP address (IPv4 or IPv6).
+
+    Uses the ipaddress module to perform validation. This function properly
+    handles both IPv4 addresses and IPv6 addresses.
+
+    Args:
+        string: The string to validate as an IP address
+
+    Returns:
+        bool: True if the string is a valid IP address, False otherwise
+    """
+    try:
+        ip_address(string)
+    except ValueError:
+        return False
+    return True
+
+
+def is_sha256hash(string: str) -> bool:
+    """
+    Validate if a string is a valid SHA-256 hash.
+
+    A SHA-256 hash is a string of exactly 64 hexadecimal characters
+    (0-9, a-f, A-F). This function checks if the input string matches
+    this pattern using a regular expression.
+
+    Args:
+        string: The string to validate as a SHA-256 hash
+
+    Returns:
+        bool: True if the string is a valid SHA-256 hash, False otherwise
+    """
+    return bool(re.fullmatch(r"^[A-Fa-f0-9]{64}$", string))

docker/.version

@@ -1 +1 @@
-REACT_APP_GREEDYBEAR_VERSION="1.5.2"
+REACT_APP_GREEDYBEAR_VERSION="1.6.0"

greedybear/models.py

@@ -9,6 +9,7 @@
 class viewType(models.TextChoices):
     FEEDS_VIEW = "feeds"
     ENRICHMENT_VIEW = "enrichment"
+    COMMAND_SEQUENCE_VIEW = "command sequence"
 
 
 class iocType(models.TextChoices):

tests/__init__.py

@@ -42,11 +42,12 @@ def setUpTestData(cls):
         cls.ioc.save()
 
         cls.cmd_seq = ["cd foo", "ls -la"]
+        cls.hash = sha256("\n".join(cls.cmd_seq).encode()).hexdigest()
         cls.command_sequence = CommandSequence.objects.create(
             first_seen=cls.current_time,
             last_seen=cls.current_time,
             commands=cls.cmd_seq,
-            commands_hash=sha256("\n".join(cls.cmd_seq).encode()).hexdigest(),
+            commands_hash=cls.hash,
             cluster=11,
         )
         cls.command_sequence.save()

tests/test_models.py

@@ -1,5 +1,3 @@
-from hashlib import sha256
-
 from greedybear.models import Statistics, iocType, viewType
 
 from . import CustomTestCase
@@ -34,7 +32,7 @@ def test_command_sequence_model(self):
         self.assertEqual(self.command_sequence.first_seen, self.current_time)
         self.assertEqual(self.command_sequence.last_seen, self.current_time)
         self.assertEqual(self.command_sequence.commands, self.cmd_seq)
-        self.assertEqual(self.command_sequence.commands_hash, sha256("\n".join(self.cmd_seq).encode()).hexdigest())
+        self.assertEqual(self.command_sequence.commands_hash, self.hash)
         self.assertEqual(self.command_sequence.cluster, 11)
 
     def test_cowrie_session_model(self):

tests/test_views.py

@@ -1,3 +1,4 @@
+from api.views.utils import is_ip_address, is_sha256hash
 from greedybear.consts import FEEDS_LICENSE
 from greedybear.models import GeneralHoneypot, Statistics, viewType
 from rest_framework.test import APIClient
@@ -156,6 +157,7 @@ def setUpClass(self):
 
     @classmethod
     def tearDownClass(self):
+        super(StatisticsViewTestCase, self).tearDownClass()
         Statistics.objects.all().delete()
 
     def test_200_feeds_sources(self):
@@ -209,3 +211,96 @@ def test_200_active_general_honeypots(self):
         response = self.client.get("/api/general_honeypot?onlyActive=true")
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response.json(), ["Heralding", "Ciscoasa"])
+
+
+class CommandSequenceViewTestCase(CustomTestCase):
+    """Test cases for the command_sequence_view."""
+
+    def setUp(self):
+        # setup client
+        self.client = APIClient()
+        self.client.force_authenticate(user=self.superuser)
+
+    def test_missing_query_parameter(self):
+        """Test that view returns BadRequest when query parameter is missing."""
+        response = self.client.get("/api/command_sequence")
+        self.assertEqual(response.status_code, 400)
+
+    def test_invalid_query_parameter(self):
+        """Test that view returns BadRequest when query parameter is invalid."""
+        response = self.client.get("/api/command_sequence?query=invalid-input}")
+        self.assertEqual(response.status_code, 400)
+
+    def test_ip_address_query(self):
+        """Test view with a valid IP address query."""
+        response = self.client.get("/api/command_sequence?query=140.246.171.141")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("executed_commands", response.data)
+        self.assertIn("executed_by", response.data)
+
+    def test_ip_address_query_with_similar(self):
+        """Test view with a valid IP address query including similar sequences."""
+        response = self.client.get("/api/command_sequence?query=140.246.171.141&include_similar")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("executed_commands", response.data)
+        self.assertIn("executed_by", response.data)
+
+    def test_nonexistent_ip_address(self):
+        """Test that view returns 404 for IP with no sequences."""
+        response = self.client.get("/api/command_sequence?query=10.0.0.1")
+        self.assertEqual(response.status_code, 404)
+
+    def test_hash_query(self):
+        """Test view with a valid hash query."""
+        response = self.client.get(f"/api/command_sequence?query={self.hash}")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("commands", response.data)
+        self.assertIn("iocs", response.data)
+
+    def test_hash_query_with_similar(self):
+        """Test view with a valid hash query including similar sequences."""
+        response = self.client.get(f"/api/command_sequence?query={self.hash}&include_similar")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("commands", response.data)
+        self.assertIn("iocs", response.data)
+
+    def test_nonexistent_hash(self):
+        """Test that view returns 404 for nonexistent hash."""
+        response = self.client.get(f"/api/command_sequence?query={'f' * 64}")
+        self.assertEqual(response.status_code, 404)
+
+
+class ValidationHelpersTestCase(CustomTestCase):
+    """Test cases for the validation helper functions."""
+
+    def test_is_ip_address_valid_ipv4(self):
+        """Test that is_ip_address returns True for valid IPv4 addresses."""
+        self.assertTrue(is_ip_address("192.168.1.1"))
+        self.assertTrue(is_ip_address("10.0.0.1"))
+        self.assertTrue(is_ip_address("127.0.0.1"))
+
+    def test_is_ip_address_valid_ipv6(self):
+        """Test that is_ip_address returns True for valid IPv6 addresses."""
+        self.assertTrue(is_ip_address("::1"))
+        self.assertTrue(is_ip_address("2001:db8::1"))
+        self.assertTrue(is_ip_address("fe80::1ff:fe23:4567:890a"))
+
+    def test_is_ip_address_invalid(self):
+        """Test that is_ip_address returns False for invalid IP addresses."""
+        self.assertFalse(is_ip_address("not-an-ip"))
+        self.assertFalse(is_ip_address("256.256.256.256"))
+        self.assertFalse(is_ip_address("192.168.0"))
+        self.assertFalse(is_ip_address("2001:xyz::1"))
+
+    def test_is_sha256hash_valid(self):
+        """Test that is_sha256hash returns True for valid SHA-256 hashes."""
+        self.assertTrue(is_sha256hash("a" * 64))
+        self.assertTrue(is_sha256hash("1234567890abcdef" * 4))
+        self.assertTrue(is_sha256hash("A" * 64))
+
+    def test_is_sha256hash_invalid(self):
+        """Test that is_sha256hash returns False for invalid SHA-256 hashes."""
+        self.assertFalse(is_sha256hash("a" * 63))  # Too short
+        self.assertFalse(is_sha256hash("a" * 65))  # Too long
+        self.assertFalse(is_sha256hash("z" * 64))  # Invalid chars
+        self.assertFalse(is_sha256hash("not-a-hash"))