From 27fcec4c01e56cfbce753c073ed94a392e4913d5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rub=C3=A9n=20De=20la=20Torre=20Vico?= Date: Wed, 22 Jul 2026 12:19:43 +0200 Subject: [PATCH] feat(mcp): add basic users related tools --- .../changelog.d/users-read-tools.added.md | 1 + .../prowler_app/models/users.py | 171 ++++++++++++++++++ .../prowler_app/tools/users.py | 118 ++++++++++++ 3 files changed, 290 insertions(+) create mode 100644 mcp_server/changelog.d/users-read-tools.added.md create mode 100644 mcp_server/prowler_mcp_server/prowler_app/models/users.py create mode 100644 mcp_server/prowler_mcp_server/prowler_app/tools/users.py diff --git a/mcp_server/changelog.d/users-read-tools.added.md b/mcp_server/changelog.d/users-read-tools.added.md new file mode 100644 index 0000000000..3bc55952d8 --- /dev/null +++ b/mcp_server/changelog.d/users-read-tools.added.md @@ -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 diff --git a/mcp_server/prowler_mcp_server/prowler_app/models/users.py b/mcp_server/prowler_mcp_server/prowler_app/models/users.py new file mode 100644 index 0000000000..ecfc0b9cfe --- /dev/null +++ b/mcp_server/prowler_mcp_server/prowler_app/models/users.py @@ -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), + ) diff --git a/mcp_server/prowler_mcp_server/prowler_app/tools/users.py b/mcp_server/prowler_mcp_server/prowler_app/tools/users.py new file mode 100644 index 0000000000..a7f910c365 --- /dev/null +++ b/mcp_server/prowler_mcp_server/prowler_app/tools/users.py @@ -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()