BBOT 3.0 - blazed_elijah

bbot/__init__.py

@@ -1,6 +1,2 @@
 # version placeholder (replaced by poetry-dynamic-versioning)
 __version__ = "v0.0.0"
-
-from .scanner import Scanner, Preset
-
-__all__ = ["Scanner", "Preset"]

bbot/constants.py

@@ -0,0 +1,72 @@
+SCAN_STATUS_QUEUED = 0
+SCAN_STATUS_NOT_STARTED = 1
+SCAN_STATUS_STARTING = 2
+SCAN_STATUS_RUNNING = 3
+SCAN_STATUS_FINISHING = 4
+SCAN_STATUS_ABORTING = 5
+SCAN_STATUS_FINISHED = 6
+SCAN_STATUS_FAILED = 7
+SCAN_STATUS_ABORTED = 8
+
+
+SCAN_STATUSES = {
+    "QUEUED": SCAN_STATUS_QUEUED,
+    "NOT_STARTED": SCAN_STATUS_NOT_STARTED,
+    "STARTING": SCAN_STATUS_STARTING,
+    "RUNNING": SCAN_STATUS_RUNNING,
+    "FINISHING": SCAN_STATUS_FINISHING,
+    "ABORTING": SCAN_STATUS_ABORTING,
+    "FINISHED": SCAN_STATUS_FINISHED,
+    "FAILED": SCAN_STATUS_FAILED,
+    "ABORTED": SCAN_STATUS_ABORTED,
+}
+
+SCAN_STATUS_CODES = {v: k for k, v in SCAN_STATUSES.items()}
+
+
+def is_valid_scan_status(status):
+    """
+    Check if a status is a valid scan status
+    """
+    return status in SCAN_STATUSES
+
+
+def is_valid_scan_status_code(status):
+    """
+    Check if a status is a valid scan status code
+    """
+    return status in SCAN_STATUS_CODES
+
+
+def get_scan_status_name(status):
+    """
+    Convert a numeric scan status code to a string status name
+    """
+    try:
+        if isinstance(status, str):
+            if not is_valid_scan_status(status):
+                raise ValueError(f"Invalid scan status: {status}")
+            return status
+        elif isinstance(status, int):
+            return SCAN_STATUS_CODES[status]
+        else:
+            raise ValueError(f"Invalid scan status: {status} (must be int or str)")
+    except KeyError:
+        raise ValueError(f"Invalid scan status: {status}")
+
+
+def get_scan_status_code(status):
+    """
+    Convert a scan status string to a numeric status code
+    """
+    try:
+        if isinstance(status, int):
+            if not is_valid_scan_status_code(status):
+                raise ValueError(f"Invalid scan status code: {status}")
+            return status
+        elif isinstance(status, str):
+            return SCAN_STATUSES[status]
+        else:
+            raise ValueError(f"Invalid scan status: {status} (must be int or str)")
+    except KeyError:
+        raise ValueError(f"Invalid scan status: {status}")

bbot/core/config/logger.py

@@ -2,6 +2,7 @@
 import sys
 import atexit
 import logging
+import threading
 from copy import copy
 import multiprocessing
 import logging.handlers
@@ -93,7 +94,10 @@ def cleanup_logging(self):
 
         # Stop queue listener
         with suppress(Exception):
-            self.listener.stop()
+            stop_thread = threading.Thread(target=self.listener.stop)
+            stop_thread.daemon = True
+            stop_thread.start()
+            stop_thread.join()
 
     def setup_queue_handler(self, logging_queue=None, log_level=logging.DEBUG):
         if logging_queue is None:

bbot/core/event/base.py

@@ -11,6 +11,7 @@
 
 from pathlib import Path
 from typing import Optional
+from zoneinfo import ZoneInfo
 from copy import copy, deepcopy
 from contextlib import suppress
 from radixtarget import RadixTarget
@@ -40,6 +41,7 @@
     validators,
     get_file_extension,
 )
+from bbot.models.helpers import utc_datetime_validator
 from bbot.core.helpers.web.envelopes import BaseEnvelope
 
 
@@ -401,14 +403,15 @@ def host_filterable(self):
     @property
     def port(self):
         self.host
+        if self._port:
+            return self._port
         if getattr(self, "parsed_url", None):
             if self.parsed_url.port is not None:
                 return self.parsed_url.port
             elif self.parsed_url.scheme == "https":
                 return 443
             elif self.parsed_url.scheme == "http":
                 return 80
-        return self._port
 
     @property
     def netloc(self):
@@ -811,7 +814,7 @@ def __contains__(self, other):
             return bool(radixtarget.search(other.host))
         return False
 
-    def json(self, mode="json", siem_friendly=False):
+    def json(self, mode="json"):
         """
         Serializes the event object to a JSON-compatible dictionary.
 
@@ -820,7 +823,6 @@ def json(self, mode="json", siem_friendly=False):
 
         Parameters:
             mode (str): Specifies the data serialization mode. Default is "json". Other options include "graph", "human", and "id".
-            siem_friendly (bool): Whether to format the JSON in a way that's friendly to SIEM ingestion by Elastic, Splunk, etc. This ensures the value of "data" is always the same type (a dictionary).
 
         Returns:
             dict: JSON-serializable dictionary representation of the event object.
@@ -837,10 +839,12 @@ def json(self, mode="json", siem_friendly=False):
             data = data_attr
         else:
             data = smart_decode(self.data)
-        if siem_friendly:
-            j["data"] = {self.type: data}
-        else:
+        if isinstance(data, str):
             j["data"] = data
+        elif isinstance(data, dict):
+            j["data_json"] = data
+        else:
+            raise ValueError(f"Invalid data type: {type(data)}")
         # host, dns children
         if self.host:
             j["host"] = str(self.host)
@@ -858,7 +862,7 @@ def json(self, mode="json", siem_friendly=False):
         if self.scan:
             j["scan"] = self.scan.id
         # timestamp
-        j["timestamp"] = self.timestamp.isoformat()
+        j["timestamp"] = utc_datetime_validator(self.timestamp).timestamp()
         # parent event
         parent_id = self.parent_id
         if parent_id:
@@ -867,8 +871,7 @@ def json(self, mode="json", siem_friendly=False):
         if parent_uuid:
             j["parent_uuid"] = parent_uuid
         # tags
-        if self.tags:
-            j.update({"tags": list(self.tags)})
+        j.update({"tags": sorted(self.tags)})
         # parent module
         if self.module:
             j.update({"module": str(self.module)})
@@ -1084,9 +1087,10 @@ def __init__(self, *args, **kwargs):
                     parent_path = parent.data.get("path", None)
                     if parent_path is not None:
                         self.data["path"] = parent_path
-                # inherit closest host
+                # inherit closest host+port
                 if parent.host:
                     self.data["host"] = str(parent.host)
+                    self._port = parent.port
                     # we do this to refresh the hash
                     self.data = self.data
                     break
@@ -1097,6 +1101,7 @@ def __init__(self, *args, **kwargs):
 
 class DictPathEvent(DictEvent):
     def sanitize_data(self, data):
+        data = super().sanitize_data(data)
         new_data = dict(data)
         new_data["path"] = str(new_data["path"])
         file_blobs = getattr(self.scan, "_file_blobs", False)
@@ -1550,19 +1555,23 @@ class VULNERABILITY(ClosestHostEvent):
         "HIGH": "🟥",
         "MEDIUM": "🟧",
         "LOW": "🟨",
+        "INFO": "🟦",
         "UNKNOWN": "⬜",
     }
 
     def sanitize_data(self, data):
+        data = super().sanitize_data(data)
         self.add_tag(data["severity"].lower())
         return data
 
     class _data_validator(BaseModel):
         host: Optional[str] = None
         severity: str
+        name: str
         description: 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)
@@ -1577,6 +1586,7 @@ class FINDING(ClosestHostEvent):
 
     class _data_validator(BaseModel):
         host: Optional[str] = None
+        name: str
         description: str
         url: Optional[str] = None
         path: Optional[str] = None
@@ -1595,6 +1605,11 @@ class _data_validator(BaseModel):
         _validate_url = field_validator("url")(validators.validate_url)
         _validate_host = field_validator("host")(validators.validate_host)
 
+    def _sanitize_data(self, data):
+        data = super()._sanitize_data(data)
+        data["technology"] = data["technology"].lower()
+        return data
+
     def _data_id(self):
         # dedupe by host+port+tech
         tech = self.data.get("technology", "")
@@ -1724,6 +1739,7 @@ class MOBILE_APP(DictEvent):
     _always_emit = True
 
     def _sanitize_data(self, data):
+        data = super()._sanitize_data(data)
         if isinstance(data, str):
             data = {"url": data}
         if "url" not in data:
@@ -1886,7 +1902,7 @@ def make_event(
         )
 
 
-def event_from_json(j, siem_friendly=False):
+def event_from_json(j):
     """
     Creates an event object from a JSON dictionary.
 
@@ -1917,10 +1933,12 @@ def event_from_json(j, siem_friendly=False):
             "context": j.get("discovery_context", None),
             "dummy": True,
         }
-        if siem_friendly:
-            data = j["data"][event_type]
-        else:
-            data = j["data"]
+        data = j.get("data_json", None)
+        if data is None:
+            data = j.get("data", None)
+        if data is None:
+            json_pretty = json.dumps(j, indent=2)
+            raise ValueError(f"data or data_json must be provided. JSON: {json_pretty}")
         kwargs["data"] = data
         event = make_event(**kwargs)
         event_uuid = j.get("uuid", None)
@@ -1929,7 +1947,12 @@ def event_from_json(j, siem_friendly=False):
 
         resolved_hosts = j.get("resolved_hosts", [])
         event._resolved_hosts = set(resolved_hosts)
-        event.timestamp = datetime.datetime.fromisoformat(j["timestamp"])
+
+        # accept both isoformat and unix timestamp
+        try:
+            event.timestamp = datetime.datetime.fromtimestamp(j["timestamp"], ZoneInfo("UTC"))
+        except Exception:
+            event.timestamp = datetime.datetime.fromisoformat(j["timestamp"])
         event.scope_distance = j["scope_distance"]
         parent_id = j.get("parent", None)
         if parent_id is not None:

bbot/defaults.yml

@@ -1,5 +1,14 @@
 ### BASIC OPTIONS ###
 
+# NOTE: If used in a preset, these options must be nested underneath "config:" like so:
+# config:
+#   home: ~/.bbot
+#   keep_scans: 20
+#   scope:
+#     strict: true
+#   dns:
+#     minimal: true
+
 # BBOT working directory
 home: ~/.bbot
 # How many scan results to keep before cleaning up the older ones
@@ -15,7 +24,7 @@ folder_blobs: false
 
 scope:
   # strict scope means only exact DNS names are considered in-scope
-  # subdomains are not included unless they are explicitly provided in the target list
+  # their subdomains are not included unless explicitly whitelisted
   strict: false
   # Filter by scope distance which events are displayed in the output
   # 0 == show only in-scope events (affiliates are always shown)

bbot/models/helpers.py

@@ -0,0 +1,20 @@
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+
+def utc_datetime_validator(d: datetime) -> datetime:
+    """
+    Converts all dates into UTC
+    """
+    if d.tzinfo is not None:
+        return d.astimezone(ZoneInfo("UTC"))
+    else:
+        return d.replace(tzinfo=ZoneInfo("UTC"))
+
+
+def utc_now() -> datetime:
+    return datetime.now(ZoneInfo("UTC"))
+
+
+def utc_now_timestamp() -> datetime:
+    return utc_now().timestamp()