update USERNAME event type to fall back to EMAIL_ADDRESS

bbot/core/event/base.py

@@ -1091,6 +1091,17 @@ class HASHED_PASSWORD(BaseEvent):
 class USERNAME(BaseEvent):
     _always_emit = True
 
+    def __new__(cls, data, *args, **kwargs):
+        # if the data is an email, emit as an email instead
+        if validators.soft_validate(data, "email"):
+            log.critical(f"{data} is an email")
+            tags = set(kwargs.get("tags", []))
+            # add affiliate tag so the event is always emitted regardless of scope distance
+            tags.add("affiliate")
+            kwargs["tags"] = tags
+            return EMAIL_ADDRESS(data, *args, **kwargs)
+        return super().__new__(cls)
+
 
 class SOCIAL(DictEvent):
     _always_emit = True

bbot/modules/dehashed.py

@@ -6,7 +6,7 @@
 class dehashed(credential_leak):
     watched_events = ["DNS_NAME"]
     produced_events = ["PASSWORD", "HASHED_PASSWORD", "USERNAME"]
-    flags = ["passive"]
+    flags = ["passive", "safe", "email-enum"]
     meta = {"description": "Execute queries against dehashed.com for exposed credentials", "auth_required": True}
     options = {"username": "", "api_key": ""}
     options_desc = {"username": "Email Address associated with your API key", "api_key": "DeHashed API Key"}

bbot/test/test_step_2/module_tests/test_module_dehashed.py

@@ -7,7 +7,7 @@
             "id": "4363462346",
             "email": "[email protected]",
             "ip_address": "",
-            "username": "",
+            "username": "[email protected]",
             "password": "",
             "hashed_password": "$2a$12$pVmwJ7pXEr3mE.DmCCE4fOUDdeadbeefd2KuCy/tq1ZUFyEOH2bve",
             "name": "Bob Smith",
@@ -46,8 +46,11 @@ async def setup_before_prep(self, module_test):
         )
 
     def check(self, module_test, events):
-        assert len(events) == 7
+        for e in events:
+            module_test.log.critical(e)
+        assert len(events) == 8
         assert 1 == len([e for e in events if e.type == "EMAIL_ADDRESS" and e.data == "[email protected]"])
+        assert 1 == len([e for e in events if e.type == "EMAIL_ADDRESS" and e.data == "[email protected]" and e.scope_distance == 1 and "affiliate" in e.tags] and e.source.data == "[email protected]")
         assert 1 == len([e for e in events if e.type == "EMAIL_ADDRESS" and e.data == "[email protected]"])
         assert 1 == len(
             [