Allow to query for passwords in CowrieSession API - Closes #607

.gitignore

@@ -7,3 +7,4 @@ __pycache__/
 mlmodels/
 # JetBrains IDEs (PyCharm, IntelliJ, etc.)
 .idea/
+.DS_Store

api/views/cowrie_session.py

@@ -9,7 +9,7 @@
 from django.conf import settings
 from django.http import Http404, HttpResponseBadRequest
 from greedybear.consts import GET
-from greedybear.models import IOC, CommandSequence, CowrieSession, Statistics, viewType
+from greedybear.models import 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
@@ -24,13 +24,17 @@
 def cowrie_session_view(request):
     """
     Retrieve Cowrie honeypot session data including command sequences, credentials, and session details.
-    Queries can be performed using either an IP address to find all sessions from that source,
-    or a SHA-256 hash to find sessions containing a specific command sequence.
+    Queries can be performed using an IP address, SHA-256 hash or password.
 
     Args:
         request: The HTTP request object containing query parameters
-        query (str, required): The search term, can be either an IP address or the SHA-256 hash of a command sequence.
-            SHA-256 hashes should match command sequences generated using Python's "\\n".join(sequence) format.
+        query (str, required): The search term, can be:
+            - An IP address to find all sessions from that
+            source
+            - A SHA-256 hash of a command sequence
+            (generated using Python's "\\n".join(sequence) format)
+            - A password string to find all sessions where
+            that password was used
         include_similar (bool, optional): When "true", expands the result to include all sessions that executed
             command sequences belonging to the same cluster(s) as command sequences found in the initial query result.
             Requires CLUSTER_COWRIE_COMMAND_SEQUENCES enabled in configuration. Default: false
@@ -74,19 +78,32 @@ def cowrie_session_view(request):
     if not observable:
         return HttpResponseBadRequest("Missing required 'query' parameter")
 
+    if "<" in observable or ">" in observable or observable.count("}") > observable.count("{"):
+        return HttpResponseBadRequest("Invalid query parameter: contains invalid characters")
+    looks_like_ip = "." in observable or ":" in observable or "/" in observable
+    if looks_like_ip and not is_ip_address(observable):
+        return HttpResponseBadRequest(f"Invalid IP address format: {observable}")
+
+    if len(observable) == 64 and not is_sha256hash(observable):
+        return HttpResponseBadRequest("Invalid hash format: must be 64 hexadecimal characters")
+
     if is_ip_address(observable):
         sessions = CowrieSession.objects.filter(source__name=observable, duration__gt=0).prefetch_related("source", "commands")
         if not sessions.exists():
             raise Http404(f"No information found for IP: {observable}")
 
-    elif is_sha256hash(observable):
+    elif len(observable) == 64 and is_sha256hash(observable):
         try:
             commands = CommandSequence.objects.get(commands_hash=observable.lower())
         except CommandSequence.DoesNotExist as exc:
             raise Http404(f"No command sequences found with hash: {observable}") from exc
         sessions = CowrieSession.objects.filter(commands=commands, duration__gt=0).prefetch_related("source", "commands")
     else:
-        return HttpResponseBadRequest("Query must be a valid IP address or SHA-256 hash")
+        password_pattern = f" | {observable}"
+        sessions = CowrieSession.objects.filter(duration__gt=0, credentials__icontains=password_pattern).prefetch_related("source", "commands")
+
+        if not sessions.exists():
+            raise Http404(f"No sessions found with password: {observable}")
 
     if include_similar:
         commands = set(s.commands for s in sessions if s.commands)

greedybear/migrations/0001_initial.py

@@ -46,9 +46,7 @@ class Migration(migrations.Migration):
                 ("last_seen", models.DateTimeField(default=datetime.datetime.utcnow)),
                 (
                     "days_seen",
-                    django.contrib.postgres.fields.ArrayField(
-                        base_field=models.DateField(), blank=True, size=None
-                    ),
+                    django.contrib.postgres.fields.ArrayField(base_field=models.DateField(), blank=True, size=None),
                 ),
                 ("number_of_days_seen", models.IntegerField(default=1)),
                 ("times_seen", models.IntegerField(default=1)),

greedybear/migrations/0004_alter_id_field.py

@@ -13,22 +13,16 @@ class Migration(migrations.Migration):
         migrations.AlterField(
             model_name="ioc",
             name="id",
-            field=models.BigAutoField(
-                auto_created=True, primary_key=True, serialize=False, verbose_name="ID"
-            ),
+            field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID"),
         ),
         migrations.AlterField(
             model_name="sensors",
             name="id",
-            field=models.BigAutoField(
-                auto_created=True, primary_key=True, serialize=False, verbose_name="ID"
-            ),
+            field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID"),
         ),
         migrations.AlterField(
             model_name="statistics",
             name="id",
-            field=models.BigAutoField(
-                auto_created=True, primary_key=True, serialize=False, verbose_name="ID"
-            ),
+            field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID"),
         ),
     ]

greedybear/migrations/0005_clients.py

@@ -8,9 +8,7 @@ def create_default_clients(apps, schema_editor):
     # version than this migration expects. We use the historical version.
     Client = apps.get_model("durin", "Client")
     # for pygreedybear, custom token_ttl
-    Client.objects.update_or_create(
-        name="pygreedybear", token_ttl=timedelta(weeks=4 * 12 * 10)
-    )
+    Client.objects.update_or_create(name="pygreedybear", token_ttl=timedelta(weeks=4 * 12 * 10))
     # others, default token_ttl
     Client.objects.update_or_create(name="web-browser")
 
@@ -23,6 +21,4 @@ class Migration(migrations.Migration):
         ("durin", "0001_initial"),
     ]
 
-    operations = [
-        migrations.RunPython(create_default_clients, migrations.RunPython.noop)
-    ]
+    operations = [migrations.RunPython(create_default_clients, migrations.RunPython.noop)]

greedybear/migrations/0010_alter_ioc_related_ioc.py

@@ -6,13 +6,13 @@
 class Migration(migrations.Migration):
 
     dependencies = [
-        ('greedybear', '0009_alter_ioc_general_field'),
+        ("greedybear", "0009_alter_ioc_general_field"),
     ]
 
     operations = [
         migrations.AlterField(
-            model_name='ioc',
-            name='related_ioc',
-            field=models.ManyToManyField(blank=True, to='greedybear.ioc'),
+            model_name="ioc",
+            name="related_ioc",
+            field=models.ManyToManyField(blank=True, to="greedybear.ioc"),
         ),
     ]

greedybear/models.py

@@ -3,6 +3,7 @@
 from datetime import datetime
 
 from django.contrib.postgres import fields as pg_fields
+from django.contrib.postgres.indexes import GinIndex
 from django.db import models
 
 
@@ -67,7 +68,12 @@ def __str__(self):
 class CommandSequence(models.Model):
     first_seen = models.DateTimeField(blank=False, default=datetime.now)
     last_seen = models.DateTimeField(blank=False, default=datetime.now)
-    commands = pg_fields.ArrayField(models.CharField(max_length=1024, blank=True), blank=False, null=False, default=list)
+    commands = pg_fields.ArrayField(
+        models.CharField(max_length=1024, blank=True),
+        blank=False,
+        null=False,
+        default=list,
+    )
     commands_hash = models.CharField(max_length=64, unique=True, blank=True, null=True)
     cluster = models.IntegerField(blank=True, null=True)
 
@@ -81,7 +87,12 @@ class CowrieSession(models.Model):
     start_time = models.DateTimeField(blank=True, null=True)
     duration = models.FloatField(blank=True, null=True)
     login_attempt = models.BooleanField(blank=False, null=False, default=False)
-    credentials = pg_fields.ArrayField(models.CharField(max_length=256, blank=True), blank=False, null=False, default=list)
+    credentials = pg_fields.ArrayField(
+        models.CharField(max_length=256, blank=True),
+        blank=False,
+        null=False,
+        default=list,
+    )
     command_execution = models.BooleanField(blank=False, null=False, default=False)
     interaction_count = models.IntegerField(blank=False, null=False, default=0)
     source = models.ForeignKey(IOC, on_delete=models.CASCADE, blank=False, null=False)
@@ -90,6 +101,7 @@ class CowrieSession(models.Model):
     class Meta:
         indexes = [
             models.Index(fields=["source"]),
+            GinIndex(fields=["credentials"], name="greedybear_credentials_gin_idx"),
         ]
 
 

tests/test_views.py

@@ -47,8 +47,14 @@ def test_for_vaild_registered_ip(self):
         self.assertEqual(response.json()["ioc"]["general_honeypot"][1], self.ciscoasa.name)  # FEEDS
         self.assertEqual(response.json()["ioc"]["scanner"], self.ioc.scanner)
         self.assertEqual(response.json()["ioc"]["payload_request"], self.ioc.payload_request)
-        self.assertEqual(response.json()["ioc"]["recurrence_probability"], self.ioc.recurrence_probability)
-        self.assertEqual(response.json()["ioc"]["expected_interactions"], self.ioc.expected_interactions)
+        self.assertEqual(
+            response.json()["ioc"]["recurrence_probability"],
+            self.ioc.recurrence_probability,
+        )
+        self.assertEqual(
+            response.json()["ioc"]["expected_interactions"],
+            self.ioc.expected_interactions,
+        )
 
     def test_for_invalid_authentication(self):
         """Check for a invalid authentication"""
@@ -420,7 +426,7 @@ def setUp(self):
         self.client = APIClient()
         self.client.force_authenticate(user=self.superuser)
 
-    # # # # # Basic IP Query Test # # # # #
+    # Basic IP Query Test
     def test_ip_address_query(self):
         """Test view with a valid IP address query."""
         response = self.client.get("/api/cowrie_session?query=140.246.171.141")
@@ -454,6 +460,34 @@ def test_ip_address_query_with_credentials(self):
         self.assertEqual(len(response.data["credentials"]), 1)
         self.assertEqual(response.data["credentials"][0], "root | root")
 
+    # Password Query Tests
+    def test_password_query(self):
+        """Test view with a valid password query."""
+        response = self.client.get("/api/cowrie_session?query=root&include_credentials=true")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("query", response.data)
+        self.assertIn("credentials", response.data)
+        self.assertEqual(response.data["query"], "root")
+        self.assertEqual(len(response.data["credentials"]), 1)
+        self.assertEqual(response.data["credentials"][0], "root | root")
+
+    def test_password_query_not_found(self):
+        """Test view with a password that doesn't exist."""
+        response = self.client.get("/api/cowrie_session?query=nonexistentpassword")
+        self.assertEqual(response.status_code, 404)
+
+    def test_password_query_with_session_data(self):
+        """Test password query with session data included."""
+        response = self.client.get("/api/cowrie_session?query=root&include_session_data=true")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("query", response.data)
+        self.assertIn("sessions", response.data)
+        self.assertEqual(response.data["query"], "root")
+        self.assertEqual(len(response.data["sessions"]), 1)
+        self.assertIn("time", response.data["sessions"][0])
+        self.assertEqual(response.data["sessions"][0]["source"], "140.246.171.141")
+        self.assertEqual(response.data["sessions"][0]["credentials"][0], "root | root")
+
     def test_ip_address_query_with_sessions(self):
         """Test view with a valid IP address query including session data."""
         response = self.client.get("/api/cowrie_session?query=140.246.171.141&include_session_data=true")
@@ -558,9 +592,9 @@ def test_nonexistent_hash(self):
         self.assertEqual(response.status_code, 404)
 
     def test_hash_wrong_length(self):
-        """Test that hashes with incorrect length are rejected."""
+        """Test that hashes with incorrect length are treated as password queries."""
         response = self.client.get("/api/cowrie_session?query=" + "a" * 32)  # 32 chars instead of 64
-        self.assertEqual(response.status_code, 400)
+        self.assertEqual(response.status_code, 404)
 
     def test_hash_invalid_characters(self):
         """Test that hashes with invalid characters are rejected."""