Restore microsoft bucket module

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])

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 = """<?xml version="1.0" encoding="UTF-8"?>
     <ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>vpn-static</Name><Prefix></Prefix><Marker></Marker><MaxKeys>1000</MaxKeys><IsTruncated>false</IsTruncated><Contents><Key>style.css</Key><LastModified>2017-03-18T06:41:59.000Z</LastModified><ETag>&quot;bf9e72bdab09b785f05ff0395023cc35&quot;</ETag><Size>429</Size><StorageClass>STANDARD</StorageClass></Contents></ListBucketResult>"""
 
@@ -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"]

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"])