Expose credential_count in IOC feeds and enable filtering by credential usage . Closes GreedyBear-Project#1294 (GreedyBear-Project#1295)

* added credential_count

Signed-off-by: Drona Raj Gyawali <dronarajgyawali@gmail.com>

* endpoint only for advanc. feed

Signed-off-by: Drona Raj Gyawali <dronarajgyawali@gmail.com>

* added edgecases

---------

Signed-off-by: Drona Raj Gyawali <dronarajgyawali@gmail.com>

Author: drona-gyawali

api/serializers.py

@@ -132,6 +132,8 @@ class FeedsRequestSerializer(serializers.Serializer):
     ioc_type = serializers.ChoiceField(choices=["ip", "domain", "all"])
     max_age = serializers.IntegerField(min_value=1)
     min_days_seen = serializers.IntegerField(min_value=1)
+    min_credential_count = serializers.IntegerField(required=False, min_value=1)
+    max_credential_count = serializers.IntegerField(required=False, min_value=0)
     include_reputation = serializers.ListField(child=serializers.CharField(max_length=120))
     exclude_reputation = serializers.ListField(child=serializers.CharField(max_length=120))
     feed_size = serializers.IntegerField(min_value=1)
@@ -167,6 +169,13 @@ def validate_ordering(self, ordering):
         logger.debug(f"FeedsRequestSerializer - validation ordering: '{ordering}'")
         return ordering_validation(ordering)
 
+    def validate(self, data):
+        min_cc = data.get("min_credential_count")
+        max_cc = data.get("max_credential_count")
+        if min_cc is not None and max_cc is not None and min_cc > max_cc:
+            raise serializers.ValidationError("min_credential_count must be less than or equal to max_credential_count")
+        return data
+
 
 class ASNFeedsOrderingSerializer(FeedsRequestSerializer):
     ALLOWED_ORDERING_FIELDS = frozenset(

api/views/feeds.py

@@ -127,6 +127,8 @@ def feeds_advanced(request):
         attack_type (str): Type of attack to filter. (supported: `scanner`, `payload_request`, `all`; default: `all`)
         max_age (int): Maximum number of days since last occurrence. E.g. an IOC that was last seen 4 days ago is excluded by default. (default: 3)
         min_days_seen (int): Minimum number of days on which an IOC must have been seen. (default: 1)
+        min_credential_count (int, optional): Filter IOCs with at least this many distinct credentials. (default: no filter)
+        max_credential_count (int, optional): Filter IOCs with at most this many distinct credentials. (default: no filter)
         include_reputation (str): `;`-separated list of reputation values to include, e.g. `known attacker` or `known attacker;` to include IOCs without reputation. (default: include all)
         exclude_reputation (str): `;`-separated list of reputation values to exclude, e.g. `mass scanner` or `mass scanner;bot, crawler`. (default: exclude none)
         feed_size (int): Number of IOC items to return. (default: 5000)
@@ -154,6 +156,7 @@ def feeds_advanced(request):
         tag_key=request.query_params.get("tag_key", "").strip(),
         tag_value=request.query_params.get("tag_value", "").strip(),
         include_sensors=True,
+        include_credential_count=True,
     )
     if paginate:
         paginator = CustomPageNumberPagination()

api/views/utils.py

@@ -131,6 +131,8 @@ def __init__(self, query_params: dict):
         self.start_date = query_params.get("start_date")
         self.end_date = query_params.get("end_date")
         self.country_code = query_params.get("country_code")
+        self.min_credential_count = query_params.get("min_credential_count")
+        self.max_credential_count = query_params.get("max_credential_count")
 
     def apply_default_filters(self, query_params):
         if not query_params:
@@ -177,7 +179,15 @@ def get_valid_feed_types() -> frozenset[str]:
 
 
 def get_queryset(
-    request, feed_params, valid_feed_types, is_aggregated=False, serializer_class=FeedsRequestSerializer, tag_key="", tag_value="", include_sensors=False
+    request,
+    feed_params,
+    valid_feed_types,
+    is_aggregated=False,
+    serializer_class=FeedsRequestSerializer,
+    tag_key="",
+    tag_value="",
+    include_sensors=False,
+    include_credential_count=False,
 ):
     """
     Build a queryset to filter IOC data based on the request parameters.
@@ -199,7 +209,8 @@ def get_queryset(
         tag_value (str, optional): Filter IOCs by tag value (case-insensitive substring). Only passed from feeds_advanced.
         include_sensors (bool, optional): If True, annotates sensors_json for each IOC.
             Only passed from authenticated views like feeds_advanced. Default: False.
-
+        include_credential_count (bool, optional): If True, annotates credential Count for each IOC.
+            Only passed from authenticated views like feeds_advanced. Default: False.
     Returns:
         QuerySet: The filtered queryset of IOC data.
     """
@@ -257,6 +268,16 @@ def get_queryset(
 
     iocs = IOC.objects.filter(**query_dict).exclude(ip_reputation__in=feed_params.exclude_reputation).annotate(value=F("name")).distinct()
 
+    # credential count filtering is only available on the advanced feed
+    if include_credential_count:
+        iocs = iocs.annotate(credential_count=Count("credentials", distinct=True))
+        min_credential_count = serializer.validated_data.get("min_credential_count")
+        max_credential_count = serializer.validated_data.get("max_credential_count")
+        if min_credential_count is not None:
+            iocs = iocs.filter(credential_count__gte=min_credential_count)
+        if max_credential_count is not None:
+            iocs = iocs.filter(credential_count__lte=max_credential_count)
+
     # apply feed type filter as union;
     if "all" not in feed_params.feed_types:
         type_filter = Q()
@@ -355,6 +376,7 @@ def feeds_response(request=None, iocs=None, feed_params=None, valid_feed_types=N
                 "first_seen",
                 "last_seen",
                 "attack_count",
+                "credential_count",
                 "interaction_count",
                 "scanner",
                 "payload_request",
@@ -384,11 +406,13 @@ def feeds_response(request=None, iocs=None, feed_params=None, valid_feed_types=N
             if isinstance(iocs, list):
                 has_tags_annotation = bool(iocs) and hasattr(iocs[0], "tags_json")
                 has_sensors_annotation = include_sensors and bool(iocs) and hasattr(iocs[0], "sensors_json")
+                has_credential_count = bool(iocs) and hasattr(iocs[0], "credential_count")
             else:
                 has_tags_annotation = "tags_json" in getattr(iocs, "query", type("", (), {"annotations": {}})()).annotations
                 has_sensors_annotation = include_sensors and "sensors_json" in getattr(iocs, "query", type("", (), {"annotations": {}})()).annotations
-
+                has_credential_count = "credential_count" in getattr(iocs, "query", type("", (), {"annotations": {}})()).annotations
             required_fields = tuple(("tags_json" if f == "tags" else f) for f in required_fields if f != "tags" or has_tags_annotation)
+            required_fields = tuple(f for f in required_fields if f != "credential_count" or has_credential_count)
             if has_sensors_annotation:
                 required_fields = (*required_fields, "sensors_json")
 

tests/api/views/test_feeds_advanced_view.py

@@ -8,7 +8,7 @@
 from rest_framework.test import APIClient
 
 from api.throttles import SharedFeedRateThrottle
-from greedybear.models import IOC, AutonomousSystem, IocType, Sensor, ShareToken
+from greedybear.models import IOC, AutonomousSystem, Credential, IocType, Sensor, ShareToken
 from tests import CustomTestCase
 
 
@@ -115,6 +115,11 @@ def test_200_feed_contains_attacker_country_code(self):
         self.assertIsNotNone(target_ioc)
         self.assertEqual(target_ioc["attacker_country_code"], "NP")
 
+    def test_400_invalid_credential_count_range(self):
+        """min_credential_count greater than max_credential_count returns 400."""
+        response = self.client.get("/api/feeds/advanced/?min_credential_count=10&max_credential_count=5")
+        self.assertEqual(response.status_code, 400)
+
     def test_feeds_advanced_includes_sensors(self):
         """Sensors field appears in feeds_advanced response for authenticated users."""
         sensor = Sensor.objects.create(address="10.0.0.1", label="test-sensor")
@@ -140,6 +145,44 @@ def test_public_feeds_excludes_sensors(self):
         for ioc in iocs:
             self.assertNotIn("sensors", ioc)
 
+    def test_min_credential_count_filter(self):
+        """IOCs with fewer credentials than min_credential_count are excluded."""
+        cred1 = Credential.objects.create(username="admin", password="admin")
+        cred2 = Credential.objects.create(username="root", password="1234")
+        self.ioc.credentials.add(cred1, cred2)
+
+        response = self.client.get("/api/feeds/advanced/?min_credential_count=2")
+        self.assertEqual(response.status_code, 200)
+        iocs = response.json()["iocs"]
+        target_ioc = next((i for i in iocs if i["value"] == self.ioc.name), None)
+        self.assertIsNotNone(target_ioc)
+        self.assertGreaterEqual(target_ioc["credential_count"], 2)
+
+    def test_max_credential_count_filter(self):
+        """IOCs with more credentials than max_credential_count are excluded."""
+        cred1 = Credential.objects.create(username="admin", password="admin")
+        cred2 = Credential.objects.create(username="root", password="1234")
+        self.ioc.credentials.add(cred1, cred2)
+
+        response = self.client.get("/api/feeds/advanced/?max_credential_count=1")
+        self.assertEqual(response.status_code, 200)
+        iocs = response.json()["iocs"]
+        target_ioc = next((i for i in iocs if i["value"] == self.ioc.name), None)
+        # self.ioc has 2 credentials so it should be excluded
+        self.assertIsNone(target_ioc)
+
+    def test_credential_count_in_response(self):
+        """credential_count field is present in JSON response when filtering by credential count."""
+        cred1 = Credential.objects.create(username="admin", password="admin")
+        self.ioc.credentials.add(cred1)
+        response = self.client.get("/api/feeds/advanced/")
+        self.assertEqual(response.status_code, 200)
+        iocs = response.json()["iocs"]
+        target_ioc = next((i for i in iocs if i["value"] == self.ioc.name), None)
+        self.assertIsNotNone(target_ioc)
+        self.assertIn("credential_count", target_ioc)
+        self.assertEqual(target_ioc["credential_count"], 1)
+
 
 class FeedsEnhancementsTestCase(CustomTestCase):
     """Tests for advanced filtering, STIX export, and shareable feeds functionality."""