From f4b8d166213bae4e2093c7a04d0bb376669bbd25 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rub=C3=A9n=20De=20la=20Torre=20Vico?= Date: Wed, 22 Jul 2026 12:52:37 +0200 Subject: [PATCH] feat(mcp): add basic roles related tools --- mcp_server/changelog.d/roles-tools.added.md | 1 + .../prowler_app/models/roles.py | 234 +++++++++++++++++ .../prowler_app/models/users.py | 29 +-- .../prowler_app/models/utils.py | 34 +++ .../prowler_app/tools/roles.py | 241 ++++++++++++++++++ .../prowler_app/utils/api_client.py | 12 +- 6 files changed, 523 insertions(+), 28 deletions(-) create mode 100644 mcp_server/changelog.d/roles-tools.added.md create mode 100644 mcp_server/prowler_mcp_server/prowler_app/models/roles.py create mode 100644 mcp_server/prowler_mcp_server/prowler_app/models/utils.py create mode 100644 mcp_server/prowler_mcp_server/prowler_app/tools/roles.py diff --git a/mcp_server/changelog.d/roles-tools.added.md b/mcp_server/changelog.d/roles-tools.added.md new file mode 100644 index 0000000000..e59e9a2138 --- /dev/null +++ b/mcp_server/changelog.d/roles-tools.added.md @@ -0,0 +1 @@ +RBAC role tools `prowler_list_roles`, `prowler_get_role`, `prowler_get_user_roles`, `prowler_assign_role_to_user`, and `prowler_remove_role_from_user` for browsing roles and managing a user's role assignments diff --git a/mcp_server/prowler_mcp_server/prowler_app/models/roles.py b/mcp_server/prowler_mcp_server/prowler_app/models/roles.py new file mode 100644 index 0000000000..809a4a6763 --- /dev/null +++ b/mcp_server/prowler_mcp_server/prowler_app/models/roles.py @@ -0,0 +1,234 @@ +"""Data models for Prowler RBAC roles. + +This module provides Pydantic models for representing Prowler roles with +two-tier complexity: +- SimplifiedRole: For list operations with essential identification fields +- DetailedRole: Extends simplified with the capabilities the role grants and + its related users / provider groups + +It also provides UserRolesResult, used by the tools that read or change the +roles assigned to a specific user. + +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 +from prowler_mcp_server.prowler_app.models.utils import extract_relationship_ids + +# Boolean "manage_*" attributes that describe what a role is allowed to do. +# Only the enabled ones are surfaced as a flat `permissions` list. +_MANAGE_PERMISSIONS = ( + "manage_users", + "manage_account", + "manage_billing", + "manage_integrations", + "manage_providers", + "manage_scans", + "manage_ingestions", + "manage_alerts", +) + + +class SimplifiedRole(MinimalSerializerMixin, BaseModel): + """Simplified role representation for list operations. + + Includes core identification fields for efficient overview. + Used by list_roles() tool. + """ + + model_config = ConfigDict(frozen=True) + + id: str = Field( + description="Unique UUIDv4 identifier for this role in Prowler database" + ) + name: str = Field(description="Human-readable name of the role") + permission_state: str | None = Field( + default=None, + description="Summary of the role's permissions: 'unlimited' (all), 'limited' (some), or 'none'", + ) + + @classmethod + def from_api_response(cls, data: dict[str, Any]) -> "SimplifiedRole": + """Transform a JSON:API role resource into a simplified model. + + Args: + data: Role data from API response['data'] (single item or list item) + + Returns: + SimplifiedRole instance + """ + attributes = data["attributes"] + + return cls( + id=data["id"], + name=attributes["name"], + permission_state=attributes.get("permission_state"), + ) + + +class DetailedRole(SimplifiedRole): + """Detailed role representation with granted capabilities and relationships. + + Extends SimplifiedRole with the concrete management capabilities the role + grants, its visibility scope, and the IDs of related users and provider + groups. Used by get_role(), get_user_roles() and the role assignment tools. + """ + + model_config = ConfigDict(frozen=True) + + permissions: list[str] | None = Field( + default=None, + description="Management capabilities granted by this role (only the enabled ones), e.g. ['manage_users', 'manage_scans']", + ) + unlimited_visibility: bool | None = Field( + default=None, + description="Whether the role can see all providers (True) or only those in its provider groups (False)", + ) + provider_group_ids: list[str] | None = Field( + default=None, + description="UUIDv4 identifiers of the provider groups this role is scoped to. An empty list means the role is not scoped to any provider group.", + ) + user_ids: list[str] | None = Field( + default=None, + description="UUIDv4 identifiers of the users this role is assigned to. An empty list means the role is not assigned to any user.", + ) + inserted_at: str | None = Field( + default=None, description="ISO 8601 timestamp when the role was created" + ) + updated_at: str | None = Field( + default=None, description="ISO 8601 timestamp when the role was last modified" + ) + + def _should_exclude(self, key: str, value: Any) -> bool: + """Keep fields whose "empty" form carries meaning. + + ``unlimited_visibility`` is kept even when ``False``, and the + relationship lists are kept even when empty so that an empty + ``user_ids``/``provider_group_ids`` explicitly signals "not assigned to + any user / not scoped to any provider group" instead of looking like an + omitted, unknown field to an agent. + """ + if key in ("unlimited_visibility", "user_ids", "provider_group_ids"): + return value is None + return super()._should_exclude(key, value) + + @classmethod + def from_api_response(cls, data: dict[str, Any]) -> "DetailedRole": + """Transform a JSON:API role resource into a detailed model. + + Args: + data: Role data from API response['data'] or an included role + + Returns: + DetailedRole instance with all fields populated + """ + attributes = data["attributes"] + relationships = data.get("relationships", {}) + + permissions = [ + permission + for permission in _MANAGE_PERMISSIONS + if attributes.get(permission) + ] + + return cls( + id=data["id"], + name=attributes["name"], + permission_state=attributes.get("permission_state"), + permissions=permissions, + unlimited_visibility=attributes.get("unlimited_visibility"), + provider_group_ids=extract_relationship_ids( + relationships, "provider_groups" + ), + user_ids=extract_relationship_ids(relationships, "users"), + inserted_at=attributes.get("inserted_at"), + updated_at=attributes.get("updated_at"), + ) + + +class RolesListResponse(BaseModel): + """Response model for list_roles() with pagination metadata. + + Follows the established pattern from ScansListResponse and UsersListResponse. + """ + + roles: list[SimplifiedRole] + total_num_roles: int + total_num_pages: int + current_page: int + + @classmethod + def from_api_response(cls, response: dict[str, Any]) -> "RolesListResponse": + """Transform a JSON:API list response into a roles list with pagination. + + Args: + response: Full API response with data and meta + + Returns: + RolesListResponse with simplified roles and pagination metadata + """ + data = response.get("data", []) + meta = response.get("meta", {}) + pagination = meta.get("pagination", {}) + + roles = [SimplifiedRole.from_api_response(item) for item in data] + + return cls( + roles=roles, + total_num_roles=pagination.get("count", 0), + total_num_pages=pagination.get("pages", 0), + current_page=pagination.get("page", 1), + ) + + +class UserRolesResult(MinimalSerializerMixin, BaseModel): + """The roles currently assigned to a user. + + Used by get_user_roles() to report a user's roles, and by the assignment + tools to report the authoritative role set after a change (with `changed` + and `message` describing the outcome). + """ + + user_id: str = Field(description="UUIDv4 identifier of the user") + total_num_roles: int = Field( + description="Number of roles currently assigned to the user" + ) + roles: list[DetailedRole] = Field( + description="The roles currently assigned to the user, with their granted capabilities" + ) + changed: bool | None = Field( + default=None, + description="For assignment operations: whether this call actually modified the user's roles", + ) + message: str | None = Field( + default=None, + description="For assignment operations: human-readable description of the outcome", + ) + + def _should_exclude(self, key: str, value: Any) -> bool: + """Always include the roles list, even when empty (explicit 'no roles').""" + if key == "roles": + return False + return super()._should_exclude(key, value) + + @classmethod + def build( + cls, + user_id: str, + roles: list[DetailedRole], + changed: bool | None = None, + message: str | None = None, + ) -> "UserRolesResult": + """Assemble a result from a user's role list, filling the count.""" + return cls( + user_id=user_id, + total_num_roles=len(roles), + roles=roles, + changed=changed, + message=message, + ) diff --git a/mcp_server/prowler_mcp_server/prowler_app/models/users.py b/mcp_server/prowler_mcp_server/prowler_app/models/users.py index ecfc0b9cfe..18dc22528f 100644 --- a/mcp_server/prowler_mcp_server/prowler_app/models/users.py +++ b/mcp_server/prowler_mcp_server/prowler_app/models/users.py @@ -14,30 +14,7 @@ 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 [] +from prowler_mcp_server.prowler_app.models.utils import extract_relationship_ids class SimplifiedUser(MinimalSerializerMixin, BaseModel): @@ -131,8 +108,8 @@ class DetailedUser(SimplifiedUser): 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"), + role_ids=extract_relationship_ids(relationships, "roles"), + membership_ids=extract_relationship_ids(relationships, "memberships"), ) diff --git a/mcp_server/prowler_mcp_server/prowler_app/models/utils.py b/mcp_server/prowler_mcp_server/prowler_app/models/utils.py new file mode 100644 index 0000000000..ae4e6876b9 --- /dev/null +++ b/mcp_server/prowler_mcp_server/prowler_app/models/utils.py @@ -0,0 +1,34 @@ +"""Shared helpers for building models from Prowler API responses. + +Stateless utilities used by the models' ``from_api_response()`` factory methods +to read the JSON:API document structure (relationships, linkage, etc.). Keeping +them here leaves ``base.py`` focused on the base model/mixin and gives these +response-parsing helpers a single, discoverable home. +""" + +from typing import Any + + +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. Missing or empty + relationships yield an empty list. + + 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 [] diff --git a/mcp_server/prowler_mcp_server/prowler_app/tools/roles.py b/mcp_server/prowler_mcp_server/prowler_app/tools/roles.py new file mode 100644 index 0000000000..98ce8f77d1 --- /dev/null +++ b/mcp_server/prowler_mcp_server/prowler_app/tools/roles.py @@ -0,0 +1,241 @@ +"""Role (RBAC) tools for Prowler MCP Server. + +This module provides read tools for browsing roles and inspecting the roles +assigned to a user, plus two convenience tools for assigning and removing a +role from a user. The assignment tools wrap Prowler's JSON:API relationship +endpoints (which have append/replace/remove semantics and several guard rails) +behind an idempotent, single-role interface so agents do not have to reason +about those low-level details. +""" + +from typing import Any + +from pydantic import Field + +from prowler_mcp_server.prowler_app.models.roles import ( + DetailedRole, + RolesListResponse, + UserRolesResult, +) +from prowler_mcp_server.prowler_app.tools.base import BaseTool + + +class RolesTools(BaseTool): + """Tools for RBAC role operations. + + Provides tools for: + - prowler_list_roles: List the roles defined in the tenant + - prowler_get_role: Get detailed information about a specific role by ID + - prowler_get_user_roles: List the roles assigned to a specific user + - prowler_assign_role_to_user: Assign a role to a user (idempotent) + - prowler_remove_role_from_user: Remove a role from a user (idempotent) + """ + + async def list_roles( + self, + search: str | None = Field( + default=None, + description="Free-text search term to find roles (matches on name and related fields).", + ), + 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 RBAC roles defined in the authenticated tenant. + + Use this to discover which roles exist and their permission scope before + assigning one to a user. Returns LIGHTWEIGHT role information. + + Each role includes: + - id: Prowler internal UUID (v4), used with `prowler_get_role` and the assignment tools + - name: Human-readable role name + - permission_state: Summary of what the role grants ('unlimited', 'limited' or 'none') + + For the concrete capabilities a role grants and the users/provider groups + it relates to, use `prowler_get_role`. + """ + self.api_client.validate_page_size(page_size) + + params: dict[str, Any] = { + "fields[roles]": "name,permission_state", + "page[number]": page_number, + "page[size]": page_size, + } + if search: + params["filter[search]"] = search + + clean_params = self.api_client.build_filter_params(params) + + api_response = await self.api_client.get("/roles", params=clean_params) + simplified_response = RolesListResponse.from_api_response(api_response) + + return simplified_response.model_dump() + + async def get_role( + self, + role_id: str = Field( + description="Prowler's internal UUID (v4) for the role to retrieve. Use `prowler_list_roles` to find role IDs if you only know a name." + ), + ) -> dict[str, Any]: + """Retrieve detailed information about a specific role by its ID. + + Returns everything `prowler_list_roles` returns PLUS: + - permissions: The management capabilities the role grants (only the enabled ones) + - unlimited_visibility: Whether the role can see all providers or only its provider groups + - provider_group_ids: Provider groups the role is scoped to (empty list means it is scoped to no provider group) + - user_ids: Users the role is assigned to (empty list means it is assigned to no user) + - inserted_at / updated_at: Lifecycle timestamps + + The `user_ids` and `provider_group_ids` fields are always present: an + empty list means "none", not "unknown". + + Workflow: + 1. Use `prowler_list_roles` to browse roles and find the target role 'id' + 2. Use this tool with that 'id' to inspect exactly what the role grants + """ + api_response = await self.api_client.get(f"/roles/{role_id}") + detailed_role = DetailedRole.from_api_response(api_response["data"]) + + return detailed_role.model_dump() + + async def get_user_roles( + self, + user_id: str = Field( + description="Prowler's internal UUID (v4) for the user whose roles you want. Use `prowler_list_users` to find user IDs, or `prowler_get_current_user` for the caller." + ), + ) -> dict[str, Any]: + """List the roles currently assigned to a specific user. + + Returns the user's roles with the concrete capabilities each one grants, + so you can see what the user is allowed to do in the tenant. + + Workflow: + 1. Use `prowler_list_users` (or `prowler_get_current_user`) to find the user 'id' + 2. Use this tool to see which roles they hold and what those roles grant + 3. Use `prowler_assign_role_to_user` / `prowler_remove_role_from_user` to change them + """ + roles = await self._fetch_user_roles(user_id) + + return UserRolesResult.build(user_id=user_id, roles=roles).model_dump() + + async def assign_role_to_user( + self, + user_id: str = Field( + description="Prowler's internal UUID (v4) for the user to assign the role to. Use `prowler_list_users` to find user IDs." + ), + role_id: str = Field( + description="Prowler's internal UUID (v4) for the role to assign. Use `prowler_list_roles` to find role IDs." + ), + ) -> dict[str, Any]: + """Assign a role to a user, keeping their other roles intact. + + This tool is idempotent: if the user already has the role, it makes no + change and reports `changed: false`. It always returns the user's full, + up-to-date role set after the operation. + + Note: role management requires MANAGE_ACCOUNT permission. The API also + enforces guard rails (for example, it keeps at least one user with + MANAGE_ACCOUNT in the tenant); such rejections are surfaced as errors. + + Workflow: + 1. Use `prowler_list_roles` to find the role 'id' to grant + 2. Use `prowler_list_users` to find the target user 'id' + 3. Use this tool to assign the role + """ + current_roles = await self._fetch_user_roles(user_id) + if any(role.id == role_id for role in current_roles): + return UserRolesResult.build( + user_id=user_id, + roles=current_roles, + changed=False, + message=f"Role {role_id} is already assigned to user {user_id}; no change made.", + ).model_dump() + + # POST appends the role to the user's existing roles. + await self.api_client.post( + f"/users/{user_id}/relationships/roles", + json_data={"data": [{"type": "roles", "id": role_id}]}, + ) + + updated_roles = await self._fetch_user_roles(user_id) + return UserRolesResult.build( + user_id=user_id, + roles=updated_roles, + changed=True, + message=f"Role {role_id} assigned to user {user_id}.", + ).model_dump() + + async def remove_role_from_user( + self, + user_id: str = Field( + description="Prowler's internal UUID (v4) for the user to remove the role from. Use `prowler_list_users` to find user IDs." + ), + role_id: str = Field( + description="Prowler's internal UUID (v4) for the role to remove. Use `prowler_get_user_roles` to see which roles the user currently holds." + ), + ) -> dict[str, Any]: + """Remove a single role from a user, keeping their other roles intact. + + This tool is idempotent: if the user does not have the role, it makes no + change and reports `changed: false`. It always returns the user's full, + up-to-date role set after the operation. + + Note: role management requires MANAGE_ACCOUNT permission. The API also + enforces guard rails — users cannot remove their own role assignments, + and the last user holding MANAGE_ACCOUNT in the tenant cannot lose it; + such rejections are surfaced as errors. + + Workflow: + 1. Use `prowler_get_user_roles` to see the user's current roles + 2. Use this tool with the role 'id' you want to remove + """ + current_roles = await self._fetch_user_roles(user_id) + if not any(role.id == role_id for role in current_roles): + return UserRolesResult.build( + user_id=user_id, + roles=current_roles, + changed=False, + message=f"Role {role_id} is not assigned to user {user_id}; no change made.", + ).model_dump() + + # DELETE with a body removes only the listed role, leaving the rest. + await self.api_client.delete( + f"/users/{user_id}/relationships/roles", + json_data={"data": [{"type": "roles", "id": role_id}]}, + ) + + updated_roles = await self._fetch_user_roles(user_id) + return UserRolesResult.build( + user_id=user_id, + roles=updated_roles, + changed=True, + message=f"Role {role_id} removed from user {user_id}.", + ).model_dump() + + # Private helper methods + + async def _fetch_user_roles(self, user_id: str) -> list[DetailedRole]: + """Fetch the roles currently assigned to a user. + + Uses a single ``GET /users/{id}?include=roles`` request and reads the + role resources from the JSON:API ``included`` section. + + Args: + user_id: The Prowler UUID of the user + + Returns: + The user's roles as DetailedRole instances (empty list if none) + """ + response = await self.api_client.get( + f"/users/{user_id}", params={"include": "roles"} + ) + included = response.get("included", []) or [] + + return [ + DetailedRole.from_api_response(item) + for item in included + if item.get("type") == "roles" + ] diff --git a/mcp_server/prowler_mcp_server/prowler_app/utils/api_client.py b/mcp_server/prowler_mcp_server/prowler_app/utils/api_client.py index 187364bee1..3222454dcb 100644 --- a/mcp_server/prowler_mcp_server/prowler_app/utils/api_client.py +++ b/mcp_server/prowler_mcp_server/prowler_app/utils/api_client.py @@ -176,13 +176,19 @@ class ProwlerAPIClient(metaclass=SingletonMeta): ) async def delete( - self, path: str, params: dict[str, any] | None = None + self, + path: str, + params: dict[str, any] | None = None, + json_data: dict[str, any] | None = None, ) -> dict[str, any]: """Make DELETE request. Args: path: API endpoint path params: Optional query parameters + json_data: Optional JSON body data. Some JSON:API relationship + endpoints (e.g. ``/users/{id}/relationships/roles``) accept a + body listing the specific members to remove. Returns: API response as dictionary @@ -190,7 +196,9 @@ class ProwlerAPIClient(metaclass=SingletonMeta): Raises: Exception: If API request fails """ - return await self._make_request(HTTPMethod.DELETE, path, params=params) + return await self._make_request( + HTTPMethod.DELETE, path, params=params, json_data=json_data + ) async def fetch_external_url(self, url: str) -> str: """Fetch content from an allowed external URL (unauthenticated).