Cloud helpers + Storage bucket modules

bbot/core/configurator/args.py

@@ -80,8 +80,8 @@ def error(self, message):
     web spider + advanced web scan:
         bbot -t www.evilcorp.com -m httpx -f web-basic web-advanced -c web_spider_distance=2 web_spider_depth=2
 
-    subdomains + emails + portscan + screenshots + nuclei:
-        bbot -t evilcorp.com -f subdomain-enum email-enum web-basic -m naabu gowitness nuclei --allow-deadly
+    subdomains + emails + cloud buckets + portscan + screenshots + nuclei:
+        bbot -t evilcorp.com -f subdomain-enum email-enum cloud-enum web-basic -m naabu gowitness nuclei --allow-deadly
 
 """
 

bbot/core/event/base.py

@@ -637,7 +637,7 @@ class _data_validator(BaseModel):
         url: str
 
     def _words(self):
-        return self.parsed.hostname.split(".")[0]
+        return self.data["name"]
 
 
 class URL_HINT(URL_UNVERIFIED):

bbot/core/helpers/cloud/__init__.py

@@ -0,0 +1,33 @@
+import importlib
+from pathlib import Path
+
+from .base import BaseCloudProvider
+
+# dynamically load cloud provider modules
+provider_list = []
+for file in Path(__file__).parent.glob("*.py"):
+    if not file.stem in ("base", "__init__"):
+        import_path = f"bbot.core.helpers.cloud.{file.stem}"
+        module_variables = importlib.import_module(import_path, "bbot")
+        for variable in module_variables.__dict__.keys():
+            value = getattr(module_variables, variable)
+            if hasattr(value, "__mro__") and not value == BaseCloudProvider and BaseCloudProvider in value.__mro__:
+                provider_list.append(value)
+
+
+class CloudProviders:
+    def __init__(self, parent_helper):
+        self.parent_helper = parent_helper
+        self.providers = {}
+        for provider_class in provider_list:
+            provider_name = str(provider_class.__name__).lower()
+            provider = provider_class(self.parent_helper)
+            self.providers[provider_name] = provider
+            setattr(self, provider_name, provider)
+
+    def excavate(self, event):
+        for provider in self.providers.values():
+            provider.excavate(event)
+
+    def __iter__(self):
+        yield from self.providers.values()

bbot/core/helpers/cloud/aws.py

@@ -0,0 +1,23 @@
+from .base import BaseCloudProvider
+
+
+class AWS(BaseCloudProvider):
+
+    domains = [
+        "amazon-dss.com",
+        "amazonaws.com",
+        "amazonaws.com.cn",
+        "amazonaws.org",
+        "amazonses.com",
+        "amazonwebservices.com",
+        "aws",
+        "aws.a2z.com",
+        "aws.amazon.com",
+        "aws.dev",
+        "awsstatic.com",
+        "elasticbeanstalk.com",
+    ]
+    bucket_name_regex = r"[a-z0-9_][a-z0-9-\.]{1,61}[a-z0-9]"
+    regexes = {
+        "STORAGE_BUCKET": [r"(%[a-f0-9]{2})?(" + bucket_name_regex + r")\.(s3-?(?:[a-z0-9-]*\.){1,2}amazonaws\.com)"]
+    }

bbot/core/helpers/cloud/azure.py

@@ -0,0 +1,25 @@
+from .base import BaseCloudProvider
+
+
+class Azure(BaseCloudProvider):
+    domains = [
+        "windows.net",
+        "azure.com",
+        "azmk8s.io",
+        "azure-api.net",
+        "azure-mobile.net",
+        "azurecontainer.io",
+        "azurecr.io",
+        "azureedge.net",
+        "azurefd.net",
+        "azurewebsites.net",
+        "cloudapp.net",
+        "onmicrosoft.com",
+        "trafficmanager.net",
+        "vault.azure.net",
+        "visualstudio.com",
+        "vo.msecnd.net",
+    ]
+
+    bucket_name_regex = r"[a-z0-9][a-z0-9-_\.]{1,61}[a-z0-9]"
+    regexes = {"STORAGE_BUCKET": [r"(%[a-f0-9]{2})?(" + bucket_name_regex + r")\.(blob\.core\.windows\.net)"]}

bbot/core/helpers/cloud/base.py

@@ -0,0 +1,66 @@
+import re
+import logging
+
+log = logging.getLogger("bbot.helpers.cloud.provider")
+
+
+class BaseCloudProvider:
+
+    domains = []
+    regexes = {}
+
+    def __init__(self, parent_helper):
+        self.parent_helper = parent_helper
+        self.name = str(self.__class__.__name__).lower()
+        self.dummy_module = self.parent_helper._make_dummy_module(f"{self.name}_cloud", _type="scan")
+        self.bucket_name_regex = re.compile("^" + self.bucket_name_regex + "$", re.I)
+        self.signatures = {}
+        self.domain_regexes = []
+        for domain in self.domains:
+            self.domain_regexes.append(re.compile(r"^(?:[\w\-]+\.)*" + rf"{re.escape(domain)}$"))
+        for event_type, regexes in self.regexes.items():
+            self.signatures[event_type] = [re.compile(r, re.I) for r in regexes]
+
+    @property
+    def base_tags(self):
+        return {f"cloud-{self.name}"}
+
+    def excavate(self, event):
+        base_kwargs = dict(source=event, tags=self.base_tags)
+
+        # check for buckets in HTTP responses
+        if event.type == "HTTP_RESPONSE":
+            body = event.data.get("response-body", "")
+            for event_type, sigs in self.signatures.items():
+                found = set()
+                for sig in sigs:
+                    for match in sig.findall(body):
+                        kwargs = dict(base_kwargs)
+                        kwargs["event_type"] = event_type
+                        if not match in found:
+                            found.add(match)
+                            if event_type == "STORAGE_BUCKET":
+                                self.emit_bucket(match, **kwargs)
+                            else:
+                                self.emit_event(**kwargs)
+
+    def emit_bucket(self, match, **kwargs):
+        _, bucket_name, bucket_domain = match
+        kwargs["data"] = {"name": bucket_name, "url": f"https://{bucket_name}.{bucket_domain}"}
+        self.emit_event(**kwargs)
+
+    def emit_event(self, *args, **kwargs):
+        excavate_module = self.parent_helper.scan.modules.get("excavate", None)
+        if excavate_module:
+            event = self.dummy_module.make_event(*args, **kwargs)
+            excavate_module.emit_event(event)
+
+    def is_valid_bucket(self, bucket_name):
+        return self.bucket_name_regex.match(bucket_name)
+
+    def tag_event(self, event):
+        if event.host and isinstance(event.host, str):
+            for r in self.domain_regexes:
+                if r.match(event.host):
+                    event.tags.update(self.base_tags)
+                    break

bbot/core/helpers/cloud/gcp.py

@@ -0,0 +1,13 @@
+from .base import BaseCloudProvider
+
+
+class GCP(BaseCloudProvider):
+    domains = [
+        "googleapis.cn",
+        "googleapis.com",
+        "cloud.google.com",
+        "gcp.gvt2.com",
+    ]
+
+    bucket_name_regex = r"[a-z0-9][a-z0-9-_\.]{1,61}[a-z0-9]"
+    regexes = {"STORAGE_BUCKET": [r"(%[a-f0-9]{2})?(" + bucket_name_regex + r")\.(storage\.googleapis\.com)"]}

bbot/core/helpers/helper.py

@@ -6,11 +6,13 @@
 from .dns import DNSHelper
 from .diff import HttpCompare
 from .wordcloud import WordCloud
+from .cloud import CloudProviders
 from .interactsh import Interactsh
 from .threadpool import as_completed
 from ...modules.base import BaseModule
 from .depsinstaller import DepsInstaller
 
+
 log = logging.getLogger("bbot.core.helpers")
 
 
@@ -48,6 +50,9 @@ def __init__(self, config, scan=None):
         self.word_cloud = WordCloud(self)
         self.dummy_modules = {}
 
+        # cloud helpers
+        self.cloud = CloudProviders(self)
+
     def interactsh(self):
         return Interactsh(self)
 

bbot/modules/bucket_aws.py

@@ -0,0 +1,81 @@
+from bbot.modules.base import BaseModule
+
+
+class bucket_aws(BaseModule):
+    watched_events = ["DNS_NAME"]
+    produced_events = ["STORAGE_BUCKET"]
+    flags = ["active", "safe", "cloud-enum"]
+    meta = {"description": "Check for S3 buckets related to target"}
+    options = {"max_threads": 10, "permutations": False}
+    options_desc = {
+        "max_threads": "Maximum number of threads for HTTP requests",
+        "permutations": "Whether to try permutations",
+    }
+
+    cloud_helper_name = "aws"
+    delimiters = ("", ".", "-")
+    base_domains = ["s3.amazonaws.com"]
+
+    def setup(self):
+        self.buckets_tried = set()
+        self.cloud_helper = getattr(self.helpers.cloud, self.cloud_helper_name)
+        self.permutations = self.config.get("permutations", False)
+        return True
+
+    def handle_event(self, event):
+        buckets = set()
+        base = event.data
+        stem = self.helpers.domain_stem(base)
+        for b in [base, stem]:
+            split = b.split(".")
+            for d in self.delimiters:
+                bucket_name = d.join(split)
+                if self.valid_bucket_name(bucket_name):
+                    buckets.add(bucket_name)
+
+        if self.permutations:
+            for b in [stem, base]:
+                for mutation in self.helpers.word_cloud.mutations(b):
+                    for d in self.delimiters:
+                        bucket_name = d.join(mutation)
+                        if self.valid_bucket_name(bucket_name):
+                            buckets.add(bucket_name)
+
+        for bucket_name in buckets:
+            for base_domain in self.base_domains:
+                self.submit_task(self.check_bucket, bucket_name, base_domain, event)
+
+    def valid_bucket_name(self, bucket_name):
+        valid = self.cloud_helper.is_valid_bucket(bucket_name)
+        if valid and not self.helpers.is_ip(bucket_name):
+            bucket_hash = hash(bucket_name)
+            if not bucket_hash in self.buckets_tried:
+                self.buckets_tried.add(bucket_hash)
+                return True
+        return False
+
+    def build_url(self, bucket_name, base_domain):
+        return f"https://{bucket_name}.{base_domain}"
+
+    def gen_tags(self, bucket_name, web_response):
+        tags = []
+        status_code = getattr(web_response, "status_code", 404)
+        content = getattr(web_response, "text", "")
+        if status_code == 200 and "Contents" in content:
+            tags.append("open-bucket")
+        return tags
+
+    def check_bucket(self, bucket_name, base_domain, event):
+        url = self.build_url(bucket_name, base_domain)
+        web_response = self.helpers.request(url)
+        tags = []
+        existent_bucket, event_type = self.check_response(bucket_name, web_response)
+        if existent_bucket:
+            tags = list(set(self.gen_tags(bucket_name, web_response) + tags))
+            self.emit_event({"name": bucket_name, "url": url}, event_type, source=event, tags=tags)
+
+    def check_response(self, bucket_name, web_response):
+        status_code = getattr(web_response, "status_code", 404)
+        existent_bucket = status_code != 404
+        event_type = "STORAGE_BUCKET"
+        return existent_bucket, event_type

bbot/modules/bucket_azure.py

@@ -0,0 +1,26 @@
+from bbot.modules.bucket_aws import bucket_aws
+
+
+class bucket_azure(bucket_aws):
+    watched_events = ["DNS_NAME"]
+    produced_events = ["STORAGE_BUCKET"]
+    flags = ["active", "safe", "cloud-enum"]
+    meta = {"description": "Check for Azure storage blobs related to target"}
+    options = {"max_threads": 10, "permutations": False}
+    options_desc = {
+        "max_threads": "Maximum number of threads for HTTP requests",
+        "permutations": "Whether to try permutations",
+    }
+
+    cloud_helper_name = "azure"
+    delimiters = ("", "-")
+    base_domains = ["blob.core.windows.net"]
+
+    def gen_tags(self, *args, **kwargs):
+        return []
+
+    def check_response(self, bucket_name, web_response):
+        status_code = getattr(web_response, "status_code", 0)
+        existent_bucket = status_code != 0
+        event_type = "STORAGE_BUCKET"
+        return existent_bucket, event_type

bbot/modules/bucket_gcp.py

@@ -0,0 +1,50 @@
+from bbot.modules.bucket_aws import bucket_aws
+
+
+class bucket_gcp(bucket_aws):
+    """
+    Adapted from https://github.com/RhinoSecurityLabs/GCPBucketBrute/blob/master/gcpbucketbrute.py
+    """
+
+    watched_events = ["DNS_NAME"]
+    produced_events = ["STORAGE_BUCKET"]
+    flags = ["active", "safe", "cloud-enum"]
+    meta = {"description": "Check for Google object storage related to target"}
+    options = {"max_threads": 10, "permutations": False}
+    options_desc = {
+        "max_threads": "Maximum number of threads for HTTP requests",
+        "permutations": "Whether to try permutations",
+    }
+
+    cloud_helper_name = "gcp"
+    delimiters = ("", "-", ".", "_")
+    base_domains = ["storage.googleapis.com"]
+    bad_permissions = [
+        "storage.buckets.setIamPolicy",
+        "storage.objects.list",
+        "storage.objects.get",
+        "storage.objects.create",
+    ]
+
+    def build_url(self, bucket_name, base_domain):
+        return f"https://www.googleapis.com/storage/v1/b/{bucket_name}"
+
+    def gen_tags(self, bucket_name, web_response):
+        tags = []
+        try:
+            list_permissions = "&".join(["=".join(("permissions", p)) for p in self.bad_permissions])
+            url = f"https://www.googleapis.com/storage/v1/b/{bucket_name}/iam/testPermissions?" + list_permissions
+            permissions = self.helpers.request(url).json()
+            if isinstance(permissions, dict):
+                permissions = permissions.get("permissions", {})
+                if any(p in permissions for p in self.bad_permissions):
+                    tags.append("open-bucket")
+        except Exception as e:
+            self.warning(f'Failed to enumerate permissions for bucket "{bucket_name}": {e}')
+        return tags
+
+    def check_response(self, bucket_name, web_response):
+        status_code = getattr(web_response, "status_code", 0)
+        existent_bucket = status_code not in (0, 400, 404)
+        event_type = "STORAGE_BUCKET"
+        return existent_bucket, event_type

bbot/modules/internal/excavate.py

@@ -286,3 +286,6 @@ def handle_event(self, event):
                 [self.hostname, self.url, self.email, self.error_extractor, self.jwt, self.serialization],
                 event,
             )
+
+        # Cloud extractors
+        self.helpers.cloud.excavate(event)