Merge Vulnerabilities and Findings, add Confidence and Severity to Findings

bbot/core/event/base.py

@@ -152,7 +152,6 @@ class BaseEvent:
         "_discovery_context_regex",
         "_stats_recorded",
         "_internal",
-        "_confidence",
         "_dummy",
         "_module",
         # DNS-related attributes
@@ -181,7 +180,6 @@ def __init__(
         module=None,
         scan=None,
         tags=None,
-        confidence=100,
         timestamp=None,
         _dummy=False,
         _internal=None,
@@ -199,7 +197,6 @@ def __init__(
             module (str, optional): Module that discovered the event. Defaults to None.
             scan (Scan, optional): BBOT Scan object. Required unless _dummy is True. Defaults to None.
             tags (list of str, optional): Descriptive tags for the event. Defaults to None.
-            confidence (int, optional): Confidence level for the event, on a scale of 1-100. Defaults to 100.
             timestamp (datetime, optional): Time of event discovery. Defaults to current UTC time.
             _dummy (bool, optional): If True, disables certain data validations. Defaults to False.
             _internal (Any, optional): If specified, makes the event internal. Defaults to None.
@@ -247,7 +244,6 @@ def __init__(
             except AttributeError:
                 self.timestamp = datetime.datetime.utcnow()
 
-        self.confidence = int(confidence)
         self._internal = False
 
         # self.scan holds the instantiated scan object (for helpers, etc.)
@@ -287,27 +283,6 @@ def __init__(
     def data(self):
         return self._data
 
-    @property
-    def confidence(self):
-        return self._confidence
-
-    @confidence.setter
-    def confidence(self, confidence):
-        self._confidence = min(100, max(1, int(confidence)))
-
-    @property
-    def cumulative_confidence(self):
-        """
-        Considers the confidence of parent events. This is useful for filtering out speculative/unreliable events.
-
-        E.g. an event with a confidence of 50 whose parent is also 50 would have a cumulative confidence of 25.
-
-        A confidence of 100 will reset the cumulative confidence to 100.
-        """
-        if self._confidence == 100 or self.parent is None or self.parent is self:
-            return self._confidence
-        return int(self._confidence * self.parent.cumulative_confidence / 100)
-
     @property
     def resolved_hosts(self):
         if is_ip(self.host):
@@ -1547,54 +1522,55 @@ def redirect_location(self):
         return location
 
 
-class VULNERABILITY(ClosestHostEvent):
+class FINDING(ClosestHostEvent):
     _always_emit = True
     _quick_emit = True
     severity_colors = {
         "CRITICAL": "🟪",
         "HIGH": "🟥",
         "MEDIUM": "🟧",
         "LOW": "🟨",
-        "INFO": "🟦",
-        "UNKNOWN": "⬜",
+        "INFORMATIONAL": "⬜",
+    }
+
+    confidence_colors = {
+        "CONFIRMED": "🟣",
+        "HIGH": "🔴",
+        "MODERATE": "🟠",
+        "LOW": "🟡",
+        "UNKNOWN": "⚪",
     }
 
     def sanitize_data(self, data):
-        data = super().sanitize_data(data)
-        self.add_tag(data["severity"].lower())
+        self.add_tag(f"severity-{data['severity'].lower()}")
+        self.add_tag(f"confidence-{data['confidence'].lower()}")
         return data
 
     class _data_validator(BaseModel):
         host: Optional[str] = None
         severity: str
         name: str
         description: str
+        confidence: str
         url: Optional[str] = None
         path: Optional[str] = None
         cves: Optional[list[str]] = None
         _validate_url = field_validator("url")(validators.validate_url)
         _validate_host = field_validator("host")(validators.validate_host)
         _validate_severity = field_validator("severity")(validators.validate_severity)
+        _validate_confidence = field_validator("confidence")(validators.validate_confidence)
 
     def _pretty_string(self):
-        return f"[{self.data['severity']}] {self.data['description']}"
-
+        severity = self.data["severity"]
+        confidence = self.data["confidence"]
+        description = self.data["description"]
 
-class FINDING(ClosestHostEvent):
-    _always_emit = True
-    _quick_emit = True
-
-    class _data_validator(BaseModel):
-        host: Optional[str] = None
-        name: str
-        description: str
-        url: Optional[str] = None
-        path: Optional[str] = None
-        _validate_url = field_validator("url")(validators.validate_url)
-        _validate_host = field_validator("host")(validators.validate_host)
-
-    def _pretty_string(self):
-        return self.data["description"]
+        # Add bold formatting for CONFIRMED confidence
+        if confidence == "CONFIRMED":
+            confidence_str = f"[\033[1m{confidence}\033[0m]"
+        else:
+            confidence_str = f"[{confidence}]"
+        return f"Severity: [{severity}] Confidence: {confidence_str} {description}"
 
 
 class TECHNOLOGY(DictHostEvent):
@@ -1772,7 +1748,6 @@ def make_event(
     module=None,
     scan=None,
     tags=None,
-    confidence=100,
     dummy=False,
     internal=None,
 ):
@@ -1792,7 +1767,6 @@ def make_event(
         scan (Scan, optional): BBOT Scan object associated with the event.
         scans (List[Scan], optional): Multiple BBOT Scan objects, primarily used for unserialization.
         tags (Union[str, List[str]], optional): Descriptive tags for the event, as a list or a single string.
-        confidence (int, optional): Confidence level for the event, on a scale of 1-100. Defaults to 100.
         dummy (bool, optional): Disables data validations if set to True. Defaults to False.
         internal (Any, optional): Makes the event internal if set to True. Defaults to None.
 
@@ -1896,7 +1870,6 @@ def make_event(
             module=module,
             scan=scan,
             tags=tags,
-            confidence=confidence,
             _dummy=dummy,
             _internal=internal,
         )
@@ -1929,7 +1902,6 @@ def event_from_json(j):
         kwargs = {
             "event_type": event_type,
             "tags": j.get("tags", []),
-            "confidence": j.get("confidence", 100),
             "context": j.get("discovery_context", None),
             "dummy": True,
         }

bbot/core/helpers/validators.py

@@ -132,11 +132,19 @@ def validate_host(host: Union[str, ipaddress.IPv4Address, ipaddress.IPv6Address]
 @validator
 def validate_severity(severity: str):
     severity = str(severity).strip().upper()
-    if severity not in ("UNKNOWN", "INFO", "LOW", "MEDIUM", "HIGH", "CRITICAL"):
+    if severity not in ("INFORMATIONAL", "LOW", "MEDIUM", "HIGH", "CRITICAL"):
         raise ValueError(f"Invalid severity: {severity}")
     return severity
 
 
+@validator
+def validate_confidence(confidence: str):
+    confidence = str(confidence).strip().upper()
+    if confidence not in ("UNKNOWN", "LOW", "MODERATE", "HIGH", "CONFIRMED"):
+        raise ValueError(f"Invalid confidence: {confidence}")
+    return confidence
+
+
 @validator
 def validate_email(email: str):
     email = smart_encode_punycode(str(email).strip().lower())

bbot/modules/ajaxpro.py

@@ -10,7 +10,7 @@ class ajaxpro(BaseModule):
 
     ajaxpro_regex = re.compile(r'<script.+src="([\/a-zA-Z0-9\._]+,[a-zA-Z0-9\._]+\.ashx)"')
     watched_events = ["HTTP_RESPONSE", "URL"]
-    produced_events = ["VULNERABILITY", "TECHNOLOGY"]
+    produced_events = ["FINDING", "TECHNOLOGY"]
     flags = ["active", "safe", "web-thorough"]
     meta = {
         "description": "Check for potentially vulnerable Ajaxpro instances",
@@ -40,7 +40,7 @@ async def check_http_response_event(self, event):
         if resp_body:
             match = await self.helpers.re.search(self.ajaxpro_regex, resp_body)
             if match:
-                ajaxpro_path = match.group(0)
+                ajaxpro_path = match.group(1)
                 await self.emit_technology(event, ajaxpro_path)
                 await self.confirm_exploitability(ajaxpro_path, event)
 
@@ -71,17 +71,21 @@ async def confirm_exploitability(self, detection_url, event):
 
         probe_response = await self.helpers.request(full_url, method="POST", headers=headers, json=payload)
         if probe_response:
-            if "AjaxPro.Services.ICartService" and "MissingMethodException" in probe_response.text:
+            if (
+                "AjaxPro.Services.ICartService" in probe_response.text
+                and "MissingMethodException" in probe_response.text
+            ):
                 await self.emit_event(
                     {
                         "host": str(event.host),
                         "name": "Ajaxpro Deserialization RCE (CVE-2021-23758)",
                         "cves": ["CVE-2021-23758"],
                         "severity": "CRITICAL",
+                        "confidence": "HIGH",
                         "url": event.data if event.type == "URL" else event.data["url"],
                         "description": f"Ajaxpro Deserialization RCE (CVE-2021-23758) Trigger: [{full_url}]",
                     },
-                    "VULNERABILITY",
+                    "FINDING",
                     event,
                     context=f"{self.meta['description']} discovered Ajaxpro instance ({event.type}) at {detection_url}",
                 )

bbot/modules/aspnet_bin_exposure.py

@@ -3,7 +3,7 @@
 
 class aspnet_bin_exposure(BaseModule):
     watched_events = ["URL"]
-    produced_events = ["VULNERABILITY"]
+    produced_events = ["FINDING"]
     flags = ["active", "safe", "web-thorough"]
     meta = {
         "description": "Check for ASP.NET Security Feature Bypasses (CVE-2023-36899 and CVE-2023-36560)",
@@ -65,11 +65,12 @@ async def handle_event(self, event):
                                     {
                                         "name": "IIS Bin Directory DLL Exposure",
                                         "severity": "HIGH",
+                                        "confidence": "HIGH",
                                         "host": str(event.host),
                                         "url": normalized_url,
                                         "description": description,
                                     },
-                                    "VULNERABILITY",
+                                    "FINDING",
                                     event,
                                     context="{module} detected IIS Bin Directory DLL Exposure vulnerability",
                                 )

bbot/modules/baddns.py

@@ -8,7 +8,7 @@
 
 class baddns(BaseModule):
     watched_events = ["DNS_NAME", "DNS_NAME_UNRESOLVED"]
-    produced_events = ["FINDING", "VULNERABILITY"]
+    produced_events = ["FINDING"]
     flags = ["active", "safe", "web-basic", "baddns", "cloud-enum", "subdomain-hijack"]
     meta = {
         "description": "Check hosts for domain/subdomain takeovers",
@@ -93,12 +93,13 @@ async def handle_event(self, event):
                             data = {
                                 "severity": "MEDIUM",
                                 "name": f"BadDNS {r_dict['signature']}",
+                                "confidence": "HIGH",
                                 "description": f"{r_dict['description']}. Confidence: [{confidence}] Signature: [{r_dict['signature']}] Indicator: [{r_dict['indicator']}] Trigger: [{r_dict['trigger']}] baddns Module: [{r_dict['module']}]",
                                 "host": str(event.host),
                             }
                             await self.emit_event(
                                 data,
-                                "VULNERABILITY",
+                                "FINDING",
                                 event,
                                 tags=[f"baddns-{module_instance.name.lower()}"],
                                 context=f'{{module}}\'s "{r_dict["module"]}" module found {{event.type}}: {r_dict["description"]}',
@@ -110,6 +111,8 @@ async def handle_event(self, event):
                                     "name": f"BadDNS {r_dict['signature']}",
                                     "description": f"{r_dict['description']} Confidence: [{confidence}] Signature: [{r_dict['signature']}] Indicator: [{r_dict['indicator']}] Trigger: [{r_dict['trigger']}] baddns Module: [{r_dict['module']}]",
                                     "host": str(event.host),
+                                    "severity": "MEDIUM",
+                                    "confidence": "LOW",
                                 }
                                 await self.emit_event(
                                     data,

bbot/modules/baddns_direct.py

@@ -7,7 +7,7 @@
 
 class baddns_direct(BaseModule):
     watched_events = ["URL", "STORAGE_BUCKET"]
-    produced_events = ["FINDING", "VULNERABILITY"]
+    produced_events = ["FINDING"]
     flags = ["active", "safe", "subdomain-enum", "baddns", "cloud-enum"]
     meta = {
         "description": "Check for unusual subdomain / service takeover edge cases that require direct detection",
@@ -60,6 +60,8 @@ async def handle_event(self, event):
                         "name": f"BadDNS {r_dict['signature']}",
                         "description": f"Possible [{r_dict['signature']}] via direct BadDNS analysis. Indicator: [{r_dict['indicator']}] Trigger: [{r_dict['trigger']}] baddns Module: [{r_dict['module']}]",
                         "host": str(event.host),
+                        "severity": "HIGH",
+                        "confidence": "MODERATE",
                     }
 
                     await self.emit_event(

bbot/modules/baddns_zone.py

@@ -3,7 +3,7 @@
 
 class baddns_zone(baddns_module):
     watched_events = ["DNS_NAME"]
-    produced_events = ["FINDING", "VULNERABILITY"]
+    produced_events = ["FINDING"]
     flags = ["active", "safe", "subdomain-enum", "baddns", "cloud-enum"]
     meta = {
         "description": "Check hosts for DNS zone transfers and NSEC walks",

bbot/modules/badsecrets.py

@@ -6,7 +6,7 @@
 
 class badsecrets(BaseModule):
     watched_events = ["HTTP_RESPONSE"]
-    produced_events = ["FINDING", "VULNERABILITY", "TECHNOLOGY"]
+    produced_events = ["FINDING", "TECHNOLOGY"]
     flags = ["active", "safe", "web-basic"]
     meta = {
         "description": "Library for detecting known or weak secrets across many web frameworks",
@@ -73,10 +73,11 @@ async def handle_event(self, event):
                             "description": f"Known Secret Found. Secret Type: [{r['description']['secret']}] Secret: [{r['secret']}] Product Type: [{r['description']['product']}] Product: [{self.helpers.truncate_string(r['product'], 2000)}] Detecting Module: [{r['detecting_module']}] Details: [{r['details']}]",
                             "url": event.data["url"],
                             "host": str(event.host),
+                            "confidence": "CONFIRMED",
                         }
                         await self.emit_event(
                             data,
-                            "VULNERABILITY",
+                            "FINDING",
                             event,
                             context=f'{{module}}\'s "{r["detecting_module"]}" module found known {r["description"]["product"]} secret ({{event.type}}): "{r["secret"]}"',
                         )
@@ -96,6 +97,8 @@ async def handle_event(self, event):
                                 "description": f"Cryptographic Product identified. Product Type: [{r['description']['product']}] Product: [{self.helpers.truncate_string(r['product'], 2000)}] Detecting Module: [{r['detecting_module']}]",
                                 "url": event.data["url"],
                                 "host": str(event.host),
+                                "severity": "INFORMATIONAL",
+                                "confidence": "CONFIRMED",
                             }
                             await self.emit_event(
                                 data,

bbot/modules/bypass403.py

@@ -145,6 +145,8 @@ async def handle_event(self, event):
                     "description": f"403 Bypass MULTIPLE SIGNATURES (exceeded threshold {str(collapse_threshold)})",
                     "host": str(event.host),
                     "url": event.data,
+                    "severity": "INFORMATIONAL",
+                    "confidence": "LOW",
                 },
                 "FINDING",
                 parent=event,
@@ -158,6 +160,8 @@ async def handle_event(self, event):
                         "description": description,
                         "host": str(event.host),
                         "url": event.data,
+                        "severity": "MEDIUM",
+                        "confidence": "LOW",
                     },
                     "FINDING",
                     parent=event,