feat(bloodhound): add Active Directory attack-path injector (#316)

bloodhound/.dockerignore

@@ -0,0 +1,2 @@
+config.yml
+**/__pycache__

bloodhound/.env.sample

@@ -0,0 +1,13 @@
+# OPENAEV Environment Variables
+# base URL to reach the OpenAEV server
+# note this URL must be routable from inside the container
+# so `localhost` will most likely not work
+OPENAEV_URL=ChangeMe
+# admin account API token from the OpenAEV server
+OPENAEV_TOKEN=ChangeMe
+OPENAEV_TENANT_ID=ChangeMe
+
+# INJECTOR Environment Variables
+INJECTOR_ID=bloodhound--ChangeMe
+INJECTOR_NAME=BloodHound AD
+INJECTOR_LOG_LEVEL=error

bloodhound/Dockerfile

@@ -0,0 +1,52 @@
+FROM python:3.13-alpine AS builder
+
+ENV PIP_VERSION=25.0.1
+
+RUN apk update && apk upgrade && apk add git curl
+
+WORKDIR /opt/injector_common
+COPY --from=injector_common ./ ./
+
+WORKDIR /
+RUN git clone https://github.com/OpenAEV-Platform/client-python
+
+RUN curl -sS https://bootstrap.pypa.io/get-pip.py -o get-pip.py && \
+    python3 get-pip.py pip==${PIP_VERSION} && \
+    rm get-pip.py
+
+RUN python3 -m pip install poetry==2.3.2 \
+    && poetry config installer.re-resolve false \
+    && poetry config virtualenvs.create false
+
+ARG installdir=/opt/injector
+ADD . ${installdir}
+WORKDIR ${installdir}
+RUN poetry install && \
+    python3 -m pip install --no-cache-dir pip==${PIP_VERSION}
+
+FROM python:3.13-alpine AS runner
+
+ENV PIP_VERSION=25.0.1
+
+WORKDIR /opt/injector_common
+COPY --from=injector_common ./ ./
+
+ARG installdir=/opt/injector
+WORKDIR ${installdir}
+COPY --from=builder ${installdir} ${installdir}
+COPY --from=builder /usr/local/lib/python3.13/site-packages /usr/local/lib/python3.13/site-packages
+
+ARG PYOAEV_GIT_BRANCH_OVERRIDE
+
+RUN if [[ ${PYOAEV_GIT_BRANCH_OVERRIDE} ]] ; then \
+        echo "Forcing specific version of client-python" && \
+        apk add --no-cache git curl && \
+        curl -sS https://bootstrap.pypa.io/get-pip.py -o get-pip.py && \
+        python3 get-pip.py pip==${PIP_VERSION} && \
+        rm get-pip.py && \
+        pip install pip3-autoremove && \
+        pip-autoremove pyoaev -y && \
+        pip install git+https://github.com/OpenAEV-Platform/client-python@${PYOAEV_GIT_BRANCH_OVERRIDE} ; \
+    fi
+
+CMD ["python3", "-m", "bloodhound_injector.openaev_bloodhound"]

bloodhound/README.md

@@ -0,0 +1,30 @@
+# OpenAEV BloodHound AD Injector
+
+Runs the [BloodHound.py](https://github.com/dirkjanm/BloodHound.py) Active
+Directory collector (SharpHound-compatible) and surfaces users, computers and
+privilege-escalation attack paths (Kerberoastable, AS-REP roastable) as
+findings.
+
+## Contract
+
+- BloodHound - Collect AD attack paths: fields for domain, username, password
+  and domain controller. Produces VULNERABILITY (attack paths exist) and
+  DETECTION (enumeration detected) expectations.
+
+## Credentials
+
+The AD credentials are provided per inject through the contract fields and are
+never logged (the password argument is redacted in logs).
+
+## Development
+
+```bash
+poetry install
+poetry run python -m unittest
+```
+
+## Icon
+
+`bloodhound_injector/img/icon-bloodhound.png` must follow the injector icon
+standard (square 1:1, 512x512 PNG, solid opaque background, genuine BloodHound
+artwork) - see OpenAEV-Platform/injectors#305.

bloodhound/bloodhound_injector/__init__.py

@@ -0,0 +1,2 @@
+# OpenAEV BloodHound AD Attack-Path Injector
+__version__ = "1.0.0"

bloodhound/bloodhound_injector/configuration/config_loader.py

@@ -0,0 +1,28 @@
+from pydantic import Field
+from pyoaev.configuration import ConfigLoaderOAEV, Configuration, SettingsLoader
+
+from bloodhound_injector.configuration.injector_config_override import (
+    InjectorConfigOverride,
+)
+from bloodhound_injector.contracts_bloodhound import BloodhoundContracts
+
+
+class ConfigLoader(SettingsLoader):
+    openaev: ConfigLoaderOAEV = Field(default_factory=ConfigLoaderOAEV)
+    injector: InjectorConfigOverride = Field(default_factory=InjectorConfigOverride)
+
+    def to_daemon_config(self) -> Configuration:
+        return Configuration(
+            config_hints={
+                "openaev_url": {"data": str(self.openaev.url)},
+                "openaev_token": {"data": self.openaev.token},
+                "openaev_tenant_id": {"data": self.openaev.tenant_id},
+                "injector_id": {"data": self.injector.id},
+                "injector_name": {"data": self.injector.name},
+                "injector_type": {"data": "openaev_bloodhound"},
+                "injector_contracts": {"data": BloodhoundContracts.build_contract()},
+                "injector_log_level": {"data": self.injector.log_level},
+                "injector_icon_filepath": {"data": self.injector.icon_filepath},
+            },
+            config_base_model=self,
+        )

bloodhound/bloodhound_injector/configuration/injector_config_override.py

@@ -0,0 +1,16 @@
+from pydantic import Field
+from pyoaev.configuration import ConfigLoaderCollector
+
+
+class InjectorConfigOverride(ConfigLoaderCollector):
+    id: str = Field(
+        description="A unique UUIDv4 identifier for this injector instance.",
+    )
+    name: str = Field(
+        default="BloodHound AD",
+        description="Name of the injector.",
+    )
+    icon_filepath: str | None = Field(
+        default="bloodhound_injector/img/icon-bloodhound.png",
+        description="Path to the icon file",
+    )

bloodhound/bloodhound_injector/contracts_bloodhound.py

@@ -0,0 +1,114 @@
+from typing import List
+
+from pyoaev.contracts import ContractBuilder
+from pyoaev.contracts.contract_config import (
+    Contract,
+    ContractCardinality,
+    ContractConfig,
+    ContractElement,
+    ContractExpectations,
+    ContractOutputElement,
+    ContractOutputType,
+    ContractText,
+    Expectation,
+    ExpectationType,
+    SupportedLanguage,
+    prepare_contracts,
+)
+from pyoaev.security_domain.types import SecurityDomains
+
+TYPE = "openaev_bloodhound"
+
+AD_COLLECTION_CONTRACT = "a7d8e9f0-abbb-4ac8-8fd6-bc2d3e4f5a67"
+
+
+class BloodhoundContracts:
+    @staticmethod
+    def build_contract():
+        contract_config = ContractConfig(
+            type=TYPE,
+            label={
+                SupportedLanguage.en: "BloodHound AD",
+                SupportedLanguage.fr: "BloodHound AD",
+            },
+            color_dark="#b00020",
+            color_light="#b00020",
+            expose=True,
+        )
+
+        domain = ContractText(key="domain", label="AD domain (FQDN)", mandatory=True)
+        username = ContractText(key="username", label="Username", mandatory=True)
+        password = ContractText(key="password", label="Password", mandatory=True)
+        domain_controller = ContractText(
+            key="domain_controller",
+            label="Domain controller (host or IP)",
+            mandatory=True,
+        )
+
+        expectations = ContractExpectations(
+            key="expectations",
+            label="Expectations",
+            mandatory=False,
+            cardinality=ContractCardinality.Multiple,
+            predefinedExpectations=[
+                Expectation(
+                    expectation_type=ExpectationType.vulnerability,
+                    expectation_name="Vulnerability",
+                    expectation_description="Privilege-escalation attack paths exist.",
+                    expectation_score=100,
+                    expectation_expectation_group=False,
+                ),
+                Expectation(
+                    expectation_type=ExpectationType.detection,
+                    expectation_name="Detection",
+                    expectation_description="The AD enumeration is detected.",
+                    expectation_score=100,
+                    expectation_expectation_group=False,
+                ),
+            ],
+        )
+
+        output_users = ContractOutputElement(
+            type=ContractOutputType.Username,
+            field="users",
+            isMultiple=True,
+            isFindingCompatible=True,
+            labels=["bloodhound", "ad", "user"],
+        )
+        output_computers = ContractOutputElement(
+            type=ContractOutputType.Computer,
+            field="computers",
+            isMultiple=True,
+            isFindingCompatible=True,
+            labels=["bloodhound", "ad", "computer"],
+        )
+        output_paths = ContractOutputElement(
+            type=ContractOutputType.Vulnerability,
+            field="attack_paths",
+            isMultiple=True,
+            isFindingCompatible=True,
+            labels=["bloodhound", "ad", "attack_path"],
+        )
+
+        fields: List[ContractElement] = (
+            ContractBuilder()
+            .add_fields([domain, username, password, domain_controller, expectations])
+            .build_fields()
+        )
+
+        contract = Contract(
+            contract_id=AD_COLLECTION_CONTRACT,
+            config=contract_config,
+            label={
+                SupportedLanguage.en: "BloodHound - Collect AD attack paths",
+                SupportedLanguage.fr: "BloodHound - Collecter les chemins d'attaque AD",
+            },
+            fields=fields,
+            outputs=ContractBuilder()
+            .add_outputs([output_users, output_computers, output_paths])
+            .build_outputs(),
+            manual=False,
+            domains=[SecurityDomains.ENDPOINT.value],
+        )
+
+        return prepare_contracts([contract])

bloodhound/bloodhound_injector/helpers/bloodhound_executor.py

@@ -0,0 +1,133 @@
+"""Runs the BloodHound.py (SharpHound-compatible) AD collector and parses it.
+
+Collects Active Directory objects via `bloodhound-python` and turns the emitted
+JSON into OpenAEV findings (users, computers) plus counts that surface the
+attack surface. Kerberoastable / AS-REP-roastable accounts are flagged as
+privilege-route findings.
+"""
+
+import glob
+import json
+import os
+import subprocess
+import tempfile
+from dataclasses import dataclass, field
+from typing import Dict, List
+
+
+@dataclass
+class BloodhoundResult:
+    success: bool
+    message: str
+    outputs: Dict[str, List[str]] = field(default_factory=dict)
+
+
+class BloodhoundExecutor:
+    DEFAULT_TIMEOUT_SECONDS = 900
+
+    def __init__(self, logger=None):
+        self.logger = logger
+
+    def run_collection(
+        self,
+        domain: str,
+        username: str,
+        password: str,
+        domain_controller: str,
+    ) -> BloodhoundResult:
+        with tempfile.TemporaryDirectory() as workdir:
+            cmd = [
+                "bloodhound-python",
+                "-d",
+                domain,
+                "-u",
+                username,
+                "-p",
+                password,
+                "-dc",
+                domain_controller,
+                "-c",
+                "All",
+                "--zip",
+            ]
+            try:
+                self._run(cmd, workdir)
+            except subprocess.TimeoutExpired:
+                return BloodhoundResult(
+                    False, f"BloodHound collection timed out for {domain}"
+                )
+            except FileNotFoundError:
+                return BloodhoundResult(
+                    False, "bloodhound-python not found in the image"
+                )
+
+            outputs = self.parse_collection(workdir)
+
+        users = outputs.get("users", [])
+        computers = outputs.get("computers", [])
+        return BloodhoundResult(
+            success=True,
+            message=(
+                f"Collected {len(users)} users and {len(computers)} computers "
+                f"from {domain}"
+            ),
+            outputs=outputs,
+        )
+
+    def _run(self, cmd: List[str], cwd: str) -> subprocess.CompletedProcess:
+        if self.logger:
+            # Never log the password argument.
+            safe = [a if a != cmd[cmd.index("-p") + 1] else "***" for a in cmd]
+            self.logger.info(f"Executing: {' '.join(safe)}")
+        return subprocess.run(
+            cmd,
+            capture_output=True,
+            text=True,
+            timeout=self.DEFAULT_TIMEOUT_SECONDS,
+            cwd=cwd,
+            stdin=subprocess.DEVNULL,
+        )
+
+    @staticmethod
+    def parse_collection(workdir: str) -> Dict[str, List[str]]:
+        outputs: Dict[str, List[str]] = {}
+        outputs["users"] = BloodhoundExecutor._names(workdir, "*_users.json")
+        outputs["computers"] = BloodhoundExecutor._names(workdir, "*_computers.json")
+        privileged = BloodhoundExecutor._privileged(workdir)
+        if privileged:
+            outputs["attack_paths"] = privileged
+        return {k: v for k, v in outputs.items() if v}
+
+    @staticmethod
+    def _load(path: str) -> dict:
+        try:
+            with open(path, encoding="utf-8") as handle:
+                return json.load(handle)
+        except (OSError, json.JSONDecodeError):
+            return {}
+
+    @staticmethod
+    def _names(workdir: str, pattern: str) -> List[str]:
+        names: List[str] = []
+        for path in glob.glob(os.path.join(workdir, pattern)):
+            data = BloodhoundExecutor._load(path)
+            for entry in data.get("data", []):
+                properties = entry.get("Properties", {})
+                name = properties.get("name") or properties.get("samaccountname")
+                if name and name not in names:
+                    names.append(name)
+        return names
+
+    @staticmethod
+    def _privileged(workdir: str) -> List[str]:
+        paths: List[str] = []
+        for path in glob.glob(os.path.join(workdir, "*_users.json")):
+            data = BloodhoundExecutor._load(path)
+            for entry in data.get("data", []):
+                properties = entry.get("Properties", {})
+                name = properties.get("name", "unknown")
+                if properties.get("hasspn"):
+                    paths.append(f"Kerberoastable: {name}")
+                if properties.get("dontreqpreauth"):
+                    paths.append(f"AS-REP roastable: {name}")
+        return paths