diff --git a/bbot/core/event/base.py b/bbot/core/event/base.py index 27e9ef5a86..07eb8671cf 100644 --- a/bbot/core/event/base.py +++ b/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,7 +1522,7 @@ def redirect_location(self): return location -class VULNERABILITY(ClosestHostEvent): +class FINDING(ClosestHostEvent): _always_emit = True _quick_emit = True severity_colors = { @@ -1555,13 +1530,20 @@ class VULNERABILITY(ClosestHostEvent): "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): @@ -1569,32 +1551,26 @@ class _data_validator(BaseModel): 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, } diff --git a/bbot/core/helpers/validators.py b/bbot/core/helpers/validators.py index 97a39fae3c..0d1b63e6c7 100644 --- a/bbot/core/helpers/validators.py +++ b/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()) diff --git a/bbot/modules/ajaxpro.py b/bbot/modules/ajaxpro.py index c1ff413915..284741b5d9 100644 --- a/bbot/modules/ajaxpro.py +++ b/bbot/modules/ajaxpro.py @@ -10,7 +10,7 @@ class ajaxpro(BaseModule): ajaxpro_regex = re.compile(r' + + +]> +&{rand_entity};""" + test_url = event.parsed_url.geturl() + r = await self.generic_ssrf.helpers.curl( + url=test_url, method="POST", raw_body=post_body, headers={"Content-type": "application/xml"} + ) + if r: + self.process(event, r, subdomain_tag) + + +class generic_ssrf(BaseModule): + watched_events = ["URL"] + produced_events = ["FINDING"] + flags = ["active", "aggressive", "web-thorough"] + meta = {"description": "Check for generic SSRFs", "created_date": "2022-07-30", "author": "@liquidsec"} + options = { + "skip_dns_interaction": False, + } + options_desc = { + "skip_dns_interaction": "Do not report DNS interactions (only HTTP interaction)", + } + in_scope_only = True + + deps_apt = ["curl"] + + async def setup(self): + self.submodules = {} + self.interactsh_subdomain_tags = {} + self.parameter_subdomain_tags_map = {} + self.severity = None + self.skip_dns_interaction = self.config.get("skip_dns_interaction", False) + + if self.scan.config.get("interactsh_disable", False) is False: + try: + self.interactsh_instance = self.helpers.interactsh() + self.interactsh_domain = await self.interactsh_instance.register(callback=self.interactsh_callback) + except InteractshError as e: + self.warning(f"Interactsh failure: {e}") + return False + else: + self.warning( + "The generic_ssrf module is completely dependent on interactsh to function, but it is disabled globally. Aborting." + ) + return None + + # instantiate submodules + for m in BaseSubmodule.__subclasses__(): + if m.__name__.startswith("Generic_"): + self.verbose(f"Starting generic_ssrf submodule: {m.__name__}") + self.submodules[m.__name__] = m(self) + + return True + + async def handle_event(self, event): + for s in self.submodules.values(): + await s.test(event) + + async def interactsh_callback(self, r): + protocol = r.get("protocol").upper() + if protocol == "DNS" and self.skip_dns_interaction: + return + + full_id = r.get("full-id", None) + subdomain_tag = full_id.split(".")[0] + + if full_id: + if "." in full_id: + match = self.interactsh_subdomain_tags.get(subdomain_tag) + if not match: + return + matched_event = match[0] + matched_technique = match[1] + matched_severity = match[2] + matched_echoed_response = str(match[3]) + + triggering_param = self.parameter_subdomain_tags_map.get(subdomain_tag, None) + description = f"Out-of-band interaction: [{matched_technique}]" + if triggering_param: + self.debug(f"Found triggering parameter: {triggering_param}") + description += f" [Triggering Parameter: {triggering_param}]" + description += f" [{protocol}] Echoed Response: {matched_echoed_response}" + + self.debug(f"Emitting event with description: {description}") # Debug the final description + + confidence = "CONFIRMED" if protocol == "HTTP" else "MODERATE" + event_data = { + "host": str(matched_event.host), + "url": matched_event.data, + "description": description, + "name": "Generic SSRF Detection", + "confidence": confidence, + "severity": matched_severity, + } + + await self.emit_event( + event_data, + "FINDING", + matched_event, + context=f"{{module}} scanned {matched_event.data} and detected {{event.type}}: {matched_technique}", + ) + else: + # this is likely caused by something trying to resolve the base domain first and can be ignored + self.debug("skipping result because subdomain tag was missing") + + async def cleanup(self): + if self.scan.config.get("interactsh_disable", False) is False: + try: + await self.interactsh_instance.deregister() + self.debug( + f"successfully deregistered interactsh session with correlation_id {self.interactsh_instance.correlation_id}" + ) + except InteractshError as e: + self.warning(f"Interactsh failure: {e}") + + async def finish(self): + if self.scan.config.get("interactsh_disable", False) is False: + await self.helpers.sleep(5) + try: + for r in await self.interactsh_instance.poll(): + await self.interactsh_callback(r) + except InteractshError as e: + self.debug(f"Error in interact.sh: {e}") diff --git a/bbot/modules/git.py b/bbot/modules/git.py index 0c069bd958..2d53d7d29e 100644 --- a/bbot/modules/git.py +++ b/bbot/modules/git.py @@ -37,6 +37,8 @@ async def handle_event(self, event): "url": url, "description": description, "name": "Exposed .git config", + "severity": "MEDIUM", + "confidence": "HIGH", }, "FINDING", event, diff --git a/bbot/modules/graphql_introspection.py b/bbot/modules/graphql_introspection.py index 7a6864cfd7..df9b84fd4f 100644 --- a/bbot/modules/graphql_introspection.py +++ b/bbot/modules/graphql_introspection.py @@ -142,6 +142,8 @@ async def handle_event(self, event): "url": url, "description": f"GraphQL Schema at {url}", "path": relative_path, + "severity": "INFORMATIONAL", + "confidence": "CONFIRMED", }, "FINDING", event, diff --git a/bbot/modules/host_header.py b/bbot/modules/host_header.py index 2d664f42d0..c7da42d89c 100644 --- a/bbot/modules/host_header.py +++ b/bbot/modules/host_header.py @@ -43,14 +43,16 @@ async def interactsh_callback(self, r): return matched_event = match[0] matched_technique = match[1] - protocol = r.get("protocol").upper() + confidence = "HIGH" if protocol == "HTTP" else "MODERATE" await self.emit_event( { "host": str(matched_event.host), "url": matched_event.data["url"], "name": "Host Header Spoofing", "description": f"Spoofed Host header ({matched_technique}) [{protocol}] interaction", + "severity": "MEDIUM", + "confidence": confidence, }, "FINDING", matched_event, @@ -144,6 +146,8 @@ async def handle_event(self, event): "url": url, "description": description, "name": "Duplicate Host Header Tolerated", + "severity": "INFORMATIONAL", + "confidence": "LOW", }, "FINDING", event, @@ -187,6 +191,8 @@ async def handle_event(self, event): "url": url, "description": description, "name": "Possible Host Header Injection", + "severity": "INFORMATIONAL", + "confidence": "LOW", }, "FINDING", event, diff --git a/bbot/modules/hunt.py b/bbot/modules/hunt.py index d4a0720436..3b6c4239cc 100644 --- a/bbot/modules/hunt.py +++ b/bbot/modules/hunt.py @@ -316,6 +316,8 @@ async def handle_event(self, event): "host": str(event.host), "description": description, "name": "Potentially Interesting Parameter", + "severity": "INFORMATIONAL", + "confidence": "LOW", } url = event.data.get("url", "") if url: diff --git a/bbot/modules/iis_shortnames.py b/bbot/modules/iis_shortnames.py index fcf2b28d1e..5c53217bd6 100644 --- a/bbot/modules/iis_shortnames.py +++ b/bbot/modules/iis_shortnames.py @@ -233,13 +233,14 @@ class safety_counter_obj: description = f"IIS Shortname Vulnerability Detected. Potentially Vulnerable Method/Techniques: [{','.join(technique_strings)}]" await self.emit_event( { + "name": "IIS Shortnames", "severity": "LOW", + "confidence": "HIGH", "host": str(event.host), "url": normalized_url, "description": description, - "name": "IIS Shortnames", }, - "VULNERABILITY", + "FINDING", event, context="{module} detected low {event.type}: IIS shortname enumeration", ) @@ -345,6 +346,8 @@ class safety_counter_obj: "host": str(event.host), "url": event.data, "description": f"Possible backup file (zip) in web root: {normalized_url}{url_hint}", + "confidence": "MODERATE", + "severity": "MEDIUM", }, "FINDING", event, diff --git a/bbot/modules/internal/excavate.py b/bbot/modules/internal/excavate.py index ceb8da05ee..9c7a9fbbfb 100644 --- a/bbot/modules/internal/excavate.py +++ b/bbot/modules/internal/excavate.py @@ -105,10 +105,12 @@ def extract_params_location(location_header_value, original_parsed_url): class YaraRuleSettings: - def __init__(self, description, tags, emit_match): + def __init__(self, description, tags, emit_match, severity, confidence): self.description = description self.tags = tags self.emit_match = emit_match + self.severity = severity + self.confidence = confidence class ExcavateRule: @@ -153,6 +155,8 @@ async def preprocess(self, r, event, discovery_context): description = "" tags = [] emit_match = False + severity = "INFORMATIONAL" + confidence = "UNKNOWN" if "description" in r.meta.keys(): description = r.meta["description"] @@ -160,8 +164,12 @@ async def preprocess(self, r, event, discovery_context): tags = self.excavate.helpers.chain_lists(r.meta["tags"]) if "emit_match" in r.meta.keys(): emit_match = True + if "severity" in r.meta.keys(): + severity = r.meta["severity"] + if "confidence" in r.meta.keys(): + confidence = r.meta["confidence"] - yara_rule_settings = YaraRuleSettings(description, tags, emit_match) + yara_rule_settings = YaraRuleSettings(description, tags, emit_match, severity, confidence) yara_results = {} for h in r.strings: yara_results[h.identifier.lstrip("$")] = sorted( @@ -185,7 +193,7 @@ async def process(self, yara_results, event, yara_rule_settings, discovery_conte event : Event The event data associated with the YARA match. yara_rule_settings : YaraRuleSettings - The settings configured from YARA rule meta tags, including description, tags, and emit_match flag. + The settings configured from YARA rule meta tags, including description, severity, confidence, tags, and emit_match flag. discovery_context : DiscoveryContext The context in which the discovery is made. @@ -248,7 +256,7 @@ async def report( event : Event The parent event to which this event is related. yara_rule_settings : YaraRuleSettings - The settings configured from YARA rule meta tags, including description and tags. + The settings configured from YARA rule meta tags, including description, severity, confidence, and tags. discovery_context : DiscoveryContext The context in which the discovery is made. event_type : str, optional @@ -261,11 +269,16 @@ async def report( Returns: None """ - # If a description is not set and is needed, provide a basic one - if event_type == "FINDING" and "description" not in event_data.keys(): - event_data["name"] = f"{discovery_context} {yara_rule_settings.description}" - event_data["description"] = f"{discovery_context} {yara_rule_settings.description}" + if event_type == "FINDING": + if "description" not in event_data.keys(): + event_data["description"] = f"{discovery_context} {yara_rule_settings.description}" + if "name" not in event_data.keys(): + event_data["name"] = f"{discovery_context} {yara_rule_settings.description}" + if "severity" not in event_data.keys(): + event_data["severity"] = yara_rule_settings.severity + if "confidence" not in event_data.keys(): + event_data["confidence"] = yara_rule_settings.confidence subject = "" if isinstance(event_data, str): subject = f" {event_data}" @@ -296,6 +309,9 @@ async def process(self, yara_results, event, yara_rule_settings, discovery_conte ) if yara_rule_settings.emit_match: event_data["description"] += f" and extracted [{result}]" + event_data["severity"] = yara_rule_settings.get("severity", "LOW") + event_data["confidence"] = yara_rule_settings.get("confidence", "UNKNOWN") + await self.report(event_data, event, yara_rule_settings, discovery_context) @@ -718,7 +734,7 @@ def __init__(self, excavate): signature_component_list.append(rf"${signature_name} = {signature}") signature_component = " ".join(signature_component_list) self.yara_rules["error_detection"] = ( - f'rule error_detection {{meta: description = "contains a verbose error message" strings: {signature_component} condition: any of them}}' + f'rule error_detection {{meta: description = "contains a verbose error message" severity = "INFORMATIONAL" confidence = "MODERATE" strings: {signature_component} condition: any of them}}' ) async def process(self, yara_results, event, yara_rule_settings, discovery_context): @@ -750,7 +766,7 @@ def __init__(self, excavate): regexes_component_list.append(rf"${regex_name} = /\b{regex.pattern}/") regexes_component = " ".join(regexes_component_list) self.yara_rules["serialization_detection"] = ( - f'rule serialization_detection {{meta: description = "contains a possible serialized object" strings: {regexes_component} condition: any of them}}' + f'rule serialization_detection {{meta: description = "contains a possible serialized object" severity = "INFORMATIONAL" confidence = "MODERATE" strings: {regexes_component} condition: any of them}}' ) async def process(self, yara_results, event, yara_rule_settings, discovery_context): diff --git a/bbot/modules/lightfuzz/lightfuzz.py b/bbot/modules/lightfuzz/lightfuzz.py index 3e7f8ae460..119cd839f0 100644 --- a/bbot/modules/lightfuzz/lightfuzz.py +++ b/bbot/modules/lightfuzz/lightfuzz.py @@ -6,7 +6,7 @@ class lightfuzz(BaseModule): watched_events = ["URL", "WEB_PARAMETER"] - produced_events = ["FINDING", "VULNERABILITY"] + produced_events = ["FINDING"] flags = ["active", "aggressive", "web-thorough", "deadly"] options = { @@ -75,12 +75,13 @@ async def interactsh_callback(self, r): await self.emit_event( { "severity": "CRITICAL", + "confidence": "CONFIRMED", "host": str(details["event"].host), "url": details["event"].data["url"], "name": "Lightfuzz - OS Command Injection", "description": f"OS Command Injection (OOB Interaction) Type: [{details['type']}] Parameter Name: [{details['name']}] Probe: [{details['probe']}]", }, - "VULNERABILITY", + "FINDING", details["event"], ) else: @@ -117,11 +118,12 @@ async def run_submodule(self, submodule, event): # Append the envelope summary to the description event_data["description"] += f" Envelopes: [{envelope_summary}]" - if r["type"] == "VULNERABILITY": - event_data["severity"] = r["severity"] + event_data["severity"] = r["severity"] + event_data["confidence"] = r["confidence"] + event_data["name"] = f"Lightfuzz - {r['name']}" await self.emit_event( event_data, - r["type"], + "FINDING", event, ) diff --git a/bbot/modules/lightfuzz/submodules/cmdi.py b/bbot/modules/lightfuzz/submodules/cmdi.py index 51b256c08f..c253d24323 100644 --- a/bbot/modules/lightfuzz/submodules/cmdi.py +++ b/bbot/modules/lightfuzz/submodules/cmdi.py @@ -74,8 +74,9 @@ async def fuzz(self): if len(positive_detections) > 0: self.results.append( { - "type": "FINDING", - "name": "Lightfuzz - Possible Command Injection", + "name": "Possible Command Injection", + "severity": "CRITICAL", + "confidence": "MODERATE", "description": f"POSSIBLE OS Command Injection. {self.metadata()} Detection Method: [echo canary] CMD Probe Delimeters: [{' '.join(positive_detections)}]", } ) diff --git a/bbot/modules/lightfuzz/submodules/crypto.py b/bbot/modules/lightfuzz/submodules/crypto.py index b5bc846f64..6a0e4ee4ce 100644 --- a/bbot/modules/lightfuzz/submodules/crypto.py +++ b/bbot/modules/lightfuzz/submodules/crypto.py @@ -284,9 +284,9 @@ async def padding_oracle(self, probe_value, cookies): context = f"Lightfuzz Cryptographic Probe Submodule detected a probable padding oracle vulnerability after manipulating parameter: [{self.event.data['name']}]" self.results.append( { - "type": "VULNERABILITY", "severity": "HIGH", "name": "Padding Oracle Vulnerability", + "confidence": "HIGH", "description": f"Padding Oracle Vulnerability. Block size: [{str(block_size)}] {self.metadata()}", "context": context, } @@ -320,8 +320,9 @@ async def error_string_search(self, text_dict, baseline_text): if unique_matches: self.results.append( { - "type": "FINDING", - "name": "Lightfuzz - Possible Cryptographic Error", + "name": "Possible Cryptographic Error", + "severity": "INFORMATIONAL", + "confidence": "LOW", "description": f"Possible Cryptographic Error. {self.metadata()} Strings: [{','.join(unique_matches)}] Detection Technique(s): [{','.join(matching_techniques)}]", "context": context, } @@ -415,8 +416,9 @@ async def fuzz(self): context = f"Lightfuzz Cryptographic Probe Submodule detected a parameter ({self.event.data['name']}) to appears to drive a cryptographic operation" self.results.append( { - "type": "FINDING", - "name": "Lightfuzz - Probable Cryptographic Parameter", + "name": "Probable Cryptographic Parameter", + "severity": "INFORMATIONAL", + "confidence": "LOW", "description": f"Probable Cryptographic Parameter. {self.metadata()} Detection Technique(s): [{', '.join(confirmed_techniques)}]", "context": context, } @@ -470,8 +472,9 @@ async def fuzz(self): context = f"Lightfuzz Cryptographic Probe Submodule detected a parameter ({self.event.data['name']}) that is a likely a hash, which is connected to another parameter {additional_param_name})" self.results.append( { - "type": "FINDING", - "name": "Lightfuzz - Possible Length Extension Attack", + "name": "Possible Length Extension Attack", + "severity": "INFORMATIONAL", + "confidence": "LOW", "description": f"Possible {self.event.data['type']} parameter with {hash_instance.name.upper()} Hash as value. {self.metadata()}, linked to additional parameter [{additional_param_name}]", "context": context, } diff --git a/bbot/modules/lightfuzz/submodules/path.py b/bbot/modules/lightfuzz/submodules/path.py index 39bdd9c7c3..03249a706d 100644 --- a/bbot/modules/lightfuzz/submodules/path.py +++ b/bbot/modules/lightfuzz/submodules/path.py @@ -121,8 +121,9 @@ async def fuzz(self): if confirmations > 3: self.results.append( { - "type": "FINDING", - "name": "Lightfuzz - Possible Path Traversal", + "name": "Possible Path Traversal", + "severity": "HIGH", + "confidence": "LOW", "description": f"POSSIBLE Path Traversal. {self.metadata()} Detection Method: [{path_technique}]", } ) @@ -149,8 +150,9 @@ async def fuzz(self): if r and trigger in r.text: self.results.append( { - "type": "FINDING", - "name": "Lightfuzz - Possible Path Traversal", + "name": "Possible Path Traversal", + "severity": "HIGH", + "confidence": "MODERATE", "description": f"POSSIBLE Path Traversal. {self.metadata()} Detection Method: [Absolute Path: {path}]", } ) diff --git a/bbot/modules/lightfuzz/submodules/serial.py b/bbot/modules/lightfuzz/submodules/serial.py index ae3724e87b..58837f06be 100644 --- a/bbot/modules/lightfuzz/submodules/serial.py +++ b/bbot/modules/lightfuzz/submodules/serial.py @@ -165,8 +165,9 @@ def get_title(text): self.results.append( { - "type": "FINDING", - "name": "Lightfuzz - Possible Unsafe Deserialization", + "name": "Possible Unsafe Deserialization", + "severity": "HIGH", + "confidence": "LOW", "description": f"POSSIBLE Unsafe Deserialization. {self.metadata()} Technique: [Error Resolution (Baseline: [{payload_baseline.baseline.status_code}] {baseline_title} -> Probe: [{status_code}] {probe_title})] Serialization Payload: [{type}]", } ) @@ -183,8 +184,9 @@ def get_title(text): self.debug(f"Error string '{serialization_error}' found in response for {type}") self.results.append( { - "type": "FINDING", - "name": "Lightfuzz - Possible Unsafe Deserialization", + "name": "Possible Unsafe Deserialization", + "severity": "HIGH", + "confidence": "LOW", "description": f"POSSIBLE Unsafe Deserialization. {self.metadata()} Technique: [Differential Error Analysis] Error-String: [{serialization_error}] Payload: [{type}]", } ) diff --git a/bbot/modules/lightfuzz/submodules/sqli.py b/bbot/modules/lightfuzz/submodules/sqli.py index e0f4385f8f..869432a3e9 100644 --- a/bbot/modules/lightfuzz/submodules/sqli.py +++ b/bbot/modules/lightfuzz/submodules/sqli.py @@ -99,8 +99,9 @@ async def fuzz(self): if sqli_error_string.lower() in single_quote[3].text.lower(): self.results.append( { - "type": "FINDING", - "name": "Lightfuzz - Possible SQL Injection", + "name": "Possible SQL Injection", + "severity": "HIGH", + "confidence": "MODERATE", "description": f"Possible SQL Injection. {self.metadata()} Detection Method: [SQL Error Detection] Detected String: [{sqli_error_string}]", } ) @@ -120,8 +121,9 @@ async def fuzz(self): ): self.results.append( { - "type": "FINDING", - "name": "Lightfuzz - Possible SQL Injection", + "name": "Possible SQL Injection", + "severity": "HIGH", + "confidence": "MODERATE", "description": f"Possible SQL Injection. {self.metadata()} Detection Method: [Single Quote/Two Single Quote, Code Change ({http_compare.baseline.status_code}->{single_quote[3].status_code}->{double_single_quote[3].status_code})]", } ) @@ -181,8 +183,9 @@ async def fuzz(self): if confirmations == 3: self.results.append( { - "type": "FINDING", - "name": "Lightfuzz - Possible Blind SQL Injection", + "name": "Possible Blind SQL Injection", + "severity": "HIGH", + "confidence": "LOW", "description": f"Possible Blind SQL Injection. {self.metadata()} Detection Method: [Delay Probe ({p})]", } ) diff --git a/bbot/modules/lightfuzz/submodules/ssti.py b/bbot/modules/lightfuzz/submodules/ssti.py index d871ec03a1..187c5ca4bf 100644 --- a/bbot/modules/lightfuzz/submodules/ssti.py +++ b/bbot/modules/lightfuzz/submodules/ssti.py @@ -32,8 +32,9 @@ async def fuzz(self): if r and ("1787569" in r.text or "1,787,569" in r.text): self.results.append( { - "type": "FINDING", - "name": "Lightfuzz - Possible Server-side Template Injection", + "name": "Possible Server-side Template Injection", + "severity": "HIGH", + "confidence": "HIGH", "description": f"POSSIBLE Server-side Template Injection. {self.metadata()} Detection Method: [Integer Multiplication] Payload: [{probe_value}]", } ) diff --git a/bbot/modules/lightfuzz/submodules/xss.py b/bbot/modules/lightfuzz/submodules/xss.py index fed860191f..1bb3c48ec2 100644 --- a/bbot/modules/lightfuzz/submodules/xss.py +++ b/bbot/modules/lightfuzz/submodules/xss.py @@ -90,8 +90,9 @@ async def check_probe(self, cookies, probe, match, context): if probe_result and match in probe_result.text: self.results.append( { - "type": "FINDING", - "name": "Lightfuzz - Possible Reflected XSS", + "name": "Possible Reflected XSS", + "severity": "MEDIUM", + "confidence": "MODERATE", "description": f"Possible Reflected XSS. Parameter: [{self.event.data['name']}] Context: [{context}] Parameter Type: [{self.event.data['type']}]", } ) diff --git a/bbot/modules/medusa.py b/bbot/modules/medusa.py index f8e5d60573..847c4dd4b0 100644 --- a/bbot/modules/medusa.py +++ b/bbot/modules/medusa.py @@ -5,7 +5,7 @@ class medusa(BaseModule): watched_events = ["PROTOCOL"] - produced_events = ["VULNERABILITY"] + produced_events = ["FINDING"] flags = ["active", "aggressive", "deadly"] per_host_only = True meta = { @@ -143,9 +143,18 @@ async def handle_event(self, event): self.info(f"Medusa stderr: {result.stderr}") async for message in self.parse_output(result.stdout, snmp_version): - vuln_name = f"Valid SNMPV{snmp_version} Credentials Found!" - vuln_event = self.create_vuln_event("CRITICAL", vuln_name, message, event) - await self.emit_event(vuln_event) + await self.emit_event( + { + "name": f"Valid SNMPV{snmp_version} Credentials Found!", + "severity": "CRITICAL", + "confidence": "CONFIRMED", + "host": str(event.host), + "port": str(event.port), + "description": message, + }, + "FINDING", + parent=event, + ) # else: Medusa supports various protocols which could in theory be implemented later on. @@ -219,13 +228,3 @@ async def construct_command(self, host, port, protocol, protocol_version): ] return cmd - - def create_vuln_event(self, severity, name, description, source_event): - host = str(source_event.host) - port = str(source_event.port) - - return self.make_event( - {"severity": severity, "host": host, "port": port, "description": description, "name": name}, - "VULNERABILITY", - source_event, - ) diff --git a/bbot/modules/newsletters.py b/bbot/modules/newsletters.py index 3b23307952..17f5a1c98d 100644 --- a/bbot/modules/newsletters.py +++ b/bbot/modules/newsletters.py @@ -56,6 +56,8 @@ async def handle_event(self, event): "description": description, "url": _event.data["url"], "name": "Newsletter Submission Form", + "severity": "INFORMATIONAL", + "confidence": "LOW", } await self.emit_event( data, diff --git a/bbot/modules/ntlm.py b/bbot/modules/ntlm.py index 90c5a986c2..73cf83b6bd 100644 --- a/bbot/modules/ntlm.py +++ b/bbot/modules/ntlm.py @@ -121,6 +121,8 @@ async def handle_event(self, event): "url": url, "description": f"NTLM AUTH: {ntlm_resp_decoded}", "name": "NTLM Authentication", + "severity": "INFORMATIONAL", + "confidence": "HIGH", }, "FINDING", parent=event, diff --git a/bbot/modules/nuclei.py b/bbot/modules/nuclei.py index 3a0c863405..3aeba4feb4 100644 --- a/bbot/modules/nuclei.py +++ b/bbot/modules/nuclei.py @@ -6,7 +6,7 @@ class nuclei(BaseModule): watched_events = ["URL"] - produced_events = ["FINDING", "VULNERABILITY", "TECHNOLOGY"] + produced_events = ["FINDING", "TECHNOLOGY"] flags = ["active", "aggressive", "deadly"] meta = { "description": "Fast and customisable vulnerability scanner", @@ -176,6 +176,8 @@ async def handle_batch(self, *events): "url": url, "description": description_string, "name": f"Nuclei Vuln - {name}", + "severity": "INFORMATIONAL", + "confidence": "HIGH", }, "FINDING", parent_event, @@ -189,8 +191,9 @@ async def handle_batch(self, *events): "url": url, "description": description_string, "name": f"Nuclei Vuln - {name}", + "confidence": "HIGH", }, - "VULNERABILITY", + "FINDING", parent_event, context=f"{{module}} scanned {url} and identified {severity.lower()} {{event.type}}: {description_string}", ) diff --git a/bbot/modules/oauth.py b/bbot/modules/oauth.py index 559dea9313..dba3b52579 100644 --- a/bbot/modules/oauth.py +++ b/bbot/modules/oauth.py @@ -66,6 +66,8 @@ async def handle_event(self, event): "description": f"OpenID Connect Endpoint (domain: {source_domain}) found at {url}", "host": event.host, "url": url, + "severity": "INFORMATIONAL", + "confidence": "HIGH", }, "FINDING", parent=event, @@ -106,6 +108,8 @@ async def handle_event(self, event): "description": description, "host": event.host, "url": url, + "severity": "INFORMATIONAL", + "confidence": "LOW", }, "FINDING", parent=event, diff --git a/bbot/modules/output/asset_inventory.py b/bbot/modules/output/asset_inventory.py index 49c26fa8d7..9a2364e3b9 100644 --- a/bbot/modules/output/asset_inventory.py +++ b/bbot/modules/output/asset_inventory.py @@ -26,7 +26,6 @@ class asset_inventory(CSV): "DNS_NAME", "URL", "FINDING", - "VULNERABILITY", "TECHNOLOGY", "IP_ADDRESS", "WAF", @@ -316,12 +315,9 @@ def absorb_event(self, event): if event.type == "FINDING": location = event.data.get("url", event.data.get("host", "")) if location: - self.findings.add(f"{location}:{event.data['description']}") - - if event.type == "VULNERABILITY": - location = event.data.get("url", event.data.get("host", "")) - if location: - self.findings.add(f"{location}:{event.data['description']}:{event.data['severity']}") + self.findings.add( + f"{location}:{event.data['description']}:Severity: {event.data['severity']} Confidence: {event.data['confidence']}" + ) severity_int = severity_map.get(event.data.get("severity", "N/A"), 0) if severity_int > self.risk_rating: self.risk_rating = severity_int diff --git a/bbot/modules/output/discord.py b/bbot/modules/output/discord.py index 2aa4d21f84..934d89e670 100644 --- a/bbot/modules/output/discord.py +++ b/bbot/modules/output/discord.py @@ -8,10 +8,10 @@ class Discord(WebhookOutputModule): "created_date": "2023-08-14", "author": "@TheTechromancer", } - options = {"webhook_url": "", "event_types": ["VULNERABILITY", "FINDING"], "min_severity": "LOW", "retries": 10} + options = {"webhook_url": "", "event_types": ["FINDING"], "min_severity": "LOW", "retries": 10} options_desc = { "webhook_url": "Discord webhook URL", "event_types": "Types of events to send", - "min_severity": "Only allow VULNERABILITY events of this severity or higher", + "min_severity": "Only allow FINDING events of this severity or higher", "retries": "Number of times to retry sending the message before skipping the event", } diff --git a/bbot/modules/output/slack.py b/bbot/modules/output/slack.py index d65c816b3e..3366be9081 100644 --- a/bbot/modules/output/slack.py +++ b/bbot/modules/output/slack.py @@ -10,11 +10,11 @@ class Slack(WebhookOutputModule): "created_date": "2023-08-14", "author": "@TheTechromancer", } - options = {"webhook_url": "", "event_types": ["VULNERABILITY", "FINDING"], "min_severity": "LOW", "retries": 10} + options = {"webhook_url": "", "event_types": ["FINDING"], "min_severity": "LOW", "retries": 10} options_desc = { "webhook_url": "Discord webhook URL", "event_types": "Types of events to send", - "min_severity": "Only allow VULNERABILITY events of this severity or higher", + "min_severity": "Only allow FINDING events of this severity or higher", "retries": "Number of times to retry sending the message before skipping the event", } content_key = "text" @@ -26,7 +26,6 @@ def format_message_str(self, event): def format_message_other(self, event): event_yaml = yaml.dump(event.data) event_type = f"*`[{event.type}]`*" - if event.type in ("VULNERABILITY", "FINDING"): - event_str, color = self.get_severity_color(event) - event_type = f"{color} `{event_str}` {color}" + event_str, severity_color, confidence_color = self.get_colors(event) + event_type = f"Severity: {severity_color} Confidence: {confidence_color} {event_str}" return f"""*{event_type}*\n```\n{event_yaml}```""" diff --git a/bbot/modules/output/stdout.py b/bbot/modules/output/stdout.py index 59a121bd47..e642122feb 100644 --- a/bbot/modules/output/stdout.py +++ b/bbot/modules/output/stdout.py @@ -15,7 +15,13 @@ class Stdout(BaseOutputModule): "in_scope_only": "Whether to only show in-scope events", "accept_dupes": "Whether to show duplicate events, default True", } - vuln_severity_map = {"LOW": "HUGEWARNING", "MEDIUM": "HUGEWARNING", "HIGH": "CRITICAL", "CRITICAL": "CRITICAL"} + vuln_severity_map = { + "INFORMATIONAL": "HUGEINFO", + "LOW": "HUGEWARNING", + "MEDIUM": "HUGEWARNING", + "HIGH": "CRITICAL", + "CRITICAL": "CRITICAL", + } format_choices = ["text", "json"] async def setup(self): @@ -54,14 +60,14 @@ async def handle_text(self, event, event_json): else: event_str = self.human_event_str(event) - # log vulnerabilities in vivid colors - if event.type == "VULNERABILITY": - severity = event.data.get("severity", "INFO") + # log findings in vivid colors based on severity + if event.type == "FINDING": + severity = event.data.get("severity", "INFORMATIONAL") if severity in self.vuln_severity_map: loglevel = self.vuln_severity_map[severity] log_to_stderr(event_str, level=loglevel, logname=False) - elif event.type == "FINDING": - log_to_stderr(event_str, level="HUGEINFO", logname=False) + else: + log_to_stderr(event_str, level="HUGEINFO", logname=False) print(event_str) diff --git a/bbot/modules/output/teams.py b/bbot/modules/output/teams.py index c9a7cf1820..2ab461d5a6 100644 --- a/bbot/modules/output/teams.py +++ b/bbot/modules/output/teams.py @@ -8,11 +8,11 @@ class Teams(WebhookOutputModule): "created_date": "2023-08-14", "author": "@TheTechromancer", } - options = {"webhook_url": "", "event_types": ["VULNERABILITY", "FINDING"], "min_severity": "LOW", "retries": 10} + options = {"webhook_url": "", "event_types": ["FINDING"], "min_severity": "LOW", "retries": 10} options_desc = { "webhook_url": "Teams webhook URL", "event_types": "Types of events to send", - "min_severity": "Only allow VULNERABILITY events of this severity or higher", + "min_severity": "Only allow FINDING events of this severity or higher", "retries": "Number of times to retry sending the message before skipping the event", } @@ -46,7 +46,7 @@ def format_message_other(self, event): def get_severity_color(self, event): color = "Accent" - if event.type == "VULNERABILITY": + if event.type == "FINDING": severity = event.data.get("severity", "INFO") if severity == "CRITICAL": color = "Attention" @@ -78,7 +78,7 @@ def format_message(self, event): heading = {"type": "TextBlock", "text": f"{event.type}", "wrap": True, "size": "Large", "style": "heading"} body = adaptive_card["attachments"][0]["content"]["body"] body.append(heading) - if event.type in ("VULNERABILITY", "FINDING"): + if event.type == "FINDING": subheading = { "type": "TextBlock", "text": event.data.get("severity", "INFO"), diff --git a/bbot/modules/output/web_report.py b/bbot/modules/output/web_report.py index 92ff98289f..b6bee30e92 100644 --- a/bbot/modules/output/web_report.py +++ b/bbot/modules/output/web_report.py @@ -4,7 +4,7 @@ class web_report(BaseOutputModule): - watched_events = ["URL", "TECHNOLOGY", "FINDING", "VULNERABILITY", "VHOST"] + watched_events = ["URL", "TECHNOLOGY", "FINDING", "VHOST"] meta = { "description": "Create a markdown report with web assets", "created_date": "2023-02-08", @@ -89,7 +89,7 @@ async def report(self): if e in dedupe: continue dedupe.append(e) - self.markdown += f"\n* {e}\n" + self.markdown += f"* {e}\n" self.markdown += "\n" if self.file is not None: diff --git a/bbot/modules/paramminer_cookies.py b/bbot/modules/paramminer_cookies.py index a3b4619d45..83fce87ad5 100644 --- a/bbot/modules/paramminer_cookies.py +++ b/bbot/modules/paramminer_cookies.py @@ -8,7 +8,6 @@ class paramminer_cookies(paramminer_headers): watched_events = ["HTTP_RESPONSE", "WEB_PARAMETER"] produced_events = ["WEB_PARAMETER"] - produced_events = ["FINDING"] flags = ["active", "aggressive", "slow", "web-paramminer"] meta = { "description": "Smart brute-force to check for common HTTP cookie parameters", diff --git a/bbot/modules/paramminer_getparams.py b/bbot/modules/paramminer_getparams.py index e6f35f6235..b0a3da92e7 100644 --- a/bbot/modules/paramminer_getparams.py +++ b/bbot/modules/paramminer_getparams.py @@ -8,7 +8,6 @@ class paramminer_getparams(paramminer_headers): watched_events = ["HTTP_RESPONSE", "WEB_PARAMETER"] produced_events = ["WEB_PARAMETER"] - produced_events = ["FINDING"] flags = ["active", "aggressive", "slow", "web-paramminer"] meta = { "description": "Use smart brute-force to check for common HTTP GET parameters", diff --git a/bbot/modules/reflected_parameters.py b/bbot/modules/reflected_parameters.py index a942edd473..84b029347d 100644 --- a/bbot/modules/reflected_parameters.py +++ b/bbot/modules/reflected_parameters.py @@ -30,6 +30,8 @@ async def handle_event(self, event): "description": description, "url": url, "name": "Reflected Parameter", + "severity": "INFORMATIONAL", + "confidence": "HIGH", } await self.emit_event(data, "FINDING", event) diff --git a/bbot/modules/retirejs.py b/bbot/modules/retirejs.py index 4b58c1b23d..e7c3da3a9f 100644 --- a/bbot/modules/retirejs.py +++ b/bbot/modules/retirejs.py @@ -186,6 +186,7 @@ async def handle_event(self, event): "name": "Vulnerable JavaScript Library", "description": description, "severity": severity, + "confidence": "HIGH", "component": component, "url": event.parent.data["url"], } diff --git a/bbot/modules/shodan_idb.py b/bbot/modules/shodan_idb.py index 72fadfaf08..142ede3157 100644 --- a/bbot/modules/shodan_idb.py +++ b/bbot/modules/shodan_idb.py @@ -148,6 +148,8 @@ async def _parse_response(self, data: dict, event, ip): "host": str(event.host), "cves": vulns, "name": "Shodan - Possible Vulnerabilities", + "severity": "MEDIUM", + "confidence": "LOW", }, "FINDING", parent=event, diff --git a/bbot/modules/smuggler.py b/bbot/modules/smuggler.py index 4c7a78cf9e..67fcdd3d53 100644 --- a/bbot/modules/smuggler.py +++ b/bbot/modules/smuggler.py @@ -45,6 +45,8 @@ async def handle_event(self, event): "url": event.data, "description": description, "name": "Possible HTTP Smuggling", + "severity": "MEDIUM", + "confidence": "LOW", }, "FINDING", parent=event, diff --git a/bbot/modules/telerik.py b/bbot/modules/telerik.py index 0e4def5b5e..94d10f426c 100644 --- a/bbot/modules/telerik.py +++ b/bbot/modules/telerik.py @@ -20,7 +20,7 @@ class telerik(BaseModule): """ watched_events = ["URL", "HTTP_RESPONSE"] - produced_events = ["VULNERABILITY", "FINDING"] + produced_events = ["FINDING"] flags = ["active", "aggressive", "web-thorough"] meta = { "description": "Scan for critical Telerik vulnerabilities", @@ -247,6 +247,8 @@ async def handle_event(self, event): "url": f"{base_url}{webresource}", "description": description, "name": "Telerik Handler", + "severity": "INFORMATIONAL", + "confidence": "HIGH", }, "FINDING", event, @@ -274,18 +276,19 @@ async def handle_event(self, event): command.append(self.scan.http_proxy) output = await self.run_process(command) - description = f"[CVE-2017-11317] [{str(version)}] {webresource}" + description = f"Confirmed Vulnerable Telerik (version: {str(version)})" if "fileInfo" in output.stdout: self.debug(f"Confirmed Vulnerable Telerik (version: {str(version)}") await self.emit_event( { "severity": "CRITICAL", + "confidence": "CONFIRMED", "description": description, "host": str(event.host), "url": f"{base_url}{webresource}", "name": "Telerik RCE", }, - "VULNERABILITY", + "FINDING", event, context=f"{{module}} scanned {base_url} and identified critical {{event.type}}: {description}", ) @@ -318,6 +321,8 @@ async def handle_event(self, event): "url": f"{base_url}{dh}", "description": description, "name": "Telerik Handler", + "confidence": "CONFIRMED", + "severity": "INFORMATIONAL", }, "FINDING", event, @@ -343,6 +348,8 @@ async def handle_event(self, event): "url": f"{base_url}{spellcheckhandler}", "description": description, "name": "Telerik Handler", + "confidence": "CONFIRMED", + "severity": "INFORMATIONAL", }, "FINDING", event, @@ -363,6 +370,8 @@ async def handle_event(self, event): "url": f"{base_url}{chartimagehandler}", "description": "Telerik ChartImage AXD Handler Detected", "name": "Telerik Handler", + "confidence": "CONFIRMED", + "severity": "INFORMATIONAL", }, "FINDING", event, @@ -380,6 +389,8 @@ async def handle_event(self, event): "url": url, "description": "Telerik DialogHandler [SerializedParameters] Detected in HTTP Response", "name": "Telerik Handler", + "confidence": "CONFIRMED", + "severity": "INFORMATIONAL", }, "FINDING", event, @@ -392,6 +403,8 @@ async def handle_event(self, event): "url": url, "description": "Telerik AsyncUpload [serializedConfiguration] Detected in HTTP Response", "name": "Telerik AsyncUpload", + "confidence": "CONFIRMED", + "severity": "INFORMATIONAL", }, "FINDING", event, diff --git a/bbot/modules/templates/bucket.py b/bbot/modules/templates/bucket.py index 537910dbcc..d3e4f611a7 100644 --- a/bbot/modules/templates/bucket.py +++ b/bbot/modules/templates/bucket.py @@ -72,6 +72,8 @@ async def handle_storage_bucket(self, event): "url": url, "description": description, "name": "Open Storage Bucket", + "severity": "LOW", + "confidence": "HIGH", } await self.emit_event( event_data, diff --git a/bbot/modules/templates/webhook.py b/bbot/modules/templates/webhook.py index 79dc11750d..d5a44794f3 100644 --- a/bbot/modules/templates/webhook.py +++ b/bbot/modules/templates/webhook.py @@ -11,7 +11,8 @@ class WebhookOutputModule(BaseOutputModule): accept_dupes = False message_size_limit = 2000 content_key = "content" - vuln_severities = ["UNKNOWN", "LOW", "MEDIUM", "HIGH", "CRITICAL"] + severities = ["INFORMATIONAL", "LOW", "MEDIUM", "HIGH", "CRITICAL"] + confidences = ["UNKNOWN", "LOW", "MODERATE", "HIGH", "CONFIRMED"] # abort module after 10 failed requests (not including retries) _api_failure_abort_threshold = 10 @@ -21,10 +22,10 @@ class WebhookOutputModule(BaseOutputModule): async def setup(self): self.webhook_url = self.config.get("webhook_url", "") self.min_severity = self.config.get("min_severity", "LOW").strip().upper() - assert self.min_severity in self.vuln_severities, ( - f"min_severity must be one of the following: {','.join(self.vuln_severities)}" + assert self.min_severity in self.severities, ( + f"min_severity must be one of the following: {','.join(self.severities)}" ) - self.allowed_severities = self.vuln_severities[self.vuln_severities.index(self.min_severity) :] + self.allowed_severities = self.severities[self.severities.index(self.min_severity) :] if not self.webhook_url: self.warning("Must set Webhook URL") return False @@ -45,15 +46,15 @@ async def handle_event(self, event): def get_watched_events(self): if self._watched_events is None: - event_types = self.config.get("event_types", ["VULNERABILITY"]) + event_types = self.config.get("event_types", ["FINDING"]) if isinstance(event_types, str): event_types = [event_types] self._watched_events = set(event_types) return self._watched_events async def filter_event(self, event): - if event.type == "VULNERABILITY": - severity = event.data.get("severity", "UNKNOWN") + if event.type == "FINDING": + severity = event.data.get("severity", "INFORMATIONAL") if severity not in self.allowed_severities: return False, f"{severity} is below min_severity threshold" return True @@ -65,17 +66,20 @@ def format_message_str(self, event): def format_message_other(self, event): event_yaml = yaml.dump(event.data) event_type = f"**`[{event.type}]`**" - if event.type in ("VULNERABILITY", "FINDING"): - event_str, color = self.get_severity_color(event) - event_type = f"{color} {event_str} {color}" + if event.type == "FINDING": + event_str, severity_color, confidence_color = self.get_colors(event) + event_type = f"{severity_color} {confidence_color} {event_str}" return f"""**`{event_type}`**\n```yaml\n{event_yaml}```""" - def get_severity_color(self, event): - if event.type == "VULNERABILITY": - severity = event.data.get("severity", "UNKNOWN") - return f"{event.type} ({severity})", event.severity_colors[severity] + def get_colors(self, event): + if event.type == "FINDING": + severity = event.data.get("severity", "INFORMATIONAL") + confidence = event.data.get("confidence", "UNKNOWN") + severity_color = event.severity_colors.get(severity, "⬜") + confidence_color = event.confidence_colors.get(confidence, "⚪") + return f"{event.type} (Severity: {severity} / Confidence: {confidence})", severity_color, confidence_color else: - return event.type, "🟦" + return event.type, "🟦", "" def format_message(self, event): if isinstance(event.data, str): diff --git a/bbot/modules/trufflehog.py b/bbot/modules/trufflehog.py index e990fe6e36..b10414327c 100644 --- a/bbot/modules/trufflehog.py +++ b/bbot/modules/trufflehog.py @@ -5,7 +5,7 @@ class trufflehog(BaseModule): watched_events = ["CODE_REPOSITORY", "FILESYSTEM", "HTTP_RESPONSE", "RAW_TEXT"] - produced_events = ["FINDING", "VULNERABILITY"] + produced_events = ["FINDING"] flags = ["passive", "safe", "code-enum"] meta = { "description": "TruffleHog is a tool for finding credentials", @@ -120,15 +120,16 @@ async def handle_event(self, event): source_metadata, ) in self.execute_trufflehog(module, path): verified_str = "Verified" if verified else "Possible" - finding_type = "VULNERABILITY" if verified else "FINDING" + confidence = "CONFIRMED" if verified else "MODERATE" data = { "name": f"TruffleHog - {detector_name}", "description": f"{verified_str} Secret Found. Detector Type: [{detector_name}] Decoder Type: [{decoder_name}] Details: [{source_metadata}]", } if host: data["host"] = host - if finding_type == "VULNERABILITY": - data["severity"] = "High" + + data["severity"] = "HIGH" + data["confidence"] = confidence if description: data["description"] += f" Description: [{description}]" data["description"] += f" Raw result: [{raw_result}]" @@ -136,7 +137,7 @@ async def handle_event(self, event): data["description"] += f" RawV2 result: [{rawv2_result}]" await self.emit_event( data, - finding_type, + "FINDING", event, context=f'{{module}} searched {event.type} using "{module}" method and found {verified_str.lower()} secret ({{event.type}}): {raw_result}', ) diff --git a/bbot/modules/url_manipulation.py b/bbot/modules/url_manipulation.py index e7863f2308..ee73a3ee6a 100644 --- a/bbot/modules/url_manipulation.py +++ b/bbot/modules/url_manipulation.py @@ -84,6 +84,8 @@ async def handle_event(self, event): "host": str(event.host), "url": event.data, "name": "URL Manipulation", + "severity": "INFORMATIONAL", + "confidence": "LOW", }, "FINDING", parent=event, diff --git a/bbot/modules/wpscan.py b/bbot/modules/wpscan.py index 6b128e77c8..0bb6f112a1 100644 --- a/bbot/modules/wpscan.py +++ b/bbot/modules/wpscan.py @@ -4,7 +4,7 @@ class wpscan(BaseModule): watched_events = ["HTTP_RESPONSE", "TECHNOLOGY"] - produced_events = ["URL_UNVERIFIED", "FINDING", "VULNERABILITY", "TECHNOLOGY"] + produced_events = ["URL_UNVERIFIED", "FINDING", "TECHNOLOGY"] flags = ["active", "aggressive"] meta = { "description": "Wordpress security scanner. Highly recommended to use an API key for better results.", @@ -179,6 +179,8 @@ def parse_wp_misc(self, interesting_json, base_url, source_event): "url": url, "host": str(source_event.host), "name": "WPScan - Possible Vulnerability", + "severity": "INFORMATIONAL", + "confidence": "MODERATE", }, "FINDING", source_event, @@ -199,12 +201,13 @@ def parse_wp_version(self, version_json, url, source_event): yield self.make_event( { "severity": "HIGH", + "confidence": "MODERATE", "host": str(source_event.host), "url": url, "description": self.vulnerability_to_s(wp_vuln), "name": "WPScan - Possible Vulnerability", }, - "VULNERABILITY", + "FINDING", source_event, ) @@ -225,12 +228,13 @@ def parse_wp_themes(self, theme_json, url, source_event): yield self.make_event( { "severity": "HIGH", + "confidence": "MODERATE", "host": str(source_event.host), "url": url, "description": self.vulnerability_to_s(theme_vuln), "name": "WPScan - Possible Vulnerability", }, - "VULNERABILITY", + "FINDING", source_event, ) @@ -255,12 +259,13 @@ def parse_wp_plugins(self, plugins_json, base_url, source_event): yield self.make_event( { "severity": "HIGH", + "confidence": "MODERATE", "host": str(source_event.host), "url": url, "description": self.vulnerability_to_s(vuln), "name": "WPScan - Possible Vulnerability", }, - "VULNERABILITY", + "FINDING", source_event, ) diff --git a/bbot/test/bbot_fixtures.py b/bbot/test/bbot_fixtures.py index 48c5a91bbd..a3dc91524d 100644 --- a/bbot/test/bbot_fixtures.py +++ b/bbot/test/bbot_fixtures.py @@ -197,14 +197,27 @@ class bbot_events: url_hint = scan.make_event( "https://api.publicAPIs.org:443/hello.ash", "URL_HINT", parent=url, module=dummy_module ) - vulnerability = scan.make_event( - {"host": "evilcorp.com", "severity": "INFO", "description": "asdf", "name": "Vulnerability"}, - "VULNERABILITY", + url_hint = scan.make_event("https://api.publicAPIs.org:443/hello.ash", "URL_HINT", parent=url) + finding = scan.make_event( + { + "host": "evilcorp.com", + "severity": "INFORMATIONAL", + "confidence": "HIGH", + "description": "asdf", + "name": "Test Finding", + }, + "FINDING", parent=scan.root_event, module=dummy_module, ) finding = scan.make_event( - {"host": "evilcorp.com", "description": "asdf", "name": "Finding"}, + { + "host": "evilcorp.com", + "description": "asdf", + "name": "Finding", + "severity": "INFORMATIONAL", + "confidence": "HIGH", + }, "FINDING", parent=scan.root_event, module=dummy_module, @@ -241,7 +254,6 @@ class bbot_events: bbot_events.ipv4_url, bbot_events.ipv6_url, bbot_events.url_hint, - bbot_events.vulnerability, bbot_events.finding, bbot_events.vhost, bbot_events.http_response, diff --git a/bbot/test/test_step_1/test_db_models.py b/bbot/test/test_step_1/test_db_models.py index d453fa81e1..8c926b84da 100644 --- a/bbot/test/test_step_1/test_db_models.py +++ b/bbot/test/test_step_1/test_db_models.py @@ -34,7 +34,7 @@ def test_pydantic_models(events, bbot_scanner): ] # convert events to pydantic and back, making sure they're exactly the same - for event in ("ipv4", "http_response", "finding", "vulnerability", "storage_bucket"): + for event in ("ipv4", "http_response", "finding", "storage_bucket"): e = getattr(events, event) event_json = e.json() event_pydantic = Event(**event_json) diff --git a/bbot/test/test_step_1/test_events.py b/bbot/test/test_step_1/test_events.py index 051b561b6c..824ab6d146 100644 --- a/bbot/test/test_step_1/test_events.py +++ b/bbot/test/test_step_1/test_events.py @@ -334,61 +334,91 @@ async def test_events(events, helpers): assert "affiliate" in corrected_event4.tags test_vuln = scan.make_event( - {"host": "EVILcorp.com", "severity": "iNfo ", "description": "asdf", "name": "Vulnerability"}, - "VULNERABILITY", + { + "host": "EVILcorp.com", + "severity": "iNformational ", + "confidence": "HIGH", + "description": "asdf", + "name": "Test Finding", + }, + "FINDING", dummy=True, ) assert test_vuln.data["host"] == "evilcorp.com" - assert test_vuln.data["severity"] == "INFO" + assert test_vuln.data["severity"] == "INFORMATIONAL" test_vuln2 = scan.make_event( - {"host": "192.168.1.1", "severity": "iNfo ", "description": "asdf", "name": "Vulnerability"}, - "VULNERABILITY", + { + "host": "192.168.1.1", + "severity": "INFORMATIONAL", + "confidence": "HIGH", + "description": "asdf", + "name": "Vulnerability", + }, + "FINDING", dummy=True, ) - assert json.loads(test_vuln2.data_human)["severity"] == "INFO" + assert json.loads(test_vuln2.data_human)["severity"] == "INFORMATIONAL" assert test_vuln2.host.is_private # must have severity with pytest.raises(ValidationError, match=".*validation error.*\nseverity\n.*Field required.*"): - test_vuln = scan.make_event( - {"host": "evilcorp.com", "description": "asdf", "name": "Vulnerability"}, "VULNERABILITY", dummy=True - ) - # invalid host + test_vuln = scan.make_event({"host": "evilcorp.com", "description": "asdf"}, "FINDING", dummy=True) with pytest.raises(ValidationError, match=".*host.*\n.*Invalid host.*"): test_vuln = scan.make_event( - {"host": "!@#$", "severity": "INFO", "description": "asdf", "name": "Vulnerability"}, - "VULNERABILITY", + {"host": "!@#$", "severity": "INFORMATIONAL", "confidence": "HIGH", "description": "asdf"}, + "FINDING", dummy=True, ) # invalid severity with pytest.raises(ValidationError, match=".*severity.*\n.*Invalid severity.*"): test_vuln = scan.make_event( - {"host": "evilcorp.com", "severity": "WACK", "description": "asdf", "name": "Vulnerability"}, - "VULNERABILITY", + {"host": "evilcorp.com", "severity": "WACK", "confidence": "HIGH", "description": "asdf"}, + "FINDING", dummy=True, ) - # must have name - with pytest.raises(ValidationError, match=".*name.*\n.*Field required.*"): + # invalid confidence + with pytest.raises(ValidationError, match=".*confidence.*\n.*Invalid confidence.*"): + test_vuln = scan.make_event( + { + "host": "evilcorp.com", + "severity": "HIGH", + "confidence": "INVALID", + "description": "asdf", + "name": "Test", + }, + "FINDING", + dummy=True, + ) + # must have confidence + with pytest.raises(ValidationError, match=".*confidence.*\n.*Field required.*"): test_vuln = scan.make_event( - {"host": "evilcorp.com", "severity": "INFO", "description": "asdf"}, - "VULNERABILITY", + {"host": "evilcorp.com", "severity": "HIGH", "description": "asdf", "name": "Test"}, + "FINDING", dummy=True, ) - # port and netloc should be derived from URL - test_vuln = scan.make_event( - { - "host": "evilcorp.com", - "name": "test", - "severity": "INFO", - "description": "asdf", - "url": "http://evilcorp.com/test", - }, - "VULNERABILITY", + # test confidence colors and formatting + from bbot.core.event.base import FINDING + + expected_colors = {"CONFIRMED": "🟣", "HIGH": "🔴", "MODERATE": "🟠", "LOW": "🟡", "UNKNOWN": "⚪"} + assert FINDING.confidence_colors == expected_colors + + # test CONFIRMED gets bold formatting + confirmed_finding = scan.make_event( + {"host": "test.com", "name": "Test", "description": "Test", "severity": "HIGH", "confidence": "CONFIRMED"}, + "FINDING", dummy=True, ) - assert test_vuln.host == "evilcorp.com" - assert test_vuln.port == 80 - assert test_vuln.netloc == "evilcorp.com:80" + pretty_string = confirmed_finding._pretty_string() + assert "[\033[1mCONFIRMED\033[0m]" in pretty_string + assert f"confidence-{confirmed_finding.data['confidence'].lower()}" in confirmed_finding.tags + + # must have name + with pytest.raises(ValidationError, match=".*name.*\n.*Field required.*"): + test_vuln = scan.make_event( + {"host": "evilcorp.com", "severity": "INFORMATIONAL", "description": "asdf", "confidence": "HIGH"}, + "FINDING", + dummy=True, + ) # technology should be lowercased tech_event = scan.make_event( @@ -929,41 +959,6 @@ async def test_event_web_spider_distance(bbot_scanner): assert "spider-max" not in url_event_5.tags -def test_event_confidence(): - scan = Scanner() - # default 100 - event1 = scan.make_event("evilcorp.com", "DNS_NAME", dummy=True) - assert event1.confidence == 100 - assert event1.cumulative_confidence == 100 - # custom confidence - event2 = scan.make_event("evilcorp.com", "DNS_NAME", confidence=90, dummy=True) - assert event2.confidence == 90 - assert event2.cumulative_confidence == 90 - # max 100 - event3 = scan.make_event("evilcorp.com", "DNS_NAME", confidence=999, dummy=True) - assert event3.confidence == 100 - assert event3.cumulative_confidence == 100 - # min 1 - event4 = scan.make_event("evilcorp.com", "DNS_NAME", confidence=0, dummy=True) - assert event4.confidence == 1 - assert event4.cumulative_confidence == 1 - # first event in chain - event5 = scan.make_event("evilcorp.com", "DNS_NAME", confidence=90, parent=scan.root_event) - assert event5.confidence == 90 - assert event5.cumulative_confidence == 90 - # compounding confidence - event6 = scan.make_event("evilcorp.com", "DNS_NAME", confidence=50, parent=event5) - assert event6.confidence == 50 - assert event6.cumulative_confidence == 45 - event7 = scan.make_event("evilcorp.com", "DNS_NAME", confidence=50, parent=event6) - assert event7.confidence == 50 - assert event7.cumulative_confidence == 22 - # 100 confidence resets - event8 = scan.make_event("evilcorp.com", "DNS_NAME", confidence=100, parent=event7) - assert event8.confidence == 100 - assert event8.cumulative_confidence == 100 - - def test_event_closest_host(): scan = Scanner() # first event has a host @@ -985,14 +980,20 @@ def test_event_closest_host(): event3 = scan.make_event({"path": "/tmp/asdf.txt"}, "FILESYSTEM", parent=event2) assert not event3.host # finding automatically uses the host from the second event - finding = scan.make_event({"description": "test", "name": "Finding"}, "FINDING", parent=event3) + finding = scan.make_event( + {"description": "test", "severity": "LOW", "confidence": "MODERATE", "name": "Test Finding"}, + "FINDING", + parent=event3, + ) assert finding.data["host"] == "www.evilcorp.com" assert finding.data["url"] == "http://www.evilcorp.com/asdf" assert finding.data["path"] == "/tmp/asdf.txt" assert finding.host == "www.evilcorp.com" # same with vuln vuln = scan.make_event( - {"description": "test", "severity": "HIGH", "name": "Vulnerability"}, "VULNERABILITY", parent=event3 + {"description": "test", "severity": "HIGH", "confidence": "HIGH", "name": "Test Finding"}, + "FINDING", + parent=event3, ) assert vuln.data["host"] == "www.evilcorp.com" assert vuln.data["url"] == "http://www.evilcorp.com/asdf" @@ -1003,28 +1004,62 @@ def test_event_closest_host(): event3 = scan.make_event("wat", "ASDF", parent=scan.root_event) assert not event3.host with pytest.raises(ValueError): - finding = scan.make_event({"description": "test", "name": "Finding"}, "FINDING", parent=event3) + finding = scan.make_event( + {"description": "test", "severity": "LOW", "confidence": "MODERATE", "name": "Test Finding"}, + "FINDING", + parent=event3, + ) finding = scan.make_event( - {"path": "/tmp/asdf.txt", "description": "test", "name": "Finding"}, "FINDING", parent=event3 + { + "path": "/tmp/asdf.txt", + "description": "test", + "severity": "LOW", + "confidence": "MODERATE", + "name": "Test Finding", + }, + "FINDING", + parent=event3, ) assert finding is not None finding = scan.make_event( - {"host": "evilcorp.com", "description": "test", "name": "Finding"}, "FINDING", parent=event3 + { + "host": "evilcorp.com", + "description": "test", + "severity": "LOW", + "confidence": "MODERATE", + "name": "Test Finding", + }, + "FINDING", + parent=event3, ) assert finding is not None with pytest.raises(ValueError): vuln = scan.make_event( - {"description": "test", "severity": "HIGH", "name": "Vulnerability"}, "VULNERABILITY", parent=event3 + {"description": "test", "severity": "HIGH", "confidence": "CONFIRMED", "name": "Test Finding"}, + "FINDING", + parent=event3, ) vuln = scan.make_event( - {"path": "/tmp/asdf.txt", "description": "test", "severity": "HIGH", "name": "Vulnerability"}, - "VULNERABILITY", + { + "path": "/tmp/asdf.txt", + "description": "test", + "severity": "HIGH", + "confidence": "CONFIRMED", + "name": "Test Finding", + }, + "FINDING", parent=event3, ) assert vuln is not None vuln = scan.make_event( - {"host": "evilcorp.com", "description": "test", "severity": "HIGH", "name": "Vulnerability"}, - "VULNERABILITY", + { + "host": "evilcorp.com", + "description": "test", + "severity": "HIGH", + "confidence": "CONFIRMED", + "name": "Test Finding", + }, + "FINDING", parent=event3, ) assert vuln is not None @@ -1115,7 +1150,12 @@ def test_event_hashing(): url_event = scan.make_event("https://api.example.com/", "URL_UNVERIFIED", parent=scan.root_event) host_event_1 = scan.make_event("www.example.com", "DNS_NAME", parent=url_event) host_event_2 = scan.make_event("test.example.com", "DNS_NAME", parent=url_event) - finding_data = {"description": "Custom Yara Rule [find_string] Matched via identifier [str1]", "name": "Finding"} + finding_data = { + "description": "Custom Yara Rule [find_string] Matched via identifier [str1]", + "severity": "MEDIUM", + "confidence": "HIGH", + "name": "Finding", + } finding1 = scan.make_event(finding_data, "FINDING", parent=host_event_1) finding2 = scan.make_event(finding_data, "FINDING", parent=host_event_2) finding3 = scan.make_event(finding_data, "FINDING", parent=host_event_2) @@ -1123,16 +1163,22 @@ def test_event_hashing(): assert finding1.data == { "description": "Custom Yara Rule [find_string] Matched via identifier [str1]", "name": "Finding", + "severity": "MEDIUM", + "confidence": "HIGH", "host": "www.example.com", } assert finding2.data == { "description": "Custom Yara Rule [find_string] Matched via identifier [str1]", "name": "Finding", + "severity": "MEDIUM", + "confidence": "HIGH", "host": "test.example.com", } assert finding3.data == { "description": "Custom Yara Rule [find_string] Matched via identifier [str1]", "name": "Finding", + "severity": "MEDIUM", + "confidence": "HIGH", "host": "test.example.com", } assert finding1.id != finding2.id diff --git a/bbot/test/test_step_1/test_helpers.py b/bbot/test/test_step_1/test_helpers.py index 2c6488cb14..2800ea4846 100644 --- a/bbot/test/test_step_1/test_helpers.py +++ b/bbot/test/test_step_1/test_helpers.py @@ -443,8 +443,8 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_httpserver): with pytest.raises(ValueError): helpers.validators.validate_url("!@#$") # severities - assert helpers.validators.validate_severity(" iNfo") == "INFO" - assert helpers.validators.soft_validate(" iNfo", "severity") is True + assert helpers.validators.validate_severity(" iNformational") == "INFORMATIONAL" + assert helpers.validators.soft_validate(" iNformational", "severity") is True assert helpers.validators.soft_validate("NOPE", "severity") is False with pytest.raises(ValueError): helpers.validators.validate_severity("NOPE") diff --git a/bbot/test/test_step_1/test_manager_scope_accuracy.py b/bbot/test/test_step_1/test_manager_scope_accuracy.py index a976498f15..44163295f6 100644 --- a/bbot/test/test_step_1/test_manager_scope_accuracy.py +++ b/bbot/test/test_step_1/test_manager_scope_accuracy.py @@ -270,9 +270,7 @@ async def filter_event(self, event): async def handle_event(self, event): await self.emit_event( - {"host": str(event.host), "description": "yep", "severity": "CRITICAL", "name": "Vulnerability"}, - "VULNERABILITY", - parent=event, + {"host": str(event.host), "description": "yep", "severity": "CRITICAL", "confidence": "CONFIRMED", "name": "Test Finding"}, "FINDING", parent=event ) def custom_setup(scan): @@ -292,21 +290,21 @@ def custom_setup(scan): assert 1 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.66" and e.internal is False and e.scope_distance == 1]) assert 0 == len([e for e in events if e.type == "DNS_NAME" and e.data == "test.notrealzies"]) assert 0 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.77"]) - assert 1 == len([e for e in events if e.type == "VULNERABILITY" and e.data["host"] == "127.0.0.77" and e.internal is False and e.scope_distance == 3]) + assert 1 == len([e for e in events if e.type == "FINDING" and e.data["host"] == "127.0.0.77" and e.internal is False and e.scope_distance == 3]) assert len(all_events) == 8 assert 1 == len([e for e in all_events if e.type == "DNS_NAME" and e.data == "test.notreal" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.66" and e.internal is False and e.scope_distance == 1]) assert 2 == len([e for e in all_events if e.type == "DNS_NAME" and e.data == "test.notrealzies" and e.internal is True and e.scope_distance == 2]) assert 2 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is True and e.scope_distance == 3]) - assert 1 == len([e for e in all_events if e.type == "VULNERABILITY" and e.data["host"] == "127.0.0.77" and e.internal is False and e.scope_distance == 3]) + assert 1 == len([e for e in all_events if e.type == "FINDING" and e.data["host"] == "127.0.0.77" and e.internal is False and e.scope_distance == 3]) assert len(all_events_nodups) == 6 assert 1 == len([e for e in all_events_nodups if e.type == "DNS_NAME" and e.data == "test.notreal" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.66" and e.internal is False and e.scope_distance == 1]) assert 1 == len([e for e in all_events_nodups if e.type == "DNS_NAME" and e.data == "test.notrealzies" and e.internal is True and e.scope_distance == 2]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is True and e.scope_distance == 3]) - assert 1 == len([e for e in all_events_nodups if e.type == "VULNERABILITY" and e.data["host"] == "127.0.0.77" and e.internal is False and e.scope_distance == 3]) + assert 1 == len([e for e in all_events_nodups if e.type == "FINDING" and e.data["host"] == "127.0.0.77" and e.internal is False and e.scope_distance == 3]) for _graph_output_events in (graph_output_events, graph_output_batch_events): assert len(_graph_output_events) == 7 @@ -314,7 +312,7 @@ def custom_setup(scan): assert 1 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.66" and e.internal is False and e.scope_distance == 1]) assert 1 == len([e for e in _graph_output_events if e.type == "DNS_NAME" and e.data == "test.notrealzies" and e.internal is True and e.scope_distance == 2]) assert 1 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is True and e.scope_distance == 3]) - assert 1 == len([e for e in _graph_output_events if e.type == "VULNERABILITY" and e.data["host"] == "127.0.0.77" and e.internal is False and e.scope_distance == 3]) + assert 1 == len([e for e in _graph_output_events if e.type == "FINDING" and e.data["host"] == "127.0.0.77" and e.internal is False and e.scope_distance == 3]) # httpx/speculate IP_RANGE --> IP_ADDRESS --> OPEN_TCP_PORT --> URL, search distance = 0 events, all_events, all_events_nodups, graph_output_events, graph_output_batch_events = await do_scan( diff --git a/bbot/test/test_step_1/test_modules_basic.py b/bbot/test/test_step_1/test_modules_basic.py index 9e2e21e12e..fdafebdf15 100644 --- a/bbot/test/test_step_1/test_modules_basic.py +++ b/bbot/test/test_step_1/test_modules_basic.py @@ -359,6 +359,8 @@ async def handle_event(self, event): "url": "http://www.evilcorp.com", "description": "asdf", "name": "Finding", + "severity": "LOW", + "confidence": "MODERATE", }, "FINDING", event, diff --git a/bbot/test/test_step_1/test_presets.py b/bbot/test/test_step_1/test_presets.py index 8fb5b75858..5680b624d6 100644 --- a/bbot/test/test_step_1/test_presets.py +++ b/bbot/test/test_step_1/test_presets.py @@ -654,7 +654,7 @@ async def test_preset_module_loader(): class TestModule1(BaseModule): watched_events = ["URL", "HTTP_RESPONSE"] - produced_events = ["VULNERABILITY"] + produced_events = ["FINDING"] """ ) diff --git a/bbot/test/test_step_2/module_tests/test_module_ajaxpro.py b/bbot/test/test_step_2/module_tests/test_module_ajaxpro.py index b917de42ca..7cbbbb783c 100644 --- a/bbot/test/test_step_2/module_tests/test_module_ajaxpro.py +++ b/bbot/test/test_step_2/module_tests/test_module_ajaxpro.py @@ -35,7 +35,7 @@ def check(self, module_test, events): for e in events: if ( - e.type == "VULNERABILITY" + e.type == "FINDING" and "Ajaxpro Deserialization RCE (CVE-2021-23758)" in e.data["description"] and "http://127.0.0.1:8888/ajaxpro/AjaxPro.Services.ICartService,AjaxPro.2.ashx" in e.data["description"] diff --git a/bbot/test/test_step_2/module_tests/test_module_aspnet_bin_exposure.py b/bbot/test/test_step_2/module_tests/test_module_aspnet_bin_exposure.py index ca14ff7d03..f86578ebac 100644 --- a/bbot/test/test_step_2/module_tests/test_module_aspnet_bin_exposure.py +++ b/bbot/test/test_step_2/module_tests/test_module_aspnet_bin_exposure.py @@ -62,12 +62,12 @@ async def setup_before_prep(self, module_test): module_test.set_expect_requests(expect_args=expect_args, respond_args=respond_args) def check(self, module_test, events): - vulnerability_found = False + finding_found = False for e in events: - if e.type == "VULNERABILITY" and "IIS Bin Directory DLL Exposure" in e.data["description"]: - vulnerability_found = True + if e.type == "FINDING" and "IIS Bin Directory DLL Exposure" in e.data["description"]: + finding_found = True assert e.data["severity"] == "HIGH", "Vulnerability severity should be HIGH" assert "Detection Url" in e.data["description"], "Description should include detection URL" break - assert vulnerability_found, "No vulnerability event was found" + assert finding_found, "No finding event was found" diff --git a/bbot/test/test_step_2/module_tests/test_module_baddns.py b/bbot/test/test_step_2/module_tests/test_module_baddns.py index 877e973b2b..2d5d476d05 100644 --- a/bbot/test/test_step_2/module_tests/test_module_baddns.py +++ b/bbot/test/test_step_2/module_tests/test_module_baddns.py @@ -32,7 +32,7 @@ async def setup_after_prep(self, module_test): def check(self, module_test, events): assert any(e.data == "baddns.azurewebsites.net" for e in events), "CNAME detection failed" - assert any(e.type == "VULNERABILITY" for e in events), "Failed to emit VULNERABILITY" + assert any(e.type == "FINDING" for e in events), "Failed to emit FINDING" assert any("baddns-cname" in e.tags for e in events), "Failed to add baddns tag" @@ -61,7 +61,7 @@ def set_target(self, target): def check(self, module_test, events): assert any(e for e in events) - assert any(e.type == "VULNERABILITY" and "bigcartel.com" in e.data["description"] for e in events), ( - "Failed to emit VULNERABILITY" + assert any(e.type == "FINDING" and "bigcartel.com" in e.data["description"] for e in events), ( + "Failed to emit FINDING" ) assert any("baddns-cname" in e.tags for e in events), "Failed to add baddns tag" diff --git a/bbot/test/test_step_2/module_tests/test_module_baddns_zone.py b/bbot/test/test_step_2/module_tests/test_module_baddns_zone.py index d8138a3f7c..10da342217 100644 --- a/bbot/test/test_step_2/module_tests/test_module_baddns_zone.py +++ b/bbot/test/test_step_2/module_tests/test_module_baddns_zone.py @@ -38,7 +38,7 @@ def from_xfr(*args, **kwargs): def check(self, module_test, events): assert any(e.data == "zzzz.bad.dns" for e in events), "Zone transfer failed (1)" assert any(e.data == "asdf.bad.dns" for e in events), "Zone transfer failed (2)" - assert any(e.type == "VULNERABILITY" for e in events), "Failed to emit VULNERABILITY" + assert any(e.type == "FINDING" for e in events), "Failed to emit FINDING" assert any("baddns-zonetransfer" in e.tags for e in events), "Failed to add baddns tag" @@ -58,5 +58,5 @@ async def setup_after_prep(self, module_test): def check(self, module_test, events): assert any(e.data == "zzzz.bad.dns" for e in events), "NSEC Walk Failed (1)" assert any(e.data == "xyz.bad.dns" for e in events), "NSEC Walk Failed (2)" - assert any(e.type == "VULNERABILITY" for e in events), "Failed to emit VULNERABILITY" + assert any(e.type == "FINDING" for e in events), "Failed to emit FINDING" assert any("baddns-nsec" in e.tags for e in events), "Failed to add baddns tag" diff --git a/bbot/test/test_step_2/module_tests/test_module_badsecrets.py b/bbot/test/test_step_2/module_tests/test_module_badsecrets.py index 9eda654eb6..60c26bd7ed 100644 --- a/bbot/test/test_step_2/module_tests/test_module_badsecrets.py +++ b/bbot/test/test_step_2/module_tests/test_module_badsecrets.py @@ -79,7 +79,7 @@ def check(self, module_test, events): for e in events: if ( - e.type == "VULNERABILITY" + e.type == "FINDING" and "Known Secret Found." in e.data["description"] and "validationKey: 0F97BAE23F6F36801ABDB5F145124E00A6F795A97093D778EE5CD24F35B78B6FC4C0D0D4420657689C4F321F8596B59E83F02E296E970C4DEAD2DFE226294979 validationAlgo: SHA1 encryptionKey: 8CCFBC5B7589DD37DC3B4A885376D7480A69645DAEEC74F418B4877BEC008156 encryptionAlgo: AES" in e.data["description"] @@ -94,7 +94,7 @@ def check(self, module_test, events): IdentifyOnly = True if ( - e.type == "VULNERABILITY" + e.type == "FINDING" and "1234" in e.data["description"] and "eyJhbGciOiJIUzI1NiJ9.eyJJc3N1ZXIiOiJJc3N1ZXIiLCJVc2VybmFtZSI6IkJhZFNlY3JldHMiLCJleHAiOjE1OTMxMzM0ODMsImlhdCI6MTQ2NjkwMzA4M30.ovqRikAo_0kKJ0GVrAwQlezymxrLGjcEiW_s3UJMMCo" in e.data["description"] @@ -102,7 +102,7 @@ def check(self, module_test, events): CookieBasedDetection = True if ( - e.type == "VULNERABILITY" + e.type == "FINDING" and "keyboard cat" in e.data["description"] and "s%3A8FnPwdeM9kdGTZlWvdaVtQ0S1BCOhY5G.qys7H2oGSLLdRsEq7sqh7btOohHsaRKqyjV4LiVnBvc" in e.data["description"] @@ -110,7 +110,7 @@ def check(self, module_test, events): CookieBasedDetection_2 = True if ( - e.type == "VULNERABILITY" + e.type == "FINDING" and "Express.js Secret (cookie-session)" in e.data["description"] and "zOQU7v7aTe_3zu7tnVuHi1MJ2DU" in e.data["description"] ): @@ -122,6 +122,15 @@ def check(self, module_test, events): assert CookieBasedDetection_2, "No Express.js cookie vuln detected" assert CookieBasedDetection_3, "No Express.js (cs dual cookies) vuln detected" + # Verify that badsecrets emits CONFIRMED confidence for detected secrets + confirmed_finding = None + for e in events: + if e.type == "FINDING" and "Known Secret Found." in e.data["description"]: + confirmed_finding = e + break + if confirmed_finding: + assert confirmed_finding.data["confidence"] == "CONFIRMED" + class TestBadSecrets_customsecrets(TestBadSecrets): config_overrides = { @@ -156,7 +165,7 @@ def check(self, module_test, events): SecretFound = False for e in events: if ( - e.type == "VULNERABILITY" + e.type == "FINDING" and "Known Secret Found." in e.data["description"] and "DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF" in e.data["description"] ): diff --git a/bbot/test/test_step_2/module_tests/test_module_discord.py b/bbot/test/test_step_2/module_tests/test_module_discord.py index d1aeb5c60f..73b6e864e4 100644 --- a/bbot/test/test_step_2/module_tests/test_module_discord.py +++ b/bbot/test/test_step_2/module_tests/test_module_discord.py @@ -8,7 +8,7 @@ class TestDiscord(ModuleTestBase): modules_overrides = ["discord", "excavate", "badsecrets", "httpx"] webhook_url = "https://discord.com/api/webhooks/1234/deadbeef-P-uF-asdf" - config_overrides = {"modules": {"discord": {"webhook_url": webhook_url}}} + config_overrides = {"modules": {"discord": {"webhook_url": webhook_url, "min_severity": "INFORMATIONAL"}}} def custom_setup(self, module_test): respond_args = { @@ -34,8 +34,6 @@ def custom_response(request: httpx.Request): module_test.httpx_mock.add_callback(custom_response, url=self.webhook_url) def check(self, module_test, events): - vulns = [e for e in events if e.type == "VULNERABILITY"] findings = [e for e in events if e.type == "FINDING"] - assert len(findings) == 1 - assert len(vulns) == 2 + assert len(findings) == 3 assert module_test.request_count == 4 diff --git a/bbot/test/test_step_2/module_tests/test_module_dotnetnuke.py b/bbot/test/test_step_2/module_tests/test_module_dotnetnuke.py index 6d6fbeb1c6..b2f1bf4395 100644 --- a/bbot/test/test_step_2/module_tests/test_module_dotnetnuke.py +++ b/bbot/test/test_step_2/module_tests/test_module_dotnetnuke.py @@ -95,26 +95,23 @@ def check(self, module_test, events): if e.type == "TECHNOLOGY" and "dotnetnuke" in e.data["technology"]: dnn_technology_detection = True - if ( - e.type == "VULNERABILITY" - and "DotNetNuke Personalization Cookie Deserialization" in e.data["description"] - ): + if e.type == "FINDING" and "DotNetNuke Personalization Cookie Deserialization" in e.data["description"]: dnn_personalization_deserialization_detection = True if ( - e.type == "VULNERABILITY" + e.type == "FINDING" and "DotNetNuke DNNArticle Module GetCSS.ashx Arbitrary File Read" in e.data["description"] ): dnn_getcss_fileread_detection = True if ( - e.type == "VULNERABILITY" + e.type == "FINDING" and "DotNetNuke dnnUI_NewsArticlesSlider Module Arbitrary File Read" in e.data["description"] ): dnn_imagehandler_fileread_detection = True if ( - e.type == "VULNERABILITY" + e.type == "FINDING" and "DotNetNuke InstallWizard SuperUser Privilege Escalation" in e.data["description"] ): dnn_installwizard_privesc_detection = True @@ -168,7 +165,7 @@ def check(self, module_test, events): if e.type == "TECHNOLOGY" and "dotnetnuke" in e.data["technology"]: dnn_technology_detection = True - if e.type == "VULNERABILITY" and "DotNetNuke Blind-SSRF (CVE 2017-0929)" in e.data["description"]: + if e.type == "FINDING" and "DotNetNuke Blind-SSRF (CVE 2017-0929)" in e.data["description"]: dnn_dnnimagehandler_blindssrf = True assert dnn_technology_detection, "DNN Technology Detection Failed" diff --git a/bbot/test/test_step_2/module_tests/test_module_excavate.py b/bbot/test/test_step_2/module_tests/test_module_excavate.py index 0289e83e8d..5573edf2eb 100644 --- a/bbot/test/test_step_2/module_tests/test_module_excavate.py +++ b/bbot/test/test_step_2/module_tests/test_module_excavate.py @@ -11,7 +11,7 @@ class TestExcavate(ModuleTestBase): targets = ["http://127.0.0.1:8888/", "test.notreal", "http://127.0.0.1:8888/subdir/links.html"] modules_overrides = ["excavate", "httpx"] - config_overrides = {"web": {"spider_distance": 1, "spider_depth": 1}} + config_overrides = {"web": {"spider_distance": 1, "spider_depth": 1}, "omit_event_types": []} async def setup_before_prep(self, module_test): response_data = """ @@ -181,24 +181,24 @@ async def setup_before_prep(self, module_test): def check(self, module_test, events): found_js_url_event = False - found_badsecrets_vulnerability = False + found_badsecrets_finding = False found_excavate_jwt_finding = False for e in events: if e.type == "URL" and e.data == "http://127.0.0.1:8888/script.js": found_js_url_event = True if e.type == "FINDING" and "JWT" in e.data["description"] and str(e.module) == "excavate": found_excavate_jwt_finding = True - if e.type == "VULNERABILITY": - found_badsecrets_vulnerability = True + if e.type == "FINDING" and "BadSecrets" in e.data["name"] and str(e.module) == "badsecrets": + found_badsecrets_finding = True assert found_js_url_event, "Failed to find URL event for script.js" - assert found_badsecrets_vulnerability, "Failed to find BADSECRETs event from script.js" + assert found_badsecrets_finding, "Failed to find BADSECRETs finding from script.js" assert found_excavate_jwt_finding, "Failed to find JWT finding from script.js" class TestExcavateRedirect(TestExcavate): targets = ["http://127.0.0.1:8888/", "http://127.0.0.1:8888/relative/", "http://127.0.0.1:8888/nonhttpredirect/"] - config_overrides = {"scope": {"report_distance": 1}} + config_overrides = {"scope": {"report_distance": 1}, "omit_event_types": []} async def setup_before_prep(self, module_test): # absolute redirect @@ -265,7 +265,7 @@ def check(self, module_test, events): class TestExcavateQuerystringRemoveTrue(TestExcavate): targets = ["http://127.0.0.1:8888/"] - config_overrides = {"url_querystring_remove": True, "url_querystring_collapse": True} + config_overrides = {"url_querystring_remove": True, "url_querystring_collapse": True, "omit_event_types": []} lots_of_params = """ @@ -290,7 +290,7 @@ def check(self, module_test, events): class TestExcavateQuerystringRemoveFalse(TestExcavateQuerystringRemoveTrue): - config_overrides = {"url_querystring_remove": False, "url_querystring_collapse": True} + config_overrides = {"url_querystring_remove": False, "url_querystring_collapse": True, "omit_event_types": []} def check(self, module_test, events): assert ( @@ -306,7 +306,7 @@ def check(self, module_test, events): class TestExcavateQuerystringCollapseFalse(TestExcavateQuerystringRemoveTrue): - config_overrides = {"url_querystring_remove": False, "url_querystring_collapse": False} + config_overrides = {"url_querystring_remove": False, "url_querystring_collapse": False, "omit_event_types": []} def check(self, module_test, events): assert ( @@ -323,7 +323,7 @@ def check(self, module_test, events): class TestExcavateMaxLinksPerPage(TestExcavate): targets = ["http://127.0.0.1:8888/"] - config_overrides = {"web": {"spider_links_per_page": 10, "spider_distance": 1}} + config_overrides = {"web": {"spider_links_per_page": 10, "spider_distance": 1}, "omit_event_types": []} lots_of_links = """ @@ -1064,6 +1064,56 @@ class TestExcavateYaraCustom(TestExcavateYara): config_overrides = {"modules": {"excavate": {"custom_yara_rules": f}}} +class TestExcavateYaraConfidence(ModuleTestBase): + """Test YARA rules with confidence options.""" + + targets = ["http://127.0.0.1:8888/"] + modules_overrides = ["excavate", "httpx"] + + async def setup_before_prep(self, module_test): + yara_test_html = """ + +

CONFIRMED_SECRET_DATA

+

HIGH_CONFIDENCE_INDICATOR

+

MODERATE_RISK_PATTERN

+

LOW_CONFIDENCE_MATCH

+

UNKNOWN_PATTERN_TYPE

+ + """ + module_test.httpserver.expect_request("/").respond_with_data(yara_test_html) + + async def setup_after_prep(self, module_test): + excavate_module = module_test.scan.modules["excavate"] + excavateruleinstance = excavateTestRule(excavate_module) + + # Add YARA rules with different confidence levels + yara_rules = { + "ConfirmedRule": 'rule ConfirmedRule { meta: description = "Confirmed rule" severity = "HIGH" confidence = "CONFIRMED" strings: $text = "CONFIRMED_SECRET_DATA" condition: $text }', + "HighConfidenceRule": 'rule HighConfidenceRule { meta: description = "High confidence rule" severity = "MEDIUM" confidence = "HIGH" strings: $text = "HIGH_CONFIDENCE_INDICATOR" condition: $text }', + "ModerateConfidenceRule": 'rule ModerateConfidenceRule { meta: description = "Moderate confidence rule" severity = "LOW" confidence = "MODERATE" strings: $text = "MODERATE_RISK_PATTERN" condition: $text }', + "LowConfidenceRule": 'rule LowConfidenceRule { meta: description = "Low confidence rule" severity = "INFORMATIONAL" confidence = "LOW" strings: $text = "LOW_CONFIDENCE_MATCH" condition: $text }', + "UnknownConfidenceRule": 'rule UnknownConfidenceRule { meta: description = "Unknown confidence rule" severity = "INFORMATIONAL" confidence = "UNKNOWN" strings: $text = "UNKNOWN_PATTERN_TYPE" condition: $text }', + } + + for rule_name, rule_content in yara_rules.items(): + excavate_module.add_yara_rule(rule_name, rule_content, excavateruleinstance) + + excavate_module.yara_rules = yara.compile(source="\n".join(excavate_module.yara_rules_dict.values())) + + def check(self, module_test, events): + """Verify findings are created with correct confidence levels.""" + findings = [e for e in events if e.type == "FINDING"] + confidence_findings = {f.data.get("confidence", "UNKNOWN"): f for f in findings} + + # Verify all confidence levels are present + expected_confidences = ["CONFIRMED", "HIGH", "MODERATE", "LOW", "UNKNOWN"] + for confidence in expected_confidences: + assert confidence in confidence_findings, f"Missing finding with confidence: {confidence}" + finding = confidence_findings[confidence] + assert finding.data["confidence"] == confidence + assert f"confidence-{confidence.lower()}" in finding.tags + + class TestExcavateSpiderDedupe(ModuleTestBase): class DummyModule(BaseModule): watched_events = ["URL_UNVERIFIED"] @@ -1081,6 +1131,7 @@ async def handle_event(self, event): dummy_text = "spider" modules_overrides = ["excavate", "httpx"] targets = ["http://127.0.0.1:8888/"] + config_overrides = {"omit_event_types": []} async def setup_after_prep(self, module_test): self.dummy_module = self.DummyModule(module_test.scan) @@ -1256,6 +1307,7 @@ class TestExcavateRAWTEXT(ModuleTestBase): "modules": { "filedownload": {"output_folder": str(bbot_test_dir / "filedownload")}, }, + "omit_event_types": [], } pdf_data = r"""%PDF-1.3 @@ -1433,7 +1485,7 @@ def check(self, module_test, events): class TestExcavateBadURLs(ModuleTestBase): targets = ["http://127.0.0.1:8888/"] modules_overrides = ["excavate", "httpx", "hunt"] - config_overrides = {"interactsh_disable": True, "scope": {"report_distance": 10}} + config_overrides = {"interactsh_disable": True, "scope": {"report_distance": 10}, "omit_event_types": []} bad_url_data = """ Help diff --git a/bbot/test/test_step_2/module_tests/test_module_generic_ssrf.py b/bbot/test/test_step_2/module_tests/test_module_generic_ssrf.py new file mode 100644 index 0000000000..c4b5e5f365 --- /dev/null +++ b/bbot/test/test_step_2/module_tests/test_module_generic_ssrf.py @@ -0,0 +1,90 @@ +import re +import asyncio +from werkzeug.wrappers import Response + +from .base import ModuleTestBase + + +def extract_subdomain_tag(data): + pattern = r"http://([a-z0-9]{4})\.fakedomain\.fakeinteractsh\.com" + match = re.search(pattern, data) + if match: + return match.group(1) + + +class TestGeneric_SSRF(ModuleTestBase): + targets = ["http://127.0.0.1:8888"] + modules_overrides = ["httpx", "generic_ssrf"] + + def request_handler(self, request): + subdomain_tag = None + + if request.method == "GET": + subdomain_tag = extract_subdomain_tag(request.full_path) + elif request.method == "POST": + subdomain_tag = extract_subdomain_tag(request.data.decode()) + if subdomain_tag: + asyncio.run( + self.interactsh_mock_instance.mock_interaction( + subdomain_tag, msg=f"{request.method}: {request.data.decode()}" + ) + ) + + return Response("alive", status=200) + + async def setup_before_prep(self, module_test): + self.interactsh_mock_instance = module_test.mock_interactsh("generic_ssrf") + module_test.monkeypatch.setattr( + module_test.scan.helpers, "interactsh", lambda *args, **kwargs: self.interactsh_mock_instance + ) + + async def setup_after_prep(self, module_test): + expect_args = re.compile("/") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + + def check(self, module_test, events): + total_vulnerabilities = 0 + total_findings = 0 + + for e in events: + if e.type == "FINDING": + total_vulnerabilities += 1 + elif e.type == "FINDING": + total_findings += 1 + + assert total_vulnerabilities == 60, "Incorrect number of findings detected" + + assert any( + e.type == "FINDING" + and "Out-of-band interaction: [Generic SSRF (GET)]" + and "[Triggering Parameter: Dest]" in e.data["description"] + for e in events + ), "Failed to detect Generic SSRF (GET)" + assert any( + e.type == "FINDING" and "Out-of-band interaction: [Generic SSRF (POST)]" in e.data["description"] + for e in events + ), "Failed to detect Generic SSRF (POST)" + + # Check that HTTP interactions have CONFIRMED confidence + http_findings = [e for e in events if e.type == "FINDING" and "[HTTP]" in e.data["description"]] + if http_findings: + assert http_findings[0].data["confidence"] == "CONFIRMED" + assert any( + e.type == "FINDING" and "Out-of-band interaction: [Generic XXE] [HTTP]" in e.data["description"] + for e in events + ), "Failed to detect Generic SSRF (XXE)" + + +class TestGeneric_SSRF_httponly(TestGeneric_SSRF): + config_overrides = {"modules": {"generic_ssrf": {"skip_dns_interaction": True}}} + + def check(self, module_test, events): + total_vulnerabilities = 0 + total_findings = 0 + + for e in events: + if e.type == "FINDING": + total_vulnerabilities += 1 + + assert total_vulnerabilities == 30, "Incorrect number of vulnerabilities detected" + assert total_findings == 0, "Incorrect number of findings detected" diff --git a/bbot/test/test_step_2/module_tests/test_module_hunt.py b/bbot/test/test_step_2/module_tests/test_module_hunt.py index 867a2565c6..87b3463c1f 100644 --- a/bbot/test/test_step_2/module_tests/test_module_hunt.py +++ b/bbot/test/test_step_2/module_tests/test_module_hunt.py @@ -14,12 +14,20 @@ async def setup_after_prep(self, module_test): module_test.set_expect_requests(expect_args=expect_args, respond_args=respond_args) def check(self, module_test, events): - assert any( - e.type == "FINDING" - and e.data["description"] - == "Found potentially interesting parameter. Name: [cipher] Parameter Type: [GETPARAM] Categories: [Insecure Cryptography] Original Value: [xor]" - for e in events - ) + finding_event = None + for e in events: + if ( + e.type == "FINDING" + and e.data["description"] + == "Found potentially interesting parameter. Name: [cipher] Parameter Type: [GETPARAM] Categories: [Insecure Cryptography] Original Value: [xor]" + ): + finding_event = e + break + + assert finding_event is not None + # Hunt emits INFORMATIONAL severity and LOW confidence + assert finding_event.data["severity"] == "INFORMATIONAL" + assert finding_event.data["confidence"] == "LOW" class TestHunt_Multiple(TestHunt): diff --git a/bbot/test/test_step_2/module_tests/test_module_iis_shortnames.py b/bbot/test/test_step_2/module_tests/test_module_iis_shortnames.py index 6ca827f56a..3bd5c721b1 100644 --- a/bbot/test/test_step_2/module_tests/test_module_iis_shortnames.py +++ b/bbot/test/test_step_2/module_tests/test_module_iis_shortnames.py @@ -90,17 +90,17 @@ async def setup_after_prep(self, module_test): module_test.set_expect_requests(expect_args=expect_args, respond_args=respond_args) def check(self, module_test, events): - vulnerabilityEmitted = False + magicurl_findingEmitted = False url_hintEmitted = False zip_findingEmitted = False for e in events: - if e.type == "VULNERABILITY" and "iis-magic-url" not in e.tags: - vulnerabilityEmitted = True + if e.type == "FINDING" and "iis-magic-url" not in e.tags: + magicurl_findingEmitted = True if e.type == "URL_HINT" and e.data == "http://127.0.0.1:8888/BLSHAX~1": url_hintEmitted = True if e.type == "FINDING" and "Possible backup file (zip) in web root" in e.data["description"]: zip_findingEmitted = True - assert vulnerabilityEmitted + assert magicurl_findingEmitted assert url_hintEmitted assert zip_findingEmitted diff --git a/bbot/test/test_step_2/module_tests/test_module_lightfuzz.py b/bbot/test/test_step_2/module_tests/test_module_lightfuzz.py index 21b5169527..d18568d8a7 100644 --- a/bbot/test/test_step_2/module_tests/test_module_lightfuzz.py +++ b/bbot/test/test_step_2/module_tests/test_module_lightfuzz.py @@ -1177,6 +1177,9 @@ def check(self, module_test, events): lightfuzz_serial_detect_errorresolution = False for e in events: + print("@@@@") + print(e.type) + print(e.data) if e.type == "WEB_PARAMETER": if e.data["name"] == "TextBox1": excavate_extracted_form_parameter = True @@ -1451,7 +1454,7 @@ def check(self, module_test, events): if "HTTP Extracted Parameter [search]" in e.data["description"]: web_parameter_emitted = True - if e.type == "VULNERABILITY": + if e.type == "FINDING": if ( "OS Command Injection (OOB Interaction) Type: [GETPARAM] Parameter Name: [search] Probe: [&&]" in e.data["description"] @@ -1678,8 +1681,6 @@ def check(self, module_test, events): == "Probable Cryptographic Parameter. Parameter: [encrypted_data] Parameter Type: [POSTPARAM] Original Value: [dplyorsu8VUriMW/8DqVDU6kRwL/FDk3Q%2B4GXVGZbo0CTh9YX1YvzZZJrYe4cHxvAICyliYtp1im4fWoOa54Zg%3D%3D] Detection Technique(s): [Single-byte Mutation] Envelopes: [URL-Encoded]" ): cryptographic_parameter_finding = True - - if e.type == "VULNERABILITY": if ( e.data["description"] == "Padding Oracle Vulnerability. Block size: [16] Parameter: [encrypted_data] Parameter Type: [POSTPARAM] Original Value: [dplyorsu8VUriMW/8DqVDU6kRwL/FDk3Q%2B4GXVGZbo0CTh9YX1YvzZZJrYe4cHxvAICyliYtp1im4fWoOa54Zg%3D%3D] Envelopes: [URL-Encoded]" diff --git a/bbot/test/test_step_2/module_tests/test_module_medusa.py b/bbot/test/test_step_2/module_tests/test_module_medusa.py index 52743ebd5c..773c6733b5 100644 --- a/bbot/test/test_step_2/module_tests/test_module_medusa.py +++ b/bbot/test/test_step_2/module_tests/test_module_medusa.py @@ -44,7 +44,9 @@ async def setup_after_prep(self, module_test): await module_test.module.emit_event(protocol_event) def check(self, module_test, events): - vuln_events = [e for e in events if e.type == "VULNERABILITY"] + vuln_events = [e for e in events if e.type == "FINDING"] assert len(vuln_events) == 1 assert "VALID [SNMPV2C] CREDENTIALS FOUND: public [READ]" in vuln_events[0].data["description"] + assert vuln_events[0].data["severity"] == "CRITICAL" + assert vuln_events[0].data["confidence"] == "CONFIRMED" diff --git a/bbot/test/test_step_2/module_tests/test_module_nuclei.py b/bbot/test/test_step_2/module_tests/test_module_nuclei.py index 777ab84f16..51b5b340f9 100644 --- a/bbot/test/test_step_2/module_tests/test_module_nuclei.py +++ b/bbot/test/test_step_2/module_tests/test_module_nuclei.py @@ -53,8 +53,11 @@ def check(self, module_test, events): if e.type == "FINDING": if "Directory listing enabled" in e.data["description"]: first_run_detect = True + # Nuclei emits HIGH confidence for most findings + assert e.data["confidence"] == "HIGH" elif "Copyright" in e.data["description"]: second_run_detect = True + assert e.data["confidence"] == "HIGH" assert first_run_detect assert second_run_detect @@ -82,9 +85,7 @@ async def setup_after_prep(self, module_test): module_test.set_expect_requests(expect_args=expect_args, respond_args=respond_args) def check(self, module_test, events): - assert any( - e.type == "VULNERABILITY" and "Generic Env File Disclosure" in e.data["description"] for e in events - ) + assert any(e.type == "FINDING" and "Generic Env File Disclosure" in e.data["description"] for e in events) class TestNucleiTechnology(TestNucleiManual): diff --git a/bbot/test/test_step_2/module_tests/test_module_slack.py b/bbot/test/test_step_2/module_tests/test_module_slack.py index 1258ed5110..a2809bcaf1 100644 --- a/bbot/test/test_step_2/module_tests/test_module_slack.py +++ b/bbot/test/test_step_2/module_tests/test_module_slack.py @@ -4,4 +4,4 @@ class TestSlack(DiscordBase): modules_overrides = ["slack", "excavate", "badsecrets", "httpx"] webhook_url = "https://hooks.slack.com/services/deadbeef/deadbeef/deadbeef" - config_overrides = {"modules": {"slack": {"webhook_url": webhook_url}}} + config_overrides = {"modules": {"slack": {"webhook_url": webhook_url, "min_severity": "INFORMATIONAL"}}} diff --git a/bbot/test/test_step_2/module_tests/test_module_teams.py b/bbot/test/test_step_2/module_tests/test_module_teams.py index 3f573dc21b..788cf45f8b 100644 --- a/bbot/test/test_step_2/module_tests/test_module_teams.py +++ b/bbot/test/test_step_2/module_tests/test_module_teams.py @@ -7,7 +7,9 @@ class TestTeams(DiscordBase): modules_overrides = ["teams", "excavate", "badsecrets", "httpx"] webhook_url = "https://evilcorp.webhook.office.com/webhookb2/deadbeef@deadbeef/IncomingWebhook/deadbeef/deadbeef" - config_overrides = {"modules": {"teams": {"webhook_url": webhook_url, "retries": 5}}} + config_overrides = { + "modules": {"teams": {"webhook_url": webhook_url, "retries": 5, "min_severity": "INFORMATIONAL"}} + } async def setup_after_prep(self, module_test): self.custom_setup(module_test) @@ -32,8 +34,6 @@ def custom_response(request: httpx.Request): module_test.httpx_mock.add_callback(custom_response, url=self.webhook_url) def check(self, module_test, events): - vulns = [e for e in events if e.type == "VULNERABILITY"] findings = [e for e in events if e.type == "FINDING"] - assert len(findings) == 1 - assert len(vulns) == 2 + assert len(findings) == 3 assert module_test.request_count == 5 diff --git a/bbot/test/test_step_2/module_tests/test_module_telerik.py b/bbot/test/test_step_2/module_tests/test_module_telerik.py index c401100bbe..71fa6bf1c4 100644 --- a/bbot/test/test_step_2/module_tests/test_module_telerik.py +++ b/bbot/test/test_step_2/module_tests/test_module_telerik.py @@ -91,7 +91,7 @@ def check(self, module_test, events): telerik_axd_detection = True continue - if e.type == "VULNERABILITY" and "Confirmed Vulnerable Telerik (version: 2014.3.1024)": + if e.type == "FINDING" and "Confirmed Vulnerable Telerik (version: 2014.3.1024)" in e.data["description"]: telerik_axd_vulnerable = True continue diff --git a/bbot/test/test_step_2/module_tests/test_module_trufflehog.py b/bbot/test/test_step_2/module_tests/test_module_trufflehog.py index 37b2e7fb4e..f87fa09735 100644 --- a/bbot/test/test_step_2/module_tests/test_module_trufflehog.py +++ b/bbot/test/test_step_2/module_tests/test_module_trufflehog.py @@ -1157,7 +1157,7 @@ def check(self, module_test, events): vuln_events = [ e for e in events - if e.type == "VULNERABILITY" + if e.type == "FINDING" and ( e.data["host"] == "hub.docker.com" or e.data["host"] == "github.com" @@ -1229,7 +1229,7 @@ def check(self, module_test, events): finding_events = [ e for e in events - if e.type == e.type == "FINDING" + if e.type == "FINDING" and ( e.data["host"] == "hub.docker.com" or e.data["host"] == "github.com" @@ -1321,3 +1321,6 @@ def check(self, module_test, events): finding_events = [e for e in events if e.type == "FINDING"] assert len(finding_events) == 1 assert "Possible Secret Found" in finding_events[0].data["description"] + # Trufflehog emits HIGH severity and MODERATE confidence for possible secrets + assert finding_events[0].data["severity"] == "HIGH" + assert finding_events[0].data["confidence"] == "MODERATE" diff --git a/bbot/test/test_step_2/module_tests/test_module_web_report.py b/bbot/test/test_step_2/module_tests/test_module_web_report.py index cfaa90f217..03e9c4e080 100644 --- a/bbot/test/test_step_2/module_tests/test_module_web_report.py +++ b/bbot/test/test_step_2/module_tests/test_module_web_report.py @@ -9,7 +9,7 @@ class TestWebReport(ModuleTestBase): async def setup_before_prep(self, module_test): # trufflehog --> FINDING # wappalyzer --> TECHNOLOGY - # badsecrets --> VULNERABILITY + # badsecrets --> FINDING respond_args = {"response_data": web_body} module_test.set_expect_requests(respond_args=respond_args) @@ -17,7 +17,9 @@ def check(self, module_test, events): report_file = module_test.scan.home / "web_report.html" with open(report_file) as f: report_content = f.read() - assert "
  • [CRITICAL] Known Secret Found" in report_content + assert "
  • Severity: [CRITICAL] Confidence: [" in report_content + assert "CONFIRMED" in report_content + assert "Known Secret Found" in report_content assert ( """

    URL