From 6fe182ebd4c6119f2cf8b255a634d81e144c831f Mon Sep 17 00:00:00 2001 From: TheTechromancer Date: Sat, 10 Jan 2026 10:29:28 -0500 Subject: [PATCH] restore microsoft bucket --- bbot/modules/bucket_microsoft.py | 37 ++++++++++++ .../module_tests/test_module_bucket_amazon.py | 5 +- .../test_module_bucket_microsoft.py | 58 +++++++++++++++++++ 3 files changed, 99 insertions(+), 1 deletion(-) create mode 100644 bbot/modules/bucket_microsoft.py create mode 100644 bbot/test/test_step_2/module_tests/test_module_bucket_microsoft.py diff --git a/bbot/modules/bucket_microsoft.py b/bbot/modules/bucket_microsoft.py new file mode 100644 index 0000000000..88cf887e85 --- /dev/null +++ b/bbot/modules/bucket_microsoft.py @@ -0,0 +1,37 @@ +from bbot.modules.templates.bucket import bucket_template + + +class bucket_microsoft(bucket_template): + watched_events = ["DNS_NAME", "STORAGE_BUCKET"] + produced_events = ["STORAGE_BUCKET", "FINDING"] + flags = ["active", "safe", "cloud-enum", "web-basic"] + meta = { + "description": "Check for Azure storage blobs related to target", + "created_date": "2022-11-04", + "author": "@TheTechromancer", + } + options = {"permutations": False} + options_desc = { + "permutations": "Whether to try permutations", + } + + cloudcheck_provider_name = "Microsoft" + delimiters = ("", "-") + base_domains = ["blob.core.windows.net"] + # Dirbusting is required to know whether a bucket is public + supports_open_check = False + + def build_bucket_request(self, bucket_name, base_domain, region): + url = self.build_url(bucket_name, base_domain, region) + url = url.strip("/") + f"/{bucket_name}?restype=container" + return url, {} + + def check_bucket_exists(self, bucket_name, response): + status_code = getattr(response, "status_code", 0) + existent_bucket = status_code != 0 + self.critical(f"Bucket exists: {existent_bucket} (response: {response})") + return existent_bucket, set() + + def clean_bucket_url(self, url): + # only return root URL + return "/".join(url.split("/")[:3]) diff --git a/bbot/test/test_step_2/module_tests/test_module_bucket_amazon.py b/bbot/test/test_step_2/module_tests/test_module_bucket_amazon.py index 822017ac5b..692b571e36 100644 --- a/bbot/test/test_step_2/module_tests/test_module_bucket_amazon.py +++ b/bbot/test/test_step_2/module_tests/test_module_bucket_amazon.py @@ -21,6 +21,8 @@ class Bucket_Amazon_Base(ModuleTestBase): random_bucket_2 = f"{random_bucket_name_2}.s3-ap-southeast-2.amazonaws.com" random_bucket_3 = f"{random_bucket_name_3}.s3.amazonaws.com" + nonexistent_is_404 = True + open_bucket_body = """ vpn-static1000falsestyle.css2017-03-18T06:41:59.000Z"bf9e72bdab09b785f05ff0395023cc35"429STANDARD""" @@ -66,7 +68,8 @@ async def setup_after_prep(self, module_test): url=self.url_3, text="", ) - module_test.httpx_mock.add_response(url=re.compile(".*"), text="", status_code=404) + if self.nonexistent_is_404: + module_test.httpx_mock.add_response(url=re.compile(".*"), text="", status_code=404) def check(self, module_test, events): storage_buckets = [e for e in events if e.type == "STORAGE_BUCKET"] diff --git a/bbot/test/test_step_2/module_tests/test_module_bucket_microsoft.py b/bbot/test/test_step_2/module_tests/test_module_bucket_microsoft.py new file mode 100644 index 0000000000..463f79033b --- /dev/null +++ b/bbot/test/test_step_2/module_tests/test_module_bucket_microsoft.py @@ -0,0 +1,58 @@ +from .test_module_bucket_amazon import * +from .base import ModuleTestBase + + +class TestBucket_Microsoft(Bucket_Amazon_Base): + provider = "microsoft" + random_bucket_1 = f"{random_bucket_name_1}.blob.core.windows.net" + random_bucket_2 = f"{random_bucket_name_2}.blob.core.windows.net" + random_bucket_3 = f"{random_bucket_name_3}.blob.core.windows.net" + + nonexistent_is_404 = False + + 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}/{random_bucket_name_3}?restype=container" + + +class TestBucket_Microsoft_NoDup(ModuleTestBase): + targets = ["tesla.com"] + module_name = "bucket_microsoft" + config_overrides = {"cloudcheck": True} + + async def setup_before_prep(self, module_test): + module_test.httpx_mock.add_response( + url="https://tesla.blob.core.windows.net/tesla?restype=container", + text="", + ) + await module_test.mock_dns( + { + "tesla.com": {"A": ["1.2.3.4"]}, + "tesla.blob.core.windows.net": {"A": ["1.2.3.4"]}, + } + ) + + def check(self, module_test, events): + assert 1 == len([e for e in events if e.type == "STORAGE_BUCKET"]) + bucket_event = [e for e in events if e.type == "STORAGE_BUCKET"][0] + assert bucket_event.data["name"] == "tesla" + assert bucket_event.data["url"] == "https://tesla.blob.core.windows.net/" + assert ( + bucket_event.discovery_context + == f"bucket_azure tried bucket variations of {event.data} and found {{event.type}} at {url}" + ) + + +class TestBucket_Microsoft_NoDup(TestBucket_Microsoft_NoDup): + """ + This tests _suppress_chain_dupes functionality to make sure it works as expected + """ + + async def setup_after_prep(self, module_test): + from bbot.core.event.base import STORAGE_BUCKET + + module_test.monkeypatch.setattr(STORAGE_BUCKET, "_suppress_chain_dupes", False) + + def check(self, module_test, events): + assert 2 == len([e for e in events if e.type == "STORAGE_BUCKET"])