Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
cbc5f58998 | ||
|
|
975c3f082d | ||
|
|
c37e2b90dc | ||
|
|
37ebd9b6fd | ||
|
|
a28487cbff | ||
|
|
68471d2a0e | ||
|
|
d41b2eaa0f | ||
|
|
b480907484 | ||
|
|
6d7bc8a86e | ||
|
|
8bfca81e4b | ||
|
|
931612443a | ||
|
|
3074f02a63 | ||
|
|
5cfc22040a | ||
|
|
48ba1692e1 | ||
|
|
a8b12813f9 | ||
|
|
85c36bb812 | ||
|
|
ce037318cd | ||
|
|
356036fe1f | ||
|
|
286685a4f3 | ||
|
|
9daca2e4df | ||
|
|
3ca3a977a9 | ||
|
|
561a1390be |
@@ -111,6 +111,7 @@ jobs:
|
||||
with:
|
||||
files: |
|
||||
api/**
|
||||
.trivyignore.yaml
|
||||
.github/actions/trivy-scan/**
|
||||
.github/actions/grype-scan/**
|
||||
.grype.yaml
|
||||
|
||||
@@ -103,6 +103,7 @@ jobs:
|
||||
with:
|
||||
files: |
|
||||
mcp_server/**
|
||||
.trivyignore.yaml
|
||||
.github/actions/trivy-scan/**
|
||||
.github/actions/grype-scan/**
|
||||
.grype.yaml
|
||||
|
||||
@@ -10,6 +10,7 @@ on:
|
||||
- 'Dockerfile*'
|
||||
- 'pyproject.toml'
|
||||
- 'uv.lock'
|
||||
- '.trivyignore.yaml'
|
||||
- '.github/workflows/sdk-container-checks.yml'
|
||||
pull_request:
|
||||
branches:
|
||||
@@ -116,6 +117,7 @@ jobs:
|
||||
Dockerfile*
|
||||
pyproject.toml
|
||||
uv.lock
|
||||
.trivyignore.yaml
|
||||
.github/workflows/sdk-container-checks.yml
|
||||
.github/actions/trivy-scan/**
|
||||
.github/actions/grype-scan/**
|
||||
|
||||
@@ -104,6 +104,7 @@ jobs:
|
||||
with:
|
||||
files: |
|
||||
ui/**
|
||||
.trivyignore.yaml
|
||||
.github/actions/trivy-scan/**
|
||||
.github/actions/grype-scan/**
|
||||
.grype.yaml
|
||||
|
||||
@@ -9,6 +9,14 @@ ignore:
|
||||
|
||||
# Modules compiled into the Trivy binary we ship.
|
||||
# Only a Trivy rebuild by its vendor can change these; the version is pinned in our Dockerfile.
|
||||
# CVE-2026-71556 is the same temporary exception documented in .trivyignore.yaml:
|
||||
# Trivy 0.73.0 still embeds go-git 5.19.1, while the 5.19.2 fix is merged only on
|
||||
# Trivy main. Remove this entry with the Trivy exception by 2026-09-15.
|
||||
# https://github.com/aquasecurity/trivy/blob/v0.73.0/go.mod#L46
|
||||
# https://github.com/aquasecurity/trivy/commit/a2edba9a03987ba0d2ebc8212c1a9a1e6979497b
|
||||
- vulnerability: CVE-2026-71556
|
||||
package:
|
||||
name: github.com/go-git/go-git/v5
|
||||
- vulnerability: CVE-2026-56852
|
||||
package:
|
||||
name: golang.org/x/text
|
||||
@@ -25,6 +33,19 @@ ignore:
|
||||
package:
|
||||
name: Microsoft.Bcl.Memory
|
||||
|
||||
# The .NET runtime bundled inside the PowerShell tarball the Dockerfile pins.
|
||||
# CVE-2026-62901 is the same temporary exception documented in .trivyignore.yaml:
|
||||
# fixed in .NET 9.0.19 / 10.0.11 (2026-08-11), but no published PowerShell release
|
||||
# ships a patched runtime yet (7.5.9 bundles 9.0.18; 7.6.4 bundles 10.0.x < 10.0.11).
|
||||
# pwsh runs only local M365 module cmdlets; nothing listens for inbound WebSocket
|
||||
# connections. Remove with the Trivy exception by 2026-09-15.
|
||||
- vulnerability: CVE-2026-62901
|
||||
package:
|
||||
name: Microsoft.NETCore.App.Runtime.linux-x64
|
||||
- vulnerability: CVE-2026-62901
|
||||
package:
|
||||
name: Microsoft.NETCore.App.Runtime.linux-arm64
|
||||
|
||||
|
||||
# The CPython interpreter, compiled into the official base image.
|
||||
# TEMPORARY, unlike the entries above: moving to Python 3.13 clears seven of these, and
|
||||
|
||||
@@ -118,8 +118,36 @@ vulnerabilities:
|
||||
- "pkg:npm/ip-address"
|
||||
expired_at: 2027-01-31
|
||||
|
||||
# CVE-2026-62901 is a DoS in System.Net.WebSockets (unchecked input for loop condition,
|
||||
# CWE-606), fixed in .NET 9.0.19 / 10.0.11 (published 2026-08-11). The vulnerable runtime
|
||||
# ships inside the PowerShell tarball the Dockerfile pins: 7.5.9 is the latest 7.5.x and
|
||||
# bundles .NET 9.0.18; 7.6.4 bundles .NET 10.0.x < 10.0.11, so no published PowerShell
|
||||
# release contains the fix yet. Prowler only invokes pwsh locally to run M365 module
|
||||
# cmdlets; the image does not accept inbound WebSocket connections, so the DoS path is
|
||||
# not reachable from the network. Remove this temporary suppression as soon as a
|
||||
# PowerShell release shipping .NET 9.0.19+ is available.
|
||||
- id: CVE-2026-62901
|
||||
purls:
|
||||
- "pkg:nuget/Microsoft.NETCore.App.Runtime.linux-x64"
|
||||
- "pkg:nuget/Microsoft.NETCore.App.Runtime.linux-arm64"
|
||||
expired_at: 2026-09-15
|
||||
|
||||
# Modules compiled into the Trivy binary the images ship. The binary is pinned by version
|
||||
# and verified by checksum in the Dockerfile; only a rebuild by its vendor moves these.
|
||||
# CVE-2026-71556 affects go-git worktree operations that can follow symlinks outside a
|
||||
# cloned repository. Trivy 0.72.0 contains go-git 5.19.1, and even the latest published
|
||||
# Trivy release, 0.73.0, still pins that vulnerable version:
|
||||
# https://github.com/aquasecurity/trivy/blob/v0.73.0/go.mod#L46
|
||||
# Trivy main already contains the 5.19.2 fix, but no published release includes it yet:
|
||||
# https://github.com/aquasecurity/trivy/commit/a2edba9a03987ba0d2ebc8212c1a9a1e6979497b
|
||||
# Prowler invokes Trivy only with `fs` on an existing local path or with `image`; it does
|
||||
# not ask Trivy to clone or mutate a Git worktree, so the affected path is not reachable.
|
||||
# Remove this temporary suppression as soon as a fixed Trivy release is available.
|
||||
- id: CVE-2026-71556
|
||||
purls:
|
||||
- "pkg:golang/github.com/go-git/go-git/v5"
|
||||
expired_at: 2026-09-15
|
||||
|
||||
- id: CVE-2026-56852
|
||||
purls:
|
||||
- "pkg:golang/golang.org/x/text"
|
||||
@@ -140,4 +168,3 @@ vulnerabilities:
|
||||
purls:
|
||||
- "pkg:golang/stdlib"
|
||||
expired_at: 2026-12-31
|
||||
|
||||
|
||||
@@ -126,12 +126,12 @@ Every AWS provider scan will enqueue an Attack Paths ingestion job automatically
|
||||
|
||||
| Provider | Checks | Services | [Compliance Frameworks](https://docs.prowler.com/user-guide/compliance/tutorials/compliance) | [Categories](https://docs.prowler.com/user-guide/cli/tutorials/misc#categories) | Support | Interface |
|
||||
|---|---|---|---|---|---|---|
|
||||
| AWS | 621 | 86 | 47 | 19 | Official | UI, API, CLI |
|
||||
| AWS | 639 | 86 | 47 | 19 | Official | UI, API, CLI |
|
||||
| Azure | 191 | 22 | 21 | 16 | Official | UI, API, CLI |
|
||||
| GCP | 109 | 20 | 19 | 12 | Official | UI, API, CLI |
|
||||
| Kubernetes | 92 | 7 | 8 | 11 | Official | UI, API, CLI |
|
||||
| GitHub | 24 | 3 | 2 | 5 | Official | UI, API, CLI |
|
||||
| M365 | 111 | 10 | 6 | 10 | Official | UI, API, CLI |
|
||||
| M365 | 143 | 10 | 6 | 10 | Official | UI, API, CLI |
|
||||
| OCI | 52 | 14 | 5 | 10 | Official | UI, API, CLI |
|
||||
| Alibaba Cloud | 63 | 9 | 6 | 9 | Official | UI, API, CLI |
|
||||
| Cloudflare | 29 | 3 | 2 | 5 | Official | UI, API, CLI |
|
||||
@@ -147,7 +147,7 @@ Every AWS provider scan will enqueue an Attack Paths ingestion job automatically
|
||||
| Huawei Cloud [Contact us](https://prowler.com/contact) | 25 | 10 | 1 | 6 | Unofficial | CLI |
|
||||
| E2E Networks [Contact us](https://prowler.com/contact) | 27 | 6 | 0 | 2 | Unofficial | CLI |
|
||||
| Scaleway [Contact us](https://prowler.com/contact) | 1 | 1 | 1 | 1 | Unofficial | CLI |
|
||||
| StackIT [Contact us](https://prowler.com/contact) | 7 | 2 | 1 | 3 | Unofficial | CLI |
|
||||
| StackIT [Contact us](https://prowler.com/contact) | 8 | 2 | 1 | 3 | Unofficial | CLI |
|
||||
| NHN | 6 | 2 | 2 | 0 | Unofficial | CLI |
|
||||
|
||||
> [!Note]
|
||||
|
||||
@@ -44,6 +44,11 @@ The main server orchestrates three sub-servers with prefixed namespacing:
|
||||
mcp_server/prowler_mcp_server/
|
||||
├── server.py # Main orchestrator
|
||||
├── main.py # CLI entry point
|
||||
├── lib/
|
||||
│ ├── server.py # ProwlerMCP, the base class of every sub-server
|
||||
│ ├── errors.py # Exception types and the one error renderer
|
||||
│ ├── logger.py
|
||||
│ └── analytics.py
|
||||
├── prowler_hub/
|
||||
├── prowler_app/
|
||||
│ ├── tools/ # Tool implementations
|
||||
@@ -59,6 +64,8 @@ The MCP Server uses two patterns for tool registration:
|
||||
1. **Direct Decorators** (Prowler Hub/Docs): Tools are registered using `@mcp.tool()` decorators
|
||||
2. **Auto-Discovery** (`prowler_app`): All public methods of `BaseTool` subclasses are auto-registered
|
||||
|
||||
Both funnel through `ProwlerMCP.tool` (`lib/server.py`), which is what applies the error contract to every tool no matter how it was registered. Build sub-servers with `ProwlerMCP`, never `FastMCP` directly.
|
||||
|
||||
## Adding Tools to the `prowler_app` Sub-Server
|
||||
|
||||
### Step 1: Create the Tool Class
|
||||
@@ -120,12 +127,10 @@ class NewFeatureTools(BaseTool):
|
||||
|
||||
Returns complete feature details including configuration and metadata.
|
||||
"""
|
||||
try:
|
||||
response = await self.api_client.get(f"/api/v1/features/{feature_id}")
|
||||
return DetailedFeature.from_api_response(response["data"]).model_dump()
|
||||
except Exception as e:
|
||||
self.logger.error(f"Failed to get feature {feature_id}: {e}")
|
||||
return {"error": str(e), "status": "failed"}
|
||||
# No try/except: a failure here raises, and the tool wrapper turns it into a
|
||||
# ToolError the client sees as `isError: true`. See "Error Handling" below.
|
||||
response = await self.api_client.get(f"/api/v1/features/{feature_id}")
|
||||
return DetailedFeature.from_api_response(response["data"]).model_dump()
|
||||
```
|
||||
|
||||
### Step 2: Create the Models
|
||||
@@ -369,18 +374,101 @@ async def search_items(self, status: str = Field(...)) -> dict:
|
||||
|
||||
### Error Handling
|
||||
|
||||
Return structured error responses instead of raising exceptions:
|
||||
Let failures raise. Every sub-server is a `ProwlerMCP` (`prowler_mcp_server/lib/server.py`),
|
||||
whose `tool()` wraps whatever it registers in `tool_errors`, turning any exception into a
|
||||
`ToolError`. The client sees `isError: true` and a message it can act on.
|
||||
|
||||
That wrapping is not something you apply — the two registration styles (the `@mcp.tool()`
|
||||
decorators, and the direct `mcp.tool(fn)` call `BaseTool` uses) both funnel through
|
||||
`ProwlerMCP.tool`. Build sub-servers with `ProwlerMCP`, never `FastMCP` directly: masking
|
||||
is on everywhere, so a tool that escaped the funnel would answer `Error calling tool 'x'`
|
||||
with no detail at all.
|
||||
|
||||
Never `return {"error": ...}`: a returned payload is `isError: false` at the MCP protocol
|
||||
level, so the client is told the call succeeded and only finds out otherwise if it happens
|
||||
to inspect the right key.
|
||||
|
||||
```python
|
||||
async def get_item(self, item_id: str) -> dict:
|
||||
try:
|
||||
response = await self.api_client.get(f"/api/v1/items/{item_id}")
|
||||
return DetailedItem.from_api_response(response["data"]).model_dump()
|
||||
except Exception as e:
|
||||
self.logger.error(f"Failed to get item {item_id}: {e}")
|
||||
return {"error": str(e), "status": "failed"}
|
||||
"""A rejected request, a timeout and a malformed payload all raise from here.
|
||||
|
||||
Each is rendered with the API's own words plus what it implies about retrying.
|
||||
"""
|
||||
response = await self.api_client.get(f"/api/v1/items/{item_id}")
|
||||
return DetailedItem.from_api_response(response["data"]).model_dump()
|
||||
```
|
||||
|
||||
Raise `ToolError` whenever the message is one you wrote for the caller. Its text reaches
|
||||
the client verbatim, so anything they need in order to recover has to be *in* the message
|
||||
— an error carries nothing else:
|
||||
|
||||
```python
|
||||
from fastmcp.exceptions import ToolError
|
||||
|
||||
if not data:
|
||||
raise ToolError(
|
||||
f"Item '{item_id}' was not found. Use prowler_list_items to find valid IDs."
|
||||
)
|
||||
```
|
||||
|
||||
**Do not raise `ValueError` from a tool.** The two are not interchangeable: anything that
|
||||
is not a `ToolError` is described as a bug in this server. That is right for a model
|
||||
factory rejecting an API payload or a pydantic `ValidationError`, and wrong for a
|
||||
refusal — so the exception type is what carries the distinction:
|
||||
|
||||
```text
|
||||
Date range cannot exceed 2 days. Requested range: 2025-01-01 to 2025-01-10 (10 days)
|
||||
|
||||
The Prowler MCP Server hit an unexpected ValueError: Missing pagination metadata in API
|
||||
response. This is a bug in the server, not something you can fix by changing the
|
||||
arguments.
|
||||
```
|
||||
|
||||
If you surface an exception yourself — into a `ToolError` you build, or into a field of a
|
||||
structured result — pass it through `render_tool_error(e)` rather than `str(e)`, so the
|
||||
same failure is never described two ways. Pass `warn=False` when the result already
|
||||
reports the outcome.
|
||||
|
||||
#### Deciding between an error and a result
|
||||
|
||||
Ask two questions, in order:
|
||||
|
||||
1. **Did the tool finish its own job?** `test_integration_connection`'s job is to run the
|
||||
check and report what happened, so `connected: false` is the job finished.
|
||||
`get_finding_details`' job is to return the finding, so no finding means it did not.
|
||||
2. **Is the reported state a fact about the remote world or about our call?** The world
|
||||
(Jira refused the credentials, 3 of 40 items failed, a discovery found nothing) is a
|
||||
**result**. Our call (403, connection reset, invalid UUID, a bug in a model factory) is
|
||||
an **error**.
|
||||
|
||||
One rule overrides both: **if a write may have partially landed, that fact travels in a
|
||||
successful structured result, never in an error.** An agent reads `isError: true` as
|
||||
"nothing happened, safe to retry"; reporting "I may have created 17 Jira issues" that way
|
||||
invites a duplicate dispatch.
|
||||
|
||||
#### What the client reads
|
||||
|
||||
`render_tool_error` describes the failure in one plain sentence: the call, the status and
|
||||
whatever the API said, with the field named when it named one.
|
||||
|
||||
```text
|
||||
GET /findings/b1ca536c failed with HTTP 404. No Finding matches the given query.
|
||||
POST /integrations failed with HTTP 400. This field may not be blank. (/data/attributes/configuration/bucket_name); Enter a valid URL.
|
||||
Date range cannot exceed 2 days. Requested range: 2025-01-01 to 2025-01-10 (10 days)
|
||||
```
|
||||
|
||||
Nothing is added that the status code already implies. The one exception is a request that
|
||||
could have changed something and never came back with a verdict — a 5xx or a timeout on a
|
||||
write — which gets a warning, because an agent otherwise reads any failure as "nothing
|
||||
happened" and sends the write again:
|
||||
|
||||
```text
|
||||
DELETE /integrations/i1 failed with HTTP 500. A server error occurred. It may have been carried out anyway, so check the current state before retrying.
|
||||
```
|
||||
|
||||
Every server sets `mask_error_details=True`. That costs nothing, because `ToolError`
|
||||
bypasses masking; it only stops raw internals escaping from code paths outside a tool.
|
||||
|
||||
### Parameter Descriptions
|
||||
|
||||
Use Pydantic `Field()` with clear descriptions. This also helps LLMs understand
|
||||
|
||||
@@ -184,7 +184,7 @@ Prowler enables security scanning of Kubernetes clusters, supporting both **in-c
|
||||
```
|
||||
|
||||
<Note>
|
||||
By default, Prowler scans all namespaces in the active Kubernetes context. Use the `--context`flag to specify the context to be scanned and `--namespaces` to restrict scanning to specific namespaces.
|
||||
By default, Prowler scans all namespaces in the active Kubernetes context. Use the `--context` flag to specify the context to be scanned and `--namespaces` to restrict scanning to specific namespaces.
|
||||
|
||||
</Note>
|
||||
## Microsoft 365
|
||||
|
||||
|
After Width: | Height: | Size: 179 KiB |
|
After Width: | Height: | Size: 203 KiB |
|
After Width: | Height: | Size: 215 KiB |
|
After Width: | Height: | Size: 190 KiB |
|
After Width: | Height: | Size: 166 KiB |
|
After Width: | Height: | Size: 201 KiB |
|
After Width: | Height: | Size: 190 KiB |
|
After Width: | Height: | Size: 182 KiB |
|
After Width: | Height: | Size: 199 KiB |
|
After Width: | Height: | Size: 184 KiB |
|
After Width: | Height: | Size: 204 KiB |
|
After Width: | Height: | Size: 194 KiB |
|
After Width: | Height: | Size: 195 KiB |
|
After Width: | Height: | Size: 196 KiB |
|
After Width: | Height: | Size: 173 KiB |
|
After Width: | Height: | Size: 191 KiB |
@@ -4,13 +4,13 @@ title: "Custom Checks Metadata"
|
||||
|
||||
In certain organizations, the severity of specific checks might differ from the default values defined in the check's metadata. For instance, while `s3_bucket_level_public_access_block` could be deemed `critical` for some organizations, others might assign a different severity level to it.
|
||||
|
||||
The custom metadata option offers a means to override default metadata set by Prowler
|
||||
The custom metadata option offers a means to override default metadata set by Prowler.
|
||||
|
||||
You can utilize `--custom-checks-metadata-file` followed by the path to your custom checks metadata YAML file.
|
||||
|
||||
## Available Fields
|
||||
|
||||
The list of supported check's metadata fields that can be override are listed as follows:
|
||||
The list of supported check's metadata fields that can be overridden are listed as follows:
|
||||
|
||||
- Severity
|
||||
- CheckTitle
|
||||
|
||||
@@ -322,7 +322,7 @@ The Mutelist Table must have the following columns:
|
||||
|
||||
- Checks (String): This field can contain either a Prowler Check Name or an `*` (which applies to all the scanned checks).
|
||||
|
||||
- Regions (List): This field contains a list of regions where this mutelist rule is applied (it can also contains an `*` to apply all scanned regions).
|
||||
- Regions (List): This field contains a list of regions where this mutelist rule is applied (it can also contain an `*` to apply all scanned regions).
|
||||
|
||||
- Resources (List): This field contains a list of regular expressions (regex) that applies to the resources that are wanted to be muted.
|
||||
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
title: 'Scanning Multiple AWS Accounts with Prowler'
|
||||
---
|
||||
|
||||
Prowler enables security scanning across multiple AWS accounts by utilizing the [Assume Role feature](/user-guide/providers/aws/role-assumption) and [integration with AWS Organizations feature](/user-guide/providers/aws/organizations).
|
||||
Prowler enables security scanning across multiple AWS accounts by utilizing the [Assume Role feature](/user-guide/providers/aws/role-assumption) and [integration with AWS Organizations feature](/user-guide/providers/aws/organizations).
|
||||
|
||||
This approach allows execution from a single account with permissions to assume roles in the target accounts.
|
||||
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
title: 'AWS Security Hub Integration with Prowler'
|
||||
---
|
||||
|
||||
Prowler natively supports **official integration** with [AWS Security Hub](https://aws.amazon.com/security-hub), allowing security findings to be sent directly. This integration enables **Prowler** to import its findings into AWS Security Hub.
|
||||
Prowler natively supports **official integration** with [AWS Security Hub](https://aws.amazon.com/security-hub), allowing security findings to be sent directly. This integration enables **Prowler** to import its findings into AWS Security Hub.
|
||||
|
||||
To activate the integration, follow these steps in at least one AWS region within your AWS account:
|
||||
|
||||
|
||||
@@ -39,10 +39,12 @@ When using service principal authentication, add these **Application Permissions
|
||||
|
||||
**Microsoft Graph API Permissions:**
|
||||
|
||||
- `AccessReview.Read.All`: Required for `entra_access_review_guest_users_configured` and `entra_access_review_privileged_roles_configured` checks. A Microsoft Entra ID P2 or Microsoft Entra ID Governance license is also required.
|
||||
- `AuditLog.Read.All`: Required for Entra service.
|
||||
- `Directory.Read.All`: Required for all services.
|
||||
- `OnPremDirectorySynchronization.Read.All`: Required for `entra_seamless_sso_disabled` check (hybrid deployments).
|
||||
- `Policy.Read.All`: Required for all services.
|
||||
- `RoleManagementPolicy.Read.Directory`: Required for `entra_pim_global_administrator_approval_required` and `entra_pim_privileged_role_administrator_approval_required` checks. A Microsoft Entra ID P2 or Microsoft Entra ID Governance license is also required.
|
||||
- `SecurityIdentitiesHealth.Read.All`: Required for `defenderidentity_health_issues_no_open` check.
|
||||
- `SecurityIdentitiesSensors.Read.All`: Required for `defenderidentity_health_issues_no_open` check.
|
||||
- `SharePointTenantSettings.Read.All`: Required for SharePoint service.
|
||||
@@ -110,10 +112,12 @@ Browser and Azure CLI authentication methods limit scanning capabilities to chec
|
||||
|
||||
3. Search and select the required permissions:
|
||||
|
||||
- `AccessReview.Read.All`: Required for `entra_access_review_guest_users_configured` and `entra_access_review_privileged_roles_configured` checks. A Microsoft Entra ID P2 or Microsoft Entra ID Governance license is also required
|
||||
- `AuditLog.Read.All`: Required for Entra service
|
||||
- `Directory.Read.All`: Required for all services
|
||||
- `OnPremDirectorySynchronization.Read.All`: Required for `entra_seamless_sso_disabled` check (hybrid deployments)
|
||||
- `Policy.Read.All`: Required for all services
|
||||
- `RoleManagementPolicy.Read.Directory`: Required for `entra_pim_global_administrator_approval_required` and `entra_pim_privileged_role_administrator_approval_required` checks. A Microsoft Entra ID P2 or Microsoft Entra ID Governance license is also required
|
||||
- `SecurityIdentitiesHealth.Read.All`: Required for `defenderidentity_health_issues_no_open` check
|
||||
- `SecurityIdentitiesSensors.Read.All`: Required for `defenderidentity_health_issues_no_open` check
|
||||
- `SharePointTenantSettings.Read.All`: Required for SharePoint service
|
||||
|
||||
@@ -136,7 +136,7 @@ To view all `new` findings that have not been seen prior to this scan, click the
|
||||
## Step 9: Download the Outputs
|
||||
Once a scan is complete, navigate to the `Scans` section to download the output files generated by Prowler:
|
||||
|
||||
You can download the output files generated by Prowler as a single `zip` file. This archive contains the CSV, JSON-OSCF, and HTML reports detailing the findings.
|
||||
You can download the output files generated by Prowler as a single `zip` file. This archive contains the CSV, JSON-OCSF, and HTML reports detailing the findings.
|
||||
|
||||
To download these files, click the **Download** button. This button becomes available only after the scan has finished.
|
||||
|
||||
|
||||
@@ -66,7 +66,7 @@ The Prowler wizard walks you through the entire flow: deploying both roles from
|
||||
<Frame>
|
||||
<img src="/images/organizations/aws-console-org-id.png" alt="AWS Organizations Console showing the Organization ID in the left sidebar" />
|
||||
</Frame>
|
||||
- **Name** (optional): A display name for the organization. If left blank, Prowler uses the name stored in AWS.
|
||||
- **Name** (optional): A display name for the organization. If left blank, Prowler uses the AWS organization ID.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/organization-details-form.png" alt="Organization Details form with Organization ID and Name fields" />
|
||||
@@ -280,7 +280,7 @@ Open the row actions menu on the organization row on the **Providers** page.
|
||||
|
||||
| Action | What it does |
|
||||
|--------|--------------|
|
||||
| **Edit Organization Name** | Renames the organization in Prowler. Leave it blank to fall back to the name stored in AWS. |
|
||||
| **Edit Organization Name** | Renames the organization in Prowler. Leave it blank to fall back to the AWS organization ID. |
|
||||
| **Update Credentials** | Reopens the Authentication Details step to store a new Role ARN. |
|
||||
| **Edit Scan Schedule** | Applies one schedule to every connected account in the organization. |
|
||||
| **Test Connections (N)** | Re-tests every account in the organization. |
|
||||
|
||||
@@ -1,11 +1,467 @@
|
||||
---
|
||||
title: 'Azure Management Groups'
|
||||
description: 'Onboard all Azure subscriptions in your management groups through a single guided wizard'
|
||||
tag: "Coming Soon"
|
||||
description: 'Discover the Azure subscriptions in your management groups and onboard the ones you select through a single guided wizard'
|
||||
---
|
||||
|
||||
Onboarding Azure management groups through a single guided wizard is coming soon to Prowler Cloud.
|
||||
import { VersionBadge } from "/snippets/version-badge.mdx"
|
||||
import { SubscriptionBanner } from "/snippets/subscription-banner.mdx"
|
||||
|
||||
Today, Azure subscriptions are onboarded individually. See [Getting Started with Azure](/user-guide/providers/azure/getting-started-azure) and [Bulk Provider Provisioning](/user-guide/tutorials/bulk-provider-provisioning) to automate onboarding multiple subscriptions.
|
||||
<VersionBadge version="5.39.0" />
|
||||
|
||||
Keep an eye on the [changelog](https://github.com/prowler-cloud/prowler/releases) for updates.
|
||||
Prowler Cloud discovers every Azure subscription in your Microsoft Entra tenant and onboards the eligible ones you select, through a single guided wizard. Instead of connecting subscriptions one by one, you can discover every Management Group and subscription under the tenant root Management Group, select the ones you want to monitor, test connectivity, and launch scans — all from the Prowler Cloud UI.
|
||||
|
||||
<SubscriptionBanner>
|
||||
For Command-Line Interface (CLI) scanning of several subscriptions, see [Configuring Specific Subscription Scans in Prowler](/user-guide/providers/azure/subscriptions).
|
||||
</SubscriptionBanner>
|
||||
|
||||
To follow this guide you need an active [Prowler Cloud](https://cloud.prowler.com) account and an Azure service principal with read access granted **at the tenant root Management Group**.
|
||||
|
||||
## Overview
|
||||
|
||||
### Individual Subscriptions vs Azure Management Groups
|
||||
|
||||
| Approach | Best for | How it works |
|
||||
|----------|----------|--------------|
|
||||
| **Individual subscriptions** | A few Azure subscriptions | Connect each subscription one by one with its own service principal credential. |
|
||||
| **Azure Management Groups** | 10+ subscriptions, or any tenant organized into Management Groups | Connect once with a tenant-wide credential, discover every Management Group and subscription automatically, and scan them in bulk. |
|
||||
|
||||
### How It Works
|
||||
|
||||
Onboarding runs in four stages:
|
||||
|
||||
1. **Grant read access** to one service principal at the tenant root Management Group.
|
||||
2. **Discover** — Prowler walks the Management Group hierarchy and returns every Management Group and subscription under the tenant root.
|
||||
3. **Select and connect** — choose the subscriptions to monitor. Prowler creates one provider per subscription and tests every connection.
|
||||
4. **Launch scans** — apply a scan schedule across the connected subscriptions.
|
||||
|
||||
<Note>
|
||||
**Nothing is deployed into your tenant.** Unlike AWS Organizations onboarding, Azure onboarding creates no resources in Azure. Prowler reuses the service principal credential you provide as the credential of every subscription it onboards, so one credential covers discovery and scanning once all the required permissions are granted.
|
||||
</Note>
|
||||
|
||||
<Note>
|
||||
**Onboarding always covers the whole tenant.** Prowler scopes the organization to the tenant root Management Group, which it derives from the Microsoft Entra tenant ID. The wizard therefore asks for the tenant, never for a Management Group — there is no partial onboarding from a Management Group further down the hierarchy.
|
||||
</Note>
|
||||
|
||||
## Before You Start
|
||||
|
||||
### Create a Service Principal
|
||||
|
||||
Prowler Cloud authenticates against Azure with a service principal application. To create one, follow [Creating a Prowler Service Principal Application](/user-guide/providers/azure/create-prowler-service-principal). Keep the **client ID** and the **client secret** — the wizard asks for both.
|
||||
|
||||
### Grant Read Access at the Tenant Root Management Group
|
||||
|
||||
Discovery reads the Management Group hierarchy and the subscriptions inside it. Assign the built-in **Reader** role to the service principal **on the tenant root Management Group**, so the grant covers every Management Group and subscription beneath it:
|
||||
|
||||
```bash
|
||||
TENANT_ID=<microsoft-entra-tenant-id>
|
||||
CLIENT_ID=<prowler-service-principal-client-id>
|
||||
|
||||
# Resolve the exact service principal from its client ID. A display-name search
|
||||
# is a prefix match, so it can return a different principal than intended.
|
||||
SP_OBJECT_ID=$(az ad sp show --id "$CLIENT_ID" --query id -o tsv)
|
||||
|
||||
az role assignment create \
|
||||
--role "Reader" \
|
||||
--assignee-object-id "$SP_OBJECT_ID" \
|
||||
--assignee-principal-type ServicePrincipal \
|
||||
--scope "/providers/Microsoft.Management/managementGroups/$TENANT_ID"
|
||||
```
|
||||
|
||||
<Note>
|
||||
The tenant root Management Group is named after the tenant ID, so its resource identifier is always `/providers/Microsoft.Management/managementGroups/<tenant-id>`. Creating a role assignment there requires **Owner** or **User Access Administrator** at that scope; a Global Administrator who holds neither can grant themselves access through the elevated-access toggle in Microsoft Entra ID.
|
||||
</Note>
|
||||
|
||||
### Grant the Scanning Permissions
|
||||
|
||||
Scanning each subscription additionally needs the permissions described in [Azure Authentication in Prowler](/user-guide/providers/azure/authentication#required-permissions):
|
||||
|
||||
| Permission | Scope | Used for |
|
||||
|------------|-------|----------|
|
||||
| **Reader** | Subscription, or inherited from the Management Group | Reading Azure resources during a scan. |
|
||||
| **ProwlerRole** | Subscription, or inherited from the Management Group | The read-only actions the Reader role does not cover. |
|
||||
| `AuditLog.Read.All`, `Directory.Read.All`, `Policy.Read.All` | Microsoft Entra ID (Microsoft Graph) | The Microsoft Entra ID checks. |
|
||||
|
||||
Role assignments made at the tenant root Management Group are inherited by every subscription beneath it, so assigning **Reader** and **ProwlerRole** there means every subscription you onboard is scannable without a per-subscription grant. To create the custom role, see [Assigning "ProwlerRole" Permissions at the Subscription Level](/user-guide/providers/azure/authentication#assigning-prowlerrole-permissions-at-the-subscription-level) — and set its `assignableScopes` to the Management Group, `/providers/Microsoft.Management/managementGroups/<tenant-id>`, rather than to a single subscription, so it can be assigned once for the whole tenant.
|
||||
|
||||
### Find Your Microsoft Entra Tenant ID
|
||||
|
||||
Prowler identifies the organization by the Microsoft Entra tenant ID, a Universally Unique Identifier (UUID):
|
||||
|
||||
```bash
|
||||
az account show --query tenantId -o tsv
|
||||
```
|
||||
|
||||
In the Azure portal, the same value sits on your service principal's **App registrations** > **Overview** blade as **Directory (tenant) ID**, next to the **Application (client) ID** the wizard also asks for. [Adding Azure credentials to Prowler Cloud](/user-guide/providers/azure/getting-started-azure#step-3-add-credentials-to-prowler-cloud) shows both on the annotated blade.
|
||||
|
||||
## Step 1: Start the Organization Wizard
|
||||
|
||||
### Open the Wizard
|
||||
|
||||
1. Navigate to **Providers** and click **Add Provider**.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/cloud-providers-add.png" alt="Providers page showing the Add Provider button" />
|
||||
</Frame>
|
||||
|
||||
2. Select **Microsoft Azure** as the provider.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/select-azure-provider.png" alt="Provider selection modal with Microsoft Azure highlighted" />
|
||||
</Frame>
|
||||
|
||||
3. Choose **Add Multiple Subscriptions With Azure Management Group**.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/select-azure-management-groups-method.png" alt="Method selector showing the Add Multiple Subscriptions With Azure Management Group option highlighted" />
|
||||
</Frame>
|
||||
|
||||
<Note>
|
||||
In Prowler Local Server the Management Group option is marked **Cloud** and opens an upgrade panel instead of the wizard. Management Group onboarding is a Prowler Cloud feature; the single-subscription method remains available.
|
||||
</Note>
|
||||
|
||||
### Enter Organization Details
|
||||
|
||||
- **Tenant ID**: the Microsoft Entra tenant ID (for example, `8b3c9a41-5f27-4d6e-9c18-7ae204f5b6d2`). Values that are not valid UUIDs are rejected before submission.
|
||||
- **Name** (optional): a display name for the organization in Prowler. If left blank, Prowler uses the tenant ID.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/azure-organization-details-form.png" alt="Organization Details form with the Microsoft Entra tenant ID and Name fields" />
|
||||
</Frame>
|
||||
|
||||
Click **Next** to proceed to the authentication phase. Prowler matches the organization by tenant ID, so submitting a tenant that is already onboarded reuses it instead of creating a duplicate.
|
||||
|
||||
## Step 2: Authenticate With Azure
|
||||
|
||||
The **Authentication Details** step collects the service principal Prowler uses to read the Management Group hierarchy and, later, to scan each subscription:
|
||||
|
||||
- **Client ID**: the application (client) ID of the service principal, a UUID.
|
||||
- **Client Secret**: a client secret of that service principal.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/azure-authentication-details.png" alt="Authentication Details step showing the Client ID and Client Secret fields" />
|
||||
</Frame>
|
||||
|
||||
The tenant is not repeated here: Prowler takes it from the organization created in the previous step.
|
||||
|
||||
<Warning>
|
||||
Every subscription you onboard inherits this credential. Revoking it, rotating the client secret, or deleting the service principal stops the scans of every subscription in the organization.
|
||||
</Warning>
|
||||
|
||||
### Authenticate and Discover
|
||||
|
||||
Click **Authenticate**. Prowler then:
|
||||
|
||||
- Creates the organization and stores the credential securely.
|
||||
- Triggers an asynchronous discovery that walks the Management Group hierarchy.
|
||||
- Shows a **"Gathering Azure Subscriptions..."** spinner while it waits.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/azure-gathering-subscriptions.png" alt="Gathering Azure Subscriptions spinner shown while discovery runs" />
|
||||
</Frame>
|
||||
|
||||
Discovery usually takes seconds to a couple of minutes, depending on how many Management Groups and subscriptions the tenant holds.
|
||||
|
||||
#### When Discovery Takes Too Long
|
||||
|
||||
Prowler waits up to **3 minutes** for a result. Past that, the wizard stops waiting — but the discovery keeps running in Azure — and offers two actions:
|
||||
|
||||
- **Keep waiting** — resume the same discovery. Nothing is re-read from Azure.
|
||||
- **Retry** — start a fresh discovery, which reads the hierarchy again.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/azure-discovery-timeout.png" alt="Discovery timeout notice offering Keep waiting and Retry" />
|
||||
</Frame>
|
||||
|
||||
If discovery fails outright, the wizard explains why and offers **Retry discovery**. See [Troubleshooting](#troubleshooting) for each message.
|
||||
|
||||
## Step 3: Select Subscriptions to Scan
|
||||
|
||||
### Understanding the Tree View
|
||||
|
||||
Once discovery completes, the wizard renders the tenant as a hierarchical tree:
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/azure-tree-view-subscriptions.png" alt="Hierarchical tree view showing Management Groups and subscriptions with selection checkboxes" />
|
||||
</Frame>
|
||||
|
||||
- **Management Groups** nest under the tenant root; subscriptions that sit directly under the tenant root appear at the top level. The tenant root itself is not shown as a row — the whole tree is its content.
|
||||
- Each Management Group row shows its Management Group name, with the Azure display name beside it. Hover the name to read the full Azure Resource Manager (ARM) resource identifier.
|
||||
- **Selecting a Management Group** selects every selectable subscription beneath it. A Management Group whose subscriptions are only partly selected renders in an indeterminate state.
|
||||
- **Individual overrides**: deselect single subscriptions even when the parent Management Group is selected.
|
||||
- The header tracks the selection as **"X of Y subscriptions selected"**.
|
||||
- Management Group hierarchies are read up to **six levels** below the tenant root Management Group, which is Azure's own platform limit. Deeper tenants report an error at discovery — see [Troubleshooting](#troubleshooting).
|
||||
|
||||
### Blocked Subscriptions
|
||||
|
||||
A subscription is shown grayed out and cannot be selected when Azure reports it as inactive, or when onboarding it would conflict with something Prowler already stores. Hover the subscription to see the reason:
|
||||
|
||||
| Reason | What it means |
|
||||
|--------|---------------|
|
||||
| `subscription_not_enabled` | The subscription is not in the **Enabled** state in Azure — it is disabled, expired, or otherwise inactive. Nothing in Prowler conflicts with it: resolve whatever Azure reports against the subscription, billing included, then run discovery again. |
|
||||
| `organization_conflict` | The subscription is already connected under a **different** Prowler organization. |
|
||||
| `organization_node_conflict` | The subscription is already grouped under a different Management Group in Prowler — for example, it moved in Azure after it was onboarded. |
|
||||
| `provider_type_conflict` | A provider with the same identifier exists in Prowler for another cloud provider. |
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/azure-blocked-subscription.png" alt="Blocked subscription row with the reason shown in a tooltip" />
|
||||
</Frame>
|
||||
|
||||
### Management Groups With Nothing to Select
|
||||
|
||||
A Management Group that holds no subscriptions, or whose subscriptions are all blocked, is shown disabled with the note *"No subscriptions available to select in this management group."* The Management Group still expands, so you can see the blocked subscriptions it holds and why they are blocked.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/azure-inert-group.png" alt="Disabled Management Group row noting that no subscriptions are available to select" />
|
||||
</Frame>
|
||||
|
||||
### Custom Aliases
|
||||
|
||||
Each subscription row carries an editable name, prefilled with the subscription's display name. The alias is used only inside Prowler — it does not rename anything in Azure. Management Group names are read-only: Prowler stores the display name from Azure.
|
||||
|
||||
### Subscriptions That Already Have Credentials
|
||||
|
||||
Applying your selection stores the organization credential on every selected subscription. When a selected subscription is already connected to Prowler with its own credential, that credential is **overwritten** — so the wizard asks first, naming the affected subscriptions:
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/azure-replace-credentials-apply.png" alt="Replace existing credentials modal listing the subscriptions whose credentials will be replaced" />
|
||||
</Frame>
|
||||
|
||||
Click **Replace and continue** to proceed, or **Cancel** to adjust your selection.
|
||||
|
||||
<Note>
|
||||
**Your existing data is safe.** A subscription already connected as an individual provider is **linked** to the organization, never duplicated: its historical scans and findings are preserved, and it does not count twice toward your subscription.
|
||||
</Note>
|
||||
|
||||
## Step 4: Test Connections
|
||||
|
||||
Click **Test Connections** to verify that Prowler can authenticate against each selected subscription. Prowler creates one provider per subscription — identified by its Azure subscription ID — and then tests every connection.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/azure-test-connections.png" alt="Connection testing in progress with status icons on each subscription" />
|
||||
</Frame>
|
||||
|
||||
Each subscription shows a real-time status indicator:
|
||||
|
||||
- **Spinner** — test in progress
|
||||
- **Green checkmark (✓)** — connection successful
|
||||
- **Red icon (✗)** — connection failed (hover to see the error)
|
||||
|
||||
If every subscription connects successfully, you advance to the next step automatically.
|
||||
|
||||
### When Some Tests Fail
|
||||
|
||||
An error banner appears: **"There was a problem connecting to some subscriptions. Hover each subscription to check the error."** You have two options:
|
||||
|
||||
**a) Fix and retry:**
|
||||
|
||||
1. Confirm the service principal holds **Reader** and **ProwlerRole** on the failing subscriptions, or on a Management Group above them.
|
||||
2. Confirm the Microsoft Graph permissions have been granted admin consent, as described in [Azure Authentication in Prowler](/user-guide/providers/azure/authentication#assigning-required-api-permissions).
|
||||
3. Click **Test Connections** again — only the **failed subscriptions are re-tested**. Subscriptions that already passed are not tested again.
|
||||
|
||||
**b) Skip and continue:**
|
||||
|
||||
Click **Skip Connection Validation** to proceed with the subscriptions that connected successfully. Failed subscriptions stay onboarded and visible on the Providers page, but they are not scanned. This option appears only when at least one subscription connected.
|
||||
|
||||
If **no subscription** connects, the banner instead reads *"No subscriptions connected successfully. Fix the connection errors and retry before launching scans."* and you cannot proceed. Fix the underlying problem — see [Troubleshooting](#troubleshooting) — and retry.
|
||||
|
||||
## Step 5: Launch Scans
|
||||
|
||||
The Organizations wizard uses the same schedule controls described in [Scan Scheduling](/user-guide/tutorials/prowler-scan-scheduling#schedule-options).
|
||||
|
||||
Click **Save**, **Save and launch scan**, or **Launch scan**, depending on the selected schedule option. A toast notification confirms whether the schedule was saved, scans were launched, or both, and links to the **Scans** page. Prowler then redirects to the **Providers** page. Scans launch only for subscriptions that passed connection testing.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/azure-launch-scan.png" alt="Launch Scan step showing the Subscriptions Connected confirmation and the scan schedule selector" />
|
||||
</Frame>
|
||||
|
||||
After launching:
|
||||
|
||||
- Scans appear on the **Scans** page as they start and complete.
|
||||
- Results populate the **Overview** and **Findings** pages.
|
||||
- On the **Providers** page, your subscriptions are grouped under the organization and, when they live in a Management Group, under that Management Group.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/azure-providers-grouping.png" alt="Providers page showing subscriptions grouped under Azure Management Groups and the organization" />
|
||||
</Frame>
|
||||
|
||||
## Manage Your Organization After Onboarding
|
||||
|
||||
Open the row actions menu on the organization row on the **Providers** page.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/azure-organization-row-actions.png" alt="Row actions menu on an Azure organization row" />
|
||||
</Frame>
|
||||
|
||||
| Action | What it does |
|
||||
|--------|--------------|
|
||||
| **Edit Organization Name** | Renames the organization in Prowler. Leave it blank to fall back to the tenant ID. |
|
||||
| **Update Credentials** | Reopens the Authentication Details step to store a new service principal credential. |
|
||||
| **Edit Scan Schedule** | Applies one schedule to every connected subscription in the organization. |
|
||||
| **Test Connections (N)** | Re-tests every subscription in the organization. |
|
||||
| **Delete Organization** | Deletes the organization and cascades to its providers. |
|
||||
|
||||
### Onboard Subscriptions Created Later
|
||||
|
||||
Subscriptions added to the tenant after onboarding are not picked up automatically. Run the wizard again with the same tenant ID: discovery returns the current hierarchy, already-connected subscriptions come back preselected, and the new ones are ready to select.
|
||||
|
||||
### Update Organization Credentials
|
||||
|
||||
Choosing **Update Credentials** re-enters the Authentication Details step. Because the organization already holds a credential, Prowler warns before overwriting it and names how many providers re-authenticate with the new one:
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/azure-replace-credentials-setup.png" alt="Replace existing credentials modal showing how many providers re-authenticate" />
|
||||
</Frame>
|
||||
|
||||
Storing a new credential runs a fresh discovery, so any discovery already in progress is discarded — discovery authenticates with the credential it started from.
|
||||
|
||||
### Delete an Organization or Management Group
|
||||
|
||||
Deleting an organization or a Management Group **cascades to every provider grouped under it**, along with their scans and findings. Both dialogs state how many providers are affected before you confirm.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/azure/azure-delete-organization.png" alt="Delete organization dialog showing how many providers are deleted with it" />
|
||||
</Frame>
|
||||
|
||||
Deletion runs in the background. Prowler confirms with a **"Deletion started"** notification; if any part of it fails, the affected rows reappear on a later refresh.
|
||||
|
||||
<Danger>
|
||||
Deleting an organization **permanently deletes every subscription provider grouped under it**, including their historical scans and findings. This action cannot be undone.
|
||||
</Danger>
|
||||
|
||||
### When Grouping Is Unavailable
|
||||
|
||||
If Prowler cannot read the hierarchy while loading the Providers page, a notice reads *"Organization grouping is incomplete. Some providers may appear ungrouped."* Your providers are still listed, just flat. Reload the page to try again.
|
||||
|
||||
## Billing Impact
|
||||
|
||||
Each Azure subscription you connect through the Organizations wizard counts as one **provider** in your Prowler Cloud subscription.
|
||||
|
||||
- **Already-connected subscriptions**: linking an existing provider to the organization does **not** add billing. The existing provider is reused.
|
||||
- **Large tenants**: connecting a 500-subscription tenant results in up to 500 providers on your subscription. Review your plan limits before proceeding.
|
||||
- **Deleted providers**: a subscription you later remove no longer counts toward your subscription.
|
||||
|
||||
For pricing details, see [Prowler Cloud Pricing](https://prowler.com/pricing).
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Those Service Principal Credentials Were Rejected
|
||||
|
||||
*"Those service principal credentials were rejected. Check the client ID and client secret, then try again."*
|
||||
|
||||
Azure refused the sign-in. Confirm the client ID belongs to the service principal you granted access to, and that the client secret is current — secrets expire, and a rotated one invalidates the old value immediately.
|
||||
|
||||
```bash
|
||||
az ad app credential list --id <client-id> --query "[].{name:displayName, expires:endDateTime}" -o table
|
||||
```
|
||||
|
||||
### The Service Principal Cannot Read the Complete Management Group Hierarchy
|
||||
|
||||
*"The service principal cannot read the complete Management Group hierarchy. Grant it the Reader role at the Management Group level, then try again."*
|
||||
|
||||
The credential authenticated but lacks read access to part of the hierarchy. Grant **Reader** at the tenant root Management Group, as described in [Grant Read Access at the Tenant Root Management Group](#grant-read-access-at-the-tenant-root-management-group), and confirm the assignment landed on the Management Group and not on a single subscription:
|
||||
|
||||
```bash
|
||||
az role assignment list \
|
||||
--assignee <client-id> \
|
||||
--scope "/providers/Microsoft.Management/managementGroups/<tenant-id>" \
|
||||
--query "[].{role:roleDefinitionName, scope:scope}" -o table
|
||||
```
|
||||
|
||||
### Azure Returned an Incomplete Management Group Hierarchy
|
||||
|
||||
*"Azure returned an incomplete Management Group hierarchy. This usually clears on a retry; if it does not, check that the service principal can read every Management Group in the tenant."*
|
||||
|
||||
Azure answered, but the hierarchy it returned was missing Management Groups that Prowler expected to find. This is usually a transient inconsistency in Azure Resource Manager, so click **Retry discovery** first.
|
||||
|
||||
If it repeats, treat it as a permissions problem. Unlike the error above, Azure did not refuse the read — a partial grant can surface as missing data rather than as an outright denial. Confirm that the **Reader** assignment sits on the tenant root Management Group and not on individual Management Groups beneath it, using the command in [The Service Principal Cannot Read the Complete Management Group Hierarchy](#the-service-principal-cannot-read-the-complete-management-group-hierarchy).
|
||||
|
||||
### The Tenant Root Management Group Could Not Be Found
|
||||
|
||||
*"The tenant root Management Group could not be found. Check the tenant ID, and that the service principal has been granted access at the tenant root."*
|
||||
|
||||
Prowler derives the tenant root Management Group from the tenant ID you entered, and could not read it. Either the tenant ID is wrong, or the service principal cannot see the root. Confirm the tenant ID and list what the credential can reach:
|
||||
|
||||
```bash
|
||||
az account management-group list --query "[].{name:name, displayName:displayName}" -o table
|
||||
```
|
||||
|
||||
### Those Credentials Belong to a Different Microsoft Entra Tenant
|
||||
|
||||
*"Those credentials belong to a different Microsoft Entra tenant. Use a service principal from the tenant you entered."*
|
||||
|
||||
The service principal is registered in another tenant. Register the application in the tenant being onboarded, or correct the tenant ID in the first step of the wizard.
|
||||
|
||||
### Azure Did Not Respond
|
||||
|
||||
*"Azure did not respond while reading the Management Group hierarchy. Nothing is wrong with your credentials — try again in a few minutes."*
|
||||
|
||||
A transient Azure Resource Manager error. Click **Retry discovery**.
|
||||
|
||||
### Azure Rate Limited the Hierarchy Read
|
||||
|
||||
*"Azure rate limited the hierarchy read. Nothing is wrong with your credentials — try again in a few minutes."*
|
||||
|
||||
Azure Resource Manager throttled Prowler's requests while it walked the hierarchy. Large tenants and back-to-back discoveries are the common causes. Wait a few minutes, then click **Retry discovery**.
|
||||
|
||||
### The Hierarchy Is Too Deep
|
||||
|
||||
*"This tenant's Management Group hierarchy is deeper than Prowler can read. Contact support so we can help you onboard it."*
|
||||
|
||||
The tenant nests Management Groups deeper than Prowler reads in one pass. Contact [Prowler Support](mailto:support@prowler.com).
|
||||
|
||||
### Authentication Failed
|
||||
|
||||
*"Authentication failed. Please verify the service principal permissions or credentials, then try again."*
|
||||
|
||||
The generic authentication failure, shown when Azure reports no more specific reason. Confirm that the service principal still exists, that its client secret has not expired, and that the Reader role assignment at the tenant root Management Group is still in place.
|
||||
|
||||
### Azure Rejected the Hierarchy Read
|
||||
|
||||
*"Azure rejected the hierarchy read. Try again, and contact support if it keeps failing."*
|
||||
|
||||
Azure refused the request for a reason Prowler could not narrow down to credentials, permissions, or throttling. Click **Retry discovery**, and contact [Prowler Support](mailto:support@prowler.com) if the failure repeats. This message names Azure as the source of the failure; the one below is shown when the failure cannot be attributed to Azure at all.
|
||||
|
||||
### Discovery Could Not Be Completed
|
||||
|
||||
*"Discovery could not be completed. Try again, and contact support if it keeps failing."*
|
||||
|
||||
Discovery failed for a reason Prowler cannot attribute to the credential or to Azure. Click **Retry discovery**, and contact [Prowler Support](mailto:support@prowler.com) if the failure repeats.
|
||||
|
||||
### Discovery Never Finishes
|
||||
|
||||
The wizard stops waiting after 3 minutes, but the discovery keeps running in Azure. Click **Keep waiting** to resume the same discovery rather than **Retry**, which starts over and re-reads the whole hierarchy.
|
||||
|
||||
## Key Concepts
|
||||
|
||||
### How Subscriptions Map to Prowler Providers
|
||||
|
||||
Each selected subscription becomes one Prowler provider:
|
||||
|
||||
| Prowler field | Comes from |
|
||||
|---------------|------------|
|
||||
| Provider identifier | The Azure subscription ID (for example, `22222222-2222-4222-8222-222222222222`). |
|
||||
| Alias | The name you typed in the tree, or the subscription's display name. |
|
||||
| Credential | A copy of the organization credential. |
|
||||
|
||||
Management Groups that hold selected subscriptions become grouping rows on the Providers page. You select subscriptions only — Prowler derives the Management Group ancestors itself.
|
||||
|
||||
### Organization Credential vs Subscription Credential
|
||||
|
||||
One credential, stored twice: on the organization, where discovery reads it, and on each subscription provider, where scans read it. That is why replacing the organization credential re-authenticates every subscription under it, and why the wizard asks before overwriting a subscription's own credential.
|
||||
|
||||
## What's Next
|
||||
|
||||
<Columns cols={2}>
|
||||
<Card title="Prowler Cloud" icon="cloud" href="/user-guide/tutorials/prowler-app">
|
||||
Full guide to using Prowler Cloud features.
|
||||
</Card>
|
||||
<Card title="Azure Subscriptions (CLI)" icon="terminal" href="/user-guide/providers/azure/subscriptions">
|
||||
CLI-based scanning of specific Azure subscriptions.
|
||||
</Card>
|
||||
<Card title="Azure Authentication" icon="key" href="/user-guide/providers/azure/authentication">
|
||||
Credential types and the permissions Prowler needs in Azure.
|
||||
</Card>
|
||||
<Card title="Bulk Provider Provisioning" icon="upload" href="/user-guide/tutorials/bulk-provider-provisioning">
|
||||
Script-based bulk provisioning for advanced automation.
|
||||
</Card>
|
||||
</Columns>
|
||||
|
||||
@@ -120,7 +120,7 @@ In Prowler Local Server the organization option is marked **Cloud** and opens an
|
||||
### Enter Organization Details
|
||||
|
||||
- **Organization ID**: the numeric ID of your Google Cloud organization (for example, `123456789012`). Non-numeric values are rejected before submission.
|
||||
- **Name** (optional): a display name for the organization in Prowler. If left blank, Prowler uses the name stored in Google Cloud.
|
||||
- **Name** (optional): a display name for the organization in Prowler. If left blank, Prowler uses the organization ID.
|
||||
|
||||
<Frame>
|
||||
<img src="/images/organizations/gcp/gcp-organization-details-form.png" alt="Organization Details form with the Google Cloud organization ID and Name fields" />
|
||||
@@ -307,7 +307,7 @@ Open the row actions menu on the organization row on the **Providers** page.
|
||||
|
||||
| Action | What it does |
|
||||
|--------|--------------|
|
||||
| **Edit Organization Name** | Renames the organization in Prowler. Leave it blank to fall back to the name stored in Google Cloud. |
|
||||
| **Edit Organization Name** | Renames the organization in Prowler. Leave it blank to fall back to the organization ID. |
|
||||
| **Update Credentials** | Reopens the Authentication Details step to store a new credential. |
|
||||
| **Edit Scan Schedule** | Applies one schedule to every connected project in the organization. |
|
||||
| **Test Connections (N)** | Re-tests every project in the organization. |
|
||||
|
||||
@@ -26,6 +26,8 @@ The Prowler MCP Server provides AI agents access to the Prowler ecosystem throug
|
||||
## CRITICAL RULES
|
||||
|
||||
### Tool Implementation
|
||||
- ALWAYS: Build sub-servers with `ProwlerMCP`, never `FastMCP` directly. It is what
|
||||
applies the error contract to every tool, whichever way it is registered
|
||||
- ALWAYS: Extend `BaseTool` ABC for Prowler tools (auto-registration)
|
||||
- ALWAYS: Use `@mcp.tool()` decorator for Hub/Docs tools
|
||||
- NEVER: Manually register BaseTool subclasses
|
||||
@@ -42,6 +44,37 @@ The Prowler MCP Server provides AI agents access to the Prowler ecosystem throug
|
||||
- ALWAYS: Use `build_filter_params()` for query parameters
|
||||
- NEVER: Create new httpx clients in tools
|
||||
|
||||
### Errors
|
||||
One rule: **`ToolError` is a message you wrote for the caller. Any other exception is
|
||||
a bug or an upstream failure**, and `render_tool_error` describes it.
|
||||
|
||||
- ALWAYS: `raise ToolError(...)` for anything the caller can act on — a rejected
|
||||
argument, a lookup that found nothing, a workflow step they must do first. Its text
|
||||
reaches the client verbatim, past `mask_error_details`
|
||||
- NEVER: `raise ValueError(...)` in a tool. It is reported as a bug in this server,
|
||||
which is correct for a model factory rejecting an API payload and wrong for a
|
||||
refusal
|
||||
- ALWAYS: Let an upstream failure propagate untouched. `ProwlerMCP.tool` wraps every
|
||||
registration, so it becomes a `ToolError` describing the call, the status and what
|
||||
the API said. There is nothing to remember to apply
|
||||
- NEVER: `return {"error": ...}` or `{"success": False}`. A returned payload is
|
||||
`isError: false`, so the client is told the call succeeded
|
||||
- NEVER: Raise a plain exception *after* a write has been accepted. `ToolError` is the
|
||||
only kind whose message reaches the client exactly as written
|
||||
- ALWAYS: `render_tool_error(e)` when you surface an exception yourself, so a failure
|
||||
is never described two different ways. Use `warn=False` when embedding it in a
|
||||
result that already reports the outcome
|
||||
- ALWAYS: Return a structured result, not an error, when a write may have partially
|
||||
landed (`status="unknown"`, `deleted="unknown"`, `safe_to_retry=False`). An agent
|
||||
reads `isError: true` as "nothing happened, safe to retry"
|
||||
- ALWAYS: Return a structured result for an outcome that *is* the tool's job to
|
||||
report: `connected: false`, an empty list, an idempotent no-op
|
||||
- NEVER: Wrap a whole tool body in `except Exception`. It reports bugs in this
|
||||
server as API failures, and the wrapper already handles the rest
|
||||
|
||||
See `prowler_mcp_server/lib/errors.py` and
|
||||
`docs/developer-guide/mcp-server.mdx` for the message format.
|
||||
|
||||
---
|
||||
|
||||
## ARCHITECTURE
|
||||
@@ -72,6 +105,9 @@ Python 3.12+ | FastMCP 3.4.4 | httpx (async) | Pydantic | uv | pytest
|
||||
```text
|
||||
mcp_server/prowler_mcp_server/
|
||||
├── server.py # Main orchestration
|
||||
├── lib/
|
||||
│ ├── server.py # ProwlerMCP: base class of every sub-server
|
||||
│ └── errors.py # Exception types + render_tool_error
|
||||
├── prowler_hub/server.py # Hub tools (no auth)
|
||||
├── prowler_app/
|
||||
│ ├── server.py
|
||||
@@ -113,7 +149,8 @@ make test-mcp # Run the MCP test suite exactly as CI does
|
||||
- [ ] Models use `MinimalSerializerMixin`
|
||||
- [ ] API responses transformed to simplified models
|
||||
- [ ] No hardcoded secrets
|
||||
- [ ] Error handling returns structured responses
|
||||
- [ ] Failures raise (never `return {"error": ...}`); outcomes that may have changed
|
||||
something return a structured result
|
||||
- [ ] Parameter descriptions use Pydantic `Field()`
|
||||
- [ ] Tests added under `mcp_server/tests/`, mirroring the source path below the
|
||||
package root (`prowler_mcp_server/prowler_app/tools/` -> `tests/prowler_app/tools/`),
|
||||
|
||||
@@ -0,0 +1,347 @@
|
||||
"""One way to fail: every tool failure reaches the client as a `ToolError`.
|
||||
|
||||
MCP draws a line this server used to blur. A tool that *returns* `{"error": ...}`
|
||||
produces a successful result (`isError: false`) whose failure is only discoverable by
|
||||
guessing which key to look at; a tool that *raises* produces `isError: true`, which
|
||||
every client and model already understands as "this call did not work".
|
||||
|
||||
`ToolError` is a `FastMCPError`, and `FastMCP._call_tool` re-raises those untouched
|
||||
(fastmcp/server/server.py:1241). So a message built here is what the client reads,
|
||||
verbatim, past every mount and past `mask_error_details`. That is what lets the servers
|
||||
mask by default while still telling the caller everything relevant.
|
||||
|
||||
`render_tool_error` is the single place an exception becomes that text, and
|
||||
`tool_errors` is what makes sure no tool can escape it.
|
||||
|
||||
The exception types live here rather than next to the API client because the hub and the
|
||||
documentation sub-servers must be able to raise and render them without taking a
|
||||
dependency on `prowler_app`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import functools
|
||||
import inspect
|
||||
import json
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from fastmcp.exceptions import ToolError
|
||||
|
||||
from prowler_mcp_server.lib.logger import logger
|
||||
|
||||
# Upstream bodies are not ours and may be large or HTML; enough to diagnose, not enough
|
||||
# to flood the model's context.
|
||||
_MAX_UPSTREAM_BODY = 500
|
||||
|
||||
# The one thing a status code does not say. Appended only when a request that could have
|
||||
# changed something did not come back with a verdict, because an agent reads a failure as
|
||||
# "nothing happened" and will happily send the write again.
|
||||
_UNKNOWN_OUTCOME = (
|
||||
" It may have been carried out anyway, so check the current state before retrying."
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ApiErrorDetail:
|
||||
"""One entry of a JSON:API `errors` array.
|
||||
|
||||
The API answers a rejected write with one error *per invalid field*, each naming the
|
||||
field in `source`. Keeping the whole shape is what turns "the request was invalid"
|
||||
into "these two fields were invalid, and here is which".
|
||||
|
||||
Field names are JSON:API's own. `source.pointer` and `source.parameter` are not two
|
||||
spellings of one thing: a pointer is a JSON Pointer into the request *document*
|
||||
(`/data/attributes/provider_id`, as `alerts/errors.py` and `tasks/beat.py` send),
|
||||
while a parameter is a *query parameter* name (`page[size]`, `lookback_days`, as
|
||||
`api/v1/views.py` sends). The spec has a third, `source.header`; the Prowler API
|
||||
never emits one, so there is nothing here to read it into.
|
||||
"""
|
||||
|
||||
detail: str | None = None
|
||||
title: str | None = None
|
||||
pointer: str | None = None
|
||||
"""JSON:API `source.pointer`: a JSON Pointer into the request document."""
|
||||
parameter: str | None = None
|
||||
"""JSON:API `source.parameter`: the query parameter that caused the error."""
|
||||
|
||||
@classmethod
|
||||
def from_jsonapi(cls, error: dict[str, Any]) -> ApiErrorDetail:
|
||||
"""Build from a single member of a JSON:API `errors` array.
|
||||
|
||||
`source` is optional and most errors omit it, so it supplies the location only,
|
||||
never whether there is a detail worth reporting.
|
||||
"""
|
||||
source = error.get("source", {})
|
||||
return cls(
|
||||
detail=error.get("detail"),
|
||||
title=error.get("title"),
|
||||
pointer=source.get("pointer"),
|
||||
parameter=source.get("parameter"),
|
||||
)
|
||||
|
||||
def render(self) -> str:
|
||||
"""The error text, and where the API said it is.
|
||||
|
||||
A pointer is left as-is because a leading `/` already reads as a path into the
|
||||
body. A parameter is labelled, since `(page[size])` on its own would read like
|
||||
one.
|
||||
"""
|
||||
text = self.detail or self.title or ""
|
||||
if not text:
|
||||
return ""
|
||||
if self.pointer:
|
||||
return f"{text} ({self.pointer})"
|
||||
if self.parameter:
|
||||
return f"{text} (parameter {self.parameter})"
|
||||
return text
|
||||
|
||||
|
||||
def parse_jsonapi_errors(payload: Any) -> tuple[ApiErrorDetail, ...]:
|
||||
"""Extract every error from a JSON:API error document.
|
||||
|
||||
Tolerant on purpose: this runs while handling a failure, and a body that is not the
|
||||
document it should be must not turn a useful API error into a parsing traceback.
|
||||
"""
|
||||
if not isinstance(payload, dict):
|
||||
return ()
|
||||
errors = payload.get("errors")
|
||||
if not isinstance(errors, list):
|
||||
return ()
|
||||
return tuple(
|
||||
ApiErrorDetail.from_jsonapi(error)
|
||||
for error in errors
|
||||
if isinstance(error, dict)
|
||||
)
|
||||
|
||||
|
||||
class ProwlerAPIError(Exception):
|
||||
"""An error response returned by the Prowler API.
|
||||
|
||||
Raised only when the API answered with an error status, which tells a caller
|
||||
something no plain exception can: the request reached Prowler and was
|
||||
rejected, so it changed nothing. A timeout or a dropped connection stays a
|
||||
bare exception because the request may well have been processed.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
message: str,
|
||||
status_code: int,
|
||||
*,
|
||||
method: str | None = None,
|
||||
path: str | None = None,
|
||||
errors: tuple[ApiErrorDetail, ...] = (),
|
||||
) -> None:
|
||||
super().__init__(message)
|
||||
self.status_code: int = status_code
|
||||
# Stored as plain strings so this module stays independent of the API client's
|
||||
# HTTPMethod enum; StrEnum members compare equal to their value either way.
|
||||
self.method: str | None = str(method) if method is not None else None
|
||||
self.path: str | None = path
|
||||
self.errors: tuple[ApiErrorDetail, ...] = tuple(errors)
|
||||
|
||||
@property
|
||||
def rejected(self) -> bool:
|
||||
"""The request reached Prowler and was refused, so it changed nothing."""
|
||||
return 400 <= self.status_code < 500
|
||||
|
||||
|
||||
class ProwlerTaskError(Exception):
|
||||
"""A background task this server was waiting on did not complete.
|
||||
|
||||
Separate from `ProwlerAPIError` because the API already accepted the work: the
|
||||
task exists and may still be running, so the outcome is unknown rather than refused.
|
||||
"""
|
||||
|
||||
def __init__(self, message: str, *, task_id: str, state: str) -> None:
|
||||
super().__init__(message)
|
||||
self.task_id: str = task_id
|
||||
self.state: str = state
|
||||
"""One of `timeout`, `failed` or `cancelled`."""
|
||||
|
||||
|
||||
class ProwlerAuthError(ValueError):
|
||||
"""The credentials are missing, malformed or expired.
|
||||
|
||||
Subclasses `ValueError` so that the handlers which already treat an
|
||||
authentication failure as a refusal-before-send keep working. It is matched by name
|
||||
in `render_tool_error` rather than by that base class, so it is described as the
|
||||
credential problem it is instead of falling through to the bug branch.
|
||||
"""
|
||||
|
||||
|
||||
class ProwlerHubError(Exception):
|
||||
"""The Prowler Hub answered with an error status.
|
||||
|
||||
The Hub is a separate public service with its own client, so its failures cannot be
|
||||
`ProwlerAPIError`. Everything the Hub exposes is a read.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
message: str,
|
||||
*,
|
||||
status_code: int,
|
||||
path: str,
|
||||
body: str | None = None,
|
||||
) -> None:
|
||||
super().__init__(message)
|
||||
self.status_code: int = status_code
|
||||
self.path: str = path
|
||||
self.body: str | None = body
|
||||
|
||||
|
||||
def _upstream_detail(body: str | None) -> str:
|
||||
"""A readable line from an error body that is not JSON:API.
|
||||
|
||||
The Prowler API answers with a JSON:API document, which is parsed into
|
||||
`ApiErrorDetail`. Every other host this server talks to has its own shape: the Hub
|
||||
answers `{"error": "Not found"}`, GitHub answers plain text, a proxy in between may
|
||||
answer HTML. Relaying any of those verbatim puts braces and markup in front of the
|
||||
model, so the message is pulled out when there is one and truncated when there is
|
||||
not.
|
||||
"""
|
||||
if not body:
|
||||
return ""
|
||||
text = body.strip()
|
||||
try:
|
||||
parsed = json.loads(text)
|
||||
except ValueError:
|
||||
parsed = None
|
||||
if isinstance(parsed, dict):
|
||||
for key in ("error", "message", "detail"):
|
||||
value = parsed.get(key)
|
||||
if isinstance(value, str) and value:
|
||||
return value
|
||||
return f"{text[:_MAX_UPSTREAM_BODY]}..." if len(text) > _MAX_UPSTREAM_BODY else text
|
||||
|
||||
|
||||
def render_tool_error(error: Exception, *, warn: bool = True) -> str:
|
||||
"""Describe an exception in the plainest sentence that keeps every useful detail.
|
||||
|
||||
Callers that surface a failure from anywhere other than a raised exception -- a
|
||||
message they write themselves, a field of a structured result -- go through here
|
||||
too, which is what keeps one failure from being described two different ways.
|
||||
|
||||
Pass `warn=False` when the caller already states the outcome, which a structured
|
||||
result reporting `status="unknown"` does by definition. Otherwise the generic
|
||||
warning lands next to a more specific one saying the same thing.
|
||||
|
||||
Ordered most specific first: `ProwlerAuthError` is a `ValueError` and
|
||||
`httpx.TimeoutException` is a `RequestError`, so the general branches come last.
|
||||
"""
|
||||
unknown = _UNKNOWN_OUTCOME if warn else ""
|
||||
|
||||
if isinstance(error, ProwlerAPIError):
|
||||
operation = (
|
||||
" ".join(p for p in (error.method, error.path) if p) or "The request"
|
||||
)
|
||||
details = "; ".join(text for text in (d.render() for d in error.errors) if text)
|
||||
message = f"{operation} failed with HTTP {error.status_code}."
|
||||
if details:
|
||||
message = f"{message} {details}"
|
||||
# A 4xx is a refusal, so it changed nothing and needs no warning.
|
||||
if error.rejected or error.method == "GET":
|
||||
return message
|
||||
return message + unknown
|
||||
|
||||
if isinstance(error, ProwlerTaskError):
|
||||
# The API accepted the work before the wait failed, so the outcome is open
|
||||
# whichever way the task ended.
|
||||
return f"{error}{unknown}"
|
||||
|
||||
if isinstance(error, ProwlerAuthError):
|
||||
return f"Prowler authentication failed: {error}"
|
||||
|
||||
if isinstance(error, ProwlerHubError):
|
||||
# Same shape as the API branch, with the service named because the Hub can be
|
||||
# down while the API is fine. Everything the Hub exposes is a GET.
|
||||
message = f"Prowler Hub GET {error.path} failed with HTTP {error.status_code}."
|
||||
detail = _upstream_detail(error.body)
|
||||
return f"{message} {detail}" if detail else message
|
||||
|
||||
if isinstance(error, httpx.HTTPStatusError):
|
||||
# An upstream that is not the Prowler API, such as the external-URL fetch.
|
||||
request = error.request
|
||||
message = (
|
||||
f"{request.method} {request.url} failed with HTTP "
|
||||
f"{error.response.status_code}."
|
||||
)
|
||||
detail = _upstream_detail(error.response.text)
|
||||
return f"{message} {detail}" if detail else message
|
||||
|
||||
if isinstance(error, httpx.RequestError):
|
||||
# No answer at all: a timeout, a dropped connection, a DNS failure.
|
||||
try:
|
||||
operation = f"{error.request.method} {error.request.url}"
|
||||
method = error.request.method
|
||||
except RuntimeError:
|
||||
# httpx only attaches the request once it has one, and reading it before
|
||||
# then raises. Never let that hide the failure being reported.
|
||||
operation, method = "The request", None
|
||||
suffix = "" if method == "GET" else unknown
|
||||
return f"{operation} got no answer ({type(error).__name__}: {error}).{suffix}"
|
||||
|
||||
# No `ValueError` branch, deliberately. A message written for the caller is raised
|
||||
# as a `ToolError`, which never reaches here. What is left -- a model factory
|
||||
# rejecting an API payload, a pydantic `ValidationError`, an `int()` on something
|
||||
# that is not a number -- is this server or the API breaking its own contract, and
|
||||
# saying so is the only useful thing to tell a caller who cannot fix it.
|
||||
return (
|
||||
f"The Prowler MCP Server hit an unexpected {type(error).__name__}: {error}. "
|
||||
"This is a bug in the server, not something you can fix by changing the "
|
||||
"arguments."
|
||||
)
|
||||
|
||||
|
||||
def tool_errors(fn: Callable[..., Any]) -> Callable[..., Any]:
|
||||
"""Wrap a tool so that every failure leaves it as a `ToolError`.
|
||||
|
||||
Applied by `ProwlerMCP.tool()` rather than by hand, so no registration can miss
|
||||
it. It wraps the callable handed to `mcp.tool()`, not the class attribute, so only
|
||||
the MCP boundary is normalised: a tool calling another tool internally still sees
|
||||
the real, typed exception and can branch on it.
|
||||
|
||||
Two constraints worth knowing before changing this:
|
||||
|
||||
- Never register the result with `exclude_args=`. That path
|
||||
(fastmcp/utilities/types.py) rebuilds the function from `__code__`, which on a
|
||||
wrapper is the wrapper's own. Nothing in this server passes it today.
|
||||
- `inspect.iscoroutinefunction`, not the `asyncio` one, which is deprecated from
|
||||
Python 3.14 and would be an error under this project's warning filters.
|
||||
"""
|
||||
name = getattr(fn, "__qualname__", repr(fn))
|
||||
|
||||
def mark(wrapper: Callable[..., Any]) -> Callable[..., Any]:
|
||||
"""Flag the wrapper so a test can prove every registered tool went through it."""
|
||||
wrapper.__prowler_tool_errors__ = True # ty: ignore[unresolved-attribute]
|
||||
return wrapper
|
||||
|
||||
if inspect.iscoroutinefunction(fn):
|
||||
|
||||
@functools.wraps(fn)
|
||||
async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
|
||||
try:
|
||||
return await fn(*args, **kwargs)
|
||||
except ToolError:
|
||||
raise
|
||||
except Exception as error:
|
||||
logger.exception(f"Tool {name} failed: {error}")
|
||||
raise ToolError(render_tool_error(error)) from error
|
||||
|
||||
return mark(async_wrapper)
|
||||
|
||||
@functools.wraps(fn)
|
||||
def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
|
||||
try:
|
||||
return fn(*args, **kwargs)
|
||||
except ToolError:
|
||||
raise
|
||||
except Exception as error:
|
||||
logger.exception(f"Tool {name} failed: {error}")
|
||||
raise ToolError(render_tool_error(error)) from error
|
||||
|
||||
return mark(sync_wrapper)
|
||||
@@ -0,0 +1,32 @@
|
||||
"""The FastMCP subclass every Prowler sub-server is built from."""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from fastmcp import FastMCP
|
||||
|
||||
from prowler_mcp_server.lib.errors import tool_errors
|
||||
|
||||
|
||||
class ProwlerMCP(FastMCP):
|
||||
"""A FastMCP server whose tools all report failures the same way.
|
||||
|
||||
`FastMCP.tool()` is the single funnel every registration goes through, the
|
||||
`@server.tool()` and bare `@server.tool` decorator forms, and the direct
|
||||
`mcp.tool(fn)` call `BaseTool` uses to auto-register, so applying
|
||||
`tool_errors` here covers all of them at once.
|
||||
"""
|
||||
|
||||
def tool(self, name_or_fn: Any = None, **kwargs: Any) -> Any:
|
||||
"""Register a tool, wrapped so its failures reach the client as `ToolError`."""
|
||||
if callable(name_or_fn):
|
||||
# Direct call: mcp.tool(fn), or the bare @mcp.tool decorator.
|
||||
return super().tool(tool_errors(name_or_fn), **kwargs)
|
||||
|
||||
# Parameterised decorator: @mcp.tool() or @mcp.tool(name="..."). FastMCP hands
|
||||
# back the decorator that does the registering, so the wrap goes in front of it.
|
||||
register = super().tool(name_or_fn, **kwargs)
|
||||
|
||||
def decorator(fn: Any) -> Any:
|
||||
return register(tool_errors(fn))
|
||||
|
||||
return decorator
|
||||
@@ -1,9 +1,8 @@
|
||||
from fastmcp import FastMCP
|
||||
|
||||
from prowler_mcp_server.lib.server import ProwlerMCP
|
||||
from prowler_mcp_server.prowler_app.utils.tool_loader import load_all_tools
|
||||
|
||||
# Initialize MCP server
|
||||
app_mcp_server = FastMCP("prowler-app")
|
||||
app_mcp_server = ProwlerMCP("prowler-app", mask_error_details=True)
|
||||
|
||||
# Auto-discover and load all tools from the tools package
|
||||
load_all_tools(app_mcp_server)
|
||||
|
||||
@@ -72,6 +72,9 @@ class BaseTool(ABC):
|
||||
async methods (not starting with '_') as tools. Subclasses do not need
|
||||
to override this method.
|
||||
|
||||
Failures need no handling here: `ProwlerMCP.tool` wraps whatever it is
|
||||
given, so every tool reports them the same way.
|
||||
|
||||
Args:
|
||||
mcp: The FastMCP instance to register tools with
|
||||
"""
|
||||
|
||||
@@ -1,14 +1,14 @@
|
||||
from typing import Any
|
||||
|
||||
from fastmcp import FastMCP
|
||||
from pydantic import Field
|
||||
|
||||
from prowler_mcp_server.lib.server import ProwlerMCP
|
||||
from prowler_mcp_server.prowler_documentation.search_engine import (
|
||||
ProwlerDocsSearchEngine,
|
||||
)
|
||||
|
||||
# Initialize FastMCP server
|
||||
docs_mcp_server = FastMCP("prowler-docs")
|
||||
# Initialize MCP server
|
||||
docs_mcp_server = ProwlerMCP("prowler-docs", mask_error_details=True)
|
||||
prowler_docs_search_engine = ProwlerDocsSearchEngine()
|
||||
|
||||
|
||||
|
||||
@@ -5,13 +5,13 @@ Provides access to Prowler Hub API for security checks and compliance frameworks
|
||||
"""
|
||||
|
||||
import httpx
|
||||
from fastmcp import FastMCP
|
||||
from pydantic import Field
|
||||
|
||||
from prowler_mcp_server import __version__
|
||||
from prowler_mcp_server.lib.server import ProwlerMCP
|
||||
|
||||
# Initialize FastMCP for Prowler Hub
|
||||
hub_mcp_server = FastMCP("prowler-hub")
|
||||
# Initialize MCP server for Prowler Hub
|
||||
hub_mcp_server = ProwlerMCP("prowler-hub", mask_error_details=True)
|
||||
|
||||
# API base URL
|
||||
BASE_URL = "https://hub.prowler.com/api"
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
from fastmcp import FastMCP
|
||||
from starlette.responses import JSONResponse
|
||||
|
||||
from prowler_mcp_server import __version__
|
||||
from prowler_mcp_server.lib.logger import logger
|
||||
from prowler_mcp_server.lib.server import ProwlerMCP
|
||||
|
||||
prowler_mcp_server = FastMCP("prowler-mcp-server")
|
||||
prowler_mcp_server = ProwlerMCP("prowler-mcp-server", mask_error_details=True)
|
||||
|
||||
|
||||
def setup_main_server():
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
"""Tests for the sentence a failure is described with.
|
||||
|
||||
These assert on the *text* a model reads, because that text is the whole contract: a
|
||||
`ToolError` carries nothing else. What matters is that the API's own words survive
|
||||
intact, and that a write whose outcome nobody can report says so.
|
||||
|
||||
How that sentence reaches a client -- and that no tool can escape it -- is
|
||||
`test_server.py`.
|
||||
"""
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from prowler_mcp_server.lib.errors import (
|
||||
ApiErrorDetail,
|
||||
ProwlerAPIError,
|
||||
ProwlerAuthError,
|
||||
ProwlerHubError,
|
||||
ProwlerTaskError,
|
||||
parse_jsonapi_errors,
|
||||
render_tool_error,
|
||||
)
|
||||
|
||||
MAY_HAVE_LANDED = "It may have been carried out anyway"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- parsing
|
||||
|
||||
|
||||
def test_every_error_of_the_document_is_preserved():
|
||||
"""A rejected write names one error per invalid field; all of them matter."""
|
||||
errors = parse_jsonapi_errors(
|
||||
{
|
||||
"errors": [
|
||||
{"status": "400", "detail": "This field may not be blank."},
|
||||
{"status": "400", "detail": "Enter a valid URL."},
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
assert [error.detail for error in errors] == [
|
||||
"This field may not be blank.",
|
||||
"Enter a valid URL.",
|
||||
]
|
||||
|
||||
|
||||
def test_a_query_parameter_error_is_not_dressed_up_as_a_body_path():
|
||||
"""`source.parameter` and `source.pointer` are different places, per JSON:API.
|
||||
|
||||
The API sends a parameter for a bad query string (`api/v1/views.py` answers
|
||||
`page[size]` and `lookback_days` that way) and a pointer for a bad body field.
|
||||
Rendering a parameter bare would read as though `page[size]` were a path into the
|
||||
document, which is somewhere the caller never put it.
|
||||
"""
|
||||
(error,) = parse_jsonapi_errors(
|
||||
{
|
||||
"errors": [
|
||||
{
|
||||
"detail": "invalid parameter 'page[size]'",
|
||||
"source": {"parameter": "page[size]"},
|
||||
}
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
assert error.parameter == "page[size]"
|
||||
assert error.pointer is None
|
||||
assert error.render() == "invalid parameter 'page[size]' (parameter page[size])"
|
||||
|
||||
|
||||
def test_the_field_an_error_points_at_is_kept():
|
||||
"""`source.pointer` is what turns "invalid" into "this field is invalid"."""
|
||||
(error,) = parse_jsonapi_errors(
|
||||
{
|
||||
"errors": [
|
||||
{
|
||||
"detail": "This field may not be blank.",
|
||||
"source": {"pointer": "/data/attributes/name"},
|
||||
}
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
assert error.render() == ("This field may not be blank. (/data/attributes/name)")
|
||||
|
||||
|
||||
def test_an_error_with_only_a_title_still_says_something():
|
||||
"""`detail` is the useful field, but the API does not always send one."""
|
||||
assert ApiErrorDetail(title="Not Found").render() == "Not Found"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"payload",
|
||||
[None, "not a document", {}, {"errors": "not a list"}, {"errors": [None]}],
|
||||
ids=["none", "text", "empty", "errors-not-a-list", "member-not-a-dict"],
|
||||
)
|
||||
def test_a_body_that_is_not_an_error_document_is_tolerated(payload):
|
||||
"""Parsing runs while handling a failure; it must not become the failure."""
|
||||
assert parse_jsonapi_errors(payload) == ()
|
||||
|
||||
|
||||
def test_an_error_with_nothing_in_it_renders_empty():
|
||||
"""Rendered to nothing rather than to punctuation, so composition can skip it."""
|
||||
assert ApiErrorDetail().render() == ""
|
||||
|
||||
|
||||
# --------------------------------------------------------------------- api failures
|
||||
|
||||
|
||||
def test_a_failed_read_names_the_call_and_the_reason():
|
||||
message = render_tool_error(
|
||||
ProwlerAPIError(
|
||||
"API request failed: 404 - Not found.",
|
||||
404,
|
||||
method="GET",
|
||||
path="/findings/nope",
|
||||
errors=parse_jsonapi_errors(
|
||||
{"errors": [{"detail": "No Finding matches the given query."}]}
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
assert message == (
|
||||
"GET /findings/nope failed with HTTP 404. No Finding matches the given query."
|
||||
)
|
||||
|
||||
|
||||
def test_every_error_of_a_rejected_write_reaches_the_client():
|
||||
"""The API rejects a write with one error per invalid field, and all of them help."""
|
||||
message = render_tool_error(
|
||||
ProwlerAPIError(
|
||||
"API request failed: 400 - blank",
|
||||
400,
|
||||
method="POST",
|
||||
path="/integrations",
|
||||
errors=parse_jsonapi_errors(
|
||||
{
|
||||
"errors": [
|
||||
{
|
||||
"detail": "This field may not be blank.",
|
||||
"source": {"pointer": "/data/attributes/bucket_name"},
|
||||
},
|
||||
{"detail": "Enter a valid URL."},
|
||||
]
|
||||
}
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
assert message == (
|
||||
"POST /integrations failed with HTTP 400. "
|
||||
"This field may not be blank. (/data/attributes/bucket_name); "
|
||||
"Enter a valid URL."
|
||||
)
|
||||
|
||||
|
||||
def test_a_rejected_write_gets_no_warning():
|
||||
"""A 4xx changed nothing, so there is nothing to warn about."""
|
||||
message = render_tool_error(
|
||||
ProwlerAPIError("boom", 400, method="POST", path="/scans")
|
||||
)
|
||||
|
||||
assert MAY_HAVE_LANDED not in message
|
||||
|
||||
|
||||
def test_a_write_that_hit_a_server_error_warns_it_may_have_landed():
|
||||
"""The API validates and queues before answering, so a 500 may have gone through."""
|
||||
message = render_tool_error(
|
||||
ProwlerAPIError("boom", 500, method="DELETE", path="/integrations/i1")
|
||||
)
|
||||
|
||||
assert message.startswith("DELETE /integrations/i1 failed with HTTP 500.")
|
||||
assert MAY_HAVE_LANDED in message
|
||||
|
||||
|
||||
def test_a_failed_read_never_warns():
|
||||
"""A read cannot have changed anything, whatever went wrong."""
|
||||
message = render_tool_error(
|
||||
ProwlerAPIError("boom", 500, method="GET", path="/scans")
|
||||
)
|
||||
|
||||
assert MAY_HAVE_LANDED not in message
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ no answer at all
|
||||
|
||||
|
||||
def test_a_write_that_got_no_answer_warns_it_may_have_landed():
|
||||
"""A timeout is the case the warning exists for."""
|
||||
request = httpx.Request("POST", "https://api.testing.invalid/api/v1/scans")
|
||||
message = render_tool_error(httpx.ReadTimeout("timed out", request=request))
|
||||
|
||||
assert "POST https://api.testing.invalid/api/v1/scans got no answer" in message
|
||||
assert "ReadTimeout" in message
|
||||
assert MAY_HAVE_LANDED in message
|
||||
|
||||
|
||||
def test_a_read_that_got_no_answer_does_not_warn():
|
||||
request = httpx.Request("GET", "https://api.testing.invalid/api/v1/findings")
|
||||
message = render_tool_error(httpx.ReadTimeout("timed out", request=request))
|
||||
|
||||
assert MAY_HAVE_LANDED not in message
|
||||
|
||||
|
||||
def test_a_dropped_connection_names_what_went_wrong():
|
||||
request = httpx.Request("POST", "https://api.testing.invalid/api/v1/scans")
|
||||
message = render_tool_error(httpx.ConnectError("connection reset", request=request))
|
||||
|
||||
assert "ConnectError: connection reset" in message
|
||||
|
||||
|
||||
def test_an_unfinished_task_warns_it_may_have_landed():
|
||||
"""The API already accepted the work, so the outcome is open, not refused."""
|
||||
message = render_tool_error(
|
||||
ProwlerTaskError(
|
||||
"Task t1 polling timed out after 60 seconds.", task_id="t1", state="timeout"
|
||||
)
|
||||
)
|
||||
|
||||
assert message.startswith("Task t1 polling timed out after 60 seconds.")
|
||||
assert MAY_HAVE_LANDED in message
|
||||
|
||||
|
||||
# -------------------------------------------------------------- refusals before send
|
||||
|
||||
|
||||
def test_a_stray_value_error_is_reported_as_a_bug():
|
||||
"""The distinction the previous passthrough branch could not make.
|
||||
|
||||
A model factory rejecting an API payload, or an `int()` on something that is not a
|
||||
number, is not the caller's mistake. Describing it like a validation message sends
|
||||
an agent off rewriting arguments that were never the problem.
|
||||
"""
|
||||
message = render_tool_error(
|
||||
ValueError("Missing pagination metadata in API response")
|
||||
)
|
||||
|
||||
assert "unexpected ValueError" in message
|
||||
assert "bug in the server" in message
|
||||
|
||||
|
||||
def test_an_authentication_failure_says_so():
|
||||
"""A `ValueError` subclass, so it must be recognised before the generic branch."""
|
||||
assert render_tool_error(ProwlerAuthError("Token has expired")) == (
|
||||
"Prowler authentication failed: Token has expired"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------- other hosts
|
||||
|
||||
|
||||
def test_a_hub_failure_reads_like_an_api_failure():
|
||||
"""Same sentence as the Prowler API, with the service named.
|
||||
|
||||
The Hub can be down while the API is fine, so which one failed is worth the two
|
||||
extra words -- but the shape must not differ, or the two look like two contracts.
|
||||
"""
|
||||
message = render_tool_error(
|
||||
ProwlerHubError(
|
||||
"hub failed",
|
||||
status_code=404,
|
||||
path="/check/test",
|
||||
body='{"error": "Not found"}',
|
||||
)
|
||||
)
|
||||
|
||||
assert message == "Prowler Hub GET /check/test failed with HTTP 404. Not found"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("body", "expected"),
|
||||
[
|
||||
('{"error": "Not found"}', "Not found"),
|
||||
('{"message": "Bad gateway"}', "Bad gateway"),
|
||||
('{"detail": "Rate limited"}', "Rate limited"),
|
||||
("Service Unavailable", "Service Unavailable"),
|
||||
('{"unexpected": "shape"}', '{"unexpected": "shape"}'),
|
||||
],
|
||||
ids=["error", "message", "detail", "plain-text", "unknown-json"],
|
||||
)
|
||||
def test_an_upstream_message_is_pulled_out_of_whatever_shape_it_came_in(body, expected):
|
||||
"""Hosts that are not the Prowler API each have their own error shape.
|
||||
|
||||
Relaying the raw body puts JSON braces, or a whole HTML page, in front of the model.
|
||||
"""
|
||||
message = render_tool_error(
|
||||
ProwlerHubError("hub failed", status_code=500, path="/checks", body=body)
|
||||
)
|
||||
|
||||
assert message.endswith(expected)
|
||||
|
||||
|
||||
def test_an_upstream_body_is_truncated():
|
||||
"""An HTML error page must not flood the model's context."""
|
||||
message = render_tool_error(
|
||||
ProwlerHubError("hub failed", status_code=500, path="/checks", body="x" * 2000)
|
||||
)
|
||||
|
||||
assert "x" * 500 in message
|
||||
assert "x" * 501 not in message
|
||||
|
||||
|
||||
# --------------------------------------------------------------------- server bugs
|
||||
|
||||
|
||||
def test_a_bug_in_this_server_is_reported_as_a_bug():
|
||||
"""Named as ours, so the caller stops trying to fix it by changing arguments."""
|
||||
message = render_tool_error(KeyError("attributes"))
|
||||
|
||||
assert "unexpected KeyError" in message
|
||||
assert "bug in the server" in message
|
||||
@@ -0,0 +1,170 @@
|
||||
"""Tests for the server class every sub-server is built from.
|
||||
|
||||
These drive a real `ProwlerMCP` through an in-memory MCP client rather than calling
|
||||
`tool_errors` directly, because applying that wrapper by hand is exactly what this
|
||||
class exists to make unnecessary. What matters is that a tool registered *any* of the
|
||||
ways this server registers them ends up with the error contract, and that it keeps the
|
||||
name, description and schema FastMCP publishes.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from fastmcp import Client
|
||||
from fastmcp.exceptions import ToolError
|
||||
from pydantic import Field
|
||||
|
||||
from prowler_mcp_server.lib.errors import ProwlerAPIError
|
||||
from prowler_mcp_server.lib.server import ProwlerMCP
|
||||
|
||||
|
||||
async def call(server: ProwlerMCP, name: str, arguments: dict | None = None):
|
||||
"""Call a tool the way a client does, without raising on failure."""
|
||||
async with Client(server) as client:
|
||||
return await client.call_tool(name, arguments or {}, raise_on_error=False)
|
||||
|
||||
|
||||
async def test_the_parameterised_decorator_form_is_wrapped():
|
||||
"""`@mcp.tool()` -- how the hub and documentation sub-servers register."""
|
||||
server = ProwlerMCP("test", mask_error_details=True)
|
||||
|
||||
@server.tool()
|
||||
async def failing() -> dict:
|
||||
"""A tool that fails."""
|
||||
raise ProwlerAPIError("boom", 404, method="GET", path="/x")
|
||||
|
||||
result = await call(server, "failing")
|
||||
|
||||
assert result.is_error
|
||||
assert result.content[0].text == "GET /x failed with HTTP 404."
|
||||
|
||||
|
||||
async def test_the_bare_decorator_form_is_wrapped():
|
||||
"""`@mcp.tool` without parentheses is a different code path in FastMCP."""
|
||||
server = ProwlerMCP("test", mask_error_details=True)
|
||||
|
||||
@server.tool
|
||||
async def failing() -> dict:
|
||||
"""A tool that fails."""
|
||||
raise ProwlerAPIError("boom", 500, method="GET", path="/y")
|
||||
|
||||
result = await call(server, "failing")
|
||||
|
||||
assert result.is_error
|
||||
assert "GET /y failed with HTTP 500." in result.content[0].text
|
||||
|
||||
|
||||
async def test_the_direct_call_form_is_wrapped():
|
||||
"""`mcp.tool(fn)` -- how `BaseTool` auto-registers its methods."""
|
||||
server = ProwlerMCP("test", mask_error_details=True)
|
||||
|
||||
async def failing() -> dict:
|
||||
"""A tool that fails."""
|
||||
raise ProwlerAPIError("boom", 403, method="DELETE", path="/z")
|
||||
|
||||
server.tool(failing)
|
||||
|
||||
result = await call(server, "failing")
|
||||
|
||||
assert result.is_error
|
||||
assert "DELETE /z failed with HTTP 403." in result.content[0].text
|
||||
|
||||
|
||||
async def test_a_synchronous_tool_is_wrapped():
|
||||
"""The documentation sub-server registers plain `def` tools."""
|
||||
server = ProwlerMCP("test", mask_error_details=True)
|
||||
|
||||
@server.tool()
|
||||
def failing() -> dict:
|
||||
"""A synchronous tool that fails."""
|
||||
raise KeyError("attributes")
|
||||
|
||||
result = await call(server, "failing")
|
||||
|
||||
assert result.is_error
|
||||
assert "unexpected KeyError" in result.content[0].text
|
||||
|
||||
|
||||
async def test_a_refusal_reaches_the_caller_word_for_word():
|
||||
"""A `ToolError` is passed through untouched, masking included.
|
||||
|
||||
That is the whole reason refusals are raised as one: the tool already wrote the
|
||||
sentence the caller needs, and nothing downstream improves on it.
|
||||
"""
|
||||
server = ProwlerMCP("test", mask_error_details=True)
|
||||
|
||||
@server.tool()
|
||||
async def refusing() -> dict:
|
||||
"""A tool that refuses its arguments."""
|
||||
raise ToolError(
|
||||
"Date range cannot exceed 2 days. Requested range: 2025-01-01 to "
|
||||
"2025-01-10 (10 days)"
|
||||
)
|
||||
|
||||
result = await call(server, "refusing")
|
||||
|
||||
assert result.is_error
|
||||
assert result.content[0].text == (
|
||||
"Date range cannot exceed 2 days. Requested range: 2025-01-01 to "
|
||||
"2025-01-10 (10 days)"
|
||||
)
|
||||
|
||||
|
||||
async def test_a_result_is_passed_through_untouched():
|
||||
server = ProwlerMCP("test")
|
||||
|
||||
@server.tool()
|
||||
async def succeeding(value: int) -> dict:
|
||||
"""A tool that works."""
|
||||
return {"value": value}
|
||||
|
||||
result = await call(server, "succeeding", {"value": 3})
|
||||
|
||||
assert not result.is_error
|
||||
assert result.data == {"value": 3}
|
||||
|
||||
|
||||
async def test_wrapping_does_not_disturb_the_published_tool():
|
||||
"""The wrapper must be invisible to FastMCP's schema generation.
|
||||
|
||||
A wrapper that loses the signature takes the parameters with it, and a tool with no
|
||||
parameters and no description is unusable while still looking registered.
|
||||
"""
|
||||
server = ProwlerMCP("test")
|
||||
|
||||
@server.tool()
|
||||
async def search(
|
||||
query: str = Field(description="What to search for"),
|
||||
limit: int = Field(default=10, description="How many results"),
|
||||
) -> dict:
|
||||
"""Search for things."""
|
||||
return {"query": query, "limit": limit}
|
||||
|
||||
async with Client(server) as client:
|
||||
(tool,) = await client.list_tools()
|
||||
|
||||
assert tool.name == "search"
|
||||
assert tool.description == "Search for things."
|
||||
properties = tool.inputSchema["properties"]
|
||||
assert properties["query"]["description"] == "What to search for"
|
||||
assert properties["limit"]["default"] == 10
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", ["decorated", "direct"])
|
||||
async def test_every_registration_carries_the_marker(name):
|
||||
"""The marker is what lets the contract test prove no tool slipped past."""
|
||||
server = ProwlerMCP("test")
|
||||
|
||||
async def direct() -> dict:
|
||||
"""Registered by direct call."""
|
||||
return {}
|
||||
|
||||
@server.tool()
|
||||
async def decorated() -> dict:
|
||||
"""Registered by decorator."""
|
||||
return {}
|
||||
|
||||
server.tool(direct)
|
||||
|
||||
tool = await server.get_tool(name)
|
||||
assert tool is not None, f"{name!r} was not registered at all"
|
||||
# `get_tool` is typed as the base `Tool`; only `FunctionTool` carries `fn`.
|
||||
assert getattr(getattr(tool, "fn", None), "__prowler_tool_errors__", False)
|
||||
@@ -33,6 +33,34 @@ async def test_every_sub_server_contributes_tools(mcp_root_server):
|
||||
assert tools_in_namespace(tools, "prowler_"), "Prowler App registered no tools"
|
||||
|
||||
|
||||
async def test_no_tool_disappears_between_registration_and_the_client(mcp_root_server):
|
||||
"""Every tool registered on a sub-server must still be reachable through the mount.
|
||||
|
||||
`ProwlerMCP.tool` wraps every tool before handing it to FastMCP, whether it arrived
|
||||
by decorator or by the direct call `BaseTool.register_tools` makes. A wrapper that
|
||||
loses the signature, the name or the coroutine-ness of what it wraps drops the tool
|
||||
silently: the mount still succeeds and the count is the only thing that moves.
|
||||
"""
|
||||
from prowler_mcp_server.prowler_app.server import app_mcp_server
|
||||
from prowler_mcp_server.prowler_documentation.server import docs_mcp_server
|
||||
from prowler_mcp_server.prowler_hub.server import hub_mcp_server
|
||||
|
||||
async with Client(mcp_root_server) as client:
|
||||
tools = await client.list_tools()
|
||||
|
||||
for namespace, sub_server in (
|
||||
("prowler_hub_", hub_mcp_server),
|
||||
("prowler_docs_", docs_mcp_server),
|
||||
("prowler_", app_mcp_server),
|
||||
):
|
||||
expected = len(await sub_server.list_tools())
|
||||
published = len(tools_in_namespace(tools, namespace))
|
||||
assert published == expected, (
|
||||
f"'{namespace}' publishes {published} tools but its sub-server registered "
|
||||
f"{expected}"
|
||||
)
|
||||
|
||||
|
||||
async def test_every_tool_is_namespaced(mcp_root_server):
|
||||
"""Tool names are a published interface; nothing may escape the namespaces."""
|
||||
async with Client(mcp_root_server) as client:
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
CMMC 2.0 universal compliance framework (`cmmc_2.0`) with the 149 official requirements from 32 CFR Part 170 — Level 1 (15, 48 CFR 52.204-21), Level 2 (110, NIST SP 800-171 Rev 2) and Level 3 (24, NIST SP 800-172) — with AWS, Azure, GCP, Alibaba Cloud, Oracle Cloud and M365 check mappings and config guardrails
|
||||
@@ -0,0 +1 @@
|
||||
GitHub `organization_repository_creation_limited` check now reports low severity for FAIL findings when repository creation is provably limited to private/internal visibility, instead of always reporting high
|
||||
@@ -0,0 +1 @@
|
||||
`entra_authentication_method_email_otp_disabled`, `entra_authentication_method_authenticator_show_context`, `entra_pim_global_administrator_approval_required`, `entra_pim_privileged_role_administrator_approval_required`, `entra_access_review_guest_users_configured` and `entra_access_review_privileged_roles_configured` checks for M365 provider covering CIS Microsoft 365 Foundations Benchmark v7.0.0 authentication method, PIM approval and access review controls
|
||||
@@ -0,0 +1 @@
|
||||
7 M365 entra checks covering CIS Microsoft 365 Foundations Benchmark v7.0.0 Conditional Access (5.2.2.x) and idle session timeout controls
|
||||
@@ -170,7 +170,10 @@
|
||||
{
|
||||
"Id": "1.3.2",
|
||||
"Description": "Idle session timeout allows the configuration of a setting which will timeout inactive users after a pre-determined amount of time. When a user reaches the set idle timeout session, they'll get a notification that they're about to be signed out. They must choose to stay signed in or they'll be automatically signed out of all Microsoft 365 web apps. Combined with a Conditional Access rule this will only impact unmanaged devices. A managed device is considered a device managed by Intune MDM or joined to a domain (Entra ID or Hybrid joined). The following Microsoft 365 web apps are supported. - Outlook Web App - OneDrive - SharePoint - Microsoft Fabric - Microsoft365.com and other start pages - Microsoft 365 web apps (Word, Excel, PowerPoint) - Microsoft 365 Admin Center - M365 Defender Portal - Microsoft Purview Compliance Portal The recommended setting is 3 hours (or less) for unmanaged devices. Note: Idle session timeout doesn't affect Microsoft 365 desktop and mobile apps.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_idle_session_timeout_configured",
|
||||
"entra_conditional_access_policy_app_enforced_restrictions"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "1 Microsoft 365 admin center",
|
||||
@@ -1783,7 +1786,9 @@
|
||||
{
|
||||
"Id": "5.2.2.8",
|
||||
"Description": "Microsoft Entra ID Protection sign-in risk detects risks in real-time and offline. A risky sign-in is an indicator for a sign-in attempt that might not have been performed by the legitimate owner of a user account. Note: While Identity Protection also provides two risk policies with limited conditions, Microsoft highly recommends setting up risk-based policies in Conditional Access as opposed to the \"legacy method\" for the following benefits: - Enhanced diagnostic data - Report-only mode integration - Graph API support - Use more Conditional Access attributes like sign-in frequency in the policy",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_conditional_access_policy_block_high_medium_sign_in_risk"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -1896,7 +1901,9 @@
|
||||
{
|
||||
"Id": "5.2.2.13",
|
||||
"Description": "Sign-in frequency defines the time period before a user is asked to sign in again when attempting to access a resource. The Microsoft Entra ID default configuration for user sign-in frequency is a rolling window of 90 days. The recommended state for all users is to enforce periodic reauthentication for 7 days or less.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_conditional_access_policy_sign_in_frequency_all_users"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -1917,7 +1924,9 @@
|
||||
{
|
||||
"Id": "5.2.2.14",
|
||||
"Description": "Microsoft Entra ID Conditional Access allows an organization to configure Named locations and configure whether those locations are trusted or untrusted. These settings provide organizations the means to specify Geographical locations for use in conditional access policies, or define actual IP addresses and IP ranges and whether or not those IP addresses and/or ranges are trusted by the organization. The recommended state is to define at least one trusted, IP range named location.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_conditional_access_trusted_named_location_exists"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -1938,7 +1947,9 @@
|
||||
{
|
||||
"Id": "5.2.2.15",
|
||||
"Description": "Conditional Access Policies can be used to block access from geographic locations that are deemed out-of-scope for your organization or application. The scope and variables for this policy should be carefully examined and defined. The recommended state is to configure at least one policy to block access from untrusted locations.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_conditional_access_policy_untrusted_locations_blocked"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -1959,7 +1970,9 @@
|
||||
{
|
||||
"Id": "5.2.2.16",
|
||||
"Description": "Token Protection is a Conditional Access session control that attempts to reduce token replay attacks by ensuring only device bound sign-in session tokens, like Primary Refresh Tokens (PRTs), are accepted by Microsoft Entra ID when applications request access to protected resources. When a user registers a supported device with Microsoft Entra, a PRT is issued and cryptographically bound to that device. This binding ensures that even if a threat actor steals the token, it can't be used from another device. With Token Protection enforced, Microsoft Entra validates that only these bound sign-in session tokens are used by supported applications. The recommended state is to enforce Token Protection for Office 365 Exchange Online, Office 365 SharePoint Online and Microsoft Teams Services.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_conditional_access_policy_token_protection_enforced"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -1980,7 +1993,9 @@
|
||||
{
|
||||
"Id": "5.2.2.17",
|
||||
"Description": "Authentication transfer is a flow that lets users seamlessly transfer authenticated state from one device to another. For example, users might see a QR code in the desktop version of Outlook that, when scanned on their mobile device, transfers their authenticated state to the mobile device. The recommended state is to block Authentication transfer.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_conditional_access_policy_authentication_transfer_blocked"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -2001,7 +2016,9 @@
|
||||
{
|
||||
"Id": "5.2.3.1",
|
||||
"Description": "Microsoft provides supporting settings to enhance the configuration of the Microsoft Authenticator application. These settings provide users with additional information and context when they receive MFA passwordless and push requests, including the geographic location of the request, the requesting application, and a requirement for number matching. The recommended state is Enabled for the following: - Show application name in push and passwordless notifications - Show geographic location in push and passwordless notifications Note: On February 27, 2023 Microsoft started enforcing number matching tenant-wide for all users using Microsoft Authenticator.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_authentication_method_authenticator_show_context"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -2135,7 +2152,9 @@
|
||||
{
|
||||
"Id": "5.2.3.7",
|
||||
"Description": "Authentication methods support a wide variety of scenarios for signing in to Microsoft 365 resources. Some of these methods are inherently more secure than others but require more investment in time to get users enrolled and operational. The email one-time passcode feature is a way to authenticate B2B collaboration users when they can't be authenticated through other means, such as Microsoft Entra ID, Microsoft account (MSA), or social identity providers. When a B2B guest user tries to redeem your invitation or sign in to your shared resources, they can request a temporary passcode, which is sent to their email address. Then they enter this passcode to continue signing in. The recommended state is to Disable email OTP.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_authentication_method_email_otp_disabled"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -2349,7 +2368,9 @@
|
||||
{
|
||||
"Id": "5.3.2",
|
||||
"Description": "Access reviews enable administrators to establish an efficient automated process for reviewing group memberships, access to enterprise applications, and role assignments. These reviews can be scheduled to recur regularly, with flexible options for delegating the task of reviewing membership to different members of the organization. When configured for guest users, access reviews can automatically remove access if no reviewer responds within the review period, enforcing a fail-closed posture for external identities.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_access_review_guest_users_configured"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -2370,7 +2391,9 @@
|
||||
{
|
||||
"Id": "5.3.3",
|
||||
"Description": "Access reviews in Microsoft Entra Privileged Identity Management (PIM) enable administrators to periodically validate whether users still require their privileged role assignments. These reviews can be scheduled to recur on a regular cadence and can be delegated to reviewers other than the role holders themselves, such as security auditors.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_access_review_privileged_roles_configured"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -2391,7 +2414,9 @@
|
||||
{
|
||||
"Id": "5.3.4",
|
||||
"Description": "Microsoft Entra Privileged Identity Management can be used to audit roles, allow just in time activation of roles and allow for periodic role attestation. Requiring approval before activation allows one of the selected approvers to first review and then approve the activation prior to PIM granted the role. The approver doesn't have to be a group member or owner. The recommended state is Require approval to activate for the Global Administrator role.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_pim_global_administrator_approval_required"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -2412,7 +2437,9 @@
|
||||
{
|
||||
"Id": "5.3.5",
|
||||
"Description": "Microsoft Entra Privileged Identity Management can be used to audit roles, allow just in time activation of roles and allow for periodic role attestation. Requiring approval before activation allows one of the selected approvers to first review and then approve the activation prior to PIM granted the role. The approver doesn't have to be a group member or owner. The recommended state is Require approval to activate for the Privileged Role Administrator role.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_pim_privileged_role_administrator_approval_required"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
from typing import List
|
||||
|
||||
from prowler.lib.check.models import Check, CheckReportGithub
|
||||
from prowler.lib.check.models import Check, CheckReportGithub, Severity
|
||||
from prowler.providers.github.services.organization.organization_client import (
|
||||
organization_client,
|
||||
)
|
||||
@@ -15,8 +15,19 @@ def _join_human_readable(items: List[str]) -> str:
|
||||
return ", ".join(items[:-1]) + f" and {items[-1]}"
|
||||
|
||||
|
||||
PUBLIC_CREATION_TYPES = {"all", "public"}
|
||||
NON_PUBLIC_CREATION_TYPES = {"private", "internal"}
|
||||
PUBLIC_DISABLED_CREATION_TYPES = NON_PUBLIC_CREATION_TYPES | {"none"}
|
||||
KNOWN_CREATION_TYPES = PUBLIC_CREATION_TYPES | PUBLIC_DISABLED_CREATION_TYPES
|
||||
|
||||
|
||||
class organization_repository_creation_limited(Check):
|
||||
"""Check if repository creation is limited to trusted organization members."""
|
||||
"""Check if repository creation is limited to trusted organization members.
|
||||
|
||||
FAIL severity scales with the visibility members can create: high when public
|
||||
repository creation is (or may be) allowed, low when it is provably limited to
|
||||
private/internal repositories.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[CheckReportGithub]:
|
||||
findings = []
|
||||
@@ -48,12 +59,19 @@ class organization_repository_creation_limited(Check):
|
||||
org, "members_allowed_repository_creation_type", None
|
||||
)
|
||||
|
||||
normalized_type = creation_type.lower() if creation_type else ""
|
||||
|
||||
type_flags = []
|
||||
enabled_types = []
|
||||
|
||||
if global_creation is not None:
|
||||
if global_creation:
|
||||
enabled_types.append("repositories of any type")
|
||||
public_known_disabled = (
|
||||
public_creation is False
|
||||
or normalized_type in PUBLIC_DISABLED_CREATION_TYPES
|
||||
)
|
||||
if not public_known_disabled:
|
||||
enabled_types.append("repositories of any type")
|
||||
else:
|
||||
type_flags.append(False)
|
||||
|
||||
@@ -70,7 +88,6 @@ class organization_repository_creation_limited(Check):
|
||||
enabled_types.append(label)
|
||||
|
||||
if creation_type:
|
||||
normalized_type = creation_type.lower()
|
||||
if normalized_type == "none":
|
||||
type_flags.append(False)
|
||||
else:
|
||||
@@ -97,7 +114,28 @@ class organization_repository_creation_limited(Check):
|
||||
unique_enabled = list(dict.fromkeys(enabled_types))
|
||||
allowed_desc = _join_human_readable(unique_enabled)
|
||||
if allowed_desc:
|
||||
report.status_extended = f"Organization {org.name} allows members to create {allowed_desc}."
|
||||
public_allowed = (
|
||||
public_creation is True
|
||||
or normalized_type in PUBLIC_CREATION_TYPES
|
||||
)
|
||||
non_public_allowed = (
|
||||
private_creation is True
|
||||
or internal_creation is True
|
||||
or normalized_type in NON_PUBLIC_CREATION_TYPES
|
||||
)
|
||||
public_known = (
|
||||
public_creation is not None
|
||||
or normalized_type in KNOWN_CREATION_TYPES
|
||||
)
|
||||
|
||||
if not public_allowed and non_public_allowed and public_known:
|
||||
report.check_metadata.Severity = Severity.low
|
||||
report.status_extended = (
|
||||
f"Organization {org.name} allows members to create {allowed_desc}. "
|
||||
"Public repository creation is disabled."
|
||||
)
|
||||
else:
|
||||
report.status_extended = f"Organization {org.name} allows members to create {allowed_desc}."
|
||||
else:
|
||||
report.status_extended = f"Organization {org.name} does not have enough data to confirm repository creation restrictions."
|
||||
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_access_review_guest_users_configured",
|
||||
"CheckTitle": "Access review for guest users is configured",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "An **access review** scoped to **guest users** should exist and be active so that external guest access is periodically recertified by reviewers. Requires a Microsoft Entra ID P2 license.",
|
||||
"Risk": "Without recurring access reviews for guests, external accounts accumulate over time and retain access long after it is needed, expanding the attack surface and the risk of orphaned or abused guest accounts.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/id-governance/create-access-review"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **ID Governance** > **Access reviews** > **New access review**\n3. Set the scope to **Guest users only**\n4. Assign at least one non-guest reviewer, enable mail notifications and reminders\n5. Set a recurring schedule and create the review",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Create a recurring access review scoped to guest users with assigned reviewers so guest access is periodically recertified.",
|
||||
"Url": "https://hub.prowler.com/check/entra_access_review_guest_users_configured"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e5"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "Requires Microsoft Entra ID P2."
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
import re
|
||||
from typing import List
|
||||
from urllib.parse import unquote
|
||||
|
||||
from prowler.lib.check.models import Check, CheckReportM365
|
||||
from prowler.providers.m365.services.entra.entra_client import entra_client
|
||||
|
||||
ACTIVE_STATUSES = {"InProgress"}
|
||||
ALLOWED_RECURRENCE_PATTERNS = {"weekly", "absoluteMonthly"}
|
||||
GUEST_USER_PREDICATE = re.compile(r"\busertype\s+eq\s+(['\"])guest\1", re.IGNORECASE)
|
||||
NEGATED_GUEST_USER_PREDICATE = re.compile(
|
||||
r"\bnot\s*\(\s*usertype\s+eq\s+(['\"])guest\1\s*\)", re.IGNORECASE
|
||||
)
|
||||
|
||||
|
||||
class entra_access_review_guest_users_configured(Check):
|
||||
"""Check if an access review for guest users is configured and fail-closed.
|
||||
|
||||
An access review scoped to guest users should exist, be active, recurring,
|
||||
and have primary reviewers assigned. It should also be fail-closed: if
|
||||
reviewers do not respond, access is removed
|
||||
(``defaultDecisionEnabled`` with ``defaultDecision`` = Deny and
|
||||
``autoApplyDecisionsEnabled``), with mail notifications and reminders enabled.
|
||||
|
||||
- PASS: A compliant recurring access review scoped to guest users exists.
|
||||
- FAIL: No compliant recurring access review scoped to guest users exists.
|
||||
"""
|
||||
|
||||
def _targets_guest_users(self, definition) -> bool:
|
||||
"""Determine whether an access review targets guest users.
|
||||
|
||||
Portal-created reviews use a principal-resource-memberships scope where the
|
||||
guest filter (``userType eq 'Guest'``) lives in the principal scopes and the
|
||||
top-level scope query is empty, so both are inspected.
|
||||
|
||||
This bounded matcher recognizes the equality predicate and its direct
|
||||
``not(...)`` negation. It does not interpret other compound or nested OData
|
||||
boolean semantics.
|
||||
|
||||
Args:
|
||||
definition: The access review definition to evaluate.
|
||||
|
||||
Returns:
|
||||
bool: True if any scope contains the guest-user equality predicate.
|
||||
"""
|
||||
queries = [definition.scope_query] + definition.principal_scope_queries
|
||||
for query in queries:
|
||||
decoded_query = unquote(query)
|
||||
if NEGATED_GUEST_USER_PREDICATE.search(decoded_query):
|
||||
continue
|
||||
if GUEST_USER_PREDICATE.search(decoded_query):
|
||||
return True
|
||||
return False
|
||||
|
||||
def _is_recurring_with_reviewers(self, definition) -> bool:
|
||||
"""Determine whether recurrence and primary reviewers are configured."""
|
||||
return (
|
||||
definition.recurrence_pattern_type in ALLOWED_RECURRENCE_PATTERNS
|
||||
and definition.recurrence_range_type == "noEnd"
|
||||
and definition.has_primary_reviewers
|
||||
)
|
||||
|
||||
def _is_fail_closed(self, definition) -> bool:
|
||||
"""Determine whether an access review definition is fail-closed.
|
||||
|
||||
Args:
|
||||
definition: The access review definition to evaluate.
|
||||
|
||||
Returns:
|
||||
bool: True if the review enables and denies access by default,
|
||||
auto-applies decisions, and has mail notifications and reminders enabled.
|
||||
"""
|
||||
return (
|
||||
definition.default_decision == "Deny"
|
||||
and definition.default_decision_enabled
|
||||
and definition.auto_apply_enabled
|
||||
and definition.mail_notifications_enabled
|
||||
and definition.reminders_enabled
|
||||
)
|
||||
|
||||
def execute(self) -> List[CheckReportM365]:
|
||||
"""Evaluate whether a compliant access review for guest users exists.
|
||||
|
||||
Searches the tenant's access review definitions for an active, recurring,
|
||||
reviewer-assigned, fail-closed review scoped to guest users.
|
||||
|
||||
Returns:
|
||||
List[CheckReportM365]: A single report indicating whether a compliant
|
||||
access review scoped to guest users is configured.
|
||||
"""
|
||||
findings = []
|
||||
definitions = entra_client.access_review_definitions
|
||||
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource={},
|
||||
resource_name="Access Review Definitions",
|
||||
resource_id="accessReviewDefinitions",
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"No compliant recurring access review scoped to guest users is configured "
|
||||
"with assigned primary reviewers."
|
||||
)
|
||||
|
||||
for definition in definitions:
|
||||
if (
|
||||
definition.status in ACTIVE_STATUSES
|
||||
and self._targets_guest_users(definition)
|
||||
and self._is_fail_closed(definition)
|
||||
and self._is_recurring_with_reviewers(definition)
|
||||
):
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=definition,
|
||||
resource_name=definition.display_name or "Access Review",
|
||||
resource_id=definition.id,
|
||||
)
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
f"Access review '{definition.display_name or definition.id}' for "
|
||||
"guest users is active, recurring, reviewer-assigned, and "
|
||||
"fail-closed."
|
||||
)
|
||||
break
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_access_review_privileged_roles_configured",
|
||||
"CheckTitle": "Access review for privileged roles is configured",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "An **access review** scoped to **privileged (PIM) directory roles** should exist and be active so privileged role assignments are periodically recertified by reviewers. Requires a Microsoft Entra ID P2 license.",
|
||||
"Risk": "Without recurring access reviews of privileged roles, standing privileged assignments accumulate and persist beyond their need, increasing the risk of privilege misuse and expanding the impact of a compromised administrator account.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/id-governance/privileged-identity-management/pim-create-roles-and-resource-roles-review"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **ID Governance** > **Privileged Identity Management** > **Microsoft Entra Roles** > **Access reviews** > **New**\n3. Scope the review to the privileged roles, set the assignment type to Eligible and Active\n4. Assign reviewers, enable notifications, and set a recurring schedule\n5. Create the review",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Create a recurring access review scoped to privileged roles with assigned reviewers so privileged role assignments are periodically recertified.",
|
||||
"Url": "https://hub.prowler.com/check/entra_access_review_privileged_roles_configured"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e5"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "Requires Microsoft Entra ID P2."
|
||||
}
|
||||
@@ -0,0 +1,117 @@
|
||||
from typing import List
|
||||
|
||||
from prowler.lib.check.models import Check, CheckReportM365
|
||||
from prowler.providers.m365.services.entra.entra_client import entra_client
|
||||
|
||||
ACTIVE_STATUSES = {"InProgress"}
|
||||
ALLOWED_RECURRENCE_PATTERNS = {"weekly", "absoluteMonthly"}
|
||||
# Markers that indicate the review targets directory role assignments (PIM roles).
|
||||
PRIVILEGED_SCOPE_MARKERS = ("roledefinition", "rolemanagement", "roleassignment")
|
||||
|
||||
|
||||
class entra_access_review_privileged_roles_configured(Check):
|
||||
"""Check if a compliant recurring access review for privileged roles exists.
|
||||
|
||||
An access review scoped to privileged (PIM) directory roles should exist, be
|
||||
active, recurring, and have primary reviewers assigned. It should use
|
||||
``defaultDecision`` = None with ``autoApplyDecisionsEnabled``, mail
|
||||
notifications, and reminders enabled.
|
||||
|
||||
- PASS: A compliant recurring access review scoped to privileged roles exists.
|
||||
- FAIL: No compliant recurring access review scoped to privileged roles exists.
|
||||
"""
|
||||
|
||||
def _targets_privileged_roles(self, definition) -> bool:
|
||||
"""Determine whether an access review targets privileged directory roles.
|
||||
|
||||
For PIM role reviews the role reference lives in the resource scopes, so both
|
||||
the scope query and the resource scope queries are inspected for markers that
|
||||
indicate directory role assignments.
|
||||
|
||||
Args:
|
||||
definition: The access review definition to evaluate.
|
||||
|
||||
Returns:
|
||||
bool: True if any of the review's scope queries reference privileged
|
||||
(PIM) directory roles.
|
||||
"""
|
||||
queries = [definition.scope_query] + definition.resource_scope_queries
|
||||
return any(
|
||||
marker in query.lower()
|
||||
for query in queries
|
||||
for marker in PRIVILEGED_SCOPE_MARKERS
|
||||
)
|
||||
|
||||
def _has_required_decision_settings(self, definition) -> bool:
|
||||
"""Determine whether an access review has the required decision settings.
|
||||
|
||||
Args:
|
||||
definition: The access review definition to evaluate.
|
||||
|
||||
Returns:
|
||||
bool: True if the review makes no default decision, auto-applies
|
||||
decisions, and has mail notifications and reminders enabled.
|
||||
"""
|
||||
return (
|
||||
definition.default_decision == "None"
|
||||
and definition.auto_apply_enabled
|
||||
and definition.mail_notifications_enabled
|
||||
and definition.reminders_enabled
|
||||
)
|
||||
|
||||
def _is_recurring_with_reviewers(self, definition) -> bool:
|
||||
"""Determine whether recurrence and primary reviewers are configured."""
|
||||
return (
|
||||
definition.recurrence_pattern_type in ALLOWED_RECURRENCE_PATTERNS
|
||||
and definition.recurrence_range_type == "noEnd"
|
||||
and definition.has_primary_reviewers
|
||||
)
|
||||
|
||||
def execute(self) -> List[CheckReportM365]:
|
||||
"""Evaluate whether a compliant access review for privileged roles exists.
|
||||
|
||||
Searches the tenant's access review definitions for an active, recurring,
|
||||
reviewer-assigned review scoped to privileged (PIM) directory roles.
|
||||
|
||||
Returns:
|
||||
List[CheckReportM365]: A single report indicating whether a compliant
|
||||
access review scoped to privileged roles is configured.
|
||||
"""
|
||||
findings = []
|
||||
definitions = entra_client.access_review_definitions
|
||||
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource={},
|
||||
resource_name="Access Review Definitions",
|
||||
resource_id="accessReviewDefinitions",
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"No compliant recurring access review scoped to privileged roles is "
|
||||
"configured with assigned primary reviewers."
|
||||
)
|
||||
|
||||
for definition in definitions:
|
||||
if (
|
||||
definition.status in ACTIVE_STATUSES
|
||||
and self._targets_privileged_roles(definition)
|
||||
and self._has_required_decision_settings(definition)
|
||||
and self._is_recurring_with_reviewers(definition)
|
||||
):
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=definition,
|
||||
resource_name=definition.display_name or "Access Review",
|
||||
resource_id=definition.id,
|
||||
)
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
f"Access review '{definition.display_name or definition.id}' for "
|
||||
"privileged roles is active, recurring, reviewer-assigned, and "
|
||||
"configured with no default decision."
|
||||
)
|
||||
break
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_authentication_method_authenticator_show_context",
|
||||
"CheckTitle": "Microsoft Authenticator shows application name and geographic location",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "The Microsoft Authenticator authentication method should have the **Show application name** (**displayAppInformationRequiredState**) and **Show geographic location** (**displayLocationInformationRequiredState**) feature settings enabled. This adds context to push and passwordless notifications so users can detect and reject fraudulent approval requests.",
|
||||
"Risk": "Without application-name and location context, users approving **MFA** push notifications cannot tell a legitimate sign-in from an attacker-initiated one, making the tenant more vulnerable to MFA fatigue and prompt-bombing attacks.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/identity/authentication/how-to-mfa-additional-context"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Protection** > **Authentication methods** > **Policies** > **Microsoft Authenticator**\n3. Select **Configure**\n4. Set **Show application name in push and passwordless notifications** to **Enabled**, Target **All users**\n5. Set **Show geographic location in push and passwordless notifications** to **Enabled**, Target **All users**\n6. Click **Save**",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Enable the application-name and geographic-location context settings for Microsoft Authenticator so users receive additional information to identify and reject fraudulent sign-in approvals.",
|
||||
"Url": "https://hub.prowler.com/check/entra_authentication_method_authenticator_show_context"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
from typing import List
|
||||
|
||||
from prowler.lib.check.models import Check, CheckReportM365
|
||||
from prowler.providers.m365.services.entra.entra_client import entra_client
|
||||
|
||||
|
||||
class entra_authentication_method_authenticator_show_context(Check):
|
||||
"""Check if Microsoft Authenticator shows application name and geographic location.
|
||||
|
||||
The Microsoft Authenticator method should be enabled with the
|
||||
``displayAppInformationRequiredState`` and
|
||||
``displayLocationInformationRequiredState`` feature settings enabled, so users see
|
||||
the app name and sign-in location context in push and passwordless notifications.
|
||||
|
||||
- PASS: Both application name and geographic location context are shown.
|
||||
- FAIL: Application name and/or geographic location context is not shown.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[CheckReportM365]:
|
||||
"""Execute the Microsoft Authenticator show context check.
|
||||
|
||||
Verifies that the Microsoft Authenticator method is enabled and configured to
|
||||
display both the application name and the geographic location of the sign-in in
|
||||
push and passwordless notifications.
|
||||
|
||||
Returns:
|
||||
List[CheckReportM365]: A list with a single report describing whether
|
||||
Microsoft Authenticator shows application name and geographic location
|
||||
context, or an empty list when the authentication methods policy settings
|
||||
are not available.
|
||||
"""
|
||||
findings = []
|
||||
settings = entra_client.authentication_methods_policy_settings
|
||||
if not settings:
|
||||
return findings
|
||||
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=settings,
|
||||
resource_name="Microsoft Authenticator Method",
|
||||
resource_id="microsoftAuthenticator",
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"Microsoft Authenticator does not show both application name and "
|
||||
"geographic location in notifications."
|
||||
)
|
||||
|
||||
if (
|
||||
settings.authenticator_state == "enabled"
|
||||
and settings.authenticator_display_app_information_state == "enabled"
|
||||
and settings.authenticator_display_location_information_state == "enabled"
|
||||
):
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
"Microsoft Authenticator shows application name and geographic "
|
||||
"location in notifications."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_authentication_method_email_otp_disabled",
|
||||
"CheckTitle": "Email One-Time Passcode authentication method is disabled in the tenant",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "Microsoft Entra tenant's authentication methods policy should have the **Email OTP** authentication method disabled. Email one-time passcodes depend on the security of the recipient mailbox, which is typically a lower-assurance channel, and should not be used as a primary or fallback multi-factor authentication method for members.",
|
||||
"Risk": "**Email OTP** is vulnerable to mailbox compromise and **phishing**. If an attacker gains access to a user's mailbox, they can intercept one-time passcodes and bypass **MFA**, gaining unauthorized access to the tenant.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/identity/authentication/howto-authentication-email-otp"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Protection** > **Authentication methods** > **Policies**\n3. Select **Email OTP** and set its status to **Disabled**, then click **Save**\n4. Ensure users have phishing-resistant MFA methods configured (e.g., FIDO2, Microsoft Authenticator)",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Disable the Email OTP authentication method and require phishing-resistant MFA methods such as FIDO2 security keys or Microsoft Authenticator across the tenant.",
|
||||
"Url": "https://hub.prowler.com/check/entra_authentication_method_email_otp_disabled"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
from typing import List
|
||||
|
||||
from prowler.lib.check.models import Check, CheckReportM365
|
||||
from prowler.providers.m365.services.entra.entra_client import entra_client
|
||||
|
||||
|
||||
class entra_authentication_method_email_otp_disabled(Check):
|
||||
"""
|
||||
Ensure that the Email One-Time Passcode (OTP) authentication method is disabled.
|
||||
|
||||
This check verifies that the tenant's authentication methods policy has the Email OTP
|
||||
method disabled. Email OTP relies on the security of the mailbox, which is often a
|
||||
lower-assurance channel and is unsuitable as a primary or fallback MFA method.
|
||||
|
||||
- PASS: Email OTP authentication method is disabled.
|
||||
- FAIL: Email OTP authentication method is enabled.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[CheckReportM365]:
|
||||
"""Execute the Email OTP authentication method check.
|
||||
|
||||
Returns:
|
||||
A list with a single report containing the result of the check.
|
||||
"""
|
||||
findings = []
|
||||
configs = entra_client.authentication_method_configurations
|
||||
|
||||
email_config = configs.get("Email")
|
||||
|
||||
if email_config:
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=email_config,
|
||||
resource_name="Email OTP Authentication Method",
|
||||
resource_id=entra_client.tenant_domain,
|
||||
)
|
||||
|
||||
if email_config.state == "disabled":
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
"Email OTP authentication method is disabled in the tenant."
|
||||
)
|
||||
elif email_config.state == "enabled":
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"Email OTP authentication method is enabled in the tenant."
|
||||
)
|
||||
else:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"Email OTP authentication method state could not be determined; "
|
||||
"treating as enabled/non-compliant."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_conditional_access_policy_authentication_transfer_blocked",
|
||||
"CheckTitle": "Conditional Access policy blocks authentication transfer",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "At least one enabled Conditional Access policy should target the **authentication transfer** flow for **all users** and **all resources** and set the grant control to **Block**. Authentication transfer lets a user move an authenticated session from one device to another (for example, by scanning a QR code), which can be abused to move a session onto an attacker-controlled device.",
|
||||
"Risk": "If authentication transfer is not blocked, an attacker who can present a transfer prompt (e.g., a malicious QR code) may capture an authenticated session and gain access to the victim's resources without needing their credentials or **MFA**.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/identity/conditional-access/concept-conditional-access-conditions"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Protection** > **Conditional Access** > **Policies** > **New policy**\n3. Under **Users**, include **All users** (exclude only documented break-glass accounts)\n4. Under **Target resources**, select **All resources**\n5. Under **Conditions** > **Authentication flows**, enable and select **Authentication transfer**\n6. Under **Grant**, select **Block access**\n7. Set the policy to **On** and click **Create**",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Create a Conditional Access policy that blocks the authentication transfer flow for all users and all resources, excluding only documented break-glass accounts.",
|
||||
"Url": "https://hub.prowler.com/check/entra_conditional_access_policy_authentication_transfer_blocked"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
from prowler.lib.check.models import Check, CheckReportM365
|
||||
from prowler.providers.m365.services.entra.entra_client import entra_client
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
ConditionalAccessGrantControl,
|
||||
ConditionalAccessPolicyState,
|
||||
TransferMethod,
|
||||
)
|
||||
|
||||
|
||||
class entra_conditional_access_policy_authentication_transfer_blocked(Check):
|
||||
"""Check if at least one Conditional Access policy blocks authentication transfer.
|
||||
|
||||
This check ensures that at least one enabled Conditional Access policy targets
|
||||
the authentication transfer flow and blocks access, preventing an authenticated
|
||||
session from being seamlessly transferred to another (potentially attacker
|
||||
controlled) device.
|
||||
|
||||
- PASS: An enabled Conditional Access policy blocks authentication transfer.
|
||||
- FAIL: No Conditional Access policy restricts authentication transfer.
|
||||
"""
|
||||
|
||||
def execute(self) -> list[CheckReportM365]:
|
||||
"""Execute the check to verify authentication transfer is blocked by a Conditional Access policy.
|
||||
|
||||
Returns:
|
||||
A list of reports containing the result of the check.
|
||||
"""
|
||||
findings = []
|
||||
|
||||
for policy in entra_client.conditional_access_policies.values():
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=policy,
|
||||
resource_name=policy.display_name,
|
||||
resource_id=policy.id,
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"Conditional Access Policy '{policy.display_name}' does not block authentication transfer."
|
||||
|
||||
authentication_flows = policy.conditions.authentication_flows
|
||||
blocks_authentication_transfer = (
|
||||
policy.state != ConditionalAccessPolicyState.DISABLED
|
||||
and "All" in policy.conditions.user_conditions.included_users
|
||||
and "All"
|
||||
in policy.conditions.application_conditions.included_applications
|
||||
and not policy.conditions.application_conditions.excluded_applications
|
||||
and authentication_flows
|
||||
and TransferMethod.AUTHENTICATION_TRANSFER
|
||||
in authentication_flows.transfer_methods
|
||||
and ConditionalAccessGrantControl.BLOCK
|
||||
in policy.grant_controls.built_in_controls
|
||||
)
|
||||
if blocks_authentication_transfer:
|
||||
if policy.state == ConditionalAccessPolicyState.ENABLED_FOR_REPORTING:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"Conditional Access Policy '{policy.display_name}' reports authentication transfer but does not block it."
|
||||
else:
|
||||
report.status = "PASS"
|
||||
report.status_extended = f"Conditional Access Policy '{policy.display_name}' blocks authentication transfer."
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
@@ -0,0 +1,38 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_conditional_access_policy_block_high_medium_sign_in_risk",
|
||||
"CheckTitle": "Conditional Access policy blocks high and medium sign-in risk",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "high",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "At least one enabled Conditional Access policy should target **all users** and **all resources** with the **sign-in risk** condition set to **High** and **Medium** and the grant control set to **Block**. Microsoft Entra ID Protection evaluates sign-in risk in real time; blocking risky sign-ins prevents access from sessions that are likely not performed by the legitimate account owner.",
|
||||
"Risk": "Without a **sign-in risk** based **Conditional Access** policy, sign-ins flagged as high or medium risk (indicative of compromised credentials, anomalous locations, or malicious IPs) are allowed, giving attackers a path to access resources with stolen credentials.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/id-protection/concept-identity-protection-policies",
|
||||
"https://learn.microsoft.com/en-us/entra/identity/conditional-access/concept-conditional-access-conditions#sign-in-risk"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Protection** > **Conditional Access** > **Policies** > **New policy**\n3. Under **Users**, include **All users** (exclude only documented break-glass accounts)\n4. Under **Target resources**, select **All resources**\n5. Under **Conditions** > **Sign-in risk**, select **High** and **Medium**\n6. Under **Grant**, select **Block access**\n7. Set the policy to **On** and click **Create**",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Create a risk-based Conditional Access policy that blocks high and medium sign-in risk for all users and all resources, excluding only documented break-glass accounts. Microsoft recommends configuring risk-based policies in Conditional Access rather than in the legacy Identity Protection policies.",
|
||||
"Url": "https://hub.prowler.com/check/entra_conditional_access_policy_block_high_medium_sign_in_risk"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e5"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
from prowler.lib.check.models import Check, CheckReportM365
|
||||
from prowler.providers.m365.services.entra.entra_client import entra_client
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
ConditionalAccessGrantControl,
|
||||
ConditionalAccessPolicyState,
|
||||
RiskLevel,
|
||||
)
|
||||
|
||||
|
||||
class entra_conditional_access_policy_block_high_medium_sign_in_risk(Check):
|
||||
"""Check if a Conditional Access policy blocks high and medium sign-in risk.
|
||||
|
||||
This check ensures that at least one enabled Conditional Access policy targets
|
||||
all users and all resources with the sign-in risk condition set to High and
|
||||
Medium, and blocks access, mitigating risky sign-in attempts detected by
|
||||
Microsoft Entra ID Protection.
|
||||
|
||||
- PASS: An enabled Conditional Access policy blocks high and medium sign-in risk.
|
||||
- FAIL: No Conditional Access policy blocks high and medium sign-in risk.
|
||||
"""
|
||||
|
||||
def execute(self) -> list[CheckReportM365]:
|
||||
"""Execute the check to verify high and medium sign-in risk is blocked.
|
||||
|
||||
Returns:
|
||||
A list of reports containing the result of the check.
|
||||
"""
|
||||
findings = []
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource={},
|
||||
resource_name="Conditional Access Policies",
|
||||
resource_id="conditionalAccessPolicies",
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"No Conditional Access Policy blocks high and medium sign-in risk."
|
||||
)
|
||||
|
||||
for policy in entra_client.conditional_access_policies.values():
|
||||
if policy.state == ConditionalAccessPolicyState.DISABLED:
|
||||
continue
|
||||
|
||||
if "All" not in policy.conditions.user_conditions.included_users:
|
||||
continue
|
||||
|
||||
if (
|
||||
"All"
|
||||
not in policy.conditions.application_conditions.included_applications
|
||||
):
|
||||
continue
|
||||
|
||||
if policy.conditions.application_conditions.excluded_applications:
|
||||
continue
|
||||
|
||||
if not {RiskLevel.HIGH, RiskLevel.MEDIUM}.issubset(
|
||||
set(policy.conditions.sign_in_risk_levels)
|
||||
):
|
||||
continue
|
||||
|
||||
if (
|
||||
ConditionalAccessGrantControl.BLOCK
|
||||
in policy.grant_controls.built_in_controls
|
||||
):
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=policy,
|
||||
resource_name=policy.display_name,
|
||||
resource_id=policy.id,
|
||||
)
|
||||
if policy.state == ConditionalAccessPolicyState.ENABLED_FOR_REPORTING:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"Conditional Access Policy '{policy.display_name}' reports high and medium sign-in risk but does not block it."
|
||||
else:
|
||||
report.status = "PASS"
|
||||
report.status_extended = f"Conditional Access Policy '{policy.display_name}' blocks high and medium sign-in risk."
|
||||
break
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_conditional_access_policy_sign_in_frequency_all_users",
|
||||
"CheckTitle": "Conditional Access policy enforces sign-in frequency for all users",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "At least one enabled Conditional Access policy should target **all users** and **all resources** and enforce a **sign-in frequency of 7 days or less**. Sign-in frequency defines how long an authenticated session remains valid before the user must reauthenticate. The Microsoft Entra default is a rolling 90-day window, which is too permissive.",
|
||||
"Risk": "Long-lived sessions increase the window in which a stolen token or hijacked session can be used. Without an enforced reauthentication interval, a compromised session can remain valid for up to 90 days, extending an attacker's access.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/identity/conditional-access/concept-session-lifetime"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Protection** > **Conditional Access** > **Policies** > **New policy**\n3. Under **Users**, include **All users** (exclude only documented break-glass accounts)\n4. Under **Target resources**, select **All resources**\n5. Under **Session**, enable **Sign-in frequency** and set it to **7 days** or less\n6. Set the policy to **On** and click **Create**",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Create a Conditional Access policy that enforces a sign-in frequency of 7 days or less for all users and all resources, excluding only documented break-glass accounts.",
|
||||
"Url": "https://hub.prowler.com/check/entra_conditional_access_policy_sign_in_frequency_all_users"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
@@ -0,0 +1,146 @@
|
||||
from prowler.lib.check.models import Check, CheckReportM365
|
||||
from prowler.providers.m365.services.entra.entra_client import entra_client
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
ClientAppType,
|
||||
ConditionalAccessPolicyState,
|
||||
SignInFrequencyInterval,
|
||||
SignInFrequencyType,
|
||||
)
|
||||
|
||||
# Maximum allowed reauthentication window (CIS: 7 days or less).
|
||||
MAX_SIGN_IN_FREQUENCY_DAYS = 7
|
||||
MAX_SIGN_IN_FREQUENCY_HOURS = MAX_SIGN_IN_FREQUENCY_DAYS * 24
|
||||
|
||||
|
||||
class entra_conditional_access_policy_sign_in_frequency_all_users(Check):
|
||||
"""Check if a Conditional Access policy enforces sign-in frequency for all users.
|
||||
|
||||
This check ensures that at least one enabled Conditional Access policy targets
|
||||
all users and all resources and enforces a sign-in frequency of 7 days or less,
|
||||
limiting how long an authenticated session remains valid before reauthentication.
|
||||
|
||||
- PASS: An enabled Conditional Access policy enforces sign-in frequency of 7 days
|
||||
or less for all users.
|
||||
- FAIL: No Conditional Access policy enforces sign-in frequency of 7 days or less
|
||||
for all users.
|
||||
"""
|
||||
|
||||
def _is_within_limit(self, sign_in_frequency) -> bool:
|
||||
"""Return True if the sign-in frequency is 7 days or less."""
|
||||
if not sign_in_frequency or not sign_in_frequency.is_enabled:
|
||||
return False
|
||||
|
||||
if sign_in_frequency.interval == SignInFrequencyInterval.EVERY_TIME:
|
||||
return True
|
||||
|
||||
if sign_in_frequency.frequency is None:
|
||||
return False
|
||||
|
||||
if sign_in_frequency.type == SignInFrequencyType.DAYS:
|
||||
return sign_in_frequency.frequency <= MAX_SIGN_IN_FREQUENCY_DAYS
|
||||
|
||||
if sign_in_frequency.type == SignInFrequencyType.HOURS:
|
||||
return sign_in_frequency.frequency <= MAX_SIGN_IN_FREQUENCY_HOURS
|
||||
|
||||
return False
|
||||
|
||||
def _has_restricted_scope(self, conditions) -> bool:
|
||||
"""Return True when policy conditions exclude part of the session scope."""
|
||||
platforms = conditions.platform_conditions
|
||||
platform_restricted = bool(
|
||||
platforms
|
||||
and (
|
||||
platforms.exclude_platforms
|
||||
or (
|
||||
platforms.include_platforms
|
||||
and "all" not in platforms.include_platforms
|
||||
)
|
||||
)
|
||||
)
|
||||
client_app_types = conditions.client_app_types or []
|
||||
client_app_restricted = bool(
|
||||
client_app_types and ClientAppType.ALL not in client_app_types
|
||||
)
|
||||
locations = conditions.locations
|
||||
location_restricted = bool(
|
||||
locations
|
||||
and (
|
||||
locations.exclude_locations
|
||||
or (
|
||||
locations.include_locations
|
||||
and "All" not in locations.include_locations
|
||||
)
|
||||
)
|
||||
)
|
||||
device_conditions = conditions.device_conditions
|
||||
device_restricted = bool(
|
||||
device_conditions
|
||||
and (
|
||||
device_conditions.device_filter_mode
|
||||
or device_conditions.device_filter_rule
|
||||
)
|
||||
)
|
||||
authentication_flows = conditions.authentication_flows
|
||||
authentication_flow_restricted = bool(
|
||||
authentication_flows and authentication_flows.transfer_methods
|
||||
)
|
||||
user_conditions = conditions.user_conditions
|
||||
user_restricted = bool(
|
||||
user_conditions.excluded_users
|
||||
or user_conditions.excluded_groups
|
||||
or user_conditions.excluded_roles
|
||||
or user_conditions.excluded_guests_or_external_users
|
||||
)
|
||||
application_restricted = bool(
|
||||
conditions.application_conditions.included_user_actions
|
||||
)
|
||||
|
||||
return bool(
|
||||
platform_restricted
|
||||
or client_app_restricted
|
||||
or location_restricted
|
||||
or device_restricted
|
||||
or authentication_flow_restricted
|
||||
or conditions.insider_risk_levels
|
||||
or user_restricted
|
||||
or application_restricted
|
||||
)
|
||||
|
||||
def execute(self) -> list[CheckReportM365]:
|
||||
"""Execute the check to verify sign-in frequency is enforced for all users.
|
||||
|
||||
Returns:
|
||||
A list of reports containing the result of the check.
|
||||
"""
|
||||
findings = []
|
||||
|
||||
for policy in entra_client.conditional_access_policies.values():
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=policy,
|
||||
resource_name=policy.display_name,
|
||||
resource_id=policy.id,
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"Conditional Access Policy '{policy.display_name}' does not enforce a sign-in frequency of 7 days or less for all users."
|
||||
|
||||
enforces_sign_in_frequency = (
|
||||
policy.state != ConditionalAccessPolicyState.DISABLED
|
||||
and "All" in policy.conditions.user_conditions.included_users
|
||||
and "All"
|
||||
in policy.conditions.application_conditions.included_applications
|
||||
and not policy.conditions.application_conditions.excluded_applications
|
||||
and not policy.conditions.sign_in_risk_levels
|
||||
and not policy.conditions.user_risk_levels
|
||||
and not self._has_restricted_scope(policy.conditions)
|
||||
and self._is_within_limit(policy.session_controls.sign_in_frequency)
|
||||
)
|
||||
if enforces_sign_in_frequency:
|
||||
if policy.state == ConditionalAccessPolicyState.ENABLED_FOR_REPORTING:
|
||||
report.status_extended = f"Conditional Access Policy '{policy.display_name}' enforces sign-in frequency but is in report-only mode."
|
||||
else:
|
||||
report.status = "PASS"
|
||||
report.status_extended = f"Conditional Access Policy '{policy.display_name}' enforces a sign-in frequency of 7 days or less for all users."
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_conditional_access_policy_token_protection_enforced",
|
||||
"CheckTitle": "Conditional Access policy enforces Token Protection",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "high",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "At least one enabled Conditional Access policy should enforce **Token Protection** (the **secureSignInSession** session control) for targeted users. Token Protection binds sign-in session tokens (such as Primary Refresh Tokens) to the device, reducing the effectiveness of token theft and replay attacks.",
|
||||
"Risk": "Without **Token Protection**, stolen sign-in tokens can be replayed from an attacker's device to impersonate the user and bypass authentication, including **MFA**, because the token is not cryptographically bound to the original device.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/identity/conditional-access/concept-token-protection"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Protection** > **Conditional Access** > **Policies** > **New policy**\n3. Under **Users**, include the targeted users (exclude only documented break-glass accounts)\n4. Under **Target resources**, select the supported applications (e.g., Exchange Online, SharePoint Online)\n5. Under **Conditions** > **Device platforms**, include Windows\n6. Under **Session**, enable **Require token protection for sign-in sessions**\n7. Set the policy to **On** and click **Create**",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Create a Conditional Access policy that enforces Token Protection (secure sign-in session) for supported applications on Windows devices to mitigate token theft and replay.",
|
||||
"Url": "https://hub.prowler.com/check/entra_conditional_access_policy_token_protection_enforced"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
from prowler.lib.check.models import Check, CheckReportM365
|
||||
from prowler.providers.m365.services.entra.entra_client import entra_client
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
ClientAppType,
|
||||
ConditionalAccessPolicyState,
|
||||
)
|
||||
|
||||
# Applications that must be covered by the Token Protection policy.
|
||||
EXCHANGE_ONLINE_APP_ID = "00000002-0000-0ff1-ce00-000000000000"
|
||||
SHAREPOINT_ONLINE_APP_ID = "00000003-0000-0ff1-ce00-000000000000"
|
||||
TEAMS_APP_ID = "cc15fd57-2c6c-4117-a88c-83b1d56b4bbe"
|
||||
REQUIRED_APP_IDS = {EXCHANGE_ONLINE_APP_ID, SHAREPOINT_ONLINE_APP_ID, TEAMS_APP_ID}
|
||||
|
||||
|
||||
class entra_conditional_access_policy_token_protection_enforced(Check):
|
||||
"""Check if a Conditional Access policy enforces Token Protection.
|
||||
|
||||
Token Protection is a Conditional Access session control that reduces token
|
||||
replay attacks by requiring device-bound sign-in session tokens. At least one
|
||||
enabled Conditional Access policy should enable the ``secureSignInSession``
|
||||
session control for the supported applications (Exchange Online, SharePoint
|
||||
Online, Teams), target the Windows platform, and apply to mobile apps and desktop
|
||||
clients.
|
||||
|
||||
- PASS: An enabled Conditional Access policy enforces Token Protection for the
|
||||
supported applications on Windows desktop/mobile clients.
|
||||
- FAIL: No Conditional Access policy enforces Token Protection with the required
|
||||
scope.
|
||||
"""
|
||||
|
||||
def _applications_covered(
|
||||
self, included_applications, excluded_applications
|
||||
) -> bool:
|
||||
"""Check whether the policy covers every required application.
|
||||
|
||||
Args:
|
||||
included_applications: Application IDs included by the policy.
|
||||
excluded_applications: Application IDs excluded by the policy.
|
||||
|
||||
Returns:
|
||||
True if all required applications are included and none are excluded.
|
||||
"""
|
||||
if REQUIRED_APP_IDS.intersection(excluded_applications):
|
||||
return False
|
||||
if "All" in included_applications:
|
||||
return True
|
||||
return REQUIRED_APP_IDS.issubset(set(included_applications))
|
||||
|
||||
def _windows_targeted(self, conditions) -> bool:
|
||||
"""Check whether the policy targets Windows devices.
|
||||
|
||||
Args:
|
||||
conditions: Conditional Access policy conditions.
|
||||
|
||||
Returns:
|
||||
True if Windows is included in the targeted platforms.
|
||||
"""
|
||||
platform_conditions = conditions.platform_conditions
|
||||
if not platform_conditions:
|
||||
return False
|
||||
return "windows" in platform_conditions.include_platforms
|
||||
|
||||
def _desktop_clients_targeted(self, conditions) -> bool:
|
||||
"""Check whether the policy targets desktop-capable client apps.
|
||||
|
||||
Args:
|
||||
conditions: Conditional Access policy conditions.
|
||||
|
||||
Returns:
|
||||
True if all clients or mobile apps and desktop clients are targeted.
|
||||
"""
|
||||
client_app_types = conditions.client_app_types or []
|
||||
return (
|
||||
ClientAppType.ALL in client_app_types
|
||||
or ClientAppType.MOBILE_APPS_AND_DESKTOP_CLIENTS in client_app_types
|
||||
)
|
||||
|
||||
def execute(self) -> list[CheckReportM365]:
|
||||
"""Execute the Token Protection Conditional Access policy check.
|
||||
|
||||
Returns:
|
||||
A list containing the Token Protection policy evaluation report.
|
||||
"""
|
||||
findings = []
|
||||
|
||||
for policy in entra_client.conditional_access_policies.values():
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=policy,
|
||||
resource_name=policy.display_name,
|
||||
resource_id=policy.id,
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"Conditional Access Policy '{policy.display_name}' does not enforce Token Protection for the supported applications on Windows desktop and mobile clients."
|
||||
|
||||
enforces_token_protection = (
|
||||
policy.state != ConditionalAccessPolicyState.DISABLED
|
||||
and policy.conditions.user_conditions.included_users
|
||||
and policy.session_controls.secure_sign_in_session_enabled
|
||||
and self._applications_covered(
|
||||
policy.conditions.application_conditions.included_applications,
|
||||
policy.conditions.application_conditions.excluded_applications,
|
||||
)
|
||||
and self._windows_targeted(policy.conditions)
|
||||
and self._desktop_clients_targeted(policy.conditions)
|
||||
)
|
||||
if enforces_token_protection:
|
||||
if policy.state == ConditionalAccessPolicyState.ENABLED_FOR_REPORTING:
|
||||
report.status_extended = f"Conditional Access Policy '{policy.display_name}' enables Token Protection but is in report-only mode."
|
||||
else:
|
||||
report.status = "PASS"
|
||||
report.status_extended = f"Conditional Access Policy '{policy.display_name}' enforces Token Protection for the supported applications on Windows desktop and mobile clients."
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_conditional_access_policy_untrusted_locations_blocked",
|
||||
"CheckTitle": "Conditional Access policy blocks access from untrusted locations",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "At least one enabled Conditional Access policy should target **all users** and **all resources**, include **all network locations**, exclude **trusted locations**, and set the grant control to **Block**. This blocks access originating from locations that are not explicitly trusted by the organization.",
|
||||
"Risk": "Without a location-based block policy, users can authenticate from any geography or network, including regions where the organization has no presence, giving attackers operating from those locations a viable path to access resources with stolen credentials.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/identity/conditional-access/concept-assignment-network"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Protection** > **Conditional Access** > **Policies** > **New policy**\n3. Under **Users**, include **All users** (exclude only documented break-glass accounts)\n4. Under **Target resources**, select **All resources**\n5. Under **Network**, include **Any location** and exclude **All trusted locations** (or selected trusted named locations)\n6. Under **Grant**, select **Block access**\n7. Set the policy to **On** and click **Create**",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Create a Conditional Access policy that blocks access from all locations except trusted named locations, targeting all users and resources, excluding only documented break-glass accounts.",
|
||||
"Url": "https://hub.prowler.com/check/entra_conditional_access_policy_untrusted_locations_blocked"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
from prowler.lib.check.models import Check, CheckReportM365
|
||||
from prowler.providers.m365.services.entra.entra_client import entra_client
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
ConditionalAccessGrantControl,
|
||||
ConditionalAccessPolicyState,
|
||||
)
|
||||
|
||||
|
||||
class entra_conditional_access_policy_untrusted_locations_blocked(Check):
|
||||
"""Check if a Conditional Access policy blocks access from untrusted locations.
|
||||
|
||||
At least one enabled Conditional Access policy should target all users and all
|
||||
resources, include all network locations, exclude trusted locations, and block
|
||||
access, so that access from untrusted locations is denied.
|
||||
|
||||
- PASS: An enabled Conditional Access policy blocks access from untrusted
|
||||
locations.
|
||||
- FAIL: No Conditional Access policy blocks access from untrusted locations.
|
||||
"""
|
||||
|
||||
def _excludes_only_trusted(self, exclude_locations, trusted_location_ids) -> bool:
|
||||
"""Check whether every excluded location is trusted.
|
||||
|
||||
Args:
|
||||
exclude_locations: Location IDs excluded by the policy.
|
||||
trusted_location_ids: IDs of collected trusted named locations.
|
||||
|
||||
Returns:
|
||||
True if at least one location is excluded and all exclusions are trusted.
|
||||
"""
|
||||
if not exclude_locations:
|
||||
return False
|
||||
return all(
|
||||
location_id == "AllTrusted" or location_id in trusted_location_ids
|
||||
for location_id in exclude_locations
|
||||
)
|
||||
|
||||
def execute(self) -> list[CheckReportM365]:
|
||||
"""Execute the untrusted-locations Conditional Access policy check.
|
||||
|
||||
Returns:
|
||||
A list containing the untrusted-locations policy evaluation report.
|
||||
"""
|
||||
findings = []
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource={},
|
||||
resource_name="Conditional Access Policies",
|
||||
resource_id="conditionalAccessPolicies",
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"No Conditional Access Policy blocks access from untrusted locations."
|
||||
)
|
||||
|
||||
trusted_location_ids = {
|
||||
location.id
|
||||
for location in entra_client.named_locations
|
||||
if location.is_trusted
|
||||
}
|
||||
untrusted_location_ids = {
|
||||
location.id
|
||||
for location in entra_client.named_locations
|
||||
if not location.is_trusted
|
||||
}
|
||||
|
||||
for policy in entra_client.conditional_access_policies.values():
|
||||
if policy.state == ConditionalAccessPolicyState.DISABLED:
|
||||
continue
|
||||
|
||||
if "All" not in policy.conditions.user_conditions.included_users:
|
||||
continue
|
||||
|
||||
if (
|
||||
"All"
|
||||
not in policy.conditions.application_conditions.included_applications
|
||||
or policy.conditions.application_conditions.excluded_applications
|
||||
):
|
||||
continue
|
||||
|
||||
locations = policy.conditions.locations
|
||||
if not locations:
|
||||
continue
|
||||
|
||||
if "All" not in locations.include_locations and not any(
|
||||
location_id in untrusted_location_ids
|
||||
for location_id in locations.include_locations
|
||||
):
|
||||
continue
|
||||
|
||||
# A trusted-location exclusion must exist so trusted networks keep access.
|
||||
if not self._excludes_only_trusted(
|
||||
locations.exclude_locations, trusted_location_ids
|
||||
):
|
||||
continue
|
||||
|
||||
if (
|
||||
ConditionalAccessGrantControl.BLOCK
|
||||
not in policy.grant_controls.built_in_controls
|
||||
):
|
||||
continue
|
||||
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=policy,
|
||||
resource_name=policy.display_name,
|
||||
resource_id=policy.id,
|
||||
)
|
||||
if policy.state == ConditionalAccessPolicyState.ENABLED_FOR_REPORTING:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"Conditional Access Policy '{policy.display_name}' blocks untrusted locations but is in report-only mode."
|
||||
else:
|
||||
report.status = "PASS"
|
||||
report.status_extended = f"Conditional Access Policy '{policy.display_name}' blocks access from untrusted locations."
|
||||
break
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_conditional_access_trusted_named_location_exists",
|
||||
"CheckTitle": "A trusted IP-range named location is defined",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "low",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "Conditional Access **named locations** let an organization define **trusted** IP ranges that can be referenced by Conditional Access policies. At least one **IP-range** named location should be marked as **Trusted** and have at least one IP range defined.",
|
||||
"Risk": "Without a trusted named location, **Conditional Access** policies cannot differentiate corporate networks from untrusted networks, limiting the ability to build location-aware controls such as blocking access from untrusted locations.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/identity/conditional-access/concept-assignment-network"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Protection** > **Conditional Access** > **Named locations**\n3. Create a **new IP ranges location**, add the organization's public IP ranges\n4. Mark the location as **Trusted**\n5. Click **Create**",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Define at least one trusted IP-range named location representing the organization's corporate networks so it can be referenced by location-aware Conditional Access policies.",
|
||||
"Url": "https://hub.prowler.com/check/entra_conditional_access_trusted_named_location_exists"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
from prowler.lib.check.models import Check, CheckReportM365
|
||||
from prowler.providers.m365.services.entra.entra_client import entra_client
|
||||
|
||||
|
||||
class entra_conditional_access_trusted_named_location_exists(Check):
|
||||
"""Check if at least one trusted IP-range named location is defined.
|
||||
|
||||
Conditional Access named locations let an organization define trusted IP ranges.
|
||||
At least one IP-range named location should be marked as trusted and have at
|
||||
least one IP range defined.
|
||||
|
||||
- PASS: A trusted IP-range named location with at least one IP range exists.
|
||||
- FAIL: No trusted IP-range named location with an IP range is defined.
|
||||
"""
|
||||
|
||||
def execute(self) -> list[CheckReportM365]:
|
||||
"""Execute the trusted named location check.
|
||||
|
||||
Returns:
|
||||
A list containing the trusted named location evaluation report.
|
||||
"""
|
||||
findings = []
|
||||
for location in entra_client.named_locations:
|
||||
location_name = location.display_name or location.id
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=location,
|
||||
resource_name=location.display_name or "Named Location",
|
||||
resource_id=location.id,
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"Named location '{location_name}' is not a trusted IP-range location with at least one IP range."
|
||||
|
||||
if (
|
||||
location.is_ip_location
|
||||
and location.is_trusted
|
||||
and location.ip_ranges_count >= 1
|
||||
):
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
f"Trusted IP-range named location '{location_name}' "
|
||||
f"is defined with {location.ip_ranges_count} IP range(s)."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_idle_session_timeout_configured",
|
||||
"CheckTitle": "Idle session timeout of 3 hours or less is configured",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "An activity-based timeout policy should sign out inactive users from Microsoft 365 web apps after a period of inactivity. The web session idle timeout (**WebSessionIdleTimeout**) should be set to **3 hours or less** so that unattended sessions are terminated.",
|
||||
"Risk": "Without an idle session timeout, an unattended and unlocked session on a shared or public device remains authenticated indefinitely, allowing anyone with physical access to act as the signed-in user.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/microsoft-365/admin/manage/idle-session-timeout-web-apps"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft 365 admin center at https://admin.microsoft.com/\n2. Go to **Settings** > **Org settings** > **Security & Privacy** > **Idle session timeout**\n3. Turn on the setting and set the period of inactivity to **3 hours** or less\n4. Click **Save**",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Configure an idle session timeout of 3 hours or less so inactive users are automatically signed out of Microsoft 365 web apps.",
|
||||
"Url": "https://hub.prowler.com/check/entra_idle_session_timeout_configured"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
from typing import List
|
||||
|
||||
from prowler.lib.check.models import Check, CheckReportM365
|
||||
from prowler.providers.m365.services.entra.entra_client import entra_client
|
||||
|
||||
# CIS recommends an idle session timeout of 3 hours or less.
|
||||
MAX_IDLE_TIMEOUT_SECONDS = 3 * 60 * 60
|
||||
|
||||
|
||||
class entra_idle_session_timeout_configured(Check):
|
||||
"""Check if an idle session timeout of 3 hours or less is configured.
|
||||
|
||||
An activity-based timeout policy should sign out inactive users from Microsoft
|
||||
365 web apps after a period of inactivity. The web session idle timeout should be
|
||||
set to 3 hours or less.
|
||||
|
||||
- PASS: An activity-based timeout policy enforces an idle timeout of 3 hours or
|
||||
less.
|
||||
- FAIL: No activity-based timeout policy enforces an idle timeout of 3 hours or
|
||||
less.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[CheckReportM365]:
|
||||
"""Execute the idle session timeout policy check.
|
||||
|
||||
Returns:
|
||||
A list of reports for the collected activity-based timeout policies.
|
||||
"""
|
||||
findings = []
|
||||
policies = entra_client.activity_based_timeout_policies
|
||||
|
||||
if not policies:
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource={},
|
||||
resource_name="Activity Based Timeout Policies",
|
||||
resource_id="activityBasedTimeoutPolicies",
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"No activity-based timeout policy enforces an idle session timeout of "
|
||||
"3 hours or less."
|
||||
)
|
||||
findings.append(report)
|
||||
return findings
|
||||
|
||||
for policy in policies:
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=policy,
|
||||
resource_name=policy.display_name or "Activity Based Timeout Policy",
|
||||
resource_id=policy.id,
|
||||
)
|
||||
timeout = policy.web_session_idle_timeout_seconds
|
||||
if timeout is not None and timeout <= MAX_IDLE_TIMEOUT_SECONDS:
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
f"Activity-based timeout policy '{policy.display_name or policy.id}' "
|
||||
f"enforces an idle session timeout of {timeout // 60} minutes and "
|
||||
f"{timeout % 60} seconds."
|
||||
)
|
||||
else:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
f"Activity-based timeout policy '{policy.display_name or policy.id}' "
|
||||
"does not enforce an idle session timeout of 3 hours or less."
|
||||
)
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_pim_global_administrator_approval_required",
|
||||
"CheckTitle": "PIM requires approval to activate the Global Administrator role",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "high",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "Privileged Identity Management (PIM) should require **approval to activate** the **Global Administrator** role, with at least one approver configured. This adds a human authorization step before the most privileged role in the tenant can be used. Requires a Microsoft Entra ID P2 license.",
|
||||
"Risk": "Without approval to activate **Global Administrator**, an eligible or compromised account can self-activate the highest-privilege role without oversight, enabling immediate tenant-wide compromise.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/id-governance/privileged-identity-management/pim-how-to-change-default-settings"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **ID Governance** > **Privileged Identity Management** > **Microsoft Entra Roles** > **Roles**\n3. Select **Global Administrator** > **Role settings**\n4. Set **Require approval to activate** to **Yes** and add at least one approver\n5. Save the settings",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Require approval to activate the Global Administrator role in PIM and configure at least one approver so activations are authorized before use.",
|
||||
"Url": "https://hub.prowler.com/check/entra_pim_global_administrator_approval_required"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e5"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "Requires Microsoft Entra ID P2. If PIM role settings are unavailable (no P2), the check returns no findings for this role."
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
from typing import List
|
||||
|
||||
from prowler.lib.check.models import Check, CheckReportM365
|
||||
from prowler.providers.m365.services.entra.entra_client import entra_client
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
GLOBAL_ADMINISTRATOR_ROLE_TEMPLATE_ID,
|
||||
)
|
||||
|
||||
|
||||
class entra_pim_global_administrator_approval_required(Check):
|
||||
"""Check if PIM requires approval to activate the Global Administrator role.
|
||||
|
||||
Privileged Identity Management (PIM) should require approval to activate the
|
||||
Global Administrator role, with at least one approver configured.
|
||||
|
||||
- PASS: Approval is required to activate Global Administrator and approvers exist.
|
||||
- FAIL: Approval is not required or no approvers are configured.
|
||||
- No findings: PIM settings are unavailable.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[CheckReportM365]:
|
||||
"""Evaluate PIM approval settings for the Global Administrator role.
|
||||
|
||||
Reports whether Privileged Identity Management requires approval to activate
|
||||
the Global Administrator role and whether at least one approver is configured.
|
||||
|
||||
Returns:
|
||||
List[CheckReportM365]: A single report for the Global Administrator PIM
|
||||
role settings, or an empty list when the settings are absent.
|
||||
"""
|
||||
findings = []
|
||||
setting = entra_client.pim_role_approval_settings.get(
|
||||
GLOBAL_ADMINISTRATOR_ROLE_TEMPLATE_ID
|
||||
)
|
||||
if not setting:
|
||||
return findings
|
||||
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=setting,
|
||||
resource_name="Global Administrator PIM Role Settings",
|
||||
resource_id=GLOBAL_ADMINISTRATOR_ROLE_TEMPLATE_ID,
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"PIM does not require approval to activate the Global Administrator role."
|
||||
)
|
||||
|
||||
if setting.is_approval_required and not setting.has_approvers:
|
||||
report.status_extended = (
|
||||
"PIM requires approval to activate the Global Administrator role but "
|
||||
"no approvers are configured."
|
||||
)
|
||||
elif setting.is_approval_required and setting.has_approvers:
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
"PIM requires approval to activate the Global Administrator role and "
|
||||
"has approvers configured."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_pim_privileged_role_administrator_approval_required",
|
||||
"CheckTitle": "PIM requires approval to activate the Privileged Role Administrator role",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "high",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "Privileged Identity Management (PIM) should require **approval to activate** the **Privileged Role Administrator** role, with at least one approver configured. This role can manage role assignments, so its activation should require human authorization. Requires a Microsoft Entra ID P2 license.",
|
||||
"Risk": "Without approval to activate **Privileged Role Administrator**, an eligible or compromised account can self-activate a role that grants control over role assignments, enabling privilege escalation across the tenant.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/id-governance/privileged-identity-management/pim-how-to-change-default-settings"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **ID Governance** > **Privileged Identity Management** > **Microsoft Entra Roles** > **Roles**\n3. Select **Privileged Role Administrator** > **Role settings**\n4. Set **Require approval to activate** to **Yes** and add at least one approver\n5. Save the settings",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Require approval to activate the Privileged Role Administrator role in PIM and configure at least one approver so activations are authorized before use.",
|
||||
"Url": "https://hub.prowler.com/check/entra_pim_privileged_role_administrator_approval_required"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e5"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "Requires Microsoft Entra ID P2. If PIM role settings are unavailable (no P2), the check returns no findings for this role."
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
from typing import List
|
||||
|
||||
from prowler.lib.check.models import Check, CheckReportM365
|
||||
from prowler.providers.m365.services.entra.entra_client import entra_client
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
PRIVILEGED_ROLE_ADMINISTRATOR_ROLE_TEMPLATE_ID,
|
||||
)
|
||||
|
||||
|
||||
class entra_pim_privileged_role_administrator_approval_required(Check):
|
||||
"""Check if PIM requires approval to activate the Privileged Role Administrator role.
|
||||
|
||||
Privileged Identity Management (PIM) should require approval to activate the
|
||||
Privileged Role Administrator role, with at least one approver configured.
|
||||
|
||||
- PASS: Approval is required to activate Privileged Role Administrator and
|
||||
approvers exist.
|
||||
- FAIL: Approval is not required or no approvers are configured.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[CheckReportM365]:
|
||||
"""Evaluate PIM approval settings for the Privileged Role Administrator role.
|
||||
|
||||
Reports whether Privileged Identity Management requires approval to activate
|
||||
the Privileged Role Administrator role and whether at least one approver is
|
||||
configured.
|
||||
|
||||
Returns:
|
||||
List[CheckReportM365]: A single report for the Privileged Role
|
||||
Administrator PIM role settings, or an empty list when the settings are
|
||||
absent.
|
||||
"""
|
||||
findings = []
|
||||
setting = entra_client.pim_role_approval_settings.get(
|
||||
PRIVILEGED_ROLE_ADMINISTRATOR_ROLE_TEMPLATE_ID
|
||||
)
|
||||
if not setting:
|
||||
return findings
|
||||
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=setting,
|
||||
resource_name="Privileged Role Administrator PIM Role Settings",
|
||||
resource_id=PRIVILEGED_ROLE_ADMINISTRATOR_ROLE_TEMPLATE_ID,
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"PIM does not require approval to activate the Privileged Role "
|
||||
"Administrator role."
|
||||
)
|
||||
|
||||
if setting.is_approval_required and not setting.has_approvers:
|
||||
report.status_extended = (
|
||||
"PIM requires approval to activate the Privileged Role Administrator "
|
||||
"role but no approvers are configured."
|
||||
)
|
||||
elif setting.is_approval_required and setting.has_approvers:
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
"PIM requires approval to activate the Privileged Role Administrator "
|
||||
"role and has approvers configured."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
@@ -105,6 +105,11 @@ class Entra(M365Service):
|
||||
self._get_device_registration_policy(),
|
||||
self._get_directory_settings(),
|
||||
self._get_b2b_collaboration_policy(),
|
||||
self._get_activity_based_timeout_policies(),
|
||||
self._get_named_locations(),
|
||||
self._get_authentication_methods_policy_settings(),
|
||||
self._get_pim_role_approval_settings(),
|
||||
self._get_access_review_definitions(),
|
||||
)
|
||||
)
|
||||
|
||||
@@ -130,6 +135,17 @@ class Entra(M365Service):
|
||||
)
|
||||
self.directory_settings: Dict[str, Dict[str, str]] = attributes[14]
|
||||
self.b2b_collaboration_policy: Optional[B2BCollaborationPolicy] = attributes[15]
|
||||
self.activity_based_timeout_policies: List[ActivityBasedTimeoutPolicy] = (
|
||||
attributes[16]
|
||||
)
|
||||
self.named_locations: List[NamedLocation] = attributes[17]
|
||||
self.authentication_methods_policy_settings: Optional[
|
||||
AuthenticationMethodsPolicySettings
|
||||
] = attributes[18]
|
||||
self.pim_role_approval_settings: Dict[str, PimRoleApprovalSetting] = attributes[
|
||||
19
|
||||
]
|
||||
self.access_review_definitions: List[AccessReviewDefinition] = attributes[20]
|
||||
self.user_accounts_status = {}
|
||||
|
||||
# Resolve directory-object identifiers referenced by Conditional Access
|
||||
@@ -394,6 +410,24 @@ class Entra(M365Service):
|
||||
authentication_flows=self._parse_authentication_flows(
|
||||
raw_auth_flows_map.get(policy.id)
|
||||
),
|
||||
locations=LocationsCondition(
|
||||
include_locations=list(
|
||||
getattr(
|
||||
getattr(policy.conditions, "locations", None),
|
||||
"include_locations",
|
||||
[],
|
||||
)
|
||||
or []
|
||||
),
|
||||
exclude_locations=list(
|
||||
getattr(
|
||||
getattr(policy.conditions, "locations", None),
|
||||
"exclude_locations",
|
||||
[],
|
||||
)
|
||||
or []
|
||||
),
|
||||
),
|
||||
device_conditions=DeviceConditions(
|
||||
device_filter_mode=(
|
||||
DeviceFilterMode(
|
||||
@@ -524,6 +558,19 @@ class Entra(M365Service):
|
||||
else False
|
||||
),
|
||||
),
|
||||
secure_sign_in_session_enabled=bool(
|
||||
getattr(
|
||||
getattr(
|
||||
policy.session_controls,
|
||||
"secure_sign_in_session",
|
||||
None,
|
||||
),
|
||||
"is_enabled",
|
||||
False,
|
||||
)
|
||||
if policy.session_controls
|
||||
else False
|
||||
),
|
||||
),
|
||||
state=ConditionalAccessPolicyState(
|
||||
getattr(policy, "state", "disabled")
|
||||
@@ -1247,6 +1294,167 @@ OAuthAppInfo
|
||||
)
|
||||
return device_registration_policy
|
||||
|
||||
async def _get_pim_role_approval_settings(self):
|
||||
"""Retrieve PIM approval-to-activate settings per directory role.
|
||||
|
||||
Fetches ``policies/roleManagementPolicyAssignments`` for directory roles with
|
||||
their expanded policy rules, and extracts, per role definition, whether
|
||||
approval is required to activate and whether approvers are configured.
|
||||
|
||||
Returns:
|
||||
Dict[str, PimRoleApprovalSetting]: Keyed by role definition (template) id.
|
||||
"""
|
||||
logger.info("Entra - Getting PIM role approval settings...")
|
||||
settings: Dict[str, PimRoleApprovalSetting] = {}
|
||||
try:
|
||||
url = (
|
||||
"https://graph.microsoft.com/v1.0/policies/"
|
||||
"roleManagementPolicyAssignments?$filter=scopeId%20eq%20'/'%20and%20"
|
||||
"scopeType%20eq%20'DirectoryRole'&$expand=policy($expand=rules)"
|
||||
)
|
||||
request_info = self.client.policies.with_url(
|
||||
url
|
||||
).to_get_request_information()
|
||||
assignments = []
|
||||
while True:
|
||||
response = await self.client.request_adapter.send_primitive_async(
|
||||
request_info, "bytes", {}
|
||||
)
|
||||
if not response:
|
||||
break
|
||||
data = json.loads(response)
|
||||
page = data.get("value", []) or []
|
||||
if not page:
|
||||
break
|
||||
assignments.extend(page)
|
||||
next_link = data.get("@odata.nextLink") or data.get("nextLink")
|
||||
if not next_link:
|
||||
break
|
||||
request_info = self.client.policies.with_url(
|
||||
next_link
|
||||
).to_get_request_information()
|
||||
for assignment in assignments:
|
||||
role_id = assignment.get("roleDefinitionId")
|
||||
if not role_id:
|
||||
continue
|
||||
rules = (assignment.get("policy", {}) or {}).get("rules", []) or []
|
||||
is_approval_required = False
|
||||
has_approvers = False
|
||||
for rule in rules:
|
||||
if rule.get("id") == "Approval_EndUser_Assignment":
|
||||
setting = rule.get("setting", {}) or {}
|
||||
is_approval_required = bool(
|
||||
setting.get("isApprovalRequired", False)
|
||||
)
|
||||
for stage in setting.get("approvalStages", []) or []:
|
||||
if stage.get("primaryApprovers"):
|
||||
has_approvers = True
|
||||
break
|
||||
settings[role_id] = PimRoleApprovalSetting(
|
||||
role_definition_id=role_id,
|
||||
is_approval_required=is_approval_required,
|
||||
has_approvers=has_approvers,
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
return settings
|
||||
|
||||
async def _get_access_review_definitions(self):
|
||||
"""Retrieve access review definitions from Microsoft Entra ID Governance.
|
||||
|
||||
Fetches ``identityGovernance/accessReviews/definitions`` and captures the
|
||||
status and scope query of each definition for the access-review checks.
|
||||
|
||||
Returns:
|
||||
List[AccessReviewDefinition]: The parsed access review definitions.
|
||||
"""
|
||||
logger.info("Entra - Getting access review definitions...")
|
||||
definitions = []
|
||||
try:
|
||||
url = (
|
||||
"https://graph.microsoft.com/v1.0/identityGovernance/"
|
||||
"accessReviews/definitions"
|
||||
)
|
||||
request_info = self.client.identity_governance.with_url(
|
||||
url
|
||||
).to_get_request_information()
|
||||
raw_definitions = []
|
||||
while True:
|
||||
response = await self.client.request_adapter.send_primitive_async(
|
||||
request_info, "bytes", {}
|
||||
)
|
||||
if not response:
|
||||
break
|
||||
data = json.loads(response)
|
||||
page = data.get("value", []) or []
|
||||
if not page:
|
||||
break
|
||||
raw_definitions.extend(page)
|
||||
next_link = data.get("@odata.nextLink") or data.get("nextLink")
|
||||
if not next_link:
|
||||
break
|
||||
request_info = self.client.identity_governance.with_url(
|
||||
next_link
|
||||
).to_get_request_information()
|
||||
for definition in raw_definitions:
|
||||
scope = definition.get("scope", {}) or {}
|
||||
settings = definition.get("settings", {}) or {}
|
||||
recurrence = settings.get("recurrence", {}) or {}
|
||||
recurrence_pattern = recurrence.get("pattern", {}) or {}
|
||||
recurrence_range = recurrence.get("range", {}) or {}
|
||||
stage_settings = definition.get("stageSettings")
|
||||
if stage_settings is not None:
|
||||
has_primary_reviewers = bool(stage_settings) and all(
|
||||
bool(stage.get("reviewers", []) or [])
|
||||
for stage in stage_settings
|
||||
)
|
||||
else:
|
||||
has_primary_reviewers = bool(definition.get("reviewers", []) or [])
|
||||
definitions.append(
|
||||
AccessReviewDefinition(
|
||||
id=definition.get("id", ""),
|
||||
display_name=definition.get("displayName"),
|
||||
status=definition.get("status"),
|
||||
scope_query=str(scope.get("query", "")),
|
||||
resource_scope_queries=[
|
||||
str(resource_scope.get("query", ""))
|
||||
for resource_scope in (
|
||||
scope.get("resourceScopes", []) or []
|
||||
)
|
||||
],
|
||||
principal_scope_queries=[
|
||||
str(principal_scope.get("query", ""))
|
||||
for principal_scope in (
|
||||
scope.get("principalScopes", []) or []
|
||||
)
|
||||
],
|
||||
default_decision=settings.get("defaultDecision"),
|
||||
default_decision_enabled=bool(
|
||||
settings.get("defaultDecisionEnabled", False)
|
||||
),
|
||||
auto_apply_enabled=bool(
|
||||
settings.get("autoApplyDecisionsEnabled", False)
|
||||
),
|
||||
mail_notifications_enabled=bool(
|
||||
settings.get("mailNotificationsEnabled", False)
|
||||
),
|
||||
reminders_enabled=bool(
|
||||
settings.get("reminderNotificationsEnabled", False)
|
||||
),
|
||||
duration_in_days=settings.get("instanceDurationInDays"),
|
||||
recurrence_pattern_type=recurrence_pattern.get("type"),
|
||||
recurrence_range_type=recurrence_range.get("type"),
|
||||
has_primary_reviewers=has_primary_reviewers,
|
||||
)
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
return definitions
|
||||
|
||||
async def _get_b2b_collaboration_policy(self):
|
||||
"""Retrieve the legacy B2B collaboration (invitation domains) policy.
|
||||
|
||||
@@ -1308,6 +1516,200 @@ OAuthAppInfo
|
||||
)
|
||||
return b2b_policy
|
||||
|
||||
async def _get_named_locations(self):
|
||||
"""Retrieve Conditional Access named locations from Microsoft Entra.
|
||||
|
||||
Fetches ``identity/conditionalAccess/namedLocations`` as raw JSON to handle
|
||||
the polymorphic ipNamedLocation / countryNamedLocation types and extract
|
||||
their trust and IP-range attributes.
|
||||
|
||||
Returns:
|
||||
List[NamedLocation]: The parsed named locations.
|
||||
"""
|
||||
logger.info("Entra - Getting named locations...")
|
||||
named_locations = []
|
||||
try:
|
||||
request_info = (
|
||||
self.client.identity.conditional_access.named_locations.to_get_request_information()
|
||||
)
|
||||
raw_locations = []
|
||||
while True:
|
||||
response = await self.client.request_adapter.send_primitive_async(
|
||||
request_info, "bytes", {}
|
||||
)
|
||||
if not response:
|
||||
break
|
||||
data = json.loads(response)
|
||||
page = data.get("value", []) or []
|
||||
raw_locations.extend(page)
|
||||
next_link = data.get("@odata.nextLink") or data.get("nextLink")
|
||||
if not next_link:
|
||||
break
|
||||
request_info = (
|
||||
self.client.identity.conditional_access.named_locations.with_url(
|
||||
next_link
|
||||
).to_get_request_information()
|
||||
)
|
||||
for location in raw_locations:
|
||||
odata_type = location.get("@odata.type", "")
|
||||
ip_ranges = location.get("ipRanges", []) or []
|
||||
named_locations.append(
|
||||
NamedLocation(
|
||||
id=location.get("id", ""),
|
||||
display_name=location.get("displayName"),
|
||||
is_trusted=bool(location.get("isTrusted", False)),
|
||||
is_ip_location="ipNamedLocation" in odata_type,
|
||||
ip_ranges_count=len(ip_ranges),
|
||||
)
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
return named_locations
|
||||
|
||||
@staticmethod
|
||||
def _parse_timespan_to_seconds(value) -> "Optional[int]":
|
||||
"""Parse a .NET TimeSpan string (``[d.]hh:mm:ss[.fffffff]``) to seconds.
|
||||
|
||||
Args:
|
||||
value: The TimeSpan string (e.g. "03:00:00" or "1.00:00:00").
|
||||
|
||||
Returns:
|
||||
The total number of seconds, or None if the value cannot be parsed.
|
||||
"""
|
||||
if not value or not isinstance(value, str):
|
||||
return None
|
||||
try:
|
||||
days = 0
|
||||
remainder = value.strip()
|
||||
if remainder.startswith("-"):
|
||||
return None
|
||||
head = remainder.split(":", 1)[0]
|
||||
# A dot in the hours component denotes the days separator (d.hh).
|
||||
if "." in head:
|
||||
day_part, remainder = remainder.split(".", 1)
|
||||
days = int(day_part)
|
||||
hours, minutes, seconds = remainder.split(":")
|
||||
# Seconds may carry fractional digits (ss.fffffff); truncate them.
|
||||
seconds = seconds.split(".")[0]
|
||||
hours = int(hours)
|
||||
minutes = int(minutes)
|
||||
seconds = int(seconds)
|
||||
if days < 0 or not 0 <= hours <= 23:
|
||||
return None
|
||||
if not 0 <= minutes <= 59 or not 0 <= seconds <= 59:
|
||||
return None
|
||||
return days * 86400 + hours * 3600 + minutes * 60 + seconds
|
||||
except (ValueError, AttributeError):
|
||||
return None
|
||||
|
||||
async def _get_activity_based_timeout_policies(self):
|
||||
"""Retrieve activity-based (idle session) timeout policies from Entra.
|
||||
|
||||
Fetches ``policies/activityBasedTimeoutPolicies`` and parses each policy's
|
||||
``definition`` JSON to extract the ``WebSessionIdleTimeout`` for the idle
|
||||
session timeout check.
|
||||
|
||||
Returns:
|
||||
List[ActivityBasedTimeoutPolicy]: The parsed timeout policies.
|
||||
"""
|
||||
logger.info("Entra - Getting activity based timeout policies...")
|
||||
policies = []
|
||||
try:
|
||||
policies_builder = self.client.policies.activity_based_timeout_policies
|
||||
response = await policies_builder.get()
|
||||
while response:
|
||||
for policy in getattr(response, "value", []) or []:
|
||||
idle_timeout_seconds = None
|
||||
for definition in getattr(policy, "definition", []) or []:
|
||||
try:
|
||||
parsed = json.loads(definition)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
app_policies = (
|
||||
parsed.get("ActivityBasedTimeoutPolicy", {}).get(
|
||||
"ApplicationPolicies", []
|
||||
)
|
||||
or []
|
||||
)
|
||||
for app_policy in app_policies:
|
||||
if app_policy.get("ApplicationId") != "default":
|
||||
continue
|
||||
seconds = self._parse_timespan_to_seconds(
|
||||
app_policy.get("WebSessionIdleTimeout")
|
||||
)
|
||||
if seconds is not None:
|
||||
idle_timeout_seconds = seconds
|
||||
break
|
||||
if idle_timeout_seconds is not None:
|
||||
break
|
||||
policies.append(
|
||||
ActivityBasedTimeoutPolicy(
|
||||
id=getattr(policy, "id", ""),
|
||||
display_name=getattr(policy, "display_name", None),
|
||||
web_session_idle_timeout_seconds=idle_timeout_seconds,
|
||||
)
|
||||
)
|
||||
next_link = getattr(response, "odata_next_link", None)
|
||||
if not next_link:
|
||||
break
|
||||
response = await policies_builder.with_url(next_link).get()
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
return policies
|
||||
|
||||
async def _get_authentication_methods_policy_settings(self):
|
||||
"""Retrieve Microsoft Authenticator settings from the authentication methods policy.
|
||||
|
||||
Fetches ``policies/authenticationMethodsPolicy`` from the v1.0 Graph endpoint
|
||||
and extracts the Microsoft Authenticator state and its ``featureSettings``
|
||||
(app-information / location-information states). Parsed from raw JSON because
|
||||
``featureSettings`` lives on the derived Microsoft Authenticator configuration
|
||||
type rather than the base configuration model.
|
||||
|
||||
Returns:
|
||||
Optional[AuthenticationMethodsPolicySettings]: Parsed settings, or None.
|
||||
"""
|
||||
logger.info("Entra - Getting authentication methods policy settings...")
|
||||
settings = None
|
||||
try:
|
||||
builder = self.client.policies.authentication_methods_policy.with_url(
|
||||
"https://graph.microsoft.com/v1.0/policies/authenticationMethodsPolicy"
|
||||
)
|
||||
request_info = builder.to_get_request_information()
|
||||
response = await self.client.request_adapter.send_primitive_async(
|
||||
request_info, "bytes", {}
|
||||
)
|
||||
if response:
|
||||
data = json.loads(response)
|
||||
authenticator = {}
|
||||
for config in data.get("authenticationMethodConfigurations", []) or []:
|
||||
if config.get("id") == "MicrosoftAuthenticator":
|
||||
authenticator = config
|
||||
break
|
||||
feature_settings = authenticator.get("featureSettings", {}) or {}
|
||||
settings = AuthenticationMethodsPolicySettings(
|
||||
authenticator_state=authenticator.get("state"),
|
||||
authenticator_display_app_information_state=(
|
||||
feature_settings.get("displayAppInformationRequiredState", {})
|
||||
or {}
|
||||
).get("state"),
|
||||
authenticator_display_location_information_state=(
|
||||
feature_settings.get(
|
||||
"displayLocationInformationRequiredState", {}
|
||||
)
|
||||
or {}
|
||||
).get("state"),
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
return settings
|
||||
|
||||
async def _get_directory_settings(self):
|
||||
"""Retrieve tenant directory (group) settings from Microsoft Entra.
|
||||
|
||||
@@ -1967,6 +2369,13 @@ class AuthenticationFlows(BaseModel):
|
||||
transfer_methods: List[TransferMethod] = []
|
||||
|
||||
|
||||
class LocationsCondition(BaseModel):
|
||||
"""Model representing location conditions for Conditional Access policies."""
|
||||
|
||||
include_locations: List[str] = []
|
||||
exclude_locations: List[str] = []
|
||||
|
||||
|
||||
class Conditions(BaseModel):
|
||||
"""Model representing conditions for Conditional Access policies."""
|
||||
|
||||
@@ -1979,6 +2388,7 @@ class Conditions(BaseModel):
|
||||
platform_conditions: Optional[PlatformConditions] = None
|
||||
authentication_flows: Optional[AuthenticationFlows] = None
|
||||
device_conditions: Optional[DeviceConditions] = None
|
||||
locations: Optional[LocationsCondition] = None
|
||||
|
||||
|
||||
class PersistentBrowser(BaseModel):
|
||||
@@ -2015,6 +2425,7 @@ class SessionControls(BaseModel):
|
||||
persistent_browser: PersistentBrowser
|
||||
sign_in_frequency: SignInFrequency
|
||||
application_enforced_restrictions: Optional[ApplicationEnforcedRestrictions] = None
|
||||
secure_sign_in_session_enabled: bool = False
|
||||
|
||||
|
||||
class ConditionalAccessGrantControl(Enum):
|
||||
@@ -2076,6 +2487,10 @@ class AuthorizationPolicy(BaseModel):
|
||||
GROUP_UNIFIED_SETTINGS_TEMPLATE_ID = "62375ab9-6b52-47ed-826b-58e47e0e304b"
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID = "5cf42378-d67d-4f36-ba46-e8b86229381d"
|
||||
|
||||
# Well-known directory role template IDs.
|
||||
GLOBAL_ADMINISTRATOR_ROLE_TEMPLATE_ID = "62e90394-69f5-4237-9190-012177145e10"
|
||||
PRIVILEGED_ROLE_ADMINISTRATOR_ROLE_TEMPLATE_ID = "e8611ab8-c189-46e8-94e1-60213ab1f814"
|
||||
|
||||
|
||||
class DeviceRegistrationMembershipType(str, Enum):
|
||||
"""OData types for Entra device registration membership settings."""
|
||||
@@ -2095,6 +2510,52 @@ class DeviceRegistrationPolicy(BaseModel):
|
||||
local_admin_password_enabled: Optional[bool] = None
|
||||
|
||||
|
||||
class ActivityBasedTimeoutPolicy(BaseModel):
|
||||
"""Activity-based (idle session) timeout policy."""
|
||||
|
||||
id: str
|
||||
display_name: Optional[str] = None
|
||||
web_session_idle_timeout_seconds: Optional[int] = None
|
||||
|
||||
|
||||
class NamedLocation(BaseModel):
|
||||
"""Conditional Access named location."""
|
||||
|
||||
id: str
|
||||
display_name: Optional[str] = None
|
||||
is_trusted: bool = False
|
||||
is_ip_location: bool = False
|
||||
ip_ranges_count: int = 0
|
||||
|
||||
|
||||
class PimRoleApprovalSetting(BaseModel):
|
||||
"""PIM approval-to-activate setting for a directory role."""
|
||||
|
||||
role_definition_id: str
|
||||
is_approval_required: bool = False
|
||||
has_approvers: bool = False
|
||||
|
||||
|
||||
class AccessReviewDefinition(BaseModel):
|
||||
"""Access review definition (identityGovernance/accessReviews/definitions)."""
|
||||
|
||||
id: str
|
||||
display_name: Optional[str] = None
|
||||
status: Optional[str] = None
|
||||
scope_query: str = ""
|
||||
resource_scope_queries: List[str] = []
|
||||
principal_scope_queries: List[str] = []
|
||||
default_decision: Optional[str] = None
|
||||
default_decision_enabled: bool = False
|
||||
auto_apply_enabled: bool = False
|
||||
mail_notifications_enabled: bool = False
|
||||
reminders_enabled: bool = False
|
||||
duration_in_days: Optional[int] = None
|
||||
recurrence_pattern_type: Optional[str] = None
|
||||
recurrence_range_type: Optional[str] = None
|
||||
has_primary_reviewers: bool = False
|
||||
|
||||
|
||||
class B2BCollaborationPolicy(BaseModel):
|
||||
"""Legacy B2B collaboration (invitation domains) policy."""
|
||||
|
||||
@@ -2102,6 +2563,14 @@ class B2BCollaborationPolicy(BaseModel):
|
||||
allowed_domains: List[str] = []
|
||||
|
||||
|
||||
class AuthenticationMethodsPolicySettings(BaseModel):
|
||||
"""Microsoft Authenticator settings from the authentication methods policy."""
|
||||
|
||||
authenticator_state: Optional[str] = None
|
||||
authenticator_display_app_information_state: Optional[str] = None
|
||||
authenticator_display_location_information_state: Optional[str] = None
|
||||
|
||||
|
||||
class Organization(BaseModel):
|
||||
id: str
|
||||
name: str
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.lib.check.models import Severity
|
||||
from prowler.providers.github.services.organization.organization_service import Org
|
||||
from tests.providers.github.github_fixtures import set_mocked_github_provider
|
||||
|
||||
@@ -58,6 +59,7 @@ class Test_organization_repository_creation_limited:
|
||||
assert len(result) == 1
|
||||
assert result[0].resource_name == org_name
|
||||
assert result[0].status == "PASS"
|
||||
assert result[0].check_metadata.Severity == Severity.high
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"Organization {org_name} has disabled repository creation for members."
|
||||
@@ -94,6 +96,7 @@ class Test_organization_repository_creation_limited:
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].check_metadata.Severity == Severity.high
|
||||
assert "public repositories" in result[0].status_extended
|
||||
assert "repositories of any type" in result[0].status_extended
|
||||
|
||||
@@ -129,6 +132,7 @@ class Test_organization_repository_creation_limited:
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert result[0].check_metadata.Severity == Severity.high
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"Organization {org_name} has disabled repository creation for members."
|
||||
@@ -164,7 +168,410 @@ class Test_organization_repository_creation_limited:
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert result[0].check_metadata.Severity == Severity.high
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"Organization {org_name} has disabled repository creation for members."
|
||||
)
|
||||
|
||||
def test_repository_creation_type_none_with_private_enabled(self):
|
||||
organization_client = mock.MagicMock
|
||||
org_name = "test-organization"
|
||||
organization_client.organizations = {
|
||||
1: Org(
|
||||
id=1,
|
||||
name=org_name,
|
||||
mfa_required=None,
|
||||
members_can_create_repositories=True,
|
||||
members_allowed_repository_creation_type="none",
|
||||
members_can_create_private_repositories=True,
|
||||
),
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_github_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited.organization_client",
|
||||
new=organization_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited import (
|
||||
organization_repository_creation_limited,
|
||||
)
|
||||
|
||||
check = organization_repository_creation_limited()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].check_metadata.Severity == Severity.low
|
||||
assert "repositories of any type" not in result[0].status_extended
|
||||
assert "private repositories" in result[0].status_extended
|
||||
assert (
|
||||
"Public repository creation is disabled." in result[0].status_extended
|
||||
)
|
||||
|
||||
def test_repository_creation_public_flag_only(self):
|
||||
organization_client = mock.MagicMock
|
||||
org_name = "test-organization"
|
||||
organization_client.organizations = {
|
||||
1: Org(
|
||||
id=1,
|
||||
name=org_name,
|
||||
mfa_required=None,
|
||||
members_can_create_public_repositories=True,
|
||||
),
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_github_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited.organization_client",
|
||||
new=organization_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited import (
|
||||
organization_repository_creation_limited,
|
||||
)
|
||||
|
||||
check = organization_repository_creation_limited()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].check_metadata.Severity == Severity.high
|
||||
|
||||
def test_repository_creation_private_only(self):
|
||||
organization_client = mock.MagicMock
|
||||
org_name = "test-organization"
|
||||
organization_client.organizations = {
|
||||
1: Org(
|
||||
id=1,
|
||||
name=org_name,
|
||||
mfa_required=None,
|
||||
members_can_create_repositories=True,
|
||||
members_can_create_public_repositories=False,
|
||||
members_can_create_private_repositories=True,
|
||||
members_can_create_internal_repositories=False,
|
||||
),
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_github_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited.organization_client",
|
||||
new=organization_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited import (
|
||||
organization_repository_creation_limited,
|
||||
)
|
||||
|
||||
check = organization_repository_creation_limited()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].check_metadata.Severity == Severity.low
|
||||
assert (
|
||||
"Public repository creation is disabled." in result[0].status_extended
|
||||
)
|
||||
assert "repositories of any type" not in result[0].status_extended
|
||||
|
||||
def test_repository_creation_public_flag_unknown_is_not_downgraded(self):
|
||||
organization_client = mock.MagicMock
|
||||
org_name = "test-organization"
|
||||
organization_client.organizations = {
|
||||
1: Org(
|
||||
id=1,
|
||||
name=org_name,
|
||||
mfa_required=None,
|
||||
members_can_create_private_repositories=True,
|
||||
),
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_github_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited.organization_client",
|
||||
new=organization_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited import (
|
||||
organization_repository_creation_limited,
|
||||
)
|
||||
|
||||
check = organization_repository_creation_limited()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].check_metadata.Severity == Severity.high
|
||||
assert (
|
||||
"Public repository creation is disabled."
|
||||
not in result[0].status_extended
|
||||
)
|
||||
|
||||
def test_repository_creation_internal_only(self):
|
||||
organization_client = mock.MagicMock
|
||||
org_name = "test-organization"
|
||||
organization_client.organizations = {
|
||||
1: Org(
|
||||
id=1,
|
||||
name=org_name,
|
||||
mfa_required=None,
|
||||
members_can_create_public_repositories=False,
|
||||
members_can_create_private_repositories=False,
|
||||
members_can_create_internal_repositories=True,
|
||||
),
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_github_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited.organization_client",
|
||||
new=organization_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited import (
|
||||
organization_repository_creation_limited,
|
||||
)
|
||||
|
||||
check = organization_repository_creation_limited()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].check_metadata.Severity == Severity.low
|
||||
assert (
|
||||
"Public repository creation is disabled." in result[0].status_extended
|
||||
)
|
||||
|
||||
def test_repository_creation_private_and_internal(self):
|
||||
organization_client = mock.MagicMock
|
||||
org_name = "test-organization"
|
||||
organization_client.organizations = {
|
||||
1: Org(
|
||||
id=1,
|
||||
name=org_name,
|
||||
mfa_required=None,
|
||||
members_can_create_public_repositories=False,
|
||||
members_can_create_private_repositories=True,
|
||||
members_can_create_internal_repositories=True,
|
||||
),
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_github_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited.organization_client",
|
||||
new=organization_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited import (
|
||||
organization_repository_creation_limited,
|
||||
)
|
||||
|
||||
check = organization_repository_creation_limited()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].check_metadata.Severity == Severity.low
|
||||
assert (
|
||||
"Public repository creation is disabled." in result[0].status_extended
|
||||
)
|
||||
|
||||
def test_repository_creation_type_all(self):
|
||||
organization_client = mock.MagicMock
|
||||
org_name = "test-organization"
|
||||
organization_client.organizations = {
|
||||
1: Org(
|
||||
id=1,
|
||||
name=org_name,
|
||||
mfa_required=None,
|
||||
members_allowed_repository_creation_type="all",
|
||||
),
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_github_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited.organization_client",
|
||||
new=organization_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited import (
|
||||
organization_repository_creation_limited,
|
||||
)
|
||||
|
||||
check = organization_repository_creation_limited()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].check_metadata.Severity == Severity.high
|
||||
|
||||
def test_repository_creation_type_private(self):
|
||||
organization_client = mock.MagicMock
|
||||
org_name = "test-organization"
|
||||
organization_client.organizations = {
|
||||
1: Org(
|
||||
id=1,
|
||||
name=org_name,
|
||||
mfa_required=None,
|
||||
members_allowed_repository_creation_type="private",
|
||||
),
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_github_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited.organization_client",
|
||||
new=organization_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited import (
|
||||
organization_repository_creation_limited,
|
||||
)
|
||||
|
||||
check = organization_repository_creation_limited()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].check_metadata.Severity == Severity.low
|
||||
assert (
|
||||
"Public repository creation is disabled." in result[0].status_extended
|
||||
)
|
||||
|
||||
def test_repository_creation_type_selected_is_undeterminable(self):
|
||||
organization_client = mock.MagicMock
|
||||
org_name = "test-organization"
|
||||
organization_client.organizations = {
|
||||
1: Org(
|
||||
id=1,
|
||||
name=org_name,
|
||||
mfa_required=None,
|
||||
members_allowed_repository_creation_type="selected",
|
||||
),
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_github_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited.organization_client",
|
||||
new=organization_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited import (
|
||||
organization_repository_creation_limited,
|
||||
)
|
||||
|
||||
check = organization_repository_creation_limited()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].check_metadata.Severity == Severity.high
|
||||
|
||||
def test_repository_creation_global_flag_only(self):
|
||||
organization_client = mock.MagicMock
|
||||
org_name = "test-organization"
|
||||
organization_client.organizations = {
|
||||
1: Org(
|
||||
id=1,
|
||||
name=org_name,
|
||||
mfa_required=None,
|
||||
members_can_create_repositories=True,
|
||||
),
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_github_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited.organization_client",
|
||||
new=organization_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited import (
|
||||
organization_repository_creation_limited,
|
||||
)
|
||||
|
||||
check = organization_repository_creation_limited()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].check_metadata.Severity == Severity.high
|
||||
|
||||
def test_repository_creation_severity_does_not_leak_between_organizations(self):
|
||||
organization_client = mock.MagicMock
|
||||
public_org_name = "public-organization"
|
||||
private_org_name = "private-organization"
|
||||
organization_client.organizations = {
|
||||
1: Org(
|
||||
id=1,
|
||||
name=public_org_name,
|
||||
mfa_required=None,
|
||||
members_can_create_public_repositories=True,
|
||||
),
|
||||
2: Org(
|
||||
id=2,
|
||||
name=private_org_name,
|
||||
mfa_required=None,
|
||||
members_can_create_public_repositories=False,
|
||||
members_can_create_private_repositories=True,
|
||||
members_can_create_internal_repositories=False,
|
||||
),
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_github_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited.organization_client",
|
||||
new=organization_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.github.services.organization.organization_repository_creation_limited.organization_repository_creation_limited import (
|
||||
organization_repository_creation_limited,
|
||||
)
|
||||
|
||||
check = organization_repository_creation_limited()
|
||||
result = check.execute()
|
||||
assert len(result) == 2
|
||||
|
||||
results_by_name = {r.resource_name: r for r in result}
|
||||
assert results_by_name[public_org_name].status == "FAIL"
|
||||
assert (
|
||||
results_by_name[public_org_name].check_metadata.Severity
|
||||
== Severity.high
|
||||
)
|
||||
assert results_by_name[private_org_name].status == "FAIL"
|
||||
assert (
|
||||
results_by_name[private_org_name].check_metadata.Severity
|
||||
== Severity.low
|
||||
)
|
||||
|
||||
@@ -0,0 +1,147 @@
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
AccessReviewDefinition,
|
||||
)
|
||||
from tests.providers.m365.m365_fixtures import set_mocked_m365_provider
|
||||
|
||||
CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_access_review_guest_users_configured.entra_access_review_guest_users_configured"
|
||||
|
||||
|
||||
def _definition(
|
||||
status="InProgress",
|
||||
scope_query="/users?$filter=(userType eq 'Guest')",
|
||||
principal_scope_queries=None,
|
||||
default_decision="Deny",
|
||||
default_decision_enabled=True,
|
||||
auto_apply_enabled=True,
|
||||
mail_notifications_enabled=True,
|
||||
reminders_enabled=True,
|
||||
recurrence_pattern_type="weekly",
|
||||
recurrence_range_type="noEnd",
|
||||
has_primary_reviewers=True,
|
||||
):
|
||||
definition = {
|
||||
"id": "ar1",
|
||||
"display_name": "Guest Review",
|
||||
"status": status,
|
||||
"scope_query": scope_query,
|
||||
"principal_scope_queries": principal_scope_queries or [],
|
||||
"default_decision": default_decision,
|
||||
"auto_apply_enabled": auto_apply_enabled,
|
||||
"mail_notifications_enabled": mail_notifications_enabled,
|
||||
"reminders_enabled": reminders_enabled,
|
||||
"recurrence_pattern_type": recurrence_pattern_type,
|
||||
"recurrence_range_type": recurrence_range_type,
|
||||
"has_primary_reviewers": has_primary_reviewers,
|
||||
}
|
||||
if default_decision_enabled is not None:
|
||||
definition["default_decision_enabled"] = default_decision_enabled
|
||||
return AccessReviewDefinition(**definition)
|
||||
|
||||
|
||||
class Test_entra_access_review_guest_users_configured:
|
||||
def _run(self, definitions):
|
||||
entra_client = mock.MagicMock
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(f"{CHECK_MODULE_PATH}.entra_client", new=entra_client),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_access_review_guest_users_configured.entra_access_review_guest_users_configured import (
|
||||
entra_access_review_guest_users_configured,
|
||||
)
|
||||
|
||||
entra_client.access_review_definitions = definitions
|
||||
return entra_access_review_guest_users_configured().execute()
|
||||
|
||||
def test_no_definitions(self):
|
||||
result = self._run([])
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_active_failclosed_guest_review(self):
|
||||
result = self._run([_definition()])
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== "Access review 'Guest Review' for guest users is active, recurring, "
|
||||
"reviewer-assigned, and fail-closed."
|
||||
)
|
||||
|
||||
def test_not_active(self):
|
||||
result = self._run([_definition(status="Completed")])
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_not_fail_closed(self):
|
||||
# Active guest review but does nothing on non-response -> FAIL.
|
||||
result = self._run(
|
||||
[_definition(default_decision="None", auto_apply_enabled=False)]
|
||||
)
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
@pytest.mark.parametrize("default_decision_enabled", [False, None])
|
||||
def test_default_decision_not_enabled(self, default_decision_enabled):
|
||||
result = self._run(
|
||||
[_definition(default_decision_enabled=default_decision_enabled)]
|
||||
)
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_not_guest_scope(self):
|
||||
result = self._run([_definition(scope_query="/roleManagement/directory")])
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_guest_filter_in_principal_scopes(self):
|
||||
# Portal-created reviews keep the guest filter in principalScopes and
|
||||
# leave the top-level scope query empty.
|
||||
result = self._run(
|
||||
[
|
||||
_definition(
|
||||
scope_query="",
|
||||
principal_scope_queries=["/users?$filter=(userType eq 'Guest')"],
|
||||
)
|
||||
]
|
||||
)
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
def test_url_encoded_guest_filter(self):
|
||||
result = self._run(
|
||||
[_definition(scope_query="/users?$filter=userType%20eq%20%27Guest%27")]
|
||||
)
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
def test_mixed_case_guest_filter(self):
|
||||
result = self._run(
|
||||
[_definition(scope_query="/users?$filter=(USERTYPE EQ 'guest')")]
|
||||
)
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"scope_query",
|
||||
[
|
||||
"/users?$filter=not(userType eq 'Guest')",
|
||||
"/users?$filter=(userType ne 'Guest')",
|
||||
"/users?$filter=displayName eq 'Guest account'",
|
||||
"/groups/guest-review-members",
|
||||
],
|
||||
)
|
||||
def test_incidental_guest_text_does_not_target_guests(self, scope_query):
|
||||
result = self._run([_definition(scope_query=scope_query)])
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"definition_overrides",
|
||||
[
|
||||
{"recurrence_pattern_type": None},
|
||||
{"recurrence_pattern_type": "daily"},
|
||||
{"recurrence_range_type": None},
|
||||
{"recurrence_range_type": "endDate"},
|
||||
{"has_primary_reviewers": False},
|
||||
],
|
||||
)
|
||||
def test_invalid_recurrence_or_missing_reviewers(self, definition_overrides):
|
||||
result = self._run([_definition(**definition_overrides)])
|
||||
assert result[0].status == "FAIL"
|
||||
@@ -0,0 +1,97 @@
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
AccessReviewDefinition,
|
||||
)
|
||||
from tests.providers.m365.m365_fixtures import set_mocked_m365_provider
|
||||
|
||||
CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_access_review_privileged_roles_configured.entra_access_review_privileged_roles_configured"
|
||||
|
||||
|
||||
def _definition(
|
||||
status="InProgress",
|
||||
scope_query="",
|
||||
resource_scope_queries=None,
|
||||
default_decision="None",
|
||||
auto_apply_enabled=True,
|
||||
mail_notifications_enabled=True,
|
||||
reminders_enabled=True,
|
||||
recurrence_pattern_type="absoluteMonthly",
|
||||
recurrence_range_type="noEnd",
|
||||
has_primary_reviewers=True,
|
||||
):
|
||||
return AccessReviewDefinition(
|
||||
id="ar1",
|
||||
display_name="Privileged Roles Review",
|
||||
status=status,
|
||||
scope_query=scope_query,
|
||||
resource_scope_queries=(
|
||||
resource_scope_queries
|
||||
if resource_scope_queries is not None
|
||||
else ["/roleManagement/directory/roleDefinitions/62e90394-..."]
|
||||
),
|
||||
default_decision=default_decision,
|
||||
auto_apply_enabled=auto_apply_enabled,
|
||||
mail_notifications_enabled=mail_notifications_enabled,
|
||||
reminders_enabled=reminders_enabled,
|
||||
recurrence_pattern_type=recurrence_pattern_type,
|
||||
recurrence_range_type=recurrence_range_type,
|
||||
has_primary_reviewers=has_primary_reviewers,
|
||||
)
|
||||
|
||||
|
||||
class Test_entra_access_review_privileged_roles_configured:
|
||||
def _run(self, definitions):
|
||||
entra_client = mock.MagicMock
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(f"{CHECK_MODULE_PATH}.entra_client", new=entra_client),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_access_review_privileged_roles_configured.entra_access_review_privileged_roles_configured import (
|
||||
entra_access_review_privileged_roles_configured,
|
||||
)
|
||||
|
||||
entra_client.access_review_definitions = definitions
|
||||
return entra_access_review_privileged_roles_configured().execute()
|
||||
|
||||
def test_no_definitions(self):
|
||||
assert self._run([])[0].status == "FAIL"
|
||||
|
||||
def test_active_fail_closed_privileged(self):
|
||||
# Role reference lives in resource scopes (not top-level scope.query).
|
||||
result = self._run([_definition()])
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
def test_deny_default_decision(self):
|
||||
result = self._run([_definition(default_decision="Deny")])
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_guest_review_ignored(self):
|
||||
result = self._run(
|
||||
[
|
||||
_definition(
|
||||
scope_query="/users?$filter=(userType eq 'Guest')",
|
||||
resource_scope_queries=[],
|
||||
)
|
||||
]
|
||||
)
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"definition_overrides",
|
||||
[
|
||||
{"recurrence_pattern_type": None},
|
||||
{"recurrence_pattern_type": "daily"},
|
||||
{"recurrence_range_type": None},
|
||||
{"recurrence_range_type": "numbered"},
|
||||
{"has_primary_reviewers": False},
|
||||
],
|
||||
)
|
||||
def test_invalid_recurrence_or_missing_reviewers(self, definition_overrides):
|
||||
result = self._run([_definition(**definition_overrides)])
|
||||
assert result[0].status == "FAIL"
|
||||
@@ -0,0 +1,59 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
AuthenticationMethodsPolicySettings,
|
||||
)
|
||||
from tests.providers.m365.m365_fixtures import set_mocked_m365_provider
|
||||
|
||||
CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_authentication_method_authenticator_show_context.entra_authentication_method_authenticator_show_context"
|
||||
|
||||
|
||||
class Test_entra_authentication_method_authenticator_show_context:
|
||||
def _run(self, settings):
|
||||
entra_client = mock.MagicMock
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(f"{CHECK_MODULE_PATH}.entra_client", new=entra_client),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_authentication_method_authenticator_show_context.entra_authentication_method_authenticator_show_context import (
|
||||
entra_authentication_method_authenticator_show_context,
|
||||
)
|
||||
|
||||
entra_client.authentication_methods_policy_settings = settings
|
||||
return entra_authentication_method_authenticator_show_context().execute()
|
||||
|
||||
def test_no_settings(self):
|
||||
assert self._run(None) == []
|
||||
|
||||
def test_both_enabled(self):
|
||||
result = self._run(
|
||||
AuthenticationMethodsPolicySettings(
|
||||
authenticator_state="enabled",
|
||||
authenticator_display_app_information_state="enabled",
|
||||
authenticator_display_location_information_state="enabled",
|
||||
)
|
||||
)
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
def test_location_disabled(self):
|
||||
result = self._run(
|
||||
AuthenticationMethodsPolicySettings(
|
||||
authenticator_state="enabled",
|
||||
authenticator_display_app_information_state="enabled",
|
||||
authenticator_display_location_information_state="disabled",
|
||||
)
|
||||
)
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_context_enabled_but_authenticator_disabled(self):
|
||||
result = self._run(
|
||||
AuthenticationMethodsPolicySettings(
|
||||
authenticator_state="disabled",
|
||||
authenticator_display_app_information_state="enabled",
|
||||
authenticator_display_location_information_state="enabled",
|
||||
)
|
||||
)
|
||||
assert result[0].status == "FAIL"
|
||||
@@ -0,0 +1,139 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
AuthenticationMethodConfiguration,
|
||||
)
|
||||
from tests.providers.m365.m365_fixtures import DOMAIN, set_mocked_m365_provider
|
||||
|
||||
|
||||
class Test_entra_authentication_method_email_otp_disabled:
|
||||
def test_no_configurations(self):
|
||||
entra_client = mock.MagicMock
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.m365.services.entra.entra_authentication_method_email_otp_disabled.entra_authentication_method_email_otp_disabled.entra_client",
|
||||
new=entra_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_authentication_method_email_otp_disabled.entra_authentication_method_email_otp_disabled import (
|
||||
entra_authentication_method_email_otp_disabled,
|
||||
)
|
||||
|
||||
entra_client.authentication_method_configurations = {}
|
||||
entra_client.tenant_domain = DOMAIN
|
||||
|
||||
check = entra_authentication_method_email_otp_disabled()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 0
|
||||
|
||||
def test_email_otp_enabled(self):
|
||||
entra_client = mock.MagicMock
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.m365.services.entra.entra_authentication_method_email_otp_disabled.entra_authentication_method_email_otp_disabled.entra_client",
|
||||
new=entra_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_authentication_method_email_otp_disabled.entra_authentication_method_email_otp_disabled import (
|
||||
entra_authentication_method_email_otp_disabled,
|
||||
)
|
||||
|
||||
entra_client.authentication_method_configurations = {
|
||||
"Email": AuthenticationMethodConfiguration(
|
||||
id="Email",
|
||||
state="enabled",
|
||||
),
|
||||
}
|
||||
entra_client.tenant_domain = DOMAIN
|
||||
|
||||
check = entra_authentication_method_email_otp_disabled()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== "Email OTP authentication method is enabled in the tenant."
|
||||
)
|
||||
|
||||
def test_email_otp_disabled(self):
|
||||
entra_client = mock.MagicMock
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.m365.services.entra.entra_authentication_method_email_otp_disabled.entra_authentication_method_email_otp_disabled.entra_client",
|
||||
new=entra_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_authentication_method_email_otp_disabled.entra_authentication_method_email_otp_disabled import (
|
||||
entra_authentication_method_email_otp_disabled,
|
||||
)
|
||||
|
||||
entra_client.authentication_method_configurations = {
|
||||
"Email": AuthenticationMethodConfiguration(
|
||||
id="Email",
|
||||
state="disabled",
|
||||
),
|
||||
}
|
||||
entra_client.tenant_domain = DOMAIN
|
||||
|
||||
check = entra_authentication_method_email_otp_disabled()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== "Email OTP authentication method is disabled in the tenant."
|
||||
)
|
||||
|
||||
def test_email_otp_unknown_state(self):
|
||||
entra_client = mock.MagicMock
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.m365.services.entra.entra_authentication_method_email_otp_disabled.entra_authentication_method_email_otp_disabled.entra_client",
|
||||
new=entra_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_authentication_method_email_otp_disabled.entra_authentication_method_email_otp_disabled import (
|
||||
entra_authentication_method_email_otp_disabled,
|
||||
)
|
||||
|
||||
entra_client.authentication_method_configurations = {
|
||||
"Email": AuthenticationMethodConfiguration(
|
||||
id="Email",
|
||||
state="unknown",
|
||||
),
|
||||
}
|
||||
entra_client.tenant_domain = DOMAIN
|
||||
|
||||
check = entra_authentication_method_email_otp_disabled()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== "Email OTP authentication method state could not be determined; "
|
||||
"treating as enabled/non-compliant."
|
||||
)
|
||||
@@ -0,0 +1,149 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
ApplicationsConditions,
|
||||
AuthenticationFlows,
|
||||
ConditionalAccessGrantControl,
|
||||
ConditionalAccessPolicy,
|
||||
ConditionalAccessPolicyState,
|
||||
Conditions,
|
||||
GrantControlOperator,
|
||||
GrantControls,
|
||||
PersistentBrowser,
|
||||
SessionControls,
|
||||
SignInFrequency,
|
||||
TransferMethod,
|
||||
UsersConditions,
|
||||
)
|
||||
from tests.providers.m365.m365_fixtures import DOMAIN, set_mocked_m365_provider
|
||||
|
||||
CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_conditional_access_policy_authentication_transfer_blocked.entra_conditional_access_policy_authentication_transfer_blocked"
|
||||
|
||||
|
||||
def _make_policy(
|
||||
policy_id="policy-1",
|
||||
display_name="Block Authentication Transfer",
|
||||
state=ConditionalAccessPolicyState.ENABLED,
|
||||
included_users=None,
|
||||
included_applications=None,
|
||||
excluded_applications=None,
|
||||
transfer_methods=None,
|
||||
built_in_controls=None,
|
||||
):
|
||||
return ConditionalAccessPolicy(
|
||||
id=policy_id,
|
||||
display_name=display_name,
|
||||
conditions=Conditions(
|
||||
application_conditions=ApplicationsConditions(
|
||||
included_applications=included_applications or ["All"],
|
||||
excluded_applications=excluded_applications or [],
|
||||
included_user_actions=[],
|
||||
),
|
||||
user_conditions=UsersConditions(
|
||||
included_groups=[],
|
||||
excluded_groups=[],
|
||||
included_users=included_users or ["All"],
|
||||
excluded_users=[],
|
||||
included_roles=[],
|
||||
excluded_roles=[],
|
||||
),
|
||||
client_app_types=[],
|
||||
authentication_flows=AuthenticationFlows(
|
||||
transfer_methods=(
|
||||
transfer_methods
|
||||
if transfer_methods is not None
|
||||
else [TransferMethod.AUTHENTICATION_TRANSFER]
|
||||
)
|
||||
),
|
||||
),
|
||||
grant_controls=GrantControls(
|
||||
built_in_controls=(
|
||||
built_in_controls
|
||||
if built_in_controls is not None
|
||||
else [ConditionalAccessGrantControl.BLOCK]
|
||||
),
|
||||
operator=GrantControlOperator.OR,
|
||||
authentication_strength=None,
|
||||
),
|
||||
session_controls=SessionControls(
|
||||
persistent_browser=PersistentBrowser(is_enabled=False, mode="always"),
|
||||
sign_in_frequency=SignInFrequency(
|
||||
is_enabled=False, frequency=None, type=None, interval=None
|
||||
),
|
||||
),
|
||||
state=state,
|
||||
)
|
||||
|
||||
|
||||
class Test_entra_conditional_access_policy_authentication_transfer_blocked:
|
||||
def _run(self, policies):
|
||||
entra_client = mock.MagicMock
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(f"{CHECK_MODULE_PATH}.entra_client", new=entra_client),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_conditional_access_policy_authentication_transfer_blocked.entra_conditional_access_policy_authentication_transfer_blocked import (
|
||||
entra_conditional_access_policy_authentication_transfer_blocked,
|
||||
)
|
||||
|
||||
entra_client.conditional_access_policies = policies
|
||||
entra_client.tenant_domain = DOMAIN
|
||||
return (
|
||||
entra_conditional_access_policy_authentication_transfer_blocked().execute()
|
||||
)
|
||||
|
||||
def test_no_resources(self):
|
||||
result = self._run({})
|
||||
assert len(result) == 0
|
||||
|
||||
def test_policy_blocks_authentication_transfer(self):
|
||||
policy = _make_policy()
|
||||
result = self._run({policy.id: policy})
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== "Conditional Access Policy 'Block Authentication Transfer' blocks authentication transfer."
|
||||
)
|
||||
|
||||
def test_multiple_policies_each_produce_a_report(self):
|
||||
blocking_policy = _make_policy(policy_id="policy-blocking")
|
||||
non_blocking_policy = _make_policy(
|
||||
policy_id="policy-non-blocking",
|
||||
display_name="Allow Authentication Transfer",
|
||||
built_in_controls=[ConditionalAccessGrantControl.MFA],
|
||||
)
|
||||
|
||||
result = self._run(
|
||||
{
|
||||
blocking_policy.id: blocking_policy,
|
||||
non_blocking_policy.id: non_blocking_policy,
|
||||
}
|
||||
)
|
||||
|
||||
assert len(result) == 2
|
||||
assert [report.resource_id for report in result] == [
|
||||
"policy-blocking",
|
||||
"policy-non-blocking",
|
||||
]
|
||||
assert [report.status for report in result] == ["PASS", "FAIL"]
|
||||
|
||||
def test_policy_excluding_application_does_not_cover_all_applications(self):
|
||||
policy = _make_policy(excluded_applications=["excluded-app"])
|
||||
result = self._run({policy.id: policy})
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_policy_report_only(self):
|
||||
policy = _make_policy(state=ConditionalAccessPolicyState.ENABLED_FOR_REPORTING)
|
||||
result = self._run({policy.id: policy})
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_policy_does_not_block(self):
|
||||
policy = _make_policy(built_in_controls=[ConditionalAccessGrantControl.MFA])
|
||||
result = self._run({policy.id: policy})
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
@@ -0,0 +1,119 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
ApplicationsConditions,
|
||||
ConditionalAccessGrantControl,
|
||||
ConditionalAccessPolicy,
|
||||
ConditionalAccessPolicyState,
|
||||
Conditions,
|
||||
GrantControlOperator,
|
||||
GrantControls,
|
||||
PersistentBrowser,
|
||||
RiskLevel,
|
||||
SessionControls,
|
||||
SignInFrequency,
|
||||
UsersConditions,
|
||||
)
|
||||
from tests.providers.m365.m365_fixtures import DOMAIN, set_mocked_m365_provider
|
||||
|
||||
CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_conditional_access_policy_block_high_medium_sign_in_risk.entra_conditional_access_policy_block_high_medium_sign_in_risk"
|
||||
|
||||
|
||||
def _make_policy(
|
||||
policy_id="policy-1",
|
||||
display_name="Block Risky Sign-ins",
|
||||
state=ConditionalAccessPolicyState.ENABLED,
|
||||
included_users=None,
|
||||
included_applications=None,
|
||||
sign_in_risk_levels=None,
|
||||
built_in_controls=None,
|
||||
):
|
||||
return ConditionalAccessPolicy(
|
||||
id=policy_id,
|
||||
display_name=display_name,
|
||||
conditions=Conditions(
|
||||
application_conditions=ApplicationsConditions(
|
||||
included_applications=included_applications or ["All"],
|
||||
excluded_applications=[],
|
||||
included_user_actions=[],
|
||||
),
|
||||
user_conditions=UsersConditions(
|
||||
included_groups=[],
|
||||
excluded_groups=[],
|
||||
included_users=included_users or ["All"],
|
||||
excluded_users=[],
|
||||
included_roles=[],
|
||||
excluded_roles=[],
|
||||
),
|
||||
client_app_types=[],
|
||||
sign_in_risk_levels=(
|
||||
sign_in_risk_levels
|
||||
if sign_in_risk_levels is not None
|
||||
else [RiskLevel.HIGH, RiskLevel.MEDIUM]
|
||||
),
|
||||
),
|
||||
grant_controls=GrantControls(
|
||||
built_in_controls=(
|
||||
built_in_controls
|
||||
if built_in_controls is not None
|
||||
else [ConditionalAccessGrantControl.BLOCK]
|
||||
),
|
||||
operator=GrantControlOperator.OR,
|
||||
authentication_strength=None,
|
||||
),
|
||||
session_controls=SessionControls(
|
||||
persistent_browser=PersistentBrowser(is_enabled=False, mode="always"),
|
||||
sign_in_frequency=SignInFrequency(
|
||||
is_enabled=False, frequency=None, type=None, interval=None
|
||||
),
|
||||
),
|
||||
state=state,
|
||||
)
|
||||
|
||||
|
||||
class Test_entra_conditional_access_policy_block_high_medium_sign_in_risk:
|
||||
def _run(self, policies):
|
||||
entra_client = mock.MagicMock
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(f"{CHECK_MODULE_PATH}.entra_client", new=entra_client),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_conditional_access_policy_block_high_medium_sign_in_risk.entra_conditional_access_policy_block_high_medium_sign_in_risk import (
|
||||
entra_conditional_access_policy_block_high_medium_sign_in_risk,
|
||||
)
|
||||
|
||||
entra_client.conditional_access_policies = policies
|
||||
entra_client.tenant_domain = DOMAIN
|
||||
return (
|
||||
entra_conditional_access_policy_block_high_medium_sign_in_risk().execute()
|
||||
)
|
||||
|
||||
def test_no_policies(self):
|
||||
result = self._run({})
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_policy_blocks_high_medium(self):
|
||||
policy = _make_policy()
|
||||
result = self._run({policy.id: policy})
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== "Conditional Access Policy 'Block Risky Sign-ins' blocks high and medium sign-in risk."
|
||||
)
|
||||
|
||||
def test_policy_only_high(self):
|
||||
policy = _make_policy(sign_in_risk_levels=[RiskLevel.HIGH])
|
||||
result = self._run({policy.id: policy})
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_policy_report_only(self):
|
||||
policy = _make_policy(state=ConditionalAccessPolicyState.ENABLED_FOR_REPORTING)
|
||||
result = self._run({policy.id: policy})
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||