upload_zip_to_bloodhound
Upload ZIP data to Bloodhound for Active Directory analysis and attack path discovery during penetration testing. Wait for ingestion before querying results.
Instructions
Upload data zip to bloodhound to ingest and analyze (wait until it gets ingested before testing queries)
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| zip_path | Yes |
Implementation Reference
- src/pentestmcp/server.py:342-342 (registration)Registration of the MCP tool via FastMCP decorator.
@mcp.tool(name="upload_zip_to_bloodhound",description="Upload data zip to bloodhound to ingest and analyze (wait until it gets ingested before testing queries)") - src/pentestmcp/server.py:343-344 (handler)Handler function that delegates the upload to the bloodhound module.
def upload_zip_to_bloodhound(zip_path): return bloodhound.upload_zip(zip_path) - src/pentestmcp/bloodhound.py:129-169 (helper)Core helper function implementing the BloodHound CE API upload flow for the zip file.
def upload_zip(zip_path: str) -> Dict[str, Any]: """ Upload a SharpHound zip (or BloodHound JSON zip) using the BHCE file-upload job flow: 1) POST /api/v2/file-upload/start -> get upload id 2) POST /api/v2/file-upload/{id} -> upload bytes (Content-Type: application/zip) 3) POST /api/v2/file-upload/{id}/end -> finish job Returns the final job response JSON (or raises on error). """ # create client credentials = Credentials(token_id=config.BHE_TOKEN_ID, token_key=config.BHE_TOKEN_KEY) client = Client(scheme=BHE_SCHEME, host=config.BHE_DOMAIN, port=config.BHE_PORT, credentials=credentials) p = Path(zip_path) if not p.exists(): raise FileNotFoundError(f"{zip_path} not found") zip_bytes = p.read_bytes() # 1) create upload job resp = client._request("POST", "/api/v2/file-upload/start", content_type="application/json") # response body should contain the upload job id in JSON: { "data": { "id": "<id>" }, ... } start_payload = resp.json() upload_id = start_payload.get("data", {}).get("id") if not upload_id: raise RuntimeError(f"Could not create upload job: {start_payload}") # 2) upload the zip bytes (include body bytes when signing) upload_uri = f"/api/v2/file-upload/{upload_id}" client._request("POST", upload_uri, body=zip_bytes, content_type="application/zip") # 3) finish the job end_uri = f"/api/v2/file-upload/{upload_id}/end" end_resp = client._request("POST", end_uri, body=None, content_type="application/json") # API returns 204 No Content on success; depending on your client .json() may fail. if end_resp.status_code == 204: return {"status": "accepted", "upload_id": upload_id} else: return end_resp