Merge pull request #2750 from blacklanternsecurity/fix-CVE-2025-10282-again

Fix CVE-2025-10282, again??

Author: TheTechromancer

bbot/modules/gitlab.py

@@ -0,0 +1,31 @@
+from bbot.modules.templates.gitlab import GitLabBaseModule
+
+
+class gitlab_com(GitLabBaseModule):
+    watched_events = ["SOCIAL"]
+    produced_events = [
+        "CODE_REPOSITORY",
+    ]
+    flags = ["active", "safe", "code-enum"]
+    meta = {
+        "description": "Enumerate GitLab SaaS (gitlab.com/org) for projects and groups",
+        "created_date": "2024-03-11",
+        "author": "@TheTechromancer",
+    }
+
+    options = {"api_key": ""}
+    options_desc = {"api_key": "GitLab access token (for gitlab.com/org only)"}
+
+    # This is needed because we are consuming SOCIAL events, which aren't in scope
+    scope_distance_modifier = 2
+
+    async def handle_event(self, event):
+        await self.handle_social(event)
+
+    async def filter_event(self, event):
+        if event.data["platform"] != "gitlab":
+            return False, "platform is not gitlab"
+        _, domain = self.helpers.split_domain(event.host)
+        if domain not in self.saas_domains:
+            return False, "gitlab instance is not gitlab.com/org"
+        return True

bbot/modules/gitlab_com.py

@@ -0,0 +1,84 @@
+from bbot.modules.templates.gitlab import GitLabBaseModule
+
+
+class gitlab_onprem(GitLabBaseModule):
+    watched_events = ["HTTP_RESPONSE", "TECHNOLOGY", "SOCIAL"]
+    produced_events = [
+        "TECHNOLOGY",
+        "SOCIAL",
+        "CODE_REPOSITORY",
+        "FINDING",
+    ]
+    flags = ["active", "safe", "code-enum"]
+    meta = {
+        "description": "Detect self-hosted GitLab instances and query them for repositories",
+        "created_date": "2024-03-11",
+        "author": "@TheTechromancer",
+    }
+
+    # Optional GitLab access token (only required for gitlab.com, but still
+    # supported for on-prem installations that expose private projects).
+    options = {"api_key": ""}
+    options_desc = {"api_key": "GitLab access token (for self-hosted instances only)"}
+
+    # Allow accepting events slightly beyond configured max distance so we can
+    # discover repos on neighbouring infrastructure.
+    scope_distance_modifier = 2
+
+    async def handle_event(self, event):
+        if event.type == "HTTP_RESPONSE":
+            await self.handle_http_response(event)
+        elif event.type == "TECHNOLOGY":
+            await self.handle_technology(event)
+        elif event.type == "SOCIAL":
+            await self.handle_social(event)
+
+    async def filter_event(self, event):
+        # only accept out-of-scope SOCIAL events
+        if event.type == "HTTP_RESPONSE":
+            if event.scope_distance > self.scan.scope_search_distance:
+                return False, "event is out of scope distance"
+        elif event.type == "TECHNOLOGY":
+            if not event.data["technology"].lower().startswith("gitlab"):
+                return False, "technology is not gitlab"
+            if not self.helpers.is_ip(event.host) and self.helpers.tldextract(event.host).domain == "gitlab":
+                return False, "gitlab instance is not self-hosted"
+        elif event.type == "SOCIAL":
+            if event.data["platform"] != "gitlab":
+                return False, "platform is not gitlab"
+            _, domain = self.helpers.split_domain(event.host)
+            if domain in self.saas_domains:
+                return False, "gitlab instance is not self-hosted"
+        return True
+
+    async def handle_http_response(self, event):
+        """Identify GitLab servers from HTTP responses."""
+        headers = event.data.get("header", {})
+        if "x_gitlab_meta" in headers:
+            url = event.parsed_url._replace(path="/").geturl()
+            await self.emit_event(
+                {"host": str(event.host), "technology": "GitLab", "url": url},
+                "TECHNOLOGY",
+                parent=event,
+                context=f"{{module}} detected {{event.type}}: GitLab at {url}",
+            )
+            description = f"GitLab server at {event.host}"
+            await self.emit_event(
+                {"host": str(event.host), "description": description},
+                "FINDING",
+                parent=event,
+                context=f"{{module}} detected {{event.type}}: {description}",
+            )
+
+    async def handle_technology(self, event):
+        """Enumerate projects & groups once we know a host is GitLab."""
+        base_url = self.get_base_url(event)
+
+        # Projects owned by the authenticated user (or public projects if no
+        # authentication).
+        projects_url = self.helpers.urljoin(base_url, "api/v4/projects?simple=true")
+        await self.handle_projects_url(projects_url, event)
+
+        # Group enumeration.
+        groups_url = self.helpers.urljoin(base_url, "api/v4/groups?simple=true")
+        await self.handle_groups_url(groups_url, event)

bbot/modules/gitlab_onprem.py

@@ -0,0 +1,98 @@
+from bbot.modules.base import BaseModule
+
+
+class GitLabBaseModule(BaseModule):
+    """Common functionality for interacting with GitLab instances.
+
+    This template is intended to be inherited by two concrete modules:
+    1. ``gitlab_com``   – Handles public SaaS instances (gitlab.com / gitlab.org).
+    2. ``gitlab_onprem`` – Handles self-hosted, on-premises GitLab servers.
+
+    Both child modules share identical behaviour when talking to the GitLab
+    REST API; they only differ in which events they are willing to accept.
+    """
+
+    # domains owned by GitLab
+    saas_domains = ["gitlab.com", "gitlab.org"]
+
+    async def setup(self):
+        if self.options.get("api_key") is not None:
+            await self.require_api_key()
+        return True
+
+    async def handle_social(self, event):
+        """Enumerate projects belonging to a user or group profile."""
+        username = event.data.get("profile_name", "")
+        if not username:
+            return
+        base_url = self.get_base_url(event)
+        urls = [
+            # User-owned projects
+            self.helpers.urljoin(base_url, f"api/v4/users/{username}/projects?simple=true"),
+            # Group-owned projects
+            self.helpers.urljoin(base_url, f"api/v4/groups/{username}/projects?simple=true"),
+        ]
+        for url in urls:
+            await self.handle_projects_url(url, event)
+
+    async def handle_projects_url(self, projects_url, event):
+        for project in await self.gitlab_json_request(projects_url):
+            project_url = project.get("web_url", "")
+            if project_url:
+                code_event = self.make_event({"url": project_url}, "CODE_REPOSITORY", tags="git", parent=event)
+                await self.emit_event(
+                    code_event,
+                    context=f"{{module}} enumerated projects and found {{event.type}} at {project_url}",
+                )
+            namespace = project.get("namespace", {})
+            if namespace:
+                await self.handle_namespace(namespace, event)
+
+    async def handle_groups_url(self, groups_url, event):
+        for group in await self.gitlab_json_request(groups_url):
+            await self.handle_namespace(group, event)
+
+    async def gitlab_json_request(self, url):
+        """Helper that performs an HTTP request and safely returns JSON list."""
+        response = await self.api_request(url)
+        if response is not None:
+            try:
+                json_data = response.json()
+            except Exception:
+                return []
+            if json_data and isinstance(json_data, list):
+                return json_data
+        return []
+
+    async def handle_namespace(self, namespace, event):
+        namespace_name = namespace.get("path", "")
+        namespace_url = namespace.get("web_url", "")
+        namespace_path = namespace.get("full_path", "")
+
+        if not (namespace_name and namespace_url and namespace_path):
+            return
+
+        namespace_url = self.helpers.parse_url(namespace_url)._replace(path=f"/{namespace_path}").geturl()
+
+        social_event = self.make_event(
+            {
+                "platform": "gitlab",
+                "profile_name": namespace_path,
+                "url": namespace_url,
+            },
+            "SOCIAL",
+            parent=event,
+        )
+        await self.emit_event(
+            social_event,
+            context=f'{{module}} found GitLab namespace ({{event.type}}) "{namespace_name}" at {namespace_url}',
+        )
+
+    # ------------------------------------------------------------------
+    # Utility helpers
+    # ------------------------------------------------------------------
+    def get_base_url(self, event):
+        base_url = event.data.get("url", "")
+        if not base_url:
+            base_url = f"https://{event.host}"
+        return self.helpers.urlparse(base_url)._replace(path="/").geturl()