ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2025-12-19 05:17:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(mcp): add API key support for STDIO mode and enhance HTTP mode authentication (#8823)

Browse Source
This commit is contained in:
Rubén De la Torre Vico
2025-10-08 15:52:26 +02:00
committed by GitHub
parent 13266b8743
commit 28e81783ef
3 changed files with 69 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
mcp_server/.env.template
View File

@@ -1,5 +1,3 @@
PROWLER_APP_EMAIL="your_registered@email.com"
PROWLER_APP_PASSWORD="your_user_pass"
PROWLER_APP_TENANT_ID="optional_tenant_to_login"
PROWLER_APP_API_KEY="pk_your_api_key_here"
PROWLER_API_BASE_URL="https://api.prowler.com"
PROWLER_MCP_MODE="stdio"

42
mcp_server/README.md
View File

@@ -224,15 +224,11 @@ The Prowler MCP server supports different authentication in Prowler Cloud and Pr
#### STDIO Mode Authentication
For STDIO mode, authentication is handled via environment variables:
For STDIO mode, authentication is handled via environment variables using an API key:
```bash
# Required for Prowler Cloud and Prowler App (Self-Managed) authentication
export PROWLER_APP_EMAIL="your-email@example.com"
export PROWLER_APP_PASSWORD="your-password"
# Optional - in case not provided the first membership that was added to the user will be used. This can be found as `Organization ID` in your User Profile in Prowler App
export PROWLER_APP_TENANT_ID="your-tenant-id"
export PROWLER_APP_API_KEY="pk_your_api_key_here"
# Optional - for custom API endpoint, in case not provided Prowler Cloud API will be used
export PROWLER_API_BASE_URL="https://api.prowler.com"
@@ -240,7 +236,16 @@ export PROWLER_API_BASE_URL="https://api.prowler.com"
#### HTTP Mode Authentication
For HTTP mode (remote server), authentication is handled via Bearer tokens. You need to obtain a JWT token from Prowler Cloud/App and include the generated token in the MCP client configuration. To get a valid token, you can use the following command (replace the email and password with your own credentials):
For HTTP mode (remote server), authentication is handled via Bearer tokens. The MCP server supports both JWT tokens and API keys:
**Option 1: Using API Keys (Recommended)**
Use your Prowler API key directly in the MCP client configuration with Bearer token format:
```
Authorization: Bearer pk_your_api_key_here
```
**Option 2: Using JWT Tokens**
You need to obtain a JWT token from Prowler Cloud/App and include the generated token in the MCP client configuration. To get a valid token, you can use the following command (replace the email and password with your own credentials):
```bash
curl -X POST https://api.prowler.com/api/v1/tokens \
@@ -276,9 +281,7 @@ For local execution, configure your MCP client to launch the server directly. Be
"command": "uvx",
"args": ["/path/to/prowler/mcp_server/"],
"env": {
"PROWLER_APP_EMAIL": "your-email@example.com",
"PROWLER_APP_PASSWORD": "your-password",
"PROWLER_APP_TENANT_ID": "your-tenant-id", // Optional, this can be found as `Organization ID` in your User Profile in Prowler App,
"PROWLER_APP_API_KEY": "pk_your_api_key_here",
"PROWLER_API_BASE_URL": "https://api.prowler.com" // Optional, in case not provided Prowler Cloud API will be used
}
}
@@ -295,9 +298,7 @@ For local execution, configure your MCP client to launch the server directly. Be
"command": "docker",
"args": [
"run", "--rm", "-i",
"--env", "PROWLER_APP_EMAIL=your-email@example.com",
"--env", "PROWLER_APP_PASSWORD=your-password",
"--env", "PROWLER_APP_TENANT_ID=your-tenant-id", // Optional, this can be found as `Organization ID` in your User Profile in Prowler App
"--env", "PROWLER_APP_API_KEY=pk_your_api_key_here",
"--env", "PROWLER_API_BASE_URL=https://api.prowler.com", // Optional, in case not provided Prowler Cloud API will be used
"prowler-mcp"
]
@@ -316,6 +317,21 @@ For HTTP mode, you can configure your MCP client to connect to a remote Prowler
Example configuration for clients that support HTTP transport:
**Using API Key (Recommended):**
```json
{
"mcpServers": {
"prowler": {
"url": "http://mcp.prowler.com/mcp", // Replace with your own MCP server URL, by default when server is run in local it is http://localhost:8000/mcp
"headers": {
"Authorization": "Bearer pk_your_api_key_here"
}
}
}
}
```
**Using JWT Token:**
```json
{
"mcpServers": {

193
mcp_server/prowler_mcp_server/prowler_app/utils/auth.py
View File

@@ -4,14 +4,13 @@ import os
from datetime import datetime
from typing import Dict, Optional
import httpx
from fastmcp.server.dependencies import get_http_headers
from prowler_mcp_server import __version__
from prowler_mcp_server.lib.logger import logger
class ProwlerAppAuth:
"""Handles authentication and token management for Prowler App API."""
"""Handles authentication for Prowler App API using API keys or JWT tokens."""
def __init__(
self,
@@ -21,23 +20,17 @@ class ProwlerAppAuth:
self.base_url = base_url.rstrip("/")
logger.info(f"Using Prowler App API base URL: {self.base_url}")
self.mode = mode
self.access_token: Optional[str] = None
self.api_key: Optional[str] = None
if mode == "stdio": # STDIO mode
self.email = os.getenv("PROWLER_APP_EMAIL")
self.password = os.getenv("PROWLER_APP_PASSWORD")
self.tenant_id = os.getenv("PROWLER_APP_TENANT_ID", None)
self.api_key = os.getenv("PROWLER_APP_API_KEY")
if not self.email or not self.password:
raise ValueError(
"PROWLER_APP_EMAIL and PROWLER_APP_PASSWORD environment variables are required"
)
else:
self.email = None
self.password = None
self.tenant_id = None
if not self.api_key:
raise ValueError("PROWLER_APP_API_KEY environment variable is required")
self.access_token: Optional[str] = None
self.refresh_token: Optional[str] = None
if not self.api_key.startswith("pk_"):
raise ValueError("Prowler App API key format is incorrect")
def _parse_jwt(self, token: str) -> Optional[Dict]:
"""Parse JWT token and return payload, similar to JS parseJwt function."""
@@ -66,67 +59,29 @@ class ProwlerAppAuth:
return None
async def authenticate(self) -> str:
"""Authenticate with Prowler App API and return access token."""
logger.info("Starting authentication with Prowler App API")
if self.mode == "stdio":
async with httpx.AsyncClient() as client:
try:
# Prepare JSON:API formatted request body
auth_attributes = {"email": self.email, "password": self.password}
if self.tenant_id:
auth_attributes["tenant_id"] = self.tenant_id
request_body = {
"data": {
"type": "tokens",
"attributes": auth_attributes,
}
}
response = await client.post(
f"{self.base_url}/api/v1/tokens",
json=request_body,
headers={
"Content-Type": "application/vnd.api+json",
"Accept": "application/vnd.api+json",
},
)
response.raise_for_status()
data = response.json()
# Extract token from JSON:API response format
self.access_token = (
data.get("data", {}).get("attributes", {}).get("access")
)
self.refresh_token = (
data.get("data", {}).get("attributes", {}).get("refresh")
)
if not self.access_token:
raise ValueError("Token not found in response")
logger.info("Authentication successful")
return self.access_token
except httpx.HTTPStatusError as e:
logger.error(
f"Authentication failed with HTTP status {e.response.status_code}: {e.response.text}"
)
raise ValueError(f"Authentication failed: {e.response.text}")
except Exception as e:
logger.error(f"Authentication failed with error: {e}")
raise ValueError(f"Authentication failed: {e}")
elif self.mode == "http":
"""Authenticate and return token (API key for STDIO, API key or JWT for HTTP)."""
if self.mode == "http":
headers = get_http_headers()
access_token = headers.get("authorization", None)
authorization_header = headers.get("authorization", None)
# Validate the token
if access_token:
if access_token.startswith("Bearer "):
access_token = access_token.replace("Bearer ", "")
if not authorization_header:
raise ValueError("No authorization header provided")
payload = self._parse_jwt(access_token)
# Extract token from Bearer header
if authorization_header.startswith("Bearer "):
token = authorization_header.replace("Bearer ", "")
else:
raise ValueError(
"Invalid authorization header format. Expected 'Bearer <token>'"
)
# Check if it's an API key or JWT token
if token.startswith("pk_"):
# API key - no expiration check needed
return token
else:
# JWT token - validate and check expiration
payload = self._parse_jwt(token)
if not payload:
raise ValueError("Invalid JWT token format")
@@ -135,100 +90,30 @@ class ProwlerAppAuth:
exp = payload.get("exp", 0)
if exp <= now:
raise ValueError("Token has expired")
else:
raise ValueError("No authorization token provided")
# Don't cache tokens in HTTP mode to prevent sharing between clients
return access_token
return token
else:
raise ValueError(f"Invalid mode: {self.mode}")
async def refresh_access_token(self) -> str:
"""Refresh the access token using the refresh token."""
if not self.refresh_token:
logger.info("No refresh token available, performing full authentication")
return await self.authenticate()
logger.info("Refreshing access token")
async with httpx.AsyncClient() as client:
try:
# Prepare JSON:API formatted request body for refresh
request_body = {
"data": {
"type": "tokens",
"attributes": {"refresh": self.refresh_token},
}
}
response = await client.post(
f"{self.base_url}/api/v1/tokens/refresh",
json=request_body,
headers={
"Content-Type": "application/vnd.api+json",
"Accept": "application/vnd.api+json",
},
)
response.raise_for_status()
data = response.json()
# Extract new access token from JSON:API response
self.access_token = (
data.get("data", {}).get("attributes", {}).get("access")
)
logger.info("Token refresh successful")
return self.access_token
except httpx.HTTPStatusError as e:
logger.warning(
f"Token refresh failed, attempting re-authentication: {e}"
)
# If refresh fails, re-authenticate
return await self.authenticate()
async def get_valid_token(self) -> str:
"""Get a valid access token, checking JWT expiry."""
if self.mode == "http":
# In HTTP mode, always authenticate fresh to prevent token sharing between clients
return await self.authenticate()
"""Get a valid token (API key or JWT token)."""
if self.mode == "stdio" and self.api_key:
return self.api_key
else:
# In STDIO mode, cache tokens for efficiency
current_token = self.access_token
need_new_token = True
if current_token:
payload = self._parse_jwt(current_token)
if payload:
now = int(datetime.now().timestamp())
time_left = payload.get("exp", 0) - now
if time_left > 120: # 2 minutes margin
need_new_token = False
if need_new_token:
token = await self.authenticate()
# Verify the new token
payload = self._parse_jwt(token)
return token
else:
return current_token
return await self.authenticate()
def get_headers(self, token: str) -> Dict[str, str]:
"""Get headers for API requests with authentication."""
if token.startswith("pk_"):
authorization_header = f"Api-Key {token}"
else:
authorization_header = f"Bearer {token}"
headers = {
"Authorization": f"Bearer {token}",
"Authorization": authorization_header,
"Content-Type": "application/vnd.api+json",
"Accept": "application/vnd.api+json",
"User-Agent": f"prowler-mcp-server/{__version__}",
}
# Add tenant ID header if available
if self.tenant_id:
headers["X-Tenant-Id"] = self.tenant_id
return headers