fixing unknown asn system

Author: liquidsec

bbot/core/helpers/asn.py

@@ -21,8 +21,8 @@ def __init__(self, parent_helper):
 
     # Default record used when no ASN data can be found
     UNKNOWN_ASN = {
-        "asn": "UNKNOWN",
-        "subnet": "0.0.0.0/32",
+        "asn": "0",
+        "subnets": [],
         "name": "unknown",
         "description": "unknown",
         "country": "",

bbot/modules/report/asn.py

@@ -18,11 +18,11 @@ class asn(BaseReportModule):
 
     async def setup(self):
         self.unknown_asn = {
-            "asn": "UNKNOWN",
-            "subnet": "0.0.0.0/32",
-            "name": "unknown",
-            "description": "unknown",
-            "country": "",
+            "asn": "0",
+            "subnets": [],
+            "asn_name": "unknown",
+            "org": "unknown",
+            "country": "unknown",
         }
         # Track ASN data locally instead of relying on cache
         self.asn_data = {}  # ASN number -> ASN record mapping
@@ -78,20 +78,22 @@ async def handle_event(self, event):
                     self.processed_subnets[subnet] = asn_number
 
             emails = asn_record.get("emails", [])
-            asn_event = self.make_event(asn_number, "ASN", parent=event)
-            if asn_event:
-                await self.emit_event(
-                    asn_event,
-                    context=f"{{module}} looked up {event.data} and got {{event.type}}: AS{asn_number} ({asn_name}, {asn_desc}, {asn_country})",
-                )
+            # Don't emit ASN 0 - it's reserved and indicates unknown ASN data
+            if asn_number != "0":
+                asn_event = self.make_event(int(asn_number), "ASN", parent=event)
+                if asn_event:
+                    await self.emit_event(
+                        asn_event,
+                        context=f"{{module}} looked up {event.data} and got {{event.type}}: AS{asn_number} ({asn_name}, {asn_desc}, {asn_country})",
+                    )
 
-            for email in emails:
-                await self.emit_event(
-                    email,
-                    "EMAIL_ADDRESS",
-                    parent=asn_event,
-                    context=f"{{module}} retrieved details for AS{asn_number} and found {{event.type}}: {{event.data}}",
-                )
+                    for email in emails:
+                        await self.emit_event(
+                            email,
+                            "EMAIL_ADDRESS",
+                            parent=asn_event,
+                            context=f"{{module}} retrieved details for AS{asn_number} and found {{event.type}}: {{event.data}}",
+                        )
 
     async def report(self):
         """Generate an ASN summary table based on locally tracked ASN data."""
@@ -105,7 +107,7 @@ async def report(self):
         header = ["ASN", "Subnet Count", "Name", "Description", "Country"]
         table = []
         for asn, data in sorted_asns:
-            number = "AS" + asn if asn != "UNKNOWN" else asn
+            number = "AS" + asn if asn != "0" else asn
             table.append(
                 [
                     number,

bbot/test/test_step_2/module_tests/test_module_asn.py

@@ -39,3 +39,37 @@ def check(self, module_test, events):
 
         # Verify ASN number is a valid integer
         assert any(isinstance(e.data, int) and e.data > 0 for e in asn_events)
+
+
+class TestASNUnknownHandling(ModuleTestBase):
+    """Test ASN module behavior when API returns no data, leading to UNKNOWN_ASN usage."""
+
+    targets = ["8.8.8.8"]  # Use known public IP but mock response to test unknown ASN handling
+    module_name = "asn"
+    modules_overrides = ["asn"]
+    config_overrides = {"scope": {"report_distance": 2}}
+
+    async def setup_after_prep(self, module_test):
+        # Point ASNHelper to local test harness
+        from bbot.core.helpers.asn import ASNHelper
+
+        module_test.monkeypatch.setattr(ASNHelper, "asndb_ip_url", "http://127.0.0.1:8888/v1/ip/")
+
+        # Mock API to return 404 (no ASN data found)
+        expect_args = {"method": "GET", "uri": "/v1/ip/8.8.8.8"}
+        respond_args = {
+            "response_data": "Not Found",
+            "status": 404,
+            "content_type": "text/plain",
+        }
+        module_test.set_expect_requests(expect_args=expect_args, respond_args=respond_args)
+
+    def check(self, module_test, events):
+        # When API returns 404, ASN helper should return UNKNOWN_ASN with string "0"
+        # but NO ASN events should be emitted since ASN 0 is reserved
+        asn_events = [e for e in events if e.type == "ASN"]
+
+        # Should NOT emit any ASN events when ASN data is unknown
+        assert not asn_events, (
+            f"Should not emit any ASN events for unknown ASN data, but found: {[e.data for e in asn_events]}"
+        )