Bufferoverrun Module

.github/workflows/docs_updater.yml

@@ -3,6 +3,7 @@ name: Daily Docs Update
 on:
   schedule:
     - cron: '0 0 * * *'  # Runs daily at midnight UTC
+  workflow_dispatch:  # Allows manual triggering
 
 jobs:
   update_docs:
@@ -23,18 +24,22 @@ jobs:
       - name: Generate docs
         run: |
           poetry run bbot/scripts/docs.py
-      - name: Commit changes
-        uses: EndBug/add-and-commit@v9
-        with:
-          add: '["*.md", "docs/data/chord_graph/*.json"]'
-          author_name: "BBOT Docs Autopublish"
-          author_email: [email protected]
-          message: "Refresh module docs"
-      - name: Create Pull Request
+      - name: Commit and Push Changes
+        run: |
+          git config user.name "BBOT Docs Autopublish"
+          git config user.email "[email protected]"
+          git checkout -b update-docs
+          git add "*.md" "docs/data/chord_graph/*.json"
+          git commit -m "Refresh module docs"
+          git push -u origin update-docs --force
+        env:
+          GITHUB_TOKEN: ${{ secrets.BBOT_DOCS_UPDATER_PAT }}
+      - name: Create or Update Pull Request
         uses: peter-evans/create-pull-request@v7
         with:
           token: ${{ secrets.BBOT_DOCS_UPDATER_PAT }}
           branch: update-docs
           base: dev
           title: "Daily Docs Update"
           body: "This is an automated pull request to update the documentation."
+          update-existing: true

bbot/modules/bufferoverrun.py

@@ -0,0 +1,48 @@
+from bbot.modules.templates.subdomain_enum import subdomain_enum_apikey
+
+
+class BufferOverrun(subdomain_enum_apikey):
+    watched_events = ["DNS_NAME"]
+    produced_events = ["DNS_NAME"]
+    flags = ["subdomain-enum", "passive", "safe"]
+    meta = {
+        "description": "Query BufferOverrun's TLS API for subdomains",
+        "created_date": "2024-10-23",
+        "author": "@TheTechromancer",
+        "auth_required": True,
+    }
+    options = {"api_key": "", "commercial": False}
+    options_desc = {"api_key": "BufferOverrun API key", "commercial": "Use commercial API"}
+
+    base_url = "https://tls.bufferover.run/dns"
+    commercial_base_url = "https://bufferover-run-tls.p.rapidapi.com/ipv4/dns"
+
+    async def setup(self):
+        self.commercial = self.config.get("commercial", False)
+        return await super().setup()
+
+    def prepare_api_request(self, url, kwargs):
+        if self.commercial:
+            kwargs["headers"]["x-rapidapi-host"] = "bufferover-run-tls.p.rapidapi.com"
+            kwargs["headers"]["x-rapidapi-key"] = self.api_key
+        else:
+            kwargs["headers"]["x-api-key"] = self.api_key
+        return url, kwargs
+
+    async def request_url(self, query):
+        url = f"{self.commercial_base_url if self.commercial else self.base_url}?q=.{query}"
+        return await self.api_request(url)
+
+    def parse_results(self, r, query):
+        j = r.json()
+        subdomains_set = set()
+        if isinstance(j, dict):
+            results = j.get("Results", [])
+            for result in results:
+                parts = result.split(",")
+                if len(parts) > 4:
+                    subdomain = parts[4].strip()
+                    if subdomain and subdomain.endswith(f".{query}"):
+                        subdomains_set.add(subdomain)
+            for subdomain in subdomains_set:
+                yield subdomain

bbot/test/test_step_2/module_tests/test_module_bufferoverrun.py

@@ -0,0 +1,35 @@
+from .base import ModuleTestBase
+
+
+class TestBufferOverrun(ModuleTestBase):
+    config_overrides = {"modules": {"bufferoverrun": {"api_key": "asdf", "commercial": False}}}
+
+    async def setup_before_prep(self, module_test):
+        # Mock response for non-commercial API
+        module_test.httpx_mock.add_response(
+            url="https://tls.bufferover.run/dns?q=.blacklanternsecurity.com",
+            match_headers={"x-api-key": "asdf"},
+            json={"Results": ["1.2.3.4,example.com,*,*,sub.blacklanternsecurity.com"]},
+        )
+
+    def check(self, module_test, events):
+        assert any(e.data == "sub.blacklanternsecurity.com" for e in events), "Failed to detect subdomain for free API"
+
+
+class TestBufferOverrunCommercial(ModuleTestBase):
+    modules_overrides = ["bufferoverrun"]
+    module_name = "bufferoverrun"
+    config_overrides = {"modules": {"bufferoverrun": {"api_key": "asdf", "commercial": True}}}
+
+    async def setup_before_prep(self, module_test):
+        # Mock response for commercial API
+        module_test.httpx_mock.add_response(
+            url="https://bufferover-run-tls.p.rapidapi.com/ipv4/dns?q=.blacklanternsecurity.com",
+            match_headers={"x-rapidapi-host": "bufferover-run-tls.p.rapidapi.com", "x-rapidapi-key": "asdf"},
+            json={"Results": ["5.6.7.8,blacklanternsecurity.com,*,*,sub.blacklanternsecurity.com"]},
+        )
+
+    def check(self, module_test, events):
+        assert any(
+            e.data == "sub.blacklanternsecurity.com" for e in events
+        ), "Failed to detect subdomain for commercial API"