wrote tests, shodan_port --> internetdb

bbot/modules/shodan_dns.py

@@ -1,7 +1,7 @@
-from bbot.modules.templates.subdomain_enum import subdomain_enum_apikey
+from bbot.modules.templates.shodan import shodan
 
 
-class shodan_dns(subdomain_enum_apikey):
+class shodan_dns(shodan):
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]
     flags = ["subdomain-enum", "passive", "safe"]
@@ -11,12 +11,6 @@ class shodan_dns(subdomain_enum_apikey):
 
     base_url = "https://api.shodan.io"
 
-    async def ping(self):
-        url = f"{self.base_url}/api-info?key={self.api_key}"
-        r = await self.request_with_fail_count(url)
-        resp_content = getattr(r, "text", "")
-        assert getattr(r, "status_code", 0) == 200, resp_content
-
     async def request_url(self, query):
         url = f"{self.base_url}/dns/domain/{self.helpers.quote(query)}?key={self.api_key}"
         response = await self.request_with_fail_count(url)

bbot/modules/templates/github.py

@@ -21,15 +21,15 @@ async def setup(self):
                 self.api_key = api_key
                 self.headers = {"Authorization": f"token {self.api_key}"}
                 break
+        if not self.api_key:
+            if self.auth_required:
+                return None, "No API key set"
         try:
             await self.ping()
             self.hugesuccess(f"API is ready")
             return True
         except Exception as e:
             return None, f"Error with API ({str(e).strip()})"
-        if not self.api_key:
-            if self.auth_required:
-                return None, "No API key set"
         return True
 
     async def ping(self):

bbot/modules/templates/shodan.py

@@ -0,0 +1,34 @@
+from bbot.modules.templates.subdomain_enum import subdomain_enum
+
+
+class shodan(subdomain_enum):
+    options = {"api_key": ""}
+    options_desc = {"api_key": "Shodan API key"}
+
+    base_url = "https://api.shodan.io"
+
+    async def setup(self):
+        await super().setup()
+        self.api_key = None
+        for module_name in ("shodan", "shodan_dns", "shodan_port"):
+            module_config = self.scan.config.get("modules", {}).get(module_name, {})
+            api_key = module_config.get("api_key", "")
+            if api_key:
+                self.api_key = api_key
+                break
+        if not self.api_key:
+            if self.auth_required:
+                return None, "No API key set"
+        try:
+            await self.ping()
+            self.hugesuccess(f"API is ready")
+            return True
+        except Exception as e:
+            return None, f"Error with API ({str(e).strip()})"
+        return True
+
+    async def ping(self):
+        url = f"{self.base_url}/api-info?key={self.api_key}"
+        r = await self.request_with_fail_count(url)
+        resp_content = getattr(r, "text", "")
+        assert getattr(r, "status_code", 0) == 200, resp_content

bbot/test/test_step_2/module_tests/test_module_internetdb.py

@@ -0,0 +1,55 @@
+from .base import ModuleTestBase
+
+
+class TestInternetDB(ModuleTestBase):
+    config_overrides = {"dns_resolution": True}
+
+    async def setup_before_prep(self, module_test):
+        module_test.scan.helpers.mock_dns(
+            {
+                ("blacklanternsecurity.com", "A"): "1.2.3.4",
+                ("autodiscover.blacklanternsecurity.com", "A"): "2.3.4.5",
+                ("mail.blacklanternsecurity.com", "A"): "3.4.5.6",
+            }
+        )
+
+        module_test.httpx_mock.add_response(
+            url="https://internetdb.shodan.io/1.2.3.4",
+            json={
+                "cpes": [
+                    "cpe:/a:microsoft:internet_information_services",
+                    "cpe:/a:microsoft:outlook_web_access:15.0.1367",
+                ],
+                "hostnames": [
+                    "autodiscover.blacklanternsecurity.com",
+                    "mail.blacklanternsecurity.com",
+                ],
+                "ip": "1.2.3.4",
+                "ports": [
+                    25,
+                    80,
+                    443,
+                ],
+                "tags": ["starttls", "self-signed", "eol-os"],
+                "vulns": ["CVE-2021-26857", "CVE-2021-26855"],
+            },
+        )
+
+    def check(self, module_test, events):
+        assert 9 == len([e for e in events if str(e.module) == "internetdb"])
+        assert 1 == len(
+            [e for e in events if e.type == "DNS_NAME" and e.data == "autodiscover.blacklanternsecurity.com"]
+        )
+        assert 1 == len([e for e in events if e.type == "DNS_NAME" and e.data == "mail.blacklanternsecurity.com"])
+        assert 3 == len([e for e in events if e.type == "OPEN_TCP_PORT" and str(e.module) == "internetdb"])
+        assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "blacklanternsecurity.com:443"])
+        assert 2 == len([e for e in events if e.type == "FINDING" and str(e.module) == "internetdb"])
+        assert 1 == len([e for e in events if e.type == "FINDING" and "CVE-2021-26857" in e.data["description"]])
+        assert 2 == len([e for e in events if e.type == "TECHNOLOGY" and str(e.module) == "internetdb"])
+        assert 1 == len(
+            [
+                e
+                for e in events
+                if e.type == "TECHNOLOGY" and e.data["technology"] == "cpe:/a:microsoft:outlook_web_access:15.0.1367"
+            ]
+        )

bbot/test/test_step_2/module_tests/test_module_shodan_dns.py

@@ -2,7 +2,7 @@
 
 
 class TestShodan_DNS(ModuleTestBase):
-    config_overrides = {"modules": {"shodan_dns": {"api_key": "asdf"}}}
+    config_overrides = {"modules": {"shodan": {"api_key": "asdf"}}}
 
     async def setup_before_prep(self, module_test):
         module_test.httpx_mock.add_response(