add tests

bbot/core/event/base.py

@@ -934,6 +934,7 @@ class STORAGE_BUCKET(DictEvent, URL_UNVERIFIED):
     class _data_validator(BaseModel):
         name: str
         url: str
+        _validate_url = field_validator("url")(validators.validate_url)
 
     def _words(self):
         return self.data["name"]
@@ -1009,6 +1010,7 @@ class _data_validator(BaseModel):
         severity: str
         description: str
         url: Optional[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)
 
@@ -1023,6 +1025,7 @@ class _data_validator(BaseModel):
         host: str
         description: str
         url: Optional[str] = None
+        _validate_url = field_validator("url")(validators.validate_url)
         _validate_host = field_validator("host")(validators.validate_host)
 
     def _pretty_string(self):
@@ -1034,6 +1037,7 @@ class _data_validator(BaseModel):
         host: str
         technology: str
         url: Optional[str] = None
+        _validate_url = field_validator("url")(validators.validate_url)
         _validate_host = field_validator("host")(validators.validate_host)
 
     def _data_id(self):
@@ -1050,6 +1054,7 @@ class _data_validator(BaseModel):
         host: str
         vhost: str
         url: Optional[str] = None
+        _validate_url = field_validator("url")(validators.validate_url)
         _validate_host = field_validator("host")(validators.validate_host)
 
     def _pretty_string(self):

bbot/modules/bucket_digitalocean.py

@@ -17,4 +17,4 @@ class bucket_digitalocean(bucket_template):
     regions = ["ams3", "fra1", "nyc3", "sfo2", "sfo3", "sgp1"]
 
     def build_url(self, bucket_name, base_domain, region):
-        return f"https://{bucket_name}.{region}.{base_domain}"
+        return f"https://{bucket_name}.{region}.{base_domain}/"

bbot/modules/bucket_file_enum.py

@@ -0,0 +1,34 @@
+from bbot.modules.base import BaseModule
+import xml.etree.ElementTree as ET
+
+
+class bucket_file_enum(BaseModule):
+    """
+    Enumerate files in a public bucket
+    """
+
+    watched_events = ["STORAGE_BUCKET"]
+    produced_events = ["URL_UNVERIFIED"]
+    meta = {
+        "description": "Works in conjunction with the filedownload module to download files from open storage buckets. Currently supported cloud providers: AWS"
+    }
+    flags = ["passive", "safe", "cloud-enum"]
+    scope_distance_modifier = 2
+
+    async def handle_event(self, event):
+        cloud_tags = (t for t in event.tags if t.startswith("cloud-"))
+        if any(t.endswith("-amazon") or t.endswith("-digitalocean") for t in cloud_tags):
+            await self.handle_aws(event)
+
+    async def handle_aws(self, event):
+        url = event.data["url"]
+        response = await self.helpers.request(url)
+        status_code = getattr(response, "status_code", 0)
+        if status_code == 200:
+            content = response.text
+            root = ET.fromstring(content)
+            namespace = {"s3": "http://s3.amazonaws.com/doc/2006-03-01/"}
+            keys = [key.text for key in root.findall(".//s3:Key", namespace)]
+            for key in keys:
+                bucket_file = url + "/" + key
+                self.emit_event(bucket_file, "URL_UNVERIFIED", source=event, tags="filedownload")

bbot/modules/filedownload.py

@@ -77,7 +77,7 @@ class filedownload(BaseModule):
         "max_filesize": "Cancel download if filesize is greater than this size",
     }
 
-    scope_distance_modifier = 1
+    scope_distance_modifier = 3
 
     async def setup(self):
         self.extensions = list(set([e.lower().strip(".") for e in self.options.get("extensions", [])]))
@@ -101,8 +101,11 @@ async def filter_event(self, event):
         # accept file download requests from other modules
         if "filedownload" in event.tags:
             return True
-        if self.hash_event(event) in self.urls_downloaded:
-            return False, f"Already processed {event}"
+        else:
+            if event.scope_distance > 1:
+                return False, f"{event} not within scope distance"
+            elif self.hash_event(event) in self.urls_downloaded:
+                return False, f"Already processed {event}"
         return True
 
     def hash_event(self, event):
@@ -113,7 +116,9 @@ def hash_event(self, event):
     async def handle_event(self, event):
         if event.type == "URL_UNVERIFIED":
             url_lower = event.data.lower()
-            if any(url_lower.endswith(f".{e}") for e in self.extensions):
+            extension_matches = any(url_lower.endswith(f".{e}") for e in self.extensions)
+            filedownload_requested = "filedownload" in event.tags
+            if extension_matches or filedownload_requested:
                 await self.download_file(event.data)
         elif event.type == "HTTP_RESPONSE":
             content_type = event.data["header"].get("content_type", "")

bbot/modules/templates/bucket.py

@@ -126,7 +126,7 @@ def valid_bucket_name(self, bucket_name):
         return False
 
     def build_url(self, bucket_name, base_domain, region):
-        return f"https://{bucket_name}.{base_domain}"
+        return f"https://{bucket_name}.{base_domain}/"
 
     def gen_tags_exists(self, response):
         return set()

bbot/test/conftest.py

@@ -18,14 +18,14 @@ def pytest_sessionfinish(session, exitstatus):
             logger.removeHandler(handler)
 
     # Wipe out BBOT home dir
-    shutil.rmtree("/tmp/.bbot_test", ignore_errors=True)
+    # shutil.rmtree("/tmp/.bbot_test", ignore_errors=True)
 
     yield
 
 
 @pytest.fixture
 def non_mocked_hosts() -> list:
-    return ["127.0.0.1", "localhost", "githubusercontent.com"] + interactsh_servers
+    return ["127.0.0.1", "localhost", "raw.githubusercontent.com"] + interactsh_servers
 
 
 @pytest.fixture

bbot/test/test_step_1/test_cloud_helpers.py

@@ -9,7 +9,6 @@ async def test_cloud_helpers(bbot_scanner, bbot_config):
     for provider_name in provider_names:
         assert provider_name in scan1.helpers.cloud.providers.providers
 
-    log.critical(scan1.helpers.cloud.providers.providers)
     for p in scan1.helpers.cloud.providers.providers.values():
         print(f"{p.name}: {p.domains} / {p.ranges}")
     amazon_ranges = list(scan1.helpers.cloud["amazon"].ranges)
@@ -30,12 +29,10 @@ async def test_cloud_helpers(bbot_scanner, bbot_config):
     other_event3._resolved_hosts = {"asdf.amazonaws.com"}
 
     for event in (ip_event, aws_event1, aws_event2, aws_event4, other_event2, other_event3):
-        log.critical(event)
         await scan1.helpers.cloud.tag_event(event)
         assert "cloud-amazon" in event.tags, f"{event} was not properly cloud-tagged"
 
     for event in (aws_event3, other_event1):
-        log.critical(event)
         await scan1.helpers.cloud.tag_event(event)
         assert "cloud-amazon" not in event.tags, f"{event} was improperly cloud-tagged"
         assert not any(

bbot/test/test_step_2/module_tests/test_module_bucket_amazon.py

@@ -37,9 +37,9 @@ def modules_overrides(self):
         return ["excavate", "speculate", "httpx", self.module_name]
 
     def url_setup(self):
-        self.url_1 = f"https://{self.random_bucket_1}"
-        self.url_2 = f"https://{self.random_bucket_2}"
-        self.url_3 = f"https://{self.random_bucket_3}"
+        self.url_1 = f"https://{self.random_bucket_1}/"
+        self.url_2 = f"https://{self.random_bucket_2}/"
+        self.url_3 = f"https://{self.random_bucket_3}/"
 
     def bucket_setup(self):
         self.url_setup()
@@ -83,14 +83,14 @@ def check(self, module_test, events):
                     url = e.data.get("url", "")
                     assert self.random_bucket_2 in url
                     assert not self.random_bucket_1 in url
-                    assert not f"{self.random_bucket_3}" in url
+                    assert not self.random_bucket_3 in url
         # make sure bucket mutations were found
         assert any(
             e.type == "STORAGE_BUCKET"
             and str(e.module) == self.module_name
             and f"{random_bucket_name_3}" in e.data["url"]
             for e in events
-        ), f'bucket (dev mutation) not found for module "{self.module_name}"'
+        ), f'bucket (dev mutation: {self.random_bucket_3}) not found for module "{self.module_name}"'
 
 
 class TestBucket_Amazon(Bucket_Amazon_Base):

bbot/test/test_step_2/module_tests/test_module_bucket_file_enum.py

@@ -0,0 +1,32 @@
+from .base import ModuleTestBase
+
+
+class TestBucket_File_Enum(ModuleTestBase):
+    targets = ["http://127.0.0.1:8888"]
+    modules_overrides = ["bucket_file_enum", "filedownload", "httpx", "excavate"]
+
+    open_bucket_url = "https://testbucket.s3.amazonaws.com/"
+    open_bucket_body = """<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>testbucket</Name><Prefix></Prefix><Marker></Marker><MaxKeys>1000</MaxKeys><IsTruncated>false</IsTruncated><Contents><Key>index.html</Key><LastModified>2023-05-22T23:04:38.000Z</LastModified><ETag>&quot;4a2d2d114f3abf90f8bd127c1f25095a&quot;</ETag><Size>5</Size><StorageClass>STANDARD</StorageClass></Contents><Contents><Key>test.pdf</Key><LastModified>2022-04-30T21:13:40.000Z</LastModified><ETag>&quot;723b0018c2f5a7ef06a34f84f6fa97e4&quot;</ETag><Size>388901</Size><StorageClass>STANDARD</StorageClass></Contents></ListBucketResult>"""
+
+    pdf_data = """%PDF-1.
+1 0 obj<</Pages 2 0 R>>endobj
+2 0 obj<</Kids[3 0 R]/Count 1>>endobj
+3 0 obj<</Parent 2 0 R>>endobj
+trailer <</Root 1 0 R>>"""
+
+    async def setup_before_prep(self, module_test):
+        module_test.httpserver.expect_request("/").respond_with_data(f'<a href="{self.open_bucket_url}"/>')
+        module_test.httpx_mock.add_response(
+            url=self.open_bucket_url,
+            text=self.open_bucket_body,
+        )
+        module_test.httpx_mock.add_response(
+            url=f"{self.open_bucket_url}test.pdf",
+            text=self.pdf_data,
+            headers={"Content-Type": "application/pdf"},
+        )
+
+    def check(self, module_test, events):
+        download_dir = module_test.scan.home / "filedownload"
+        files = list(download_dir.glob("*.pdf"))
+        assert any(f.name.endswith("test.pdf") for f in files), "Failed to download PDF file from open bucket"