diff --git a/bbot/core/event/base.py b/bbot/core/event/base.py index 90e2b96242..7f3f80fd79 100644 --- a/bbot/core/event/base.py +++ b/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, diff --git a/bbot/test/test_step_1/test_events.py b/bbot/test/test_step_1/test_events.py index 5f1e458ce0..d1505df88f 100644 --- a/bbot/test/test_step_1/test_events.py +++ b/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.*"):