# Directory Structure
```
├── .github
│ └── workflows
│ └── publish-to-pypi.yml
├── .gitignore
├── .python-version
├── LICENSE
├── pyproject.toml
├── README.md
└── src
└── defectdojo
├── __init__.py
├── client.py
├── engagements_tools.py
├── findings_tools.py
├── products_tools.py
└── tools.py
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.12
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
.pytest_*
uv.lock
# Virtual environments
venv
.venv
# macOS specific files
.DS_Store
# IDE / Editor specific files
.vscode/
.idea/
# Test reports
.coverage*
htmlcov/
# Type checker cache
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
#uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/latest/usage/project/#working-with-version-control
.pdm.toml
.pdm-python
.pdm-build/
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
>>>>>>> 850ad5e (Initial commit)
.mypy_cache/
.dmypy.json
dmypy.json
<<<<<<< HEAD
# Environment variables
.env*
# Log files
*.log
=======
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
# Ruff stuff:
.ruff_cache/
# PyPI configuration file
.pypirc
>>>>>>> 850ad5e (Initial commit)
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# DefectDojo MCP Server
[](https://badge.fury.io/py/defectdojo) <!-- Add this badge if/when published to PyPI -->
This project provides a [Model Context Protocol (MCP)](https://github.com/modelcontextprotocol/specification) server implementation for [DefectDojo](https://github.com/DefectDojo/django-DefectDojo), a popular open-source vulnerability management tool. It allows AI agents and other MCP clients to interact with the DefectDojo API programmatically.
## Features
This MCP server exposes tools for managing key DefectDojo entities:
* **Findings:** Fetch, search, create, update status, and add notes.
* **Products:** List available products.
* **Engagements:** List, retrieve details, create, update, and close engagements.
## Installation & Running
There are a couple of ways to run this server:
### Using `uvx` (Recommended)
`uvx` executes Python applications in temporary virtual environments, installing dependencies automatically.
```bash
uvx defectdojo-mcp
```
### Using `pip`
You can install the package into your Python environment using `pip`.
```bash
# Install directly from the cloned source code directory
pip install .
# Or, if the package is published on PyPI
pip install defectdojo-mcp
```
Once installed via pip, run the server using:
```bash
defectdojo-mcp
```
## Configuration
The server requires the following environment variables to connect to your DefectDojo instance:
* `DEFECTDOJO_API_TOKEN` (**required**): Your DefectDojo API token for authentication.
* `DEFECTDOJO_API_BASE` (**required**): The base URL of your DefectDojo instance (e.g., `https://your-defectdojo-instance.com`).
You can configure these in your MCP client's settings file. Here's an example using the `uvx` command:
```json
{
"mcpServers": {
"defectdojo": {
"command": "uvx",
"args": ["defectdojo-mcp"],
"env": {
"DEFECTDOJO_API_TOKEN": "YOUR_API_TOKEN_HERE",
"DEFECTDOJO_API_BASE": "https://your-defectdojo-instance.com"
}
}
}
}
```
If you installed the package using `pip`, the configuration would look like this:
```json
{
"mcpServers": {
"defectdojo": {
"command": "defectdojo-mcp",
"args": [],
"env": {
"DEFECTDOJO_API_TOKEN": "YOUR_API_TOKEN_HERE",
"DEFECTDOJO_API_BASE": "https://your-defectdojo-instance.com"
}
}
}
}
```
## Available Tools
The following tools are available via the MCP interface:
* `get_findings`: Retrieve findings with filtering (product_name, status, severity) and pagination (limit, offset).
* `search_findings`: Search findings using a text query, with filtering and pagination.
* `update_finding_status`: Change the status of a specific finding (e.g., Active, Verified, False Positive).
* `add_finding_note`: Add a textual note to a finding.
* `create_finding`: Create a new finding associated with a test.
* `list_products`: List products with filtering (name, prod_type) and pagination.
* `list_engagements`: List engagements with filtering (product_id, status, name) and pagination.
* `get_engagement`: Get details for a specific engagement by its ID.
* `create_engagement`: Create a new engagement for a product.
* `update_engagement`: Modify details of an existing engagement.
* `close_engagement`: Mark an engagement as completed.
*(See the original README content below for detailed usage examples of each tool)*
## Usage Examples
*(Note: These examples assume an MCP client environment capable of calling `use_mcp_tool`)*
### Get Findings
```python
# Get active, high-severity findings (limit 10)
result = await use_mcp_tool("defectdojo", "get_findings", {
"status": "Active",
"severity": "High",
"limit": 10
})
```
### Search Findings
```python
# Search for findings containing 'SQL Injection'
result = await use_mcp_tool("defectdojo", "search_findings", {
"query": "SQL Injection"
})
```
### Update Finding Status
```python
# Mark finding 123 as Verified
result = await use_mcp_tool("defectdojo", "update_finding_status", {
"finding_id": 123,
"status": "Verified"
})
```
### Add Note to Finding
```python
result = await use_mcp_tool("defectdojo", "add_finding_note", {
"finding_id": 123,
"note": "Confirmed vulnerability on staging server."
})
```
### Create Finding
```python
result = await use_mcp_tool("defectdojo", "create_finding", {
"title": "Reflected XSS in Search Results",
"test_id": 55, # ID of the associated test
"severity": "Medium",
"description": "User input in search is not properly sanitized, leading to XSS.",
"cwe": 79
})
```
### List Products
```python
# List products containing 'Web App' in their name
result = await use_mcp_tool("defectdojo", "list_products", {
"name": "Web App",
"limit": 10
})
```
### List Engagements
```python
# List 'In Progress' engagements for product ID 42
result = await use_mcp_tool("defectdojo", "list_engagements", {
"product_id": 42,
"status": "In Progress"
})
```
### Get Engagement
```python
result = await use_mcp_tool("defectdojo", "get_engagement", {
"engagement_id": 101
})
```
### Create Engagement
```python
result = await use_mcp_tool("defectdojo", "create_engagement", {
"product_id": 42,
"name": "Q2 Security Scan",
"target_start": "2025-04-01",
"target_end": "2025-04-15",
"status": "Not Started"
})
```
### Update Engagement
```python
result = await use_mcp_tool("defectdojo", "update_engagement", {
"engagement_id": 101,
"status": "In Progress",
"description": "Scan initiated."
})
```
### Close Engagement
```python
result = await use_mcp_tool("defectdojo", "close_engagement", {
"engagement_id": 101
})
```
## Development
### Setup
1. Clone the repository.
2. It's recommended to use a virtual environment:
```bash
python -m venv .venv
source .venv/bin/activate # On Windows use `.venv\Scripts\activate`
```
3. Install dependencies, including development dependencies:
```bash
pip install -e ".[dev]"
```
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
## Contributing
Contributions are welcome! Please feel free to open an issue for bugs, feature requests, or questions. If you'd like to contribute code, please open an issue first to discuss the proposed changes.
```
--------------------------------------------------------------------------------
/src/defectdojo/__init__.py:
--------------------------------------------------------------------------------
```python
import os
from mcp.server.fastmcp import FastMCP
# Import registration functions
from defectdojo import findings_tools, products_tools, engagements_tools
# Initialize FastMCP server
mcp = FastMCP("defectdojo")
# Register tools by calling functions from modules
findings_tools.register_tools(mcp)
products_tools.register_tools(mcp)
engagements_tools.register_tools(mcp)
def main():
"""Initialize and run the MCP server."""
print("Starting DefectDojo MCP server...")
mcp.run(transport='stdio')
if __name__ == "__main__":
# Initialize and run the server
main()
```
--------------------------------------------------------------------------------
/.github/workflows/publish-to-pypi.yml:
--------------------------------------------------------------------------------
```yaml
# .github/workflows/publish-to-pypi.yml
name: Publish Python Package to PyPI
on:
push:
branches:
- main
release:
types: [published]
permissions:
contents: read
jobs:
build:
name: Build distribution
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install build dependencies
run: python -m pip install build
- name: Build package
run: python -m build
- name: Store the distribution packages
uses: actions/upload-artifact@v4
with:
name: python-package-distributions
path: dist/
publish-to-pypi:
name: Publish distribution to PyPI
needs: build
runs-on: ubuntu-latest
if: github.event_name == 'release' && github.event.action == 'published'
environment:
name: pypi
url: https://pypi.org/p/defectdojo-mcp
permissions:
id-token: write
steps:
- name: Download all the dists
uses: actions/download-artifact@v4
with:
name: python-package-distributions
path: dist/
- name: Publish package distributions to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[build-system]
requires = ["setuptools>=61.0.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "defectdojo-mcp"
version = "0.1.2"
description = "DefectDojo MCP server for integrating with DefectDojo vulnerability management system"
readme = "README.md"
requires-python = ">=3.12"
authors = [
{ name = "Jamiesonio" },
]
license = { text = "MIT" }
keywords = ["defectdojo", "appsec", "mcp", "devsecops", "llm"]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"Intended Audience :: Information Technology",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.12",
"Topic :: Security",
"Topic :: Software Development :: Libraries :: Python Modules",
]
dependencies = [
"httpx>=0.28.1",
"mcp[cli]>=1.6.0",
]
[project.urls]
Homepage = "https://github.com/jamiesonio/defectdojo-mcp"
Repository = "https://github.com/jamiesonio/defectdojo-mcp"
"Issue Tracker" = "https://github.com/jamiesonio/defectdojo-mcp/issues"
[project.scripts]
defectdojo-mcp = "defectdojo:main"
[tool.setuptools]
package-dir = {"" = "src"}
[tool.setuptools.packages.find]
where = ["src"]
[project.optional-dependencies]
dev = [
"pytest>=8.0.0", # Or specify a version range
]
```
--------------------------------------------------------------------------------
/src/defectdojo/products_tools.py:
--------------------------------------------------------------------------------
```python
from typing import Any, Dict, Optional
from defectdojo.client import get_client
# --- Product Tool Definitions ---
async def list_products(name: Optional[str] = None, prod_type: Optional[int] = None,
limit: int = 50, offset: int = 0) -> Dict[str, Any]:
"""List all products with optional filtering and pagination.
Args:
name: Optional name filter (partial match)
prod_type: Optional product type ID filter
limit: Maximum number of products to return per page (default: 50)
offset: Number of records to skip (default: 0)
Returns:
Dictionary with status, data/error, and pagination metadata
"""
filters = {"limit": limit}
# Use __icontains for case-insensitive partial match if API supports it
if name:
filters["name"] = name # Or name__icontains if supported
if prod_type:
filters["prod_type"] = prod_type
if offset:
filters["offset"] = offset
client = get_client()
result = await client.get_products(filters)
if "error" in result:
return {"status": "error", "error": result["error"], "details": result.get("details", "")}
return {"status": "success", "data": result}
# --- Registration Function ---
def register_tools(mcp):
"""Register product-related tools with the MCP server instance."""
mcp.tool(name="list_products", description="List all products with optional filtering and pagination support")(list_products)
```
--------------------------------------------------------------------------------
/src/defectdojo/tools.py:
--------------------------------------------------------------------------------
```python
from mcp.server.fastmcp import FastMCP
# Import tool functions from the new modules
from .findings_tools import (
get_findings,
search_findings,
update_finding_status,
add_finding_note,
create_finding,
)
from .products_tools import list_products
from .engagements_tools import (
list_engagements,
get_engagement,
create_engagement,
update_engagement,
close_engagement,
)
# Placeholder for the MCP instance - will be set by the main script
mcp = None
# --- Registration Function ---
# This function will be called by the main script to register tools with the MCP instance
def register_tools(mcp_instance: FastMCP):
"""Registers all tools with the provided FastMCP instance."""
global mcp
mcp = mcp_instance
# Register Finding Tools
mcp.tool(
name="get_findings",
description="Get findings with filtering options and pagination support"
)(get_findings)
mcp.tool(
name="search_findings",
description="Search for findings using a text query with pagination support"
)(search_findings)
mcp.tool(
name="update_finding_status",
description="Update the status of a finding (Active, Verified, False Positive, Mitigated, Inactive)"
)(update_finding_status)
mcp.tool(
name="add_finding_note",
description="Add a note to a finding"
)(add_finding_note)
mcp.tool(
name="create_finding",
description="Create a new finding"
)(create_finding)
# Register Product Tools
mcp.tool(
name="list_products",
description="List all products with optional filtering and pagination support"
)(list_products)
# Register Engagement Tools
mcp.tool(
name="list_engagements",
description="List engagements with optional filtering and pagination support"
)(list_engagements)
mcp.tool(
name="get_engagement",
description="Get a specific engagement by ID"
)(get_engagement)
mcp.tool(
name="create_engagement",
description="Create a new engagement in DefectDojo"
# Schema inferred from type hints and docstring
)(create_engagement)
mcp.tool(
name="update_engagement",
description="Update an existing engagement"
)(update_engagement)
mcp.tool(
name="close_engagement",
description="Close an engagement"
)(close_engagement)
```
--------------------------------------------------------------------------------
/src/defectdojo/client.py:
--------------------------------------------------------------------------------
```python
import httpx
import os
from typing import Any, Dict, Optional
class DefectDojoClient:
"""Client for interacting with the DefectDojo API."""
def __init__(self, base_url: str, api_token: str):
"""Initialize the DefectDojo API client.
Args:
base_url: Base URL for the DefectDojo API
api_token: API token for authentication
"""
self.base_url = base_url
self.headers = {
"Authorization": f"Token {api_token}",
"Content-Type": "application/json"
}
# Consider adding timeout configuration
self.client = httpx.AsyncClient(headers=self.headers, timeout=30.0)
async def _request(self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None, json: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Helper method to make API requests."""
url = f"{self.base_url}{endpoint}"
try:
response = await self.client.request(method, url, params=params, json=json)
response.raise_for_status()
# Handle cases where response might be empty (e.g., 204 No Content)
if response.status_code == 204:
return {}
return response.json()
except httpx.HTTPStatusError as e:
# Log or handle specific status codes if needed
return {"error": f"HTTP error: {e.response.status_code}", "details": e.response.text}
except httpx.RequestError as e:
# Handle network errors, timeouts, etc.
return {"error": f"Request error: {str(e)}"}
except Exception as e:
# Catch unexpected errors
return {"error": f"An unexpected error occurred: {str(e)}"}
async def get_findings(self, filters: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Get findings with optional filters."""
return await self._request("GET", "/api/v2/findings/", params=filters)
async def search_findings(self, query: str, filters: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Search for findings using a text query."""
params = filters or {}
params["search"] = query
return await self._request("GET", "/api/v2/findings/", params=params)
async def update_finding(self, finding_id: int, data: Dict[str, Any]) -> Dict[str, Any]:
"""Update a finding by ID."""
return await self._request("PATCH", f"/api/v2/findings/{finding_id}/", json=data)
async def add_note_to_finding(self, finding_id: int, note: str) -> Dict[str, Any]:
"""Add a note to a finding."""
data = {"entry": note, "finding": finding_id}
return await self._request("POST", "/api/v2/notes/", json=data)
async def create_finding(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Create a new finding."""
return await self._request("POST", "/api/v2/findings/", json=data)
async def get_products(self, filters: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Get products with optional filters."""
return await self._request("GET", "/api/v2/products/", params=filters)
async def get_engagements(self, filters: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Get engagements with optional filters."""
return await self._request("GET", "/api/v2/engagements/", params=filters)
async def get_engagement(self, engagement_id: int) -> Dict[str, Any]:
"""Get a specific engagement by ID."""
return await self._request("GET", f"/api/v2/engagements/{engagement_id}/")
async def create_engagement(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Create a new engagement."""
return await self._request("POST", "/api/v2/engagements/", json=data)
async def update_engagement(self, engagement_id: int, data: Dict[str, Any]) -> Dict[str, Any]:
"""Update an existing engagement."""
return await self._request("PATCH", f"/api/v2/engagements/{engagement_id}/", json=data)
# --- Client Factory ---
def get_client(validate_token=True, base_url=None, token=None) -> DefectDojoClient:
"""Get a configured DefectDojo client.
Args:
validate_token: Whether to validate that the token is set (default: True)
base_url: Optional base URL override for testing
token: Optional token override for testing
Returns:
A configured DefectDojoClient instance
Raises:
ValueError: If DEFECTDOJO_API_TOKEN environment variable is not set and validate_token is True
"""
# Use provided values or get from environment variables.
# Ensure DEFECTDOJO_API_BASE and DEFECTDOJO_API_TOKEN are set in your environment.
actual_token = token if token is not None else os.environ.get("DEFECTDOJO_API_TOKEN")
actual_base_url = base_url if base_url is not None else os.environ.get("DEFECTDOJO_API_BASE")
if not actual_base_url:
raise ValueError("DEFECTDOJO_API_BASE environment variable or base_url argument must be provided and cannot be empty.")
# Only validate token if requested (e.g., skipped for tests)
if validate_token and not actual_token:
raise ValueError("DEFECTDOJO_API_TOKEN environment variable or token argument must be provided")
return DefectDojoClient(actual_base_url, actual_token)
```
--------------------------------------------------------------------------------
/src/defectdojo/findings_tools.py:
--------------------------------------------------------------------------------
```python
from typing import Any, Dict, List, Optional
from defectdojo.client import get_client
# --- Finding Tool Definitions ---
async def get_findings(product_name: Optional[str] = None, status: Optional[str] = None,
severity: Optional[str] = None, limit: int = 20,
offset: int = 0) -> Dict[str, Any]:
"""Get findings with optional filters and pagination.
Args:
product_name: Optional product name filter
status: Optional status filter
severity: Optional severity filter
limit: Maximum number of findings to return per page (default: 20)
offset: Number of records to skip (default: 0)
Returns:
Dictionary with status, data/error, and pagination metadata
"""
filters = {}
if product_name:
filters["product_name"] = product_name
if status:
filters["status"] = status
if severity:
filters["severity"] = severity
if limit:
filters["limit"] = limit
if offset:
filters["offset"] = offset
client = get_client()
result = await client.get_findings(filters)
if "error" in result:
return {"status": "error", "error": result["error"], "details": result.get("details", "")}
return {"status": "success", "data": result}
# --- Registration Function ---
def register_tools(mcp):
"""Register finding-related tools with the MCP server instance."""
mcp.tool(name="get_findings", description="Get findings with filtering options and pagination support")(get_findings)
mcp.tool(name="search_findings", description="Search for findings using a text query with pagination support")(search_findings)
mcp.tool(name="update_finding_status", description="Update the status of a finding (Active, Verified, False Positive, Mitigated, Inactive)")(update_finding_status)
mcp.tool(name="add_finding_note", description="Add a note to a finding")(add_finding_note)
mcp.tool(name="create_finding", description="Create a new finding")(create_finding)
async def search_findings(query: str, product_name: Optional[str] = None,
status: Optional[str] = None, severity: Optional[str] = None,
limit: int = 20, offset: int = 0) -> Dict[str, Any]:
"""Search for findings using a text query with pagination.
Args:
query: Text to search for in findings
product_name: Optional product name filter
status: Optional status filter
severity: Optional severity filter
limit: Maximum number of findings to return per page (default: 20)
offset: Number of records to skip (default: 0)
Returns:
Dictionary with status, data/error, and pagination metadata
"""
filters = {}
if product_name:
filters["product_name"] = product_name
if status:
filters["status"] = status
if severity:
filters["severity"] = severity
if limit:
filters["limit"] = limit
if offset:
filters["offset"] = offset
client = get_client()
result = await client.search_findings(query, filters)
if "error" in result:
return {"status": "error", "error": result["error"], "details": result.get("details", "")}
return {"status": "success", "data": result}
async def update_finding_status(finding_id: int, status: str) -> Dict[str, Any]:
"""Update the status of a finding.
Args:
finding_id: ID of the finding to update
status: New status for the finding (Active, Verified, False Positive, Mitigated, Inactive)
Returns:
Dictionary with status and data/error
"""
data = {"active": True} # Default to active
# Map common status values to API fields
status_lower = status.lower()
if status_lower == "false positive":
data["false_p"] = True
elif status_lower == "verified":
data["verified"] = True
elif status_lower == "mitigated":
data["active"] = False
data["mitigated"] = True # Assuming API uses 'mitigated' boolean field
elif status_lower == "inactive":
data["active"] = False
elif status_lower != "active":
# Check against API specific values if needed, or raise error for unsupported input
return {"status": "error", "error": f"Unsupported status: {status}. Use Active, Verified, False Positive, Mitigated, or Inactive."}
# Clear conflicting flags if setting a specific status
if data.get("false_p"):
data.pop("verified", None)
data.pop("active", None)
data.pop("mitigated", None)
elif data.get("verified"):
data.pop("false_p", None)
# Verified implies active usually, but check API docs if explicit setting is needed
data["active"] = True
data.pop("mitigated", None)
elif data.get("mitigated"):
data.pop("false_p", None)
data.pop("verified", None)
data["active"] = False # Mitigated implies inactive
elif not data.get("active", True): # Handling "Inactive" case
data.pop("false_p", None)
data.pop("verified", None)
data.pop("mitigated", None)
data["active"] = False
else: # Handling "Active" case (default or explicit)
data.pop("false_p", None)
data.pop("verified", None)
data.pop("mitigated", None)
data["active"] = True
client = get_client()
result = await client.update_finding(finding_id, data)
if "error" in result:
return {"status": "error", "error": result["error"], "details": result.get("details", "")}
return {"status": "success", "data": result}
async def add_finding_note(finding_id: int, note: str) -> Dict[str, Any]:
"""Add a note to a finding.
Args:
finding_id: ID of the finding to add a note to
note: Text content of the note
Returns:
Dictionary with status and data/error
"""
if not note.strip():
return {"status": "error", "error": "Note content cannot be empty"}
client = get_client()
result = await client.add_note_to_finding(finding_id, note)
if "error" in result:
return {"status": "error", "error": result["error"], "details": result.get("details", "")}
return {"status": "success", "data": result}
async def create_finding(title: str, test_id: int, severity: str, description: str,
cwe: Optional[int] = None, cvssv3: Optional[str] = None,
mitigation: Optional[str] = None, impact: Optional[str] = None,
steps_to_reproduce: Optional[str] = None) -> Dict[str, Any]:
"""Create a new finding.
Args:
title: Title of the finding
test_id: ID of the test to associate the finding with
severity: Severity level (Critical, High, Medium, Low, Info)
description: Description of the finding
cwe: Optional CWE identifier
cvssv3: Optional CVSS v3 score string
mitigation: Optional mitigation steps
impact: Optional impact description
steps_to_reproduce: Optional steps to reproduce
Returns:
Dictionary with status and data/error
"""
# Validate severity (case-insensitive check, but send capitalized)
valid_severities = ["critical", "high", "medium", "low", "info"]
normalized_severity = severity.lower()
if normalized_severity not in valid_severities:
# Use title case for user-facing error message
valid_display = [s.title() for s in valid_severities]
return {"status": "error", "error": f"Invalid severity '{severity}'. Must be one of: {', '.join(valid_display)}"}
# Use title case for API
api_severity = severity.title()
data = {
"title": title,
"test": test_id,
"severity": api_severity,
"description": description,
# Set defaults expected by API if not provided explicitly by user?
# e.g., "active": True, "verified": False? Check API docs.
"active": True,
"verified": False,
}
# Add optional fields if provided
if cwe is not None:
data["cwe"] = cwe
if cvssv3:
data["cvssv3"] = cvssv3 # Assuming API accepts the string directly
if mitigation:
data["mitigation"] = mitigation
if impact:
data["impact"] = impact
if steps_to_reproduce:
data["steps_to_reproduce"] = steps_to_reproduce
client = get_client()
result = await client.create_finding(data)
if "error" in result:
return {"status": "error", "error": result["error"], "details": result.get("details", "")}
return {"status": "success", "data": result}
```
--------------------------------------------------------------------------------
/src/defectdojo/engagements_tools.py:
--------------------------------------------------------------------------------
```python
from typing import Any, Dict, Optional, List
from defectdojo.client import get_client
# --- Engagement Tool Definitions ---
async def list_engagements(product_id: Optional[int] = None,
status: Optional[str] = None,
name: Optional[str] = None,
limit: int = 20, offset: int = 0) -> Dict[str, Any]:
"""List engagements with optional filtering and pagination.
Args:
product_id: Optional product ID filter
status: Optional status filter (e.g., 'Not Started', 'In Progress', 'Completed')
name: Optional name filter (partial match)
limit: Maximum number of engagements to return per page (default: 20)
offset: Number of records to skip (default: 0)
Returns:
Dictionary with status, data/error, and pagination metadata
"""
filters = {"limit": limit}
if product_id:
filters["product"] = product_id
if status:
# Validate against known API statuses if necessary
valid_statuses = ["Not Started", "Blocked", "Cancelled", "Completed", "In Progress", "On Hold", "Waiting for Resource"]
if status not in valid_statuses:
return {"status": "error", "error": f"Invalid status filter '{status}'. Must be one of: {', '.join(valid_statuses)}"}
filters["status"] = status
if name:
filters["name"] = name # Or name__icontains if supported
if offset:
filters["offset"] = offset
client = get_client()
result = await client.get_engagements(filters)
if "error" in result:
return {"status": "error", "error": result["error"], "details": result.get("details", "")}
return {"status": "success", "data": result}
async def get_engagement(engagement_id: int) -> Dict[str, Any]:
"""Get a specific engagement by ID.
Args:
engagement_id: ID of the engagement to retrieve
Returns:
Dictionary with status and data/error
"""
client = get_client()
result = await client.get_engagement(engagement_id)
if "error" in result:
return {"status": "error", "error": result["error"], "details": result.get("details", "")}
return {"status": "success", "data": result}
async def create_engagement(product_id: int, name: str, target_start: str, target_end: str, status: str, lead_id: int = None, description: str = None, version: str = None, build_id: str = None, commit_hash: str = None, branch_tag: str = None, engagement_type: str = None, deduplication_on_engagement: bool = None, tags: list = None):
"""
Creates a new engagement in DefectDojo.
Args:
product_id: ID of the product.
name: Name of the engagement.
target_start: Start date (YYYY-MM-DD).
target_end: End date (YYYY-MM-DD).
status: Engagement status ('Not Started', 'Blocked', 'Cancelled', 'Completed', 'In Progress', 'On Hold', 'Waiting for Resource').
lead_id: Optional ID of the engagement lead (user ID).
description: Optional engagement description.
version: Optional product version tested.
build_id: Optional build ID.
commit_hash: Optional commit hash.
branch_tag: Optional branch or tag.
engagement_type: Optional engagement type ('Interactive' or 'CI/CD').
deduplication_on_engagement: Optional flag to enable deduplication within this engagement.
tags: Optional list of tags.
Returns:
JSON response from the API.
"""
# endpoint = "/api/v2/engagements/" # Endpoint handled by client method
valid_statuses = ["Not Started", "Blocked", "Cancelled", "Completed", "In Progress", "On Hold", "Waiting for Resource"]
if status not in valid_statuses:
# Use raise ValueError for internal validation errors
raise ValueError(f"Invalid status '{status}'. Must be one of: {', '.join(valid_statuses)}")
# Validate engagement_type if provided
if engagement_type and engagement_type not in ["Interactive", "CI/CD"]:
raise ValueError(f"Invalid engagement_type '{engagement_type}'. Must be 'Interactive' or 'CI/CD'.")
data = {
"product": product_id,
"name": name,
"target_start": target_start,
"target_end": target_end,
"status": status, # Use API expected casing directly
}
# Add optional fields cleanly
if lead_id is not None: data["lead"] = lead_id
if description is not None: data["description"] = description
if version is not None: data["version"] = version
if build_id is not None: data["build_id"] = build_id
if commit_hash is not None: data["commit_hash"] = commit_hash
if branch_tag is not None: data["branch_tag"] = branch_tag
if engagement_type is not None: data["engagement_type"] = engagement_type
if deduplication_on_engagement is not None: data["deduplication_on_engagement"] = deduplication_on_engagement
if tags is not None: data["tags"] = tags # Assumes API accepts list directly
client = get_client()
result = await client.create_engagement(data)
# Return structured response
if "error" in result:
return {"status": "error", "error": result["error"], "details": result.get("details", "")}
return {"status": "success", "data": result}
async def update_engagement(engagement_id: int, name: Optional[str] = None,
target_start: Optional[str] = None, # Renamed from start_date
target_end: Optional[str] = None, # Renamed from end_date
status: Optional[str] = None,
description: Optional[str] = None,
# Add other updatable fields from API schema if needed
lead_id: Optional[int] = None,
version: Optional[str] = None,
build_id: Optional[str] = None,
commit_hash: Optional[str] = None,
branch_tag: Optional[str] = None,
engagement_type: Optional[str] = None,
deduplication_on_engagement: Optional[bool] = None,
tags: Optional[list] = None
) -> Dict[str, Any]:
"""Update an existing engagement. Only provided fields are updated.
Args:
engagement_id: ID of the engagement to update.
name: Optional new name.
target_start: Optional new start date (YYYY-MM-DD).
target_end: Optional new end date (YYYY-MM-DD).
status: Optional new status ('Not Started', 'Blocked', 'Cancelled', 'Completed', 'In Progress', 'On Hold', 'Waiting for Resource').
description: Optional new description.
lead_id: Optional new lead ID.
version: Optional new version.
build_id: Optional new build ID.
commit_hash: Optional new commit hash.
branch_tag: Optional new branch/tag.
engagement_type: Optional new engagement type ('Interactive', 'CI/CD').
deduplication_on_engagement: Optional new deduplication setting.
tags: Optional new list of tags (will replace existing tags).
Returns:
Dictionary with status and data/error.
"""
# Validate status if provided
if status:
valid_statuses = ["Not Started", "Blocked", "Cancelled", "Completed", "In Progress", "On Hold", "Waiting for Resource"]
if status not in valid_statuses:
return {"status": "error", "error": f"Invalid status '{status}'. Must be one of: {', '.join(valid_statuses)}"}
# Validate engagement_type if provided
if engagement_type and engagement_type not in ["Interactive", "CI/CD"]:
return {"status": "error", "error": f"Invalid engagement_type '{engagement_type}'. Must be 'Interactive' or 'CI/CD'."}
# Prepare data payload with only provided fields
data = {}
if name is not None: data["name"] = name
if target_start is not None: data["target_start"] = target_start
if target_end is not None: data["target_end"] = target_end
if status is not None: data["status"] = status # Send as is after validation
if description is not None: data["description"] = description
if lead_id is not None: data["lead"] = lead_id
if version is not None: data["version"] = version
if build_id is not None: data["build_id"] = build_id
if commit_hash is not None: data["commit_hash"] = commit_hash
if branch_tag is not None: data["branch_tag"] = branch_tag
if engagement_type is not None: data["engagement_type"] = engagement_type
if deduplication_on_engagement is not None: data["deduplication_on_engagement"] = deduplication_on_engagement
if tags is not None: data["tags"] = tags # PATCH usually replaces arrays
# If no fields were provided, return an error
if not data:
return {"status": "error", "error": "At least one field must be provided for update"}
client = get_client()
result = await client.update_engagement(engagement_id, data)
if "error" in result:
return {"status": "error", "error": result["error"], "details": result.get("details", "")}
return {"status": "success", "data": result}
async def close_engagement(engagement_id: int) -> Dict[str, Any]:
"""Close an engagement by setting its status to completed.
Args:
engagement_id: ID of the engagement to close
Returns:
Dictionary with status and data/error
"""
# Use the specific status string from the API schema
data = {
"status": "Completed"
}
client = get_client()
# Use the update_engagement client method
result = await client.update_engagement(engagement_id, data)
if "error" in result:
return {"status": "error", "error": result["error"], "details": result.get("details", "")}
# Check if the update was successful (API might return updated object or just status)
# Assuming success if no error is present
return {"status": "success", "data": result}
# --- Registration Function ---
def register_tools(mcp):
"""Register engagement-related tools with the MCP server instance."""
mcp.tool(name="list_engagements", description="List engagements with optional filtering and pagination support")(list_engagements)
mcp.tool(name="get_engagement", description="Get a specific engagement by ID")(get_engagement)
mcp.tool(name="create_engagement", description="Create a new engagement")(create_engagement)
mcp.tool(name="update_engagement", description="Update an existing engagement")(update_engagement)
mcp.tool(name="close_engagement", description="Close an engagement")(close_engagement)
```