Add human-readable stdout display for all event types

bbot/core/event/base.py

@@ -23,6 +23,7 @@
 from bbot.errors import *
 from .helpers import EventSeed
 from bbot.core.helpers import (
+    bytes_to_human,
     extract_words,
     is_domain,
     is_subdomain,
@@ -1182,8 +1183,7 @@ def _pretty_string(self):
         return str(self.data["asn"])
 
     def _data_human(self):
-        """Create a concise human-readable representation of ASN data."""
-        return json.dumps({"asn": self.data["asn"]}, sort_keys=True)
+        return f"AS{self.data['asn']}"
 
 
 class CODE_REPOSITORY(DictHostEvent):
@@ -1196,6 +1196,9 @@ class _data_validator(BaseModel):
     def _pretty_string(self):
         return self.data["url"]
 
+    def _data_human(self):
+        return self.data["url"]
+
 
 class IP_ADDRESS(BaseEvent):
     def __init__(self, *args, **kwargs):
@@ -1478,6 +1481,9 @@ def sanitize_data(self, data):
         data["name"] = data["name"].lower()
         return data
 
+    def _data_human(self):
+        return f"{self.data['name']} ({self.data['url']})"
+
     def _words(self):
         return self.data["name"]
 
@@ -1546,6 +1552,29 @@ def _outgoing_dedup_hash(self, event):
     def _url(self):
         return self.data["url"]
 
+    def _data_human(self):
+        param_type = self.data.get("type", "")
+        name = self.data.get("name", "")
+        original_value = self.data.get("original_value", "")
+        url = self.data.get("url", "")
+        description = self.data.get("description", "")
+        additional_params = self.data.get("additional_params", {})
+        parts = []
+        if param_type:
+            parts.append(f"[{param_type}]")
+        if original_value:
+            parts.append(f"{name}={original_value}")
+        else:
+            parts.append(name)
+        if description:
+            parts.append(f"- {description}")
+        if additional_params:
+            param_names = ", ".join(sorted(additional_params.keys()))
+            parts.append(f"Additional Params: [{param_names}]")
+        if url:
+            parts.append(f"({url})")
+        return " ".join(parts)
+
     def __str__(self):
         max_event_len = 200
         d = str(self.data)
@@ -1610,6 +1639,28 @@ def _words(self):
     def _pretty_string(self):
         return f"{self.data['hash']['header_mmh3']}:{self.data['hash']['body_mmh3']}"
 
+    def _data_human(self):
+        parts = []
+        status = self.http_status
+        if status:
+            parts.append(f"[{status}]")
+        method = self.data.get("method", "")
+        if method:
+            parts.append(method)
+        parts.append(self.data.get("url", ""))
+        if status and str(status).startswith("3"):
+            location = self.redirect_location
+            if location:
+                parts.append(f"-> {location}")
+        else:
+            title = self.http_title
+            if title:
+                parts.append(f"- [{title}]")
+        content_length = self.data.get("content_length", 0)
+        if content_length:
+            parts.append(f"({bytes_to_human(content_length)})")
+        return " ".join(parts)
+
     def _minimize(self):
         super()._minimize()
         if self._module_consumers <= 0:
@@ -1735,6 +1786,19 @@ def _pretty_string(self):
             confidence_str = f"[{confidence}]"
         return f"Severity: [{severity}] Confidence: {confidence_str} {description}"
 
+    def _data_human(self):
+        parts = []
+        parts.append(f"Severity: [{self.data['severity']}]")
+        parts.append(f"Confidence: [{self.data['confidence']}]")
+        parts.append(self.data["description"])
+        url = self.data.get("url", "")
+        if url and url not in self.data["description"]:
+            parts.append(f"({url})")
+        cves = self.data.get("cves", [])
+        if cves:
+            parts.append(f"[{', '.join(cves)}]")
+        return " ".join(parts)
+
 
 class TECHNOLOGY(DictHostEvent):
     class _data_validator(BaseModel):
@@ -1757,6 +1821,13 @@ def _data_id(self):
     def _pretty_string(self):
         return self.data["technology"]
 
+    def _data_human(self):
+        tech = self.data["technology"]
+        url = self.data.get("url", "")
+        if url:
+            return f"{tech} ({url})"
+        return tech
+
 
 class PROTOCOL(DictHostEvent):
     class _data_validator(BaseModel):
@@ -1779,11 +1850,35 @@ def port(self):
     def _pretty_string(self):
         return self.data["protocol"]
 
+    def _data_human(self):
+        protocol = self.data["protocol"]
+        port = self.data.get("port")
+        banner = self.data.get("banner", "")
+        if port:
+            result = f"{protocol}/{port}"
+        else:
+            result = protocol
+        if banner:
+            result += f" - {banner}"
+        return result
+
 
 class GEOLOCATION(BaseEvent):
     _always_emit = True
     _quick_emit = True
 
+    def _data_human(self):
+        country = self.data.get("country_name", "")
+        region = self.data.get("region_name", "")
+        city = self.data.get("city_name", "") or self.data.get("city", "")
+        lat = self.data.get("latitude", "")
+        lon = self.data.get("longitude", "")
+        location_parts = [p for p in (city, region, country) if p]
+        result = ", ".join(location_parts)
+        if lat and lon:
+            result += f" ({lat}, {lon})"
+        return result if result else super()._data_human()
+
 
 class PASSWORD(BaseEvent):
     _always_emit = True
@@ -1805,16 +1900,50 @@ class SOCIAL(DictHostEvent):
     _quick_emit = True
     _scope_distance_increment_same_host = True
 
+    def _data_human(self):
+        platform = self.data.get("platform", "")
+        profile_name = self.data.get("profile_name", "")
+        url = self.data.get("url", "")
+        parts = []
+        if platform:
+            parts.append(f"{platform}:")
+        parts.append(profile_name)
+        if url:
+            parts.append(f"({url})")
+        return " ".join(parts)
+
 
 class WEBSCREENSHOT(DictPathEvent):
     _always_emit = True
     _quick_emit = True
 
+    def _data_human(self):
+        return f"{self.data.get('url', '')} Saved to: {self.data.get('path', '')}"
+
 
 class AZURE_TENANT(DictEvent):
     _always_emit = True
     _quick_emit = True
 
+    def _data_human(self):
+        max_domains = 20
+        tenant_names = self.data.get("tenant-names", [])
+        tenant_id = self.data.get("tenant-id", "")
+        domains = self.data.get("domains", [])
+        parts = []
+        if tenant_names:
+            parts.append(", ".join(tenant_names))
+        if tenant_id:
+            parts.append(f"({tenant_id})")
+        if domains:
+            if len(domains) <= max_domains:
+                parts.append(f"- {', '.join(domains)}")
+            else:
+                shown = ", ".join(domains[:max_domains])
+                hidden = len(domains) - max_domains
+                parts.append(f"- {shown} (hiding {hidden} additional domains)")
+        return " ".join(parts) if parts else super()._data_human()
+
 
 class WAF(DictHostEvent):
     _always_emit = True
@@ -1831,8 +1960,25 @@ class _data_validator(BaseModel):
     def _pretty_string(self):
         return self.data["waf"]
 
+    def _data_human(self):
+        waf = self.data["waf"]
+        info = self.data.get("info", "")
+        url = self.data.get("url", "")
+        if info:
+            waf += f" - {info}"
+        if url:
+            waf += f" ({url})"
+        return waf
+
 
 class FILESYSTEM(DictPathEvent):
+    def _data_human(self):
+        path = self.data.get("path", "")
+        description = self.data.get("magic_description", "")
+        if description:
+            return f"{path} ({description})"
+        return path
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         if self._data_path.is_file():
@@ -1861,6 +2007,12 @@ class RAW_DNS_RECORD(DictHostEvent, DnsEvent):
     # don't emit raw DNS records for affiliates
     _always_emit_tags = ["seed"]
 
+    def _data_human(self):
+        rdtype = self.data.get("type", "")
+        host = self.data.get("host", "")
+        answer = self.data.get("answer", "")
+        return f"{rdtype} {host} -> {answer}"
+
 
 class MOBILE_APP(DictEvent):
     _always_emit = True
@@ -1890,6 +2042,13 @@ def _sanitize_data(self, data):
     def _pretty_string(self):
         return self.data["url"]
 
+    def _data_human(self):
+        app_id = self.data.get("id", "")
+        url = self.data.get("url", "")
+        if app_id and url:
+            return f"{app_id} ({url})"
+        return url or app_id
+
 
 def update_event(
     event,

bbot/test/test_step_1/test_events.py

@@ -1,4 +1,3 @@
-import json
 import random
 import ipaddress
 
@@ -359,7 +358,7 @@ async def test_events(events, helpers):
         "FINDING",
         dummy=True,
     )
-    assert json.loads(test_vuln2.data_human)["severity"] == "INFO"
+    assert test_vuln2.data_human == "Severity: [INFO] Confidence: [HIGH] asdf"
     assert test_vuln2.host.is_private
     # must have severity
     with pytest.raises(ValidationError, match=".*validation error.*\nseverity\n.*Field required.*"):