feat(mcp): add basic roles related tools

This commit is contained in:
Rubén De la Torre Vico
2026-07-22 12:52:37 +02:00
parent 27fcec4c01
commit f4b8d16621
6 changed files with 523 additions and 28 deletions
@@ -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
@@ -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,
)
@@ -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"),
)
@@ -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 []
@@ -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"
]
@@ -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).