mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-23 20:42:02 +00:00
feat(mcp): add basic users related tools
This commit is contained in:
Rubén De la Torre Vico
@@ -0,0 +1 @@
|
||||
Read-only user management tools `prowler_list_users`, `prowler_get_user`, and `prowler_get_current_user` for listing tenant users with their emails and identifying the authenticated user
|
||||
@@ -0,0 +1,171 @@
|
||||
"""Data models for Prowler users.
|
||||
|
||||
This module provides Pydantic models for representing Prowler users with
|
||||
two-tier complexity:
|
||||
- SimplifiedUser: For list operations with essential identification fields
|
||||
- DetailedUser: Extends simplified with account metadata and role/membership links
|
||||
|
||||
All models inherit from MinimalSerializerMixin to exclude None/empty values
|
||||
for optimal LLM token usage.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
from prowler_mcp_server.prowler_app.models.base import MinimalSerializerMixin
|
||||
|
||||
|
||||
def _extract_relationship_ids(
|
||||
relationships: dict[str, Any], relationship_name: str
|
||||
) -> list[str]:
|
||||
"""Extract related resource IDs from a JSON:API relationship.
|
||||
|
||||
Handles both to-one (``data`` is an object) and to-many (``data`` is a list)
|
||||
relationships, returning a flat list of IDs in either case.
|
||||
|
||||
Args:
|
||||
relationships: The ``relationships`` object from a JSON:API resource
|
||||
relationship_name: The relationship key to read (e.g. ``"roles"``)
|
||||
|
||||
Returns:
|
||||
List of related resource IDs (empty if the relationship is absent/empty)
|
||||
"""
|
||||
data = relationships.get(relationship_name, {}).get("data")
|
||||
if not data:
|
||||
return []
|
||||
if isinstance(data, list):
|
||||
return [item["id"] for item in data if item and item.get("id")]
|
||||
# to-one relationship
|
||||
return [data["id"]] if data.get("id") else []
|
||||
|
||||
|
||||
class SimplifiedUser(MinimalSerializerMixin, BaseModel):
|
||||
"""Simplified user representation for list operations.
|
||||
|
||||
Includes core identification fields for efficient overview.
|
||||
Used by list_users() tool.
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
id: str = Field(
|
||||
description="Unique UUIDv4 identifier for this user in Prowler database"
|
||||
)
|
||||
name: str = Field(description="Display name of the user")
|
||||
email: str = Field(description="Email address of the user")
|
||||
company_name: str | None = Field(
|
||||
default=None, description="Company the user belongs to, if provided"
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_api_response(cls, data: dict[str, Any]) -> "SimplifiedUser":
|
||||
"""Transform a JSON:API user resource into a simplified model.
|
||||
|
||||
Args:
|
||||
data: User data from API response['data'] (single item or list item)
|
||||
|
||||
Returns:
|
||||
SimplifiedUser instance
|
||||
"""
|
||||
attributes = data["attributes"]
|
||||
|
||||
return cls(
|
||||
id=data["id"],
|
||||
name=attributes["name"],
|
||||
email=attributes["email"],
|
||||
company_name=attributes.get("company_name"),
|
||||
)
|
||||
|
||||
|
||||
class DetailedUser(SimplifiedUser):
|
||||
"""Detailed user representation with account metadata and relationships.
|
||||
|
||||
Extends SimplifiedUser with verification status, join date, and the IDs of
|
||||
the roles and memberships associated with the user.
|
||||
Used by get_user() and get_current_user() tools.
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
is_verified: bool | None = Field(
|
||||
default=None,
|
||||
description="Whether the user has verified their email address",
|
||||
)
|
||||
date_joined: str | None = Field(
|
||||
default=None,
|
||||
description="ISO 8601 timestamp when the user joined",
|
||||
)
|
||||
role_ids: list[str] | None = Field(
|
||||
default=None,
|
||||
description="UUIDv4 identifiers of the roles assigned to the user",
|
||||
)
|
||||
membership_ids: list[str] | None = Field(
|
||||
default=None,
|
||||
description="UUIDv4 identifiers of the tenant memberships of the user",
|
||||
)
|
||||
|
||||
def _should_exclude(self, key: str, value: Any) -> bool:
|
||||
"""Always include is_verified even when it is False."""
|
||||
if key == "is_verified":
|
||||
return value is None
|
||||
return super()._should_exclude(key, value)
|
||||
|
||||
@classmethod
|
||||
def from_api_response(cls, data: dict[str, Any]) -> "DetailedUser":
|
||||
"""Transform a JSON:API user resource into a detailed model.
|
||||
|
||||
Args:
|
||||
data: User data from API response['data']
|
||||
|
||||
Returns:
|
||||
DetailedUser instance with all fields populated
|
||||
"""
|
||||
attributes = data["attributes"]
|
||||
relationships = data.get("relationships", {})
|
||||
|
||||
return cls(
|
||||
id=data["id"],
|
||||
name=attributes["name"],
|
||||
email=attributes["email"],
|
||||
company_name=attributes.get("company_name"),
|
||||
is_verified=attributes.get("is_verified"),
|
||||
date_joined=attributes.get("date_joined"),
|
||||
role_ids=_extract_relationship_ids(relationships, "roles"),
|
||||
membership_ids=_extract_relationship_ids(relationships, "memberships"),
|
||||
)
|
||||
|
||||
|
||||
class UsersListResponse(BaseModel):
|
||||
"""Response model for list_users() with pagination metadata.
|
||||
|
||||
Follows the established pattern from ScansListResponse and ProvidersListResponse.
|
||||
"""
|
||||
|
||||
users: list[SimplifiedUser]
|
||||
total_num_users: int
|
||||
total_num_pages: int
|
||||
current_page: int
|
||||
|
||||
@classmethod
|
||||
def from_api_response(cls, response: dict[str, Any]) -> "UsersListResponse":
|
||||
"""Transform a JSON:API list response into a users list with pagination.
|
||||
|
||||
Args:
|
||||
response: Full API response with data and meta
|
||||
|
||||
Returns:
|
||||
UsersListResponse with simplified users and pagination metadata
|
||||
"""
|
||||
data = response.get("data", [])
|
||||
meta = response.get("meta", {})
|
||||
pagination = meta.get("pagination", {})
|
||||
|
||||
users = [SimplifiedUser.from_api_response(item) for item in data]
|
||||
|
||||
return cls(
|
||||
users=users,
|
||||
total_num_users=pagination.get("count", 0),
|
||||
total_num_pages=pagination.get("pages", 0),
|
||||
current_page=pagination.get("page", 1),
|
||||
)
|
||||
@@ -0,0 +1,118 @@
|
||||
"""User management tools for Prowler MCP Server.
|
||||
|
||||
This module provides read-only tools for viewing the users that belong to the
|
||||
authenticated tenant, including identifying which user the current credentials
|
||||
(API key or JWT) authenticate as.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from pydantic import Field
|
||||
|
||||
from prowler_mcp_server.prowler_app.models.users import (
|
||||
DetailedUser,
|
||||
UsersListResponse,
|
||||
)
|
||||
from prowler_mcp_server.prowler_app.tools.base import BaseTool
|
||||
|
||||
|
||||
class UsersTools(BaseTool):
|
||||
"""Tools for user management operations (read-only).
|
||||
|
||||
Provides tools for:
|
||||
- prowler_list_users: List the users in the tenant with their names and emails
|
||||
- prowler_get_user: Get detailed information about a specific user by ID
|
||||
- prowler_get_current_user: Identify which user the current credentials authenticate as
|
||||
"""
|
||||
|
||||
async def list_users(
|
||||
self,
|
||||
name: str | None = Field(
|
||||
default=None,
|
||||
description="Filter by user display name. Partial match supported (case-insensitive).",
|
||||
),
|
||||
email: str | None = Field(
|
||||
default=None,
|
||||
description="Filter by user email address. Partial match supported (case-insensitive).",
|
||||
),
|
||||
page_size: int = Field(
|
||||
default=50, description="Number of results to return per page"
|
||||
),
|
||||
page_number: int = Field(
|
||||
default=1, description="Page number to retrieve (1-indexed)"
|
||||
),
|
||||
) -> dict[str, Any]:
|
||||
"""List the users that belong to the authenticated tenant.
|
||||
|
||||
Use this to see who has access to the tenant and to look up their email
|
||||
addresses. Returns LIGHTWEIGHT user information optimized for browsing.
|
||||
|
||||
Each user includes:
|
||||
- id: Prowler internal UUID (v4), used with `prowler_get_user`
|
||||
- name: Display name
|
||||
- email: Email address
|
||||
- company_name: Company the user belongs to, when set
|
||||
|
||||
To find out which user the current credentials authenticate as, use
|
||||
`prowler_get_current_user`. For a single user's roles, membership links,
|
||||
verification status and join date, use `prowler_get_user`.
|
||||
"""
|
||||
self.api_client.validate_page_size(page_size)
|
||||
|
||||
params: dict[str, Any] = {
|
||||
"fields[users]": "name,email,company_name",
|
||||
"page[number]": page_number,
|
||||
"page[size]": page_size,
|
||||
}
|
||||
|
||||
if name:
|
||||
params["filter[name__icontains]"] = name
|
||||
if email:
|
||||
params["filter[email__icontains]"] = email
|
||||
|
||||
clean_params = self.api_client.build_filter_params(params)
|
||||
|
||||
api_response = await self.api_client.get("/users", params=clean_params)
|
||||
simplified_response = UsersListResponse.from_api_response(api_response)
|
||||
|
||||
return simplified_response.model_dump()
|
||||
|
||||
async def get_user(
|
||||
self,
|
||||
user_id: str = Field(
|
||||
description="Prowler's internal UUID (v4) for the user to retrieve. Use `prowler_list_users` to find user IDs if you only know a name or email."
|
||||
),
|
||||
) -> dict[str, Any]:
|
||||
"""Retrieve detailed information about a specific user by their ID.
|
||||
|
||||
Returns everything `prowler_list_users` returns PLUS:
|
||||
- is_verified: Whether the user has verified their email address
|
||||
- date_joined: When the user joined
|
||||
- role_ids: UUIDs of the roles assigned to the user
|
||||
- membership_ids: UUIDs of the user's tenant memberships
|
||||
|
||||
Workflow:
|
||||
1. Use `prowler_list_users` to browse users and find the target user 'id'
|
||||
2. Use this tool with that 'id' to inspect the user's roles and account details
|
||||
"""
|
||||
api_response = await self.api_client.get(f"/users/{user_id}")
|
||||
detailed_user = DetailedUser.from_api_response(api_response["data"])
|
||||
|
||||
return detailed_user.model_dump()
|
||||
|
||||
async def get_current_user(self) -> dict[str, Any]:
|
||||
"""Identify which user the current credentials authenticate as.
|
||||
|
||||
Use this to determine the identity behind the credentials this MCP server
|
||||
is currently using, e.g. before performing actions on behalf of that user
|
||||
or when reporting who is connected.
|
||||
|
||||
Returns the same detailed information as `prowler_get_user`:
|
||||
- id, name, email, company_name
|
||||
- is_verified, date_joined
|
||||
- role_ids, membership_ids
|
||||
"""
|
||||
api_response = await self.api_client.get("/users/me")
|
||||
detailed_user = DetailedUser.from_api_response(api_response["data"])
|
||||
|
||||
return detailed_user.model_dump()
|
||||
Reference in New Issue
Block a user