mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-08-19 09:30:21 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b9ba136d43 | ||
|
|
3ca3a977a9 | ||
|
|
561a1390be | ||
|
|
f2a00f19aa | ||
|
|
34b4e6f016 | ||
|
|
94594d6766 | ||
|
|
6e71dee85d | ||
|
|
90712c9ad7 | ||
|
|
3672b17a00 |
@@ -0,0 +1 @@
|
||||
`/api/v1/accounts/saml/{organization_slug}/acs/` rejects non-POST requests before SAML response processing
|
||||
@@ -0,0 +1 @@
|
||||
`GET /api/v1/users/me` membership relationships identify the active tenant with `meta.active` for JWT and API key authentication
|
||||
@@ -1590,8 +1590,8 @@ class TestAPIKeyMultiTenantWorkflows:
|
||||
tenant1 = tenants_fixture[0]
|
||||
tenant2 = tenants_fixture[1]
|
||||
|
||||
Membership.objects.create(user=user, tenant=tenant1)
|
||||
Membership.objects.create(user=user, tenant=tenant2)
|
||||
membership1 = Membership.objects.create(user=user, tenant=tenant1)
|
||||
membership2 = Membership.objects.create(user=user, tenant=tenant2)
|
||||
|
||||
role1 = Role.objects.create(
|
||||
tenant_id=tenant1.id,
|
||||
@@ -1646,6 +1646,27 @@ class TestAPIKeyMultiTenantWorkflows:
|
||||
assert me_response1.json()["data"]["id"] == str(user.id)
|
||||
assert me_response2.json()["data"]["id"] == str(user.id)
|
||||
|
||||
memberships1 = {
|
||||
item["id"]: item["meta"]["active"]
|
||||
for item in me_response1.json()["data"]["relationships"]["memberships"][
|
||||
"data"
|
||||
]
|
||||
}
|
||||
memberships2 = {
|
||||
item["id"]: item["meta"]["active"]
|
||||
for item in me_response2.json()["data"]["relationships"]["memberships"][
|
||||
"data"
|
||||
]
|
||||
}
|
||||
assert memberships1 == {
|
||||
str(membership1.id): True,
|
||||
str(membership2.id): False,
|
||||
}
|
||||
assert memberships2 == {
|
||||
str(membership1.id): False,
|
||||
str(membership2.id): True,
|
||||
}
|
||||
|
||||
def test_api_key_cannot_access_different_tenant_resources(
|
||||
self, tenants_fixture, aws_provider
|
||||
):
|
||||
|
||||
@@ -14673,6 +14673,37 @@ class TestSAMLConfigurationViewSet:
|
||||
assert not SAMLConfiguration.objects.filter(id=config.id).exists()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestSAMLACSView:
|
||||
def test_get_is_not_allowed(self, client, saml_setup):
|
||||
response = client.get(
|
||||
reverse(
|
||||
"saml_acs",
|
||||
kwargs={"organization_slug": saml_setup["domain"]},
|
||||
)
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
|
||||
assert response.headers["Allow"] == "POST"
|
||||
assert "saml-acs-session" not in response.cookies
|
||||
|
||||
def test_post_is_forwarded_to_allauth(self, client, saml_setup):
|
||||
response = client.post(
|
||||
reverse(
|
||||
"saml_acs",
|
||||
kwargs={"organization_slug": saml_setup["domain"]},
|
||||
),
|
||||
data={"SAMLResponse": "test-saml-response"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_302_FOUND
|
||||
assert response.url == reverse(
|
||||
"saml_finish_acs",
|
||||
kwargs={"organization_slug": saml_setup["domain"]},
|
||||
)
|
||||
assert "saml-acs-session" in response.cookies
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestTenantFinishACSView:
|
||||
def test_dispatch_skips_if_user_not_authenticated(self, monkeypatch):
|
||||
|
||||
@@ -329,6 +329,15 @@ class TokenSwitchTenantSerializer(BaseSerializerV1):
|
||||
# Users
|
||||
|
||||
|
||||
class ActiveMembershipRelatedField(SerializerMethodResourceRelatedField):
|
||||
def to_representation(self, value):
|
||||
representation = super().to_representation(value)
|
||||
representation["meta"] = {
|
||||
"active": str(value.tenant_id) == str(self.context["request"].tenant_id),
|
||||
}
|
||||
return representation
|
||||
|
||||
|
||||
class UserSerializer(BaseModelSerializerV1):
|
||||
"""
|
||||
Serializer for the User model.
|
||||
@@ -390,6 +399,12 @@ class UserSerializer(BaseModelSerializerV1):
|
||||
)
|
||||
|
||||
|
||||
class UserMeSerializer(UserSerializer):
|
||||
memberships = ActiveMembershipRelatedField(
|
||||
many=True, read_only=True, source="memberships", method_name="get_memberships"
|
||||
)
|
||||
|
||||
|
||||
class UserIncludeSerializer(UserSerializer):
|
||||
class Meta:
|
||||
model = User
|
||||
|
||||
@@ -46,6 +46,7 @@ from api.v1.views import (
|
||||
from django.http import JsonResponse
|
||||
from django.urls import include, path
|
||||
from django.views.decorators.csrf import csrf_exempt
|
||||
from django.views.decorators.http import require_POST
|
||||
from drf_spectacular.views import SpectacularRedocView
|
||||
from rest_framework_nested import routers
|
||||
|
||||
@@ -194,7 +195,7 @@ urlpatterns = [
|
||||
),
|
||||
path(
|
||||
"accounts/saml/<organization_slug>/acs/",
|
||||
ACSView.as_view(),
|
||||
require_POST(ACSView.as_view()),
|
||||
name="saml_acs",
|
||||
),
|
||||
path(
|
||||
|
||||
@@ -238,6 +238,7 @@ from api.v1.serializers import (
|
||||
TokenSocialLoginSerializer,
|
||||
TokenSwitchTenantSerializer,
|
||||
UserCreateSerializer,
|
||||
UserMeSerializer,
|
||||
UserRoleRelationshipSerializer,
|
||||
UserSerializer,
|
||||
UserUpdateSerializer,
|
||||
@@ -1113,6 +1114,8 @@ class UserViewSet(BaseUserViewset):
|
||||
return UserCreateSerializer
|
||||
elif self.action == "partial_update":
|
||||
return UserUpdateSerializer
|
||||
elif self.action == "me":
|
||||
return UserMeSerializer
|
||||
else:
|
||||
return UserSerializer
|
||||
|
||||
@@ -1130,7 +1133,7 @@ class UserViewSet(BaseUserViewset):
|
||||
@action(detail=False, methods=["get"], url_name="me")
|
||||
def me(self, request):
|
||||
user = self.request.user
|
||||
serializer = UserSerializer(user, context=self.get_serializer_context())
|
||||
serializer = self.get_serializer(user)
|
||||
return Response(
|
||||
data=serializer.data,
|
||||
status=status.HTTP_200_OK,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -37,6 +37,7 @@
|
||||
"glue:SearchTables",
|
||||
"glue:GetMLTransforms",
|
||||
"lambda:GetFunction*",
|
||||
"lambda:GetLayerVersion",
|
||||
"logs:FilterLogEvents",
|
||||
"lightsail:GetRelationalDatabases",
|
||||
"macie2:GetMacieSession",
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
`awslambda_layer_no_secrets_in_content` check for AWS provider, scanning Lambda layer package content for hardcoded secrets
|
||||
@@ -0,0 +1 @@
|
||||
`batch_job_definition_no_secrets` check for AWS provider, scanning Batch job definition environment variables and command parameters for hardcoded secrets
|
||||
@@ -0,0 +1 @@
|
||||
7 M365 Entra checks covering CIS Microsoft 365 Foundations Benchmark v7.0.0 password protection, default user permissions, and guest invitation domain restrictions
|
||||
@@ -1166,7 +1166,9 @@
|
||||
{
|
||||
"Id": "5.1.3.1",
|
||||
"Description": "This setting allows users in the organization to create new security groups and add members to these groups in the Azure portal, API, or PowerShell. These new groups also show up in the Access Panel for all other users. If the policy setting on the group allows it, other users can create requests to join these groups. The recommended state is Users can create security groups in Azure portals, API or PowerShell set to No.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_policy_default_user_cannot_create_security_groups"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -1229,7 +1231,9 @@
|
||||
{
|
||||
"Id": "5.1.3.4",
|
||||
"Description": "All users within a Microsoft Entra organization are permitted to create new Microsoft 365 groups and add members to those groups through the Azure portal, API, or PowerShell. Newly created groups also appear in the Access Panel for all other users. When the applicable group policy settings allow it, users can submit requests to join these groups. The recommended state is No.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_policy_default_user_cannot_create_m365_groups"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -1526,7 +1530,9 @@
|
||||
{
|
||||
"Id": "5.1.6.1",
|
||||
"Description": "B2B collaboration is a feature within Microsoft Entra External ID that allows for guest invitations to an organization. Ensure users can only send invitations to specified domains. Note: This list works independently from OneDrive for Business and SharePoint Online allow/block lists. To restrict individual file sharing in SharePoint Online, set up an allow or blocklist for OneDrive for Business and SharePoint Online. For instance, in SharePoint or OneDrive users can still share with external users from prohibited domains by using Anyone links if they haven't been disabled.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_policy_guest_invitations_restricted_to_allowed_domains"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -2016,7 +2022,9 @@
|
||||
{
|
||||
"Id": "5.2.3.2",
|
||||
"Description": "With Entra Password Protection, default global banned password lists are automatically applied to all users in an Entra ID tenant. To support business and security needs, custom banned password lists can be defined. When users change or reset their passwords, these banned password lists are checked to enforce the use of strong passwords. A custom banned password list should include some of the following examples: - Brand names - Product names - Locations, such as company headquarters - Company-specific internal terms - Abbreviations that have specific company meaning",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_password_protection_custom_banned_list_enforced"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -2037,7 +2045,9 @@
|
||||
{
|
||||
"Id": "5.2.3.3",
|
||||
"Description": "Microsoft Entra Password Protection provides a global and custom banned password list. A password change request fails if there's a match in these banned password list. To protect on-premises Active Directory Domain Services (AD DS) environment, install and configure Entra Password Protection. Note: This recommendation applies to Hybrid deployments only and will have no impact unless working with on-premises Active Directory.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_password_protection_on_premises_enforced"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -2146,7 +2156,9 @@
|
||||
{
|
||||
"Id": "5.2.3.8",
|
||||
"Description": "The account lockout threshold determines how many failed login attempts are permitted prior to placing the account in a locked-out state and initiating a variable lockout duration. The recommended Lockout threshold is 10 or less.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_password_protection_lockout_threshold_limited"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
@@ -2167,7 +2179,9 @@
|
||||
{
|
||||
"Id": "5.2.3.9",
|
||||
"Description": "The account lockout duration value determines how long an account retains the status of lockout, and therefore how long before a user can continue to attempt to login after passing the lockout threshold. The recommended state is Lockout duration in seconds is at least 60.",
|
||||
"Checks": [],
|
||||
"Checks": [
|
||||
"entra_password_protection_lockout_duration_configured"
|
||||
],
|
||||
"Attributes": [
|
||||
{
|
||||
"Section": "5 Microsoft Entra admin center",
|
||||
|
||||
@@ -27,6 +27,8 @@ aws:
|
||||
max_lambda_functions: null
|
||||
# aws.max_ecs_task_definitions --> ecs_task_definitions_* checks
|
||||
max_ecs_task_definitions: null
|
||||
# aws.max_batch_job_definitions --> batch_job_definition_* checks
|
||||
max_batch_job_definitions: null
|
||||
# aws.max_codeartifact_packages --> codeartifact_packages_* checks
|
||||
max_codeartifact_packages: null
|
||||
# aws.disallowed_regions --> List of AWS regions to exclude from the scan.
|
||||
|
||||
@@ -153,6 +153,12 @@ class AWSProviderConfig(ProviderConfigBase):
|
||||
le=1_000_000,
|
||||
description="Resource scan limit for ECS task definitions. Use 0 or -1 to disable.",
|
||||
)
|
||||
max_batch_job_definitions: ResourceScanLimit = Field(
|
||||
default=None,
|
||||
ge=-1,
|
||||
le=1_000_000,
|
||||
description="Resource scan limit for Batch job definitions. Use 0 or -1 to disable.",
|
||||
)
|
||||
max_codeartifact_packages: ResourceScanLimit = Field(
|
||||
default=None,
|
||||
ge=-1,
|
||||
|
||||
+41
@@ -0,0 +1,41 @@
|
||||
{
|
||||
"Provider": "aws",
|
||||
"CheckID": "awslambda_layer_no_secrets_in_content",
|
||||
"CheckTitle": "Lambda layer content contains no hardcoded secrets",
|
||||
"CheckType": [
|
||||
"Software and Configuration Checks/AWS Security Best Practices",
|
||||
"Sensitive Data Identifications/Passwords",
|
||||
"Effects/Data Exposure"
|
||||
],
|
||||
"ServiceName": "awslambda",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "high",
|
||||
"ResourceType": "AwsLambdaLayerVersion",
|
||||
"ResourceGroup": "serverless",
|
||||
"Description": "**Lambda layer content** is analyzed for **embedded secrets** across files in the layer's package, detecting patterns like API keys, passwords, tokens, and connection strings. Findings reference file names and line numbers where potential secrets appear.",
|
||||
"Risk": "**Hardcoded secrets** undermine confidentiality and integrity: a secret baked into a layer is pulled into every function that uses it, and is not covered by a function-code-only scan. If exposed, attackers can reuse credentials to access databases, APIs, or cloud resources, enabling data exfiltration and unauthorized changes. Rotation is harder, increasing dwell time and blast radius.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://docs.aws.amazon.com/lambda/latest/dg/chapter-layers.html",
|
||||
"https://docs.aws.amazon.com/secretsmanager/latest/userguide/best-practices.html"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "aws secretsmanager create-secret --name <secret-name> --secret-string <value>\naws iam put-role-policy --role-name <function-execution-role> --policy-name allow-get-secret --policy-document '{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Action\":\"secretsmanager:GetSecretValue\",\"Resource\":\"<secret-arn>\"}]}'\n# Remove the hardcoded value from the layer's code, then:\naws lambda publish-layer-version --layer-name <layer-name> --zip-file fileb://layer.zip",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. In AWS Secrets Manager, click Store a new secret and create a secret for the value you hardcoded. Note the secret name/ARN.\n2. In IAM > Roles, open the execution role of every function that uses this layer and add an inline policy allowing secretsmanager:GetSecretValue on that secret only.\n3. Remove the hardcoded value from the layer's code and repackage it, retrieving the secret at runtime using the AWS SDK (GetSecretValue) with the secret name/ARN.\n4. Publish a new layer version and update dependent functions to use it.",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Use **AWS Secrets Manager** (or Parameter Store) to store secrets and retrieve at runtime; never put them in layer code or packaged dependencies.\n- Apply **least privilege** IAM\n- Enable **rotation**\n- Prevent secret logging; encrypt\n- Add CI/CD secret scanning",
|
||||
"Url": "https://hub.prowler.com/check/awslambda_layer_no_secrets_in_content"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"secrets"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
+164
@@ -0,0 +1,164 @@
|
||||
import fnmatch
|
||||
import os
|
||||
import tempfile
|
||||
from collections import defaultdict
|
||||
|
||||
from prowler.lib.check.models import Check, Check_Report_AWS
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.lib.utils.utils import (
|
||||
SecretsScanError,
|
||||
annotate_verified_secrets,
|
||||
detect_secrets_scan_batch,
|
||||
)
|
||||
from prowler.providers.aws.services.awslambda.awslambda_client import awslambda_client
|
||||
|
||||
|
||||
class awslambda_layer_no_secrets_in_content(Check):
|
||||
"""Check if Lambda layer content contains hardcoded secrets.
|
||||
|
||||
Scans every file inside each Lambda layer version's package with the
|
||||
secret scanner.
|
||||
|
||||
- PASS: No secrets are detected in the layer content.
|
||||
- FAIL: At least one potential secret is detected in the layer content.
|
||||
- MANUAL: The layer content could not be fetched or scanned.
|
||||
"""
|
||||
|
||||
def execute(self) -> list[Check_Report_AWS]:
|
||||
"""Execute the Lambda layer secrets scan.
|
||||
|
||||
Returns:
|
||||
list[Check_Report_AWS]: One report per Lambda layer version used by
|
||||
the audited functions, or an empty list when there are no layers.
|
||||
"""
|
||||
findings = []
|
||||
if not awslambda_client.layers:
|
||||
return findings
|
||||
|
||||
secrets_ignore_patterns = awslambda_client.audit_config.get(
|
||||
"secrets_ignore_patterns", []
|
||||
)
|
||||
# Glob patterns of file names inside the layer package to skip
|
||||
# when scanning for secrets (e.g. "*.deps.json" for .NET layers).
|
||||
secrets_ignore_files = (
|
||||
awslambda_client.audit_config.get("secrets_ignore_files", []) or []
|
||||
)
|
||||
validate = awslambda_client.audit_config.get("secrets_validate", False)
|
||||
|
||||
# Scan files of every layer version's package in batched
|
||||
# Kingfisher invocations instead of one subprocess per file per layer.
|
||||
# Each package is extracted one at a time and its files are
|
||||
# read (byte-faithfully via latin-1) before the extraction is released,
|
||||
# so only a single package is on disk at a time. Findings are keyed by
|
||||
# (layer index, package-relative file name) so they can be grouped
|
||||
# back per layer.
|
||||
layers_with_code = []
|
||||
|
||||
def code_payloads():
|
||||
for layer, layer_code in awslambda_client._get_layers_code():
|
||||
if not layer_code:
|
||||
continue
|
||||
with tempfile.TemporaryDirectory() as tmp_dir_name:
|
||||
try:
|
||||
layer_code.code_zip.extractall(tmp_dir_name)
|
||||
except Exception as error:
|
||||
# A corrupt or truncated package must not abort the
|
||||
# scan of the remaining layers: keep this layer out of
|
||||
# layers_with_code so it is reported as MANUAL below.
|
||||
logger.error(
|
||||
f"{layer.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
continue
|
||||
index = len(layers_with_code)
|
||||
layers_with_code.append(layer)
|
||||
for root, _, files in os.walk(tmp_dir_name):
|
||||
for file_name in files:
|
||||
file_path = os.path.join(root, file_name)
|
||||
relative_file_path = os.path.relpath(
|
||||
file_path, tmp_dir_name
|
||||
)
|
||||
if any(
|
||||
fnmatch.fnmatch(relative_file_path, pattern)
|
||||
for pattern in secrets_ignore_files
|
||||
):
|
||||
continue
|
||||
try:
|
||||
with open(file_path, "rb") as code_file:
|
||||
content = code_file.read().decode("latin-1")
|
||||
except Exception:
|
||||
continue
|
||||
yield (index, relative_file_path), content
|
||||
|
||||
scan_error = None
|
||||
try:
|
||||
batch_results = detect_secrets_scan_batch(
|
||||
code_payloads(),
|
||||
excluded_secrets=secrets_ignore_patterns,
|
||||
validate=validate,
|
||||
)
|
||||
except SecretsScanError as error:
|
||||
batch_results = {}
|
||||
scan_error = error
|
||||
|
||||
if scan_error:
|
||||
# The scan failed before any layer's code could be cleared. Report
|
||||
# MANUAL for every layer rather than risk a false PASS.
|
||||
for layer in awslambda_client.layers.values():
|
||||
report = Check_Report_AWS(metadata=self.metadata(), resource=layer)
|
||||
report.status = "MANUAL"
|
||||
report.status_extended = (
|
||||
f"Could not scan Lambda layer {layer.name} (version "
|
||||
f"{layer.version}) content for secrets: {scan_error}; "
|
||||
"manual review is required."
|
||||
)
|
||||
findings.append(report)
|
||||
return findings
|
||||
|
||||
findings_by_layer = defaultdict(dict)
|
||||
for (index, file_name), file_findings in batch_results.items():
|
||||
findings_by_layer[index][file_name] = file_findings
|
||||
|
||||
for index, layer in enumerate(layers_with_code):
|
||||
report = Check_Report_AWS(metadata=self.metadata(), resource=layer)
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
f"No secrets found in Lambda layer {layer.name} "
|
||||
f"(version {layer.version}) content."
|
||||
)
|
||||
|
||||
files_with_secrets = findings_by_layer.get(index)
|
||||
if files_with_secrets:
|
||||
all_secrets = []
|
||||
secrets_findings = []
|
||||
for file_name, file_findings in files_with_secrets.items():
|
||||
all_secrets.extend(file_findings)
|
||||
secrets_string = ", ".join(
|
||||
f"{secret['type']} on line {secret['line_number']}"
|
||||
for secret in file_findings
|
||||
)
|
||||
secrets_findings.append(f"{file_name}: {secrets_string}")
|
||||
|
||||
final_output_string = "; ".join(secrets_findings)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"Potential {'secrets' if len(secrets_findings) > 1 else 'secret'} found in Lambda layer {layer.name} (version {layer.version}) content -> {final_output_string}."
|
||||
annotate_verified_secrets(report, all_secrets)
|
||||
|
||||
findings.append(report)
|
||||
|
||||
# Layers whose content could not be fetched (network error, missing
|
||||
# permissions, etc.) never reach layers_with_code above, so report
|
||||
# them as MANUAL rather than silently omitting them from the scan.
|
||||
fetched_arns = {layer.arn for layer in layers_with_code}
|
||||
for layer in awslambda_client.layers.values():
|
||||
if layer.arn in fetched_arns:
|
||||
continue
|
||||
report = Check_Report_AWS(metadata=self.metadata(), resource=layer)
|
||||
report.status = "MANUAL"
|
||||
report.status_extended = (
|
||||
f"Could not retrieve content of Lambda layer {layer.name} "
|
||||
f"(version {layer.version}) to scan for secrets; manual "
|
||||
"review is required."
|
||||
)
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
@@ -17,6 +17,11 @@ from prowler.lib.resource_limit import (
|
||||
from prowler.lib.scan_filters.scan_filters import is_resource_filtered
|
||||
from prowler.providers.aws.lib.service.service import AWSService
|
||||
|
||||
# Presigned code/layer download URLs are short-lived S3 URLs, not AWS API
|
||||
# calls, so a hung request here would otherwise block a worker thread
|
||||
# indefinitely instead of failing like the surrounding boto3 calls do.
|
||||
CODE_DOWNLOAD_TIMEOUT_SECONDS = 30
|
||||
|
||||
|
||||
class Lambda(AWSService):
|
||||
def __init__(self, provider):
|
||||
@@ -25,6 +30,7 @@ class Lambda(AWSService):
|
||||
# Functions are listed first, then trimmed to the subset selected for
|
||||
# analysis before expensive per-function detail is hydrated.
|
||||
self.functions = {}
|
||||
self.layers = {}
|
||||
self.security_groups_in_use = set()
|
||||
self.regions_with_functions = set()
|
||||
self.function_limit = get_resource_scan_limit(
|
||||
@@ -32,6 +38,7 @@ class Lambda(AWSService):
|
||||
)
|
||||
self.__threading_call__(self._list_functions)
|
||||
self._select_functions_for_analysis()
|
||||
self._collect_layers()
|
||||
self._list_tags_for_resource()
|
||||
self.__threading_call__(self._get_policy)
|
||||
self.__threading_call__(self._get_function_url_config)
|
||||
@@ -106,6 +113,11 @@ class Lambda(AWSService):
|
||||
)
|
||||
}
|
||||
|
||||
def _collect_layers(self):
|
||||
for function in self.functions.values():
|
||||
for layer in function.layers:
|
||||
self.layers.setdefault(layer.arn, layer)
|
||||
|
||||
def _list_event_source_mappings(self, regional_client):
|
||||
logger.info("Lambda - Listing Event Source Mappings...")
|
||||
try:
|
||||
@@ -193,6 +205,15 @@ class Lambda(AWSService):
|
||||
f"{function.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def _download_code(self, code_location_uri):
|
||||
raw_code_zip = requests.get(
|
||||
code_location_uri, timeout=CODE_DOWNLOAD_TIMEOUT_SECONDS
|
||||
).content
|
||||
return LambdaCode(
|
||||
location=code_location_uri,
|
||||
code_zip=zipfile.ZipFile(io.BytesIO(raw_code_zip)),
|
||||
)
|
||||
|
||||
def _fetch_function_code(self, function_name, function_region):
|
||||
try:
|
||||
regional_client = self.regional_clients[function_region]
|
||||
@@ -200,18 +221,52 @@ class Lambda(AWSService):
|
||||
FunctionName=function_name
|
||||
)
|
||||
if "Location" in function_information["Code"]:
|
||||
code_location_uri = function_information["Code"]["Location"]
|
||||
raw_code_zip = requests.get(code_location_uri).content
|
||||
return LambdaCode(
|
||||
location=code_location_uri,
|
||||
code_zip=zipfile.ZipFile(io.BytesIO(raw_code_zip)),
|
||||
)
|
||||
return self._download_code(function_information["Code"]["Location"])
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
raise
|
||||
|
||||
def _get_layers_code(self):
|
||||
logger.info("Lambda - Getting Layer Code...")
|
||||
# Use a thread pool to handle the queueing and execution of the
|
||||
# _fetch_layer_code tasks, up to max_workers tasks concurrently.
|
||||
layers_to_fetch = {
|
||||
self.thread_pool.submit(
|
||||
self._fetch_layer_code, layer.arn, layer.region
|
||||
): layer
|
||||
for layer in self.layers.values()
|
||||
}
|
||||
|
||||
for fetched_layer_code in as_completed(layers_to_fetch):
|
||||
layer = layers_to_fetch[fetched_layer_code]
|
||||
try:
|
||||
layer_code = fetched_layer_code.result()
|
||||
if layer_code:
|
||||
yield layer, layer_code
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{layer.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def _fetch_layer_code(self, layer_arn, layer_region):
|
||||
try:
|
||||
regional_client = self.regional_clients[layer_region]
|
||||
# Fetch by the full layer-version ARN: layers attached to a
|
||||
# function may be owned by another account (e.g. vendor or
|
||||
# AWS-provided layers), where a bare layer name would resolve
|
||||
# against the audited account instead.
|
||||
layer_version = regional_client.get_layer_version_by_arn(Arn=layer_arn)
|
||||
if "Location" in (layer_version.get("Content") or {}):
|
||||
return self._download_code(layer_version["Content"]["Location"])
|
||||
return None
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{layer_region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
raise
|
||||
|
||||
def _get_policy(self, regional_client):
|
||||
logger.info("Lambda - Getting Policy...")
|
||||
try:
|
||||
@@ -308,6 +363,28 @@ class Layer(BaseModel):
|
||||
parts = self.arn.split(":")
|
||||
return parts[4] if len(parts) >= 5 else ""
|
||||
|
||||
@property
|
||||
def region(self) -> str:
|
||||
"""Extract the region from the layer ARN.
|
||||
|
||||
A layer can only be attached to a function in the same region, so
|
||||
this is always one of the regions already being audited.
|
||||
"""
|
||||
parts = self.arn.split(":")
|
||||
return parts[3] if len(parts) >= 4 else ""
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
"""Extract the layer name from the ARN."""
|
||||
parts = self.arn.split(":")
|
||||
return parts[6] if len(parts) >= 7 else self.arn
|
||||
|
||||
@property
|
||||
def version(self) -> str:
|
||||
"""Extract the layer version from the ARN."""
|
||||
parts = self.arn.split(":")
|
||||
return parts[7] if len(parts) >= 8 else ""
|
||||
|
||||
|
||||
class DeadLetterConfig(BaseModel):
|
||||
target_arn: str
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
"""AWS Batch service client singleton."""
|
||||
|
||||
from prowler.providers.aws.services.batch.batch_service import Batch
|
||||
from prowler.providers.common.provider import Provider
|
||||
|
||||
batch_client = Batch(Provider.get_global_provider())
|
||||
+41
@@ -0,0 +1,41 @@
|
||||
{
|
||||
"Provider": "aws",
|
||||
"CheckID": "batch_job_definition_no_secrets",
|
||||
"CheckTitle": "AWS Batch job definitions have no secrets in environment variables or command parameters",
|
||||
"CheckType": [
|
||||
"Software and Configuration Checks/AWS Security Best Practices",
|
||||
"Software and Configuration Checks/Industry and Regulatory Standards/AWS Foundational Security Best Practices",
|
||||
"Sensitive Data Identifications/Passwords",
|
||||
"TTPs/Credential Access"
|
||||
],
|
||||
"ServiceName": "batch",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "high",
|
||||
"ResourceType": "AwsBatchJobDefinition",
|
||||
"ResourceGroup": "container",
|
||||
"Description": "**AWS Batch job definitions** are analyzed for **plaintext secrets** placed in container `environment` variables and `command` parameters. It identifies values that resemble credentials (keys, tokens, passwords) within job definitions.",
|
||||
"Risk": "Exposed secrets in env vars or command parameters undermine confidentiality via logs, job metadata, and introspection.\n\nWith container or read-only API access, attackers can reuse credentials to read databases, modify records (integrity), pivot to other services, and trigger outages or unauthorized costs (availability).",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://docs.aws.amazon.com/batch/latest/userguide/job_definition_parameters.html"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "aws batch register-job-definition --job-definition-name <JOB_DEFINITION_NAME> --type container --container-properties '{\"image\":\"<IMAGE>\",\"secrets\":[{\"name\":\"<SECRET_NAME>\",\"valueFrom\":\"arn:aws:secretsmanager:<REGION>:<ACCOUNT_ID>:secret:<SECRET_NAME>-<RANDOM>\"}]}' # Register a new revision without plaintext secrets; reference Secrets Manager or SSM Parameter Store via valueFrom",
|
||||
"NativeIaC": "```yaml\nResources:\n <example_resource_name>:\n Type: AWS::Batch::JobDefinition\n Properties:\n Type: container\n ContainerProperties:\n Image: <image>\n Environment:\n - Name: DB_PASSWORD\n Value: !Ref <secret_parameter> # Reference SSM Parameter or Secrets Manager\n```",
|
||||
"Other": "1. In the AWS Console, go to Batch > Job Definitions and open your job definition\n2. Create a new revision\n3. Remove any sensitive values from Environment variables and command parameters\n4. Reference secrets from AWS Secrets Manager or SSM Parameter Store instead\n5. Save to create the new revision\n6. Update any Batch job queues to use the new job definition revision",
|
||||
"Terraform": "```hcl\nresource \"aws_batch_job_definition\" \"<example_resource_name>\" {\n name = \"<example_resource_name>\"\n type = \"container\"\n\n container_properties = jsonencode({\n image = \"<image>\"\n environment = [\n {\n name = \"DB_PASSWORD\"\n value = var.db_password # Use variable from Secrets Manager or SSM\n }\n ]\n })\n}\n```"
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Store secrets in **AWS Secrets Manager** or **SSM Parameter Store** and inject them at runtime instead of plaintext env vars.\n\nApply **least privilege** via job role, enable regular **rotation**, avoid logging secret values, and prefer **ephemeral credentials** for downstream services.",
|
||||
"Url": "https://hub.prowler.com/check/batch_job_definition_no_secrets"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"secrets"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "Only container job definitions are evaluated (containerProperties.environment and command). Multi-node parallel (nodeProperties) and EKS (eksProperties) job definitions are not analyzed."
|
||||
}
|
||||
+120
@@ -0,0 +1,120 @@
|
||||
from json import dumps
|
||||
|
||||
from prowler.lib.check.models import Check, Check_Report_AWS
|
||||
from prowler.lib.utils.utils import (
|
||||
SecretsScanError,
|
||||
annotate_verified_secrets,
|
||||
detect_secrets_scan_batch,
|
||||
)
|
||||
from prowler.providers.aws.services.batch.batch_client import batch_client
|
||||
|
||||
|
||||
class batch_job_definition_no_secrets(Check):
|
||||
"""Detect secrets in AWS Batch job definition environment variables and commands."""
|
||||
|
||||
def execute(self) -> list[Check_Report_AWS]:
|
||||
"""Scan job definitions for hardcoded secrets in env vars and commands."""
|
||||
findings = []
|
||||
|
||||
secrets_ignore_patterns = batch_client.audit_config.get(
|
||||
"secrets_ignore_patterns", []
|
||||
)
|
||||
validate = batch_client.audit_config.get("secrets_validate", False)
|
||||
|
||||
job_definitions = list(batch_client.job_definitions.values())
|
||||
|
||||
def scan_payloads():
|
||||
"""Yield index-keyed payloads for each env var and the command."""
|
||||
for jd_index, job_definition in enumerate(job_definitions):
|
||||
container = job_definition.container_properties
|
||||
|
||||
for env_index, env_var in enumerate(container.environment):
|
||||
yield (jd_index, env_index), dumps(
|
||||
{env_var.name: env_var.value}, indent=2
|
||||
)
|
||||
|
||||
if container.command:
|
||||
yield (
|
||||
(jd_index, "command"),
|
||||
" ".join(container.command),
|
||||
)
|
||||
|
||||
scan_error = None
|
||||
try:
|
||||
batch_results = detect_secrets_scan_batch(
|
||||
scan_payloads(),
|
||||
excluded_secrets=secrets_ignore_patterns,
|
||||
validate=validate,
|
||||
)
|
||||
except SecretsScanError as error:
|
||||
batch_results = {}
|
||||
scan_error = error
|
||||
|
||||
for jd_index, job_definition in enumerate(job_definitions):
|
||||
report = Check_Report_AWS(
|
||||
metadata=self.metadata(),
|
||||
resource=job_definition,
|
||||
)
|
||||
|
||||
report.resource_id = f"{job_definition.name}:{job_definition.revision}"
|
||||
report.status = "PASS"
|
||||
|
||||
extended_status_parts = []
|
||||
all_secrets = []
|
||||
|
||||
container = job_definition.container_properties
|
||||
|
||||
if scan_error and (container.environment or container.command):
|
||||
report.status = "MANUAL"
|
||||
report.status_extended = (
|
||||
f"Could not scan Batch job definition "
|
||||
f"{job_definition.name} with revision "
|
||||
f"{job_definition.revision} for secrets: "
|
||||
f"{scan_error}; manual review is required."
|
||||
)
|
||||
findings.append(report)
|
||||
continue
|
||||
|
||||
for env_index, env_var in enumerate(container.environment):
|
||||
env_secrets = batch_results.get((jd_index, env_index))
|
||||
if env_secrets:
|
||||
all_secrets.extend(env_secrets)
|
||||
secrets_string = ", ".join(
|
||||
f"{secret['type']} on the environment variable {env_var.name}"
|
||||
for secret in env_secrets
|
||||
)
|
||||
extended_status_parts.append(
|
||||
f"Secrets in environment variables -> {secrets_string}"
|
||||
)
|
||||
|
||||
if container.command:
|
||||
command_secrets = batch_results.get((jd_index, "command"))
|
||||
if command_secrets:
|
||||
all_secrets.extend(command_secrets)
|
||||
secrets_string = ", ".join(
|
||||
secret["type"] for secret in command_secrets
|
||||
)
|
||||
extended_status_parts.append(
|
||||
f"Secrets in command -> {secrets_string}"
|
||||
)
|
||||
|
||||
if extended_status_parts:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
f"Potential secrets found in Batch job definition "
|
||||
f"{job_definition.name} with revision "
|
||||
f"{job_definition.revision}: "
|
||||
+ "; ".join(extended_status_parts)
|
||||
+ "."
|
||||
)
|
||||
annotate_verified_secrets(report, all_secrets)
|
||||
else:
|
||||
report.status_extended = (
|
||||
f"No secrets found in Batch job definition "
|
||||
f"{job_definition.name} with revision "
|
||||
f"{job_definition.revision}."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
@@ -0,0 +1,105 @@
|
||||
from itertools import zip_longest
|
||||
from typing import Optional
|
||||
|
||||
from pydantic.v1 import BaseModel
|
||||
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.lib.resource_limit import get_resource_scan_limit, limit_resources
|
||||
from prowler.lib.scan_filters.scan_filters import is_resource_filtered
|
||||
from prowler.providers.aws.lib.service.service import AWSService
|
||||
|
||||
|
||||
class ContainerEnvVariable(BaseModel):
|
||||
"""An environment variable name-value pair."""
|
||||
|
||||
name: str
|
||||
value: str
|
||||
|
||||
|
||||
class BatchContainerProperties(BaseModel):
|
||||
"""Container properties for an AWS Batch job definition."""
|
||||
|
||||
image: Optional[str]
|
||||
command: list[str] = []
|
||||
environment: list[ContainerEnvVariable] = []
|
||||
|
||||
|
||||
class BatchJobDefinition(BaseModel):
|
||||
"""An AWS Batch job definition with its container properties."""
|
||||
|
||||
name: str
|
||||
arn: str
|
||||
revision: int
|
||||
region: str
|
||||
container_properties: BatchContainerProperties
|
||||
|
||||
|
||||
class Batch(AWSService):
|
||||
"""AWS Batch service client for listing job definitions."""
|
||||
|
||||
def __init__(self, provider):
|
||||
super().__init__(__class__.__name__, provider)
|
||||
self.job_definitions = {}
|
||||
self._job_definitions_by_region = {}
|
||||
self.job_definition_limit = get_resource_scan_limit(
|
||||
self.audit_config, "max_batch_job_definitions"
|
||||
)
|
||||
self.__threading_call__(self._list_job_definitions)
|
||||
self._select_job_definitions_for_analysis()
|
||||
|
||||
def _list_job_definitions(self, regional_client):
|
||||
"""List ACTIVE job definitions for a regional client."""
|
||||
logger.info("Batch - Listing Job Definitions...")
|
||||
try:
|
||||
paginator = regional_client.get_paginator("describe_job_definitions")
|
||||
regional_job_definitions = []
|
||||
# Deregistered (INACTIVE) revisions are excluded: they cannot run
|
||||
# new jobs, and reporting them would only produce noise.
|
||||
for page in paginator.paginate(status="ACTIVE"):
|
||||
for job in page.get("jobDefinitions", []):
|
||||
if self.audit_resources and not is_resource_filtered(
|
||||
job["jobDefinitionArn"], self.audit_resources
|
||||
):
|
||||
continue
|
||||
container = job.get("containerProperties", {})
|
||||
environment = [
|
||||
ContainerEnvVariable(
|
||||
name=env["name"], value=env.get("value", "")
|
||||
)
|
||||
for env in container.get("environment", [])
|
||||
]
|
||||
regional_job_definitions.append(
|
||||
BatchJobDefinition(
|
||||
name=job["jobDefinitionName"],
|
||||
arn=job["jobDefinitionArn"],
|
||||
revision=job["revision"],
|
||||
region=regional_client.region,
|
||||
container_properties=BatchContainerProperties(
|
||||
image=container.get("image"),
|
||||
command=container.get("command", []),
|
||||
environment=environment,
|
||||
),
|
||||
)
|
||||
)
|
||||
self._job_definitions_by_region[regional_client.region] = (
|
||||
regional_job_definitions
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def _select_job_definitions_for_analysis(self):
|
||||
"""Apply the global resource limit, interleaving regions fairly."""
|
||||
interleaved = [
|
||||
job_definition
|
||||
for region_batch in zip_longest(*self._job_definitions_by_region.values())
|
||||
for job_definition in region_batch
|
||||
if job_definition
|
||||
]
|
||||
self.job_definitions = {
|
||||
job_definition.arn: job_definition
|
||||
for job_definition in limit_resources(
|
||||
interleaved, self.job_definition_limit
|
||||
)
|
||||
}
|
||||
+37
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_password_protection_custom_banned_list_enforced",
|
||||
"CheckTitle": "Entra custom banned password list is enforced",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "The tenant Password Rule Settings should enforce a **custom banned password list** (**EnableBannedPasswordCheck** true with a non-empty **BannedPasswordList**). This blocks organization-specific weak or predictable passwords (e.g., company name, products, locations) in addition to Microsoft's global banned list.",
|
||||
"Risk": "Without a custom **banned password** list, users can choose passwords that are predictable for the specific organization (brand names, local terms), which are easy targets for **password spraying** and guessing attacks.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/identity/authentication/concept-password-ban-bad"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Protection** > **Authentication methods** > **Password protection**\n3. Set **Enforce custom list** to **Yes**\n4. Add organization-specific terms to the **Custom banned password list**\n5. Click **Save**",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Enable the custom banned password list and populate it with terms relevant to the organization to strengthen protection against weak passwords beyond the global banned list.",
|
||||
"Url": "https://hub.prowler.com/check/entra_password_protection_custom_banned_list_enforced"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
+63
@@ -0,0 +1,63 @@
|
||||
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 (
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID,
|
||||
)
|
||||
|
||||
|
||||
class entra_password_protection_custom_banned_list_enforced(Check):
|
||||
"""Check if the Entra custom banned password list is enforced.
|
||||
|
||||
The Password Rule Settings directory setting should enforce a custom banned
|
||||
password list (EnableBannedPasswordCheck) with a non-empty BannedPasswordList so
|
||||
that organization-specific weak passwords are rejected in addition to the global
|
||||
banned list.
|
||||
|
||||
- PASS: The custom banned password list is enforced and non-empty.
|
||||
- FAIL: The custom banned password list is not enforced or is empty.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[CheckReportM365]:
|
||||
"""Execute the custom banned password list enforcement check.
|
||||
|
||||
Evaluate whether the Password Rule Settings directory setting enforces a
|
||||
non-empty custom banned password list. When the settings object is absent,
|
||||
no finding is produced.
|
||||
|
||||
Returns:
|
||||
List[CheckReportM365]: A list with a single report when the Password Rule
|
||||
Settings exist, or an empty list when they are absent.
|
||||
"""
|
||||
findings = []
|
||||
settings = entra_client.directory_settings.get(
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID
|
||||
)
|
||||
if not settings:
|
||||
return findings
|
||||
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=settings or {},
|
||||
resource_name="Password Rule Settings",
|
||||
resource_id=PASSWORD_RULE_SETTINGS_TEMPLATE_ID,
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"The custom banned password list is not enforced in the tenant."
|
||||
)
|
||||
|
||||
if settings:
|
||||
enforced = (
|
||||
str(settings.get("EnableBannedPasswordCheck", "")).lower() == "true"
|
||||
)
|
||||
banned_list = settings.get("BannedPasswordList", "") or ""
|
||||
if enforced and banned_list.strip():
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
"The custom banned password list is enforced in the tenant."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
+37
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_password_protection_lockout_duration_configured",
|
||||
"CheckTitle": "Smart lockout duration is set to 60 seconds or more",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "low",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "The tenant Password Rule Settings should set the smart **LockoutDurationInSeconds** to **60 or more**. The lockout duration determines how long an account remains locked out before the user can attempt to sign in again.",
|
||||
"Risk": "A short lockout duration allows attackers to resume **brute-force** or **password-spray** attempts sooner, reducing the effectiveness of **smart lockout** as a throttling control.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/identity/authentication/howto-password-smart-lockout"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Protection** > **Authentication methods** > **Password protection**\n3. Set **Lockout duration in seconds** to **60** or higher\n4. Click **Save**",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Set the smart lockout duration to 60 seconds or more so locked-out accounts remain locked long enough to throttle automated password attacks.",
|
||||
"Url": "https://hub.prowler.com/check/entra_password_protection_lockout_duration_configured"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
+62
@@ -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 (
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID,
|
||||
)
|
||||
|
||||
# CIS recommends a lockout duration of 60 seconds or more.
|
||||
MIN_LOCKOUT_DURATION_SECONDS = 60
|
||||
|
||||
|
||||
class entra_password_protection_lockout_duration_configured(Check):
|
||||
"""Check if the smart lockout duration is set to 60 seconds or more.
|
||||
|
||||
The Password Rule Settings directory setting should set LockoutDurationInSeconds
|
||||
to 60 or more so a locked-out account remains locked long enough to slow down
|
||||
automated attacks.
|
||||
|
||||
- PASS: The lockout duration is 60 seconds or more.
|
||||
- FAIL: The lockout duration is less than 60 seconds or not configured.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[CheckReportM365]:
|
||||
"""Execute the smart lockout duration check.
|
||||
|
||||
Returns:
|
||||
List[CheckReportM365]: Reports for the Password Rule Settings, or an
|
||||
empty list when the settings are absent.
|
||||
"""
|
||||
findings = []
|
||||
settings = entra_client.directory_settings.get(
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID
|
||||
)
|
||||
if not settings:
|
||||
return findings
|
||||
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=settings or {},
|
||||
resource_name="Password Rule Settings",
|
||||
resource_id=PASSWORD_RULE_SETTINGS_TEMPLATE_ID,
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"The smart lockout duration is not set to 60 seconds or more."
|
||||
)
|
||||
|
||||
if settings:
|
||||
try:
|
||||
duration = int(settings.get("LockoutDurationInSeconds"))
|
||||
except (TypeError, ValueError):
|
||||
duration = None
|
||||
if duration is not None and duration >= MIN_LOCKOUT_DURATION_SECONDS:
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
f"The smart lockout duration is set to {duration} seconds, at or "
|
||||
f"above the recommended minimum of {MIN_LOCKOUT_DURATION_SECONDS}."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
+37
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_password_protection_lockout_threshold_limited",
|
||||
"CheckTitle": "Smart lockout threshold is set to 10 or less",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "The tenant Password Rule Settings should set the smart **LockoutThreshold** to **10 or less**. The lockout threshold determines how many failed sign-in attempts are permitted before an account is placed in a locked-out state.",
|
||||
"Risk": "A high lockout threshold gives attackers more attempts per account during **password spraying** and **brute-force** attacks before lockout is triggered, increasing the chance of a successful credential compromise.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/identity/authentication/howto-password-smart-lockout"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Protection** > **Authentication methods** > **Password protection**\n3. Set **Lockout threshold** to **10** or less\n4. Click **Save**",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Set the smart lockout threshold to 10 or less so accounts lock after a small number of failed sign-in attempts, limiting brute-force and password-spray attacks.",
|
||||
"Url": "https://hub.prowler.com/check/entra_password_protection_lockout_threshold_limited"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
+63
@@ -0,0 +1,63 @@
|
||||
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 (
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID,
|
||||
)
|
||||
|
||||
# CIS recommends a smart lockout threshold of 10 or less.
|
||||
MAX_LOCKOUT_THRESHOLD = 10
|
||||
|
||||
|
||||
class entra_password_protection_lockout_threshold_limited(Check):
|
||||
"""Check if the smart lockout threshold is set to 10 or less.
|
||||
|
||||
The Password Rule Settings directory setting should set LockoutThreshold to 10 or
|
||||
less so that accounts are locked after a small number of failed sign-in attempts.
|
||||
|
||||
- PASS: The lockout threshold is 10 or less.
|
||||
- FAIL: The lockout threshold is greater than 10 or not configured.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[CheckReportM365]:
|
||||
"""Execute the smart lockout threshold check.
|
||||
|
||||
Evaluate whether the Password Rule Settings directory setting limits the smart
|
||||
lockout threshold to the recommended maximum. When the settings object is
|
||||
absent, no finding is produced.
|
||||
|
||||
Returns:
|
||||
List[CheckReportM365]: A list with a single report when the Password Rule
|
||||
Settings exist, or an empty list when they are absent.
|
||||
"""
|
||||
findings = []
|
||||
settings = entra_client.directory_settings.get(
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID
|
||||
)
|
||||
if not settings:
|
||||
return findings
|
||||
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=settings or {},
|
||||
resource_name="Password Rule Settings",
|
||||
resource_id=PASSWORD_RULE_SETTINGS_TEMPLATE_ID,
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = "The smart lockout threshold is not set to 10 or less."
|
||||
|
||||
if settings:
|
||||
try:
|
||||
threshold = int(settings.get("LockoutThreshold"))
|
||||
except (TypeError, ValueError):
|
||||
threshold = None
|
||||
if threshold is not None and threshold <= MAX_LOCKOUT_THRESHOLD:
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
f"The smart lockout threshold is set to {threshold}, within the "
|
||||
f"recommended limit of {MAX_LOCKOUT_THRESHOLD}."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
+37
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_password_protection_on_premises_enforced",
|
||||
"CheckTitle": "Entra password protection is enforced on on-premises Active Directory",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "The tenant Password Rule Settings should enable **password protection on Windows Server Active Directory** (**EnableBannedPasswordCheckOnPremises**) with the mode set to **Enforced**. This extends Entra banned-password checks to on-premises password changes in hybrid environments. This control only applies to tenants with on-premises directory synchronization.",
|
||||
"Risk": "Without **on-premises** enforcement, users in hybrid environments can set weak or banned passwords directly in Active Directory, bypassing Entra password protection and weakening the organization's overall password posture.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/identity/authentication/concept-password-ban-bad-on-premises"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Protection** > **Authentication methods** > **Password protection**\n3. Set **Enable password protection on Windows Server Active Directory** to **Yes**\n4. Set **Mode** to **Enforced**\n5. Click **Save** (requires the Entra Password Protection agents deployed on-premises)",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Enable on-premises Entra password protection in Enforced mode and deploy the password protection proxy and DC agents so banned-password rules apply to on-premises password changes.",
|
||||
"Url": "https://hub.prowler.com/check/entra_password_protection_on_premises_enforced"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
+75
@@ -0,0 +1,75 @@
|
||||
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 (
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID,
|
||||
)
|
||||
|
||||
|
||||
class entra_password_protection_on_premises_enforced(Check):
|
||||
"""Check if Entra password protection is enforced on on-premises Active Directory.
|
||||
|
||||
The Password Rule Settings directory setting should enable password protection on
|
||||
Windows Server Active Directory (EnableBannedPasswordCheckOnPremises) with the
|
||||
mode set to Enforced, so banned-password rules apply to hybrid on-premises
|
||||
password changes.
|
||||
|
||||
This check applies only to hybrid tenants with on-premises synchronization.
|
||||
|
||||
- PASS: On-premises password protection is enabled and set to Enforced.
|
||||
- FAIL: On-premises password protection is disabled or set to Audit only.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[CheckReportM365]:
|
||||
"""Execute the on-premises password protection enforcement check.
|
||||
|
||||
Evaluate whether the Password Rule Settings directory setting enables and
|
||||
enforces banned-password protection for on-premises Active Directory. When the
|
||||
settings object is absent or the tenant is confirmed cloud-only, no finding is
|
||||
produced.
|
||||
|
||||
Returns:
|
||||
List[CheckReportM365]: A list with a single report when the Password Rule
|
||||
Settings exist for a hybrid or unknown tenant, or an empty list when they
|
||||
are absent or the tenant is confirmed cloud-only.
|
||||
"""
|
||||
findings = []
|
||||
organizations = entra_client.organizations or []
|
||||
if organizations and not any(
|
||||
organization.on_premises_sync_enabled for organization in organizations
|
||||
):
|
||||
return findings
|
||||
|
||||
settings = entra_client.directory_settings.get(
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID
|
||||
)
|
||||
if not settings:
|
||||
return findings
|
||||
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=settings or {},
|
||||
resource_name="Password Rule Settings",
|
||||
resource_id=PASSWORD_RULE_SETTINGS_TEMPLATE_ID,
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"On-premises password protection is not enforced in the tenant."
|
||||
)
|
||||
|
||||
if settings:
|
||||
enabled = (
|
||||
str(settings.get("EnableBannedPasswordCheckOnPremises", "")).lower()
|
||||
== "true"
|
||||
)
|
||||
mode = str(settings.get("BannedPasswordCheckOnPremisesMode", "")).lower()
|
||||
if enabled and mode == "enforced":
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
"On-premises password protection is enabled and enforced in the "
|
||||
"tenant."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
+37
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_policy_default_user_cannot_create_m365_groups",
|
||||
"CheckTitle": "Non-admin users cannot create Microsoft 365 groups",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "The tenant Group.Unified directory setting should have **EnableGroupCreation** set to false so that non-admin users cannot create Microsoft 365 groups through the portal, API, or PowerShell. Microsoft 365 group creation should be delegated to a controlled set of users.",
|
||||
"Risk": "When any user can create Microsoft 365 groups, they can provision associated resources (SharePoint sites, Teams, mailboxes) without oversight, leading to group sprawl, ungoverned data locations, and a larger attack surface.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/microsoft-365/solutions/manage-creation-of-groups"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Identity** > **Groups** > **General**\n3. Set **Users can create Microsoft 365 groups in Azure portals, API or PowerShell** to **No**\n4. Optionally grant creation rights to a specific security group\n5. Click **Save**",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Disable self-service Microsoft 365 group creation for non-admin users and delegate creation to an approved security group as needed.",
|
||||
"Url": "https://hub.prowler.com/check/entra_policy_default_user_cannot_create_m365_groups"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
+55
@@ -0,0 +1,55 @@
|
||||
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 (
|
||||
GROUP_UNIFIED_SETTINGS_TEMPLATE_ID,
|
||||
)
|
||||
|
||||
|
||||
class entra_policy_default_user_cannot_create_m365_groups(Check):
|
||||
"""Check if default users are restricted from creating Microsoft 365 groups.
|
||||
|
||||
The Group.Unified directory setting should have EnableGroupCreation set to false
|
||||
so that non-admin users cannot create Microsoft 365 groups. If the setting does
|
||||
not exist, the tenant uses the default, which allows all users to create groups.
|
||||
|
||||
- PASS: Non-admin users cannot create Microsoft 365 groups.
|
||||
- FAIL: Non-admin users are allowed to create Microsoft 365 groups.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[CheckReportM365]:
|
||||
"""Evaluate whether default users can create Microsoft 365 groups.
|
||||
|
||||
Inspects the Group.Unified directory setting to determine whether non-admin
|
||||
users are allowed to create Microsoft 365 groups. When the setting is absent
|
||||
the tenant default (group creation allowed) applies.
|
||||
|
||||
Returns:
|
||||
List[CheckReportM365]: A single report indicating whether non-admin users
|
||||
are restricted from creating Microsoft 365 groups.
|
||||
"""
|
||||
findings = []
|
||||
settings = entra_client.directory_settings.get(
|
||||
GROUP_UNIFIED_SETTINGS_TEMPLATE_ID
|
||||
)
|
||||
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=settings or {},
|
||||
resource_name="Group.Unified Settings",
|
||||
resource_id=GROUP_UNIFIED_SETTINGS_TEMPLATE_ID,
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"Non-admin users are allowed to create Microsoft 365 groups."
|
||||
)
|
||||
|
||||
if settings and str(settings.get("EnableGroupCreation", "")).lower() == "false":
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
"Non-admin users are not allowed to create Microsoft 365 groups."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
+37
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_policy_default_user_cannot_create_security_groups",
|
||||
"CheckTitle": "Non-admin users cannot create security groups",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "Microsoft Entra tenant's authorization policy should restrict **non-admin users** from creating **security groups**. Security groups can be used to grant access to resources across Microsoft 365, so their creation should be limited to administrators to preserve least privilege and prevent uncontrolled access grants.",
|
||||
"Risk": "When any user can create security groups, they may grant themselves or others access to resources, circumventing governance controls. Uncontrolled group sprawl also complicates access reviews and increases the attack surface for privilege escalation.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/identity/users/groups-self-service-management"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Identity** > **Groups** > **All groups** > **General**\n3. Under **Security groups**, set **Users can create security groups in Azure portals, API or PowerShell** to **No**\n4. Click **Save**",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Restrict security group creation to administrators by disabling self-service security group creation for non-admin users. Grant group-creation rights only to specific roles or delegated owners as required.",
|
||||
"Url": "https://hub.prowler.com/check/entra_policy_default_user_cannot_create_security_groups"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
+49
@@ -0,0 +1,49 @@
|
||||
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_policy_default_user_cannot_create_security_groups(Check):
|
||||
"""Check if default users are restricted from creating security groups.
|
||||
|
||||
This check verifies whether the authorization policy prevents non-admin users
|
||||
from creating security groups in Microsoft Entra ID.
|
||||
|
||||
- PASS: Non-admin users cannot create security groups.
|
||||
- FAIL: Non-admin users are allowed to create security groups.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[CheckReportM365]:
|
||||
"""Execute the check for security group creation restrictions.
|
||||
|
||||
This method examines the authorization policy settings to determine if
|
||||
non-admin users are allowed to create security groups. If security group
|
||||
creation is restricted, the check passes.
|
||||
|
||||
Returns:
|
||||
List[CheckReportM365]: A list containing the result of the check.
|
||||
"""
|
||||
findings = []
|
||||
auth_policy = entra_client.authorization_policy
|
||||
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=auth_policy if auth_policy else {},
|
||||
resource_name=auth_policy.name if auth_policy else "Authorization Policy",
|
||||
resource_id=auth_policy.id if auth_policy else "authorizationPolicy",
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"Non-admin users are allowed to create security groups."
|
||||
)
|
||||
|
||||
permissions = getattr(auth_policy, "default_user_role_permissions", None)
|
||||
if permissions and permissions.allowed_to_create_security_groups is False:
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
"Non-admin users are not allowed to create security groups."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
+38
@@ -0,0 +1,38 @@
|
||||
{
|
||||
"Provider": "m365",
|
||||
"CheckID": "entra_policy_guest_invitations_restricted_to_allowed_domains",
|
||||
"CheckTitle": "Guest invitations are restricted to an allow-list of domains",
|
||||
"CheckType": [],
|
||||
"ServiceName": "entra",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "The external collaboration settings should **allow invitations only to specified domains** (most restrictive). An explicit allow-list limits B2B guest invitations to trusted partner organizations, while an empty allow-list blocks invitations from all external domains.",
|
||||
"Risk": "Allowing invitations to any domain lets users invite guests from arbitrary or malicious organizations, increasing the risk of data exposure to untrusted external parties and expanding the tenant's collaboration attack surface.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://learn.microsoft.com/en-us/entra/external-id/allow-deny-list"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Navigate to the Microsoft Entra admin center at https://entra.microsoft.com/\n2. Go to **Entra ID** > **External Identities** > **External collaboration settings**\n3. Under **Collaboration restrictions**, select **Allow invitations only to the specified domains (most restrictive)**\n4. Add trusted partner domains under **Target domains**, or leave the list empty to block all external invitations\n5. Click **Save**",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Restrict guest invitations to an allow-list of trusted partner domains, or use an empty allow-list to block all external invitations.",
|
||||
"Url": "https://hub.prowler.com/check/entra_policy_guest_invitations_restricted_to_allowed_domains"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access",
|
||||
"trust-boundaries",
|
||||
"e3"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
+60
@@ -0,0 +1,60 @@
|
||||
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_policy_guest_invitations_restricted_to_allowed_domains(Check):
|
||||
"""Check if guest invitations are restricted by an allowed-domain policy.
|
||||
|
||||
The B2B collaboration policy should allow invitations only to a specified list of
|
||||
domains (most restrictive). An empty allow-list blocks invitations from every
|
||||
external domain and is compliant.
|
||||
|
||||
- PASS: Invitations are restricted to an allow-list, including an empty block-all
|
||||
list.
|
||||
- FAIL: Invitations are not restricted to an allow-list of domains.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[CheckReportM365]:
|
||||
"""Evaluate whether guest invitations are restricted to allowed domains.
|
||||
|
||||
Inspects the B2B collaboration policy to determine whether guest invitations
|
||||
are limited to an allow-list of domains. An empty allow-list blocks all external
|
||||
invitations.
|
||||
|
||||
Returns:
|
||||
List[CheckReportM365]: A single report indicating whether guest
|
||||
invitations are restricted by an allowed-domain policy, or an empty list
|
||||
when the policy is absent.
|
||||
"""
|
||||
findings = []
|
||||
policy = entra_client.b2b_collaboration_policy
|
||||
if not policy:
|
||||
return findings
|
||||
|
||||
report = CheckReportM365(
|
||||
metadata=self.metadata(),
|
||||
resource=policy,
|
||||
resource_name="B2B Collaboration Policy",
|
||||
resource_id="b2bManagementPolicy",
|
||||
)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
"Guest invitations are not restricted to an allow-list of domains."
|
||||
)
|
||||
|
||||
if policy.invitations_restricted_to_allowed_domains:
|
||||
report.status = "PASS"
|
||||
if policy.allowed_domains:
|
||||
report.status_extended = (
|
||||
"Guest invitations are restricted to an allow-list of "
|
||||
f"{len(policy.allowed_domains)} domain(s)."
|
||||
)
|
||||
else:
|
||||
report.status_extended = (
|
||||
"Guest invitations are blocked for all external domains."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
return findings
|
||||
@@ -103,6 +103,8 @@ class Entra(M365Service):
|
||||
self._get_app_registrations(),
|
||||
self._get_exchange_mailbox_permission_service_principals(),
|
||||
self._get_device_registration_policy(),
|
||||
self._get_directory_settings(),
|
||||
self._get_b2b_collaboration_policy(),
|
||||
)
|
||||
)
|
||||
|
||||
@@ -126,6 +128,8 @@ class Entra(M365Service):
|
||||
self.device_registration_policy: Optional[DeviceRegistrationPolicy] = (
|
||||
attributes[13]
|
||||
)
|
||||
self.directory_settings: Dict[str, Dict[str, str]] = attributes[14]
|
||||
self.b2b_collaboration_policy: Optional[B2BCollaborationPolicy] = attributes[15]
|
||||
self.user_accounts_status = {}
|
||||
|
||||
# Resolve directory-object identifiers referenced by Conditional Access
|
||||
@@ -1243,6 +1247,98 @@ OAuthAppInfo
|
||||
)
|
||||
return device_registration_policy
|
||||
|
||||
async def _get_b2b_collaboration_policy(self):
|
||||
"""Retrieve the legacy B2B collaboration (invitation domains) policy.
|
||||
|
||||
Fetches the legacy ``B2BManagementPolicy`` to determine whether invitations
|
||||
are restricted to an allow-list of domains.
|
||||
|
||||
Returns:
|
||||
Optional[B2BCollaborationPolicy]: The parsed policy, or None on error.
|
||||
"""
|
||||
logger.info("Entra - Getting B2B collaboration policy...")
|
||||
b2b_policy = None
|
||||
try:
|
||||
url = "https://graph.microsoft.com/beta/legacy/policies"
|
||||
builder = self.client.policies.with_url(url)
|
||||
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)
|
||||
# The legacy policy object has no string ``type`` discriminator, so
|
||||
# match on the ``B2BManagementPolicy`` block inside the definition JSON.
|
||||
for policy in data.get("value", []) or []:
|
||||
matched = False
|
||||
allowed_domains = []
|
||||
invitations_restricted = False
|
||||
for definition in policy.get("definition", []) or []:
|
||||
try:
|
||||
parsed = json.loads(definition)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
b2b_block = parsed.get("B2BManagementPolicy")
|
||||
if not b2b_block:
|
||||
continue
|
||||
matched = True
|
||||
invitation_policy = (
|
||||
b2b_block.get(
|
||||
"InvitationsAllowedAndBlockedDomainsPolicy", {}
|
||||
)
|
||||
or {}
|
||||
)
|
||||
# Allow-list mode is active whenever the AllowedDomains key is
|
||||
# present, even when empty (empty = block all external invites,
|
||||
# the most restrictive and CIS-compliant state).
|
||||
if "AllowedDomains" in invitation_policy:
|
||||
invitations_restricted = True
|
||||
allowed_domains = (
|
||||
invitation_policy.get("AllowedDomains") or []
|
||||
)
|
||||
if matched:
|
||||
b2b_policy = B2BCollaborationPolicy(
|
||||
invitations_restricted_to_allowed_domains=invitations_restricted,
|
||||
allowed_domains=allowed_domains,
|
||||
)
|
||||
break
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
return b2b_policy
|
||||
|
||||
async def _get_directory_settings(self):
|
||||
"""Retrieve tenant directory (group) settings from Microsoft Entra.
|
||||
|
||||
Fetches the ``/groupSettings`` collection and returns a mapping of each
|
||||
setting's ``templateId`` to a dict of its ``name``/``value`` pairs. This
|
||||
exposes the Group.Unified and Password Rule Settings templates used by the
|
||||
group-creation and password-protection checks.
|
||||
|
||||
Returns:
|
||||
Dict[str, Dict[str, str]]: Mapping of template ID to its name/value pairs.
|
||||
"""
|
||||
logger.info("Entra - Getting directory (group) settings...")
|
||||
directory_settings: Dict[str, Dict[str, str]] = {}
|
||||
try:
|
||||
response = await self.client.group_settings.get()
|
||||
for setting in getattr(response, "value", []) or []:
|
||||
template_id = getattr(setting, "template_id", None)
|
||||
if not template_id:
|
||||
continue
|
||||
values = {}
|
||||
for value in getattr(setting, "values", []) or []:
|
||||
name = getattr(value, "name", None)
|
||||
if name is not None:
|
||||
values[name] = getattr(value, "value", None)
|
||||
directory_settings[template_id] = values
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
return directory_settings
|
||||
|
||||
async def _get_service_principals(self):
|
||||
"""Retrieve service principals owned by the audited tenant.
|
||||
|
||||
@@ -1976,6 +2072,11 @@ class AuthorizationPolicy(BaseModel):
|
||||
guest_user_role_id: Optional[UUID]
|
||||
|
||||
|
||||
# Well-known directory setting template IDs (from /groupSettings).
|
||||
GROUP_UNIFIED_SETTINGS_TEMPLATE_ID = "62375ab9-6b52-47ed-826b-58e47e0e304b"
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID = "5cf42378-d67d-4f36-ba46-e8b86229381d"
|
||||
|
||||
|
||||
class DeviceRegistrationMembershipType(str, Enum):
|
||||
"""OData types for Entra device registration membership settings."""
|
||||
|
||||
@@ -1994,6 +2095,13 @@ class DeviceRegistrationPolicy(BaseModel):
|
||||
local_admin_password_enabled: Optional[bool] = None
|
||||
|
||||
|
||||
class B2BCollaborationPolicy(BaseModel):
|
||||
"""Legacy B2B collaboration (invitation domains) policy."""
|
||||
|
||||
invitations_restricted_to_allowed_domains: bool = False
|
||||
allowed_domains: List[str] = []
|
||||
|
||||
|
||||
class Organization(BaseModel):
|
||||
id: str
|
||||
name: str
|
||||
|
||||
+444
@@ -0,0 +1,444 @@
|
||||
import os
|
||||
import zipfile
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.aws.services.awslambda.awslambda_service import (
|
||||
LambdaCode,
|
||||
Layer,
|
||||
)
|
||||
from tests.providers.aws.services.awslambda.awslambda_service_test import (
|
||||
create_zip_file,
|
||||
)
|
||||
from tests.providers.aws.utils import (
|
||||
AWS_ACCOUNT_NUMBER,
|
||||
AWS_REGION_US_EAST_1,
|
||||
set_mocked_aws_provider,
|
||||
)
|
||||
|
||||
LAMBDA_LAYER_NAME = "test-layer"
|
||||
LAMBDA_LAYER_ARN = (
|
||||
f"arn:aws:lambda:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:layer:"
|
||||
f"{LAMBDA_LAYER_NAME}:1"
|
||||
)
|
||||
LAMBDA_UNFETCHED_LAYER_NAME = "unfetched-layer"
|
||||
LAMBDA_UNFETCHED_LAYER_ARN = (
|
||||
f"arn:aws:lambda:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:layer:"
|
||||
f"{LAMBDA_UNFETCHED_LAYER_NAME}:2"
|
||||
)
|
||||
LAMBDA_CORRUPT_LAYER_NAME = "corrupt-layer"
|
||||
LAMBDA_CORRUPT_LAYER_ARN = (
|
||||
f"arn:aws:lambda:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:layer:"
|
||||
f"{LAMBDA_CORRUPT_LAYER_NAME}:3"
|
||||
)
|
||||
LAMBDA_LAYER_CONTENT_WITH_SECRETS = """
|
||||
db_password = "Tr0ub4dor3xKq9vLmZ"
|
||||
"""
|
||||
LAMBDA_LAYER_CONTENT_WITHOUT_SECRETS = """
|
||||
def helper():
|
||||
return True
|
||||
"""
|
||||
|
||||
|
||||
def create_lambda_layer() -> Layer:
|
||||
return Layer(arn=LAMBDA_LAYER_ARN)
|
||||
|
||||
|
||||
def get_lambda_layer_code(content):
|
||||
return LambdaCode(
|
||||
location="",
|
||||
code_zip=zipfile.ZipFile(create_zip_file(content)),
|
||||
)
|
||||
|
||||
|
||||
def get_lambda_layer_code_from_files(files: dict) -> LambdaCode:
|
||||
# The check only calls code_zip.extractall(dir); mock it to drop the
|
||||
# given files into the temporary directory the check creates, so no
|
||||
# real archive needs to be built.
|
||||
code_zip = mock.MagicMock()
|
||||
|
||||
def _extractall(path):
|
||||
for name, content in files.items():
|
||||
os.makedirs(os.path.dirname(f"{path}/{name}"), exist_ok=True)
|
||||
with open(f"{path}/{name}", "w") as fd:
|
||||
fd.write(content)
|
||||
|
||||
code_zip.extractall.side_effect = _extractall
|
||||
return LambdaCode(location="", code_zip=code_zip)
|
||||
|
||||
|
||||
def mock_get_layers_code_with_nested_vendor_secret():
|
||||
yield create_lambda_layer(), get_lambda_layer_code_from_files(
|
||||
{
|
||||
"python/lib.py": LAMBDA_LAYER_CONTENT_WITHOUT_SECRETS,
|
||||
"python/vendor/package.js": 'const dbPassword = "test-vendor-password";',
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def mock_get_layers_code_with_secrets():
|
||||
yield create_lambda_layer(), get_lambda_layer_code(
|
||||
LAMBDA_LAYER_CONTENT_WITH_SECRETS
|
||||
)
|
||||
|
||||
|
||||
def mock_get_layers_code_without_secrets():
|
||||
yield create_lambda_layer(), get_lambda_layer_code(
|
||||
LAMBDA_LAYER_CONTENT_WITHOUT_SECRETS
|
||||
)
|
||||
|
||||
|
||||
def get_lambda_layer_code_with_unreadable_file() -> LambdaCode:
|
||||
# A dangling symlink is walked as a file but cannot be opened, which is
|
||||
# how an unreadable member of the layer package behaves for the check.
|
||||
code_zip = mock.MagicMock()
|
||||
|
||||
def _extractall(path):
|
||||
with open(f"{path}/readable.py", "w") as fd:
|
||||
fd.write(LAMBDA_LAYER_CONTENT_WITHOUT_SECRETS)
|
||||
os.symlink(f"{path}/does-not-exist", f"{path}/dangling.py")
|
||||
|
||||
code_zip.extractall.side_effect = _extractall
|
||||
return LambdaCode(location="", code_zip=code_zip)
|
||||
|
||||
|
||||
def mock_get_layers_code_with_unreadable_file():
|
||||
yield create_lambda_layer(), get_lambda_layer_code_with_unreadable_file()
|
||||
|
||||
|
||||
def mock_get_layers_code_empty_code():
|
||||
yield create_lambda_layer(), None
|
||||
|
||||
|
||||
def get_lambda_layer_code_with_corrupt_archive() -> LambdaCode:
|
||||
code_zip = mock.MagicMock()
|
||||
code_zip.extractall.side_effect = zipfile.BadZipFile("truncated archive")
|
||||
return LambdaCode(location="", code_zip=code_zip)
|
||||
|
||||
|
||||
def mock_get_layers_code_one_corrupt_one_clean():
|
||||
yield (
|
||||
Layer(arn=LAMBDA_CORRUPT_LAYER_ARN),
|
||||
get_lambda_layer_code_with_corrupt_archive(),
|
||||
)
|
||||
yield create_lambda_layer(), get_lambda_layer_code(
|
||||
LAMBDA_LAYER_CONTENT_WITHOUT_SECRETS
|
||||
)
|
||||
|
||||
|
||||
def mock_get_layers_code_partial_fetch_failure():
|
||||
# Only the fetchable layer is yielded; the client's failing fetch for
|
||||
# the other layer already logged and skipped it (see _get_layers_code).
|
||||
yield create_lambda_layer(), get_lambda_layer_code(
|
||||
LAMBDA_LAYER_CONTENT_WITHOUT_SECRETS
|
||||
)
|
||||
|
||||
|
||||
class Test_awslambda_layer_no_secrets_in_content:
|
||||
def test_no_layers(self):
|
||||
lambda_client = mock.MagicMock
|
||||
lambda_client.layers = {}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_aws_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content.awslambda_client",
|
||||
new=lambda_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content import (
|
||||
awslambda_layer_no_secrets_in_content,
|
||||
)
|
||||
|
||||
check = awslambda_layer_no_secrets_in_content()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 0
|
||||
|
||||
def test_layer_content_with_secrets(self):
|
||||
lambda_client = mock.MagicMock
|
||||
lambda_client.layers = {LAMBDA_LAYER_ARN: create_lambda_layer()}
|
||||
lambda_client._get_layers_code = mock_get_layers_code_with_secrets
|
||||
lambda_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_aws_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content.awslambda_client",
|
||||
new=lambda_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content import (
|
||||
awslambda_layer_no_secrets_in_content,
|
||||
)
|
||||
|
||||
check = awslambda_layer_no_secrets_in_content()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_id == LAMBDA_LAYER_NAME
|
||||
assert result[0].resource_arn == LAMBDA_LAYER_ARN
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"Potential secret found in Lambda layer {LAMBDA_LAYER_NAME} (version 1) content -> lambda_function.py: Generic Password on line 2."
|
||||
)
|
||||
assert result[0].resource_tags == []
|
||||
|
||||
def test_layer_content_without_secrets(self):
|
||||
lambda_client = mock.MagicMock
|
||||
lambda_client.layers = {LAMBDA_LAYER_ARN: create_lambda_layer()}
|
||||
lambda_client._get_layers_code = mock_get_layers_code_without_secrets
|
||||
lambda_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_aws_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content.awslambda_client",
|
||||
new=lambda_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content import (
|
||||
awslambda_layer_no_secrets_in_content,
|
||||
)
|
||||
|
||||
check = awslambda_layer_no_secrets_in_content()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_id == LAMBDA_LAYER_NAME
|
||||
assert result[0].resource_arn == LAMBDA_LAYER_ARN
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"No secrets found in Lambda layer {LAMBDA_LAYER_NAME} (version 1) content."
|
||||
)
|
||||
assert result[0].resource_tags == []
|
||||
|
||||
def test_layer_content_nested_vendor_secret_not_ignored(self):
|
||||
lambda_client = mock.MagicMock
|
||||
lambda_client.layers = {LAMBDA_LAYER_ARN: create_lambda_layer()}
|
||||
lambda_client._get_layers_code = mock_get_layers_code_with_nested_vendor_secret
|
||||
lambda_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_aws_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content.awslambda_client",
|
||||
new=lambda_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content import (
|
||||
awslambda_layer_no_secrets_in_content,
|
||||
)
|
||||
|
||||
check = awslambda_layer_no_secrets_in_content()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "python/vendor/package.js" in result[0].status_extended
|
||||
|
||||
def test_layer_content_nested_vendor_secret_ignored_by_file_pattern(self):
|
||||
lambda_client = mock.MagicMock
|
||||
lambda_client.layers = {LAMBDA_LAYER_ARN: create_lambda_layer()}
|
||||
lambda_client._get_layers_code = mock_get_layers_code_with_nested_vendor_secret
|
||||
lambda_client.audit_config = {
|
||||
"secrets_ignore_patterns": [],
|
||||
"secrets_ignore_files": ["python/vendor/*.js"],
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_aws_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content.awslambda_client",
|
||||
new=lambda_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content import (
|
||||
awslambda_layer_no_secrets_in_content,
|
||||
)
|
||||
|
||||
check = awslambda_layer_no_secrets_in_content()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
def test_layer_content_unreadable_file_is_skipped(self):
|
||||
lambda_client = mock.MagicMock
|
||||
lambda_client.layers = {LAMBDA_LAYER_ARN: create_lambda_layer()}
|
||||
lambda_client._get_layers_code = mock_get_layers_code_with_unreadable_file
|
||||
lambda_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_aws_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content.awslambda_client",
|
||||
new=lambda_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content import (
|
||||
awslambda_layer_no_secrets_in_content,
|
||||
)
|
||||
|
||||
check = awslambda_layer_no_secrets_in_content()
|
||||
result = check.execute()
|
||||
|
||||
# The unreadable file is skipped, the rest of the package is still
|
||||
# scanned, so the layer is reported instead of being dropped.
|
||||
assert len(result) == 1
|
||||
assert result[0].resource_arn == LAMBDA_LAYER_ARN
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
def test_corrupt_layer_archive_reports_manual_and_scan_continues(self):
|
||||
lambda_client = mock.MagicMock
|
||||
lambda_client.layers = {
|
||||
LAMBDA_CORRUPT_LAYER_ARN: Layer(arn=LAMBDA_CORRUPT_LAYER_ARN),
|
||||
LAMBDA_LAYER_ARN: create_lambda_layer(),
|
||||
}
|
||||
lambda_client._get_layers_code = mock_get_layers_code_one_corrupt_one_clean
|
||||
lambda_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_aws_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content.awslambda_client",
|
||||
new=lambda_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content import (
|
||||
awslambda_layer_no_secrets_in_content,
|
||||
)
|
||||
|
||||
check = awslambda_layer_no_secrets_in_content()
|
||||
result = check.execute()
|
||||
|
||||
# The corrupt archive must not abort the scan of the clean layer.
|
||||
assert len(result) == 2
|
||||
by_arn = {r.resource_arn: r for r in result}
|
||||
|
||||
assert by_arn[LAMBDA_LAYER_ARN].status == "PASS"
|
||||
|
||||
corrupt = by_arn[LAMBDA_CORRUPT_LAYER_ARN]
|
||||
assert corrupt.status == "MANUAL"
|
||||
assert "manual review is required" in corrupt.status_extended
|
||||
|
||||
def test_layer_with_empty_code_reports_manual(self):
|
||||
lambda_client = mock.MagicMock
|
||||
lambda_client.layers = {LAMBDA_LAYER_ARN: create_lambda_layer()}
|
||||
lambda_client._get_layers_code = mock_get_layers_code_empty_code
|
||||
lambda_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_aws_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content.awslambda_client",
|
||||
new=lambda_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content import (
|
||||
awslambda_layer_no_secrets_in_content,
|
||||
)
|
||||
|
||||
check = awslambda_layer_no_secrets_in_content()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "MANUAL"
|
||||
assert "Could not retrieve content" in result[0].status_extended
|
||||
|
||||
def test_partial_fetch_failure_reports_manual_for_unfetched_layer(self):
|
||||
lambda_client = mock.MagicMock
|
||||
lambda_client.layers = {
|
||||
LAMBDA_LAYER_ARN: create_lambda_layer(),
|
||||
LAMBDA_UNFETCHED_LAYER_ARN: Layer(arn=LAMBDA_UNFETCHED_LAYER_ARN),
|
||||
}
|
||||
lambda_client._get_layers_code = mock_get_layers_code_partial_fetch_failure
|
||||
lambda_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_aws_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content.awslambda_client",
|
||||
new=lambda_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content import (
|
||||
awslambda_layer_no_secrets_in_content,
|
||||
)
|
||||
|
||||
check = awslambda_layer_no_secrets_in_content()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 2
|
||||
by_arn = {r.resource_arn: r for r in result}
|
||||
|
||||
assert by_arn[LAMBDA_LAYER_ARN].status == "PASS"
|
||||
|
||||
unfetched = by_arn[LAMBDA_UNFETCHED_LAYER_ARN]
|
||||
assert unfetched.status == "MANUAL"
|
||||
assert unfetched.resource_id == LAMBDA_UNFETCHED_LAYER_NAME
|
||||
assert unfetched.region == AWS_REGION_US_EAST_1
|
||||
assert "manual review is required" in unfetched.status_extended
|
||||
|
||||
def test_scan_failure_reports_manual_not_pass(self):
|
||||
from prowler.lib.utils.utils import SecretsScanError
|
||||
|
||||
lambda_client = mock.MagicMock
|
||||
lambda_client.layers = {LAMBDA_LAYER_ARN: create_lambda_layer()}
|
||||
lambda_client._get_layers_code = mock_get_layers_code_with_secrets
|
||||
lambda_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_aws_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content.awslambda_client",
|
||||
new=lambda_client,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content.detect_secrets_scan_batch",
|
||||
side_effect=SecretsScanError("Kingfisher exited with code 1"),
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.awslambda.awslambda_layer_no_secrets_in_content.awslambda_layer_no_secrets_in_content import (
|
||||
awslambda_layer_no_secrets_in_content,
|
||||
)
|
||||
|
||||
check = awslambda_layer_no_secrets_in_content()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "MANUAL"
|
||||
assert "Could not scan" in result[0].status_extended
|
||||
@@ -15,6 +15,7 @@ from prowler.providers.aws.services.awslambda.awslambda_service import (
|
||||
AuthType,
|
||||
Function,
|
||||
Lambda,
|
||||
Layer,
|
||||
)
|
||||
from tests.providers.aws.utils import (
|
||||
AWS_ACCOUNT_NUMBER,
|
||||
@@ -44,7 +45,7 @@ def create_zip_file(code: str = "") -> io.BytesIO:
|
||||
return zip_output
|
||||
|
||||
|
||||
def mock_request_get(_):
|
||||
def mock_request_get(_, **kwargs):
|
||||
"""Mock requests.get() to get the Lambda Code in Zip Format"""
|
||||
mock_resp = mock.MagicMock
|
||||
mock_resp.status_code = 200
|
||||
@@ -680,3 +681,158 @@ class Test_Lambda_Service:
|
||||
|
||||
assert len(list(awslambda._get_function_code())) == 1
|
||||
assert len(fetched) == 1
|
||||
|
||||
def test_layer_properties_parsed_from_arn(self):
|
||||
layer = Layer(
|
||||
arn=f"arn:aws:lambda:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:layer:my-layer:3"
|
||||
)
|
||||
|
||||
assert layer.region == AWS_REGION_US_EAST_1
|
||||
assert layer.name == "my-layer"
|
||||
assert layer.version == "3"
|
||||
assert layer.account_id == AWS_ACCOUNT_NUMBER
|
||||
|
||||
def test_collect_layers_deduplicates_across_functions(self):
|
||||
layer_arn = f"arn:aws:lambda:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:layer:shared-layer:1"
|
||||
awslambda = Lambda.__new__(Lambda)
|
||||
awslambda.layers = {}
|
||||
awslambda.functions = {
|
||||
"function-1": Function(
|
||||
name="function-1",
|
||||
arn="function-1",
|
||||
security_groups=[],
|
||||
region=AWS_REGION_US_EAST_1,
|
||||
layers=[Layer(arn=layer_arn)],
|
||||
),
|
||||
"function-2": Function(
|
||||
name="function-2",
|
||||
arn="function-2",
|
||||
security_groups=[],
|
||||
region=AWS_REGION_US_EAST_1,
|
||||
layers=[Layer(arn=layer_arn)],
|
||||
),
|
||||
}
|
||||
|
||||
awslambda._collect_layers()
|
||||
|
||||
assert len(awslambda.layers) == 1
|
||||
assert awslambda.layers[layer_arn].arn == layer_arn
|
||||
|
||||
@mock_aws
|
||||
def test_get_layers_code_fetches_each_layer_once(self):
|
||||
iam_client = client("iam", region_name=AWS_REGION_US_EAST_1)
|
||||
iam_role = iam_client.create_role(
|
||||
RoleName="test-role",
|
||||
AssumeRolePolicyDocument="{}",
|
||||
)["Role"]["Arn"]
|
||||
lambda_client = client("lambda", region_name=AWS_REGION_US_EAST_1)
|
||||
layer_code = "shared_secret = 'hunter2'"
|
||||
layer_arn = lambda_client.publish_layer_version(
|
||||
LayerName="shared-layer",
|
||||
Content={"ZipFile": create_zip_file(layer_code).read()},
|
||||
CompatibleRuntimes=["python3.9"],
|
||||
)["LayerVersionArn"]
|
||||
for name in ("function-1", "function-2"):
|
||||
lambda_client.create_function(
|
||||
FunctionName=name,
|
||||
Runtime="python3.9",
|
||||
Role=iam_role,
|
||||
Handler="lambda_function.lambda_handler",
|
||||
Code={"ZipFile": create_zip_file().read()},
|
||||
PackageType="ZIP",
|
||||
Layers=[layer_arn],
|
||||
)
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.awslambda.awslambda_service.requests.get",
|
||||
new=mock_request_get,
|
||||
):
|
||||
awslambda = Lambda(
|
||||
set_mocked_aws_provider(audited_regions=[AWS_REGION_US_EAST_1])
|
||||
)
|
||||
|
||||
assert len(awslambda.layers) == 1
|
||||
assert awslambda.layers[layer_arn].name == "shared-layer"
|
||||
assert awslambda.layers[layer_arn].version == "1"
|
||||
|
||||
# moto's get_layer_version_by_arn omits Content.Location, so
|
||||
# delegate to get_layer_version, which moto implements fully.
|
||||
regional_client = awslambda.regional_clients[AWS_REGION_US_EAST_1]
|
||||
|
||||
def get_layer_version_by_arn(Arn):
|
||||
assert Arn == layer_arn
|
||||
return regional_client.get_layer_version(
|
||||
LayerName="shared-layer", VersionNumber=1
|
||||
)
|
||||
|
||||
with mock.patch.object(
|
||||
regional_client,
|
||||
"get_layer_version_by_arn",
|
||||
side_effect=get_layer_version_by_arn,
|
||||
):
|
||||
layers_fetched = list(awslambda._get_layers_code())
|
||||
assert len(layers_fetched) == 1
|
||||
fetched_layer, fetched_code = layers_fetched[0]
|
||||
assert fetched_layer.arn == layer_arn
|
||||
assert fetched_code
|
||||
|
||||
@mock_aws
|
||||
def test_get_layers_code_skips_layer_that_cannot_be_fetched(self):
|
||||
awslambda = Lambda(
|
||||
set_mocked_aws_provider(audited_regions=[AWS_REGION_US_EAST_1])
|
||||
)
|
||||
layer_arn = f"arn:aws:lambda:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:layer:missing-layer:1"
|
||||
awslambda.layers = {layer_arn: Layer(arn=layer_arn)}
|
||||
# moto answers GetLayerVersionByArn with an empty stub instead of
|
||||
# raising for an unknown layer, so the failure is forced here.
|
||||
regional_client = mock.MagicMock()
|
||||
regional_client.get_layer_version_by_arn.side_effect = Exception(
|
||||
"ResourceNotFoundException"
|
||||
)
|
||||
awslambda.regional_clients[AWS_REGION_US_EAST_1] = regional_client
|
||||
|
||||
# The lookup raises inside _fetch_layer_code; _get_layers_code must
|
||||
# log it and yield nothing rather than propagating to the check.
|
||||
assert list(awslambda._get_layers_code()) == []
|
||||
|
||||
@mock_aws
|
||||
def test_fetch_layer_code_returns_none_without_location(self):
|
||||
awslambda = Lambda(
|
||||
set_mocked_aws_provider(audited_regions=[AWS_REGION_US_EAST_1])
|
||||
)
|
||||
awslambda.regional_clients[AWS_REGION_US_EAST_1] = mock.MagicMock()
|
||||
awslambda.regional_clients[
|
||||
AWS_REGION_US_EAST_1
|
||||
].get_layer_version_by_arn.return_value = {"Content": {}}
|
||||
|
||||
layer_arn = f"arn:aws:lambda:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:layer:my-layer:1"
|
||||
assert awslambda._fetch_layer_code(layer_arn, AWS_REGION_US_EAST_1) is None
|
||||
|
||||
# An absent Content must be handled like an empty one rather than
|
||||
# raising on the membership test.
|
||||
awslambda.regional_clients[
|
||||
AWS_REGION_US_EAST_1
|
||||
].get_layer_version_by_arn.return_value = {"Content": None}
|
||||
|
||||
assert awslambda._fetch_layer_code(layer_arn, AWS_REGION_US_EAST_1) is None
|
||||
|
||||
@mock_aws
|
||||
def test_fetch_layer_code_uses_full_layer_version_arn(self):
|
||||
awslambda = Lambda(
|
||||
set_mocked_aws_provider(audited_regions=[AWS_REGION_US_EAST_1])
|
||||
)
|
||||
regional_client = mock.MagicMock()
|
||||
regional_client.get_layer_version_by_arn.return_value = {"Content": {}}
|
||||
awslambda.regional_clients[AWS_REGION_US_EAST_1] = regional_client
|
||||
|
||||
# A layer owned by another account must be fetched by its full
|
||||
# layer-version ARN, never by the bare layer name.
|
||||
foreign_layer_arn = (
|
||||
f"arn:aws:lambda:{AWS_REGION_US_EAST_1}:999999999999:"
|
||||
"layer:vendor-extension:5"
|
||||
)
|
||||
awslambda._fetch_layer_code(foreign_layer_arn, AWS_REGION_US_EAST_1)
|
||||
|
||||
regional_client.get_layer_version_by_arn.assert_called_once_with(
|
||||
Arn=foreign_layer_arn
|
||||
)
|
||||
|
||||
+535
@@ -0,0 +1,535 @@
|
||||
from unittest import mock
|
||||
from unittest.mock import patch
|
||||
|
||||
from boto3 import client
|
||||
from moto import mock_aws
|
||||
|
||||
from prowler.providers.aws.services.batch.batch_service import (
|
||||
BatchContainerProperties,
|
||||
BatchJobDefinition,
|
||||
ContainerEnvVariable,
|
||||
)
|
||||
from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
|
||||
|
||||
JOB_NAME = "test-batch-job"
|
||||
JOB_REVISION = 1
|
||||
ENV_VAR_NAME_NO_SECRETS = "host"
|
||||
ENV_VAR_VALUE_NO_SECRETS = "localhost:1234"
|
||||
ENV_VAR_NAME_WITH_KEYWORD = "DB_PASSWORD"
|
||||
# Realistic fake secrets that Kingfisher actually detects.
|
||||
ENV_VAR_VALUE_WITH_SECRETS = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.dozjgNryP4J3jVmNHl0w5N_XgL0n3I9PlFUP0THsR8U"
|
||||
ENV_VAR_NAME_WITH_KEYWORD2 = "DATABASE_PASSWORD"
|
||||
ENV_VAR_VALUE_WITH_SECRETS2 = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiI5ODc2NTQzMjEwIiwibmFtZSI6IkphbmUifQ.s5LqY8mC2pX1vN0bQwReTyUiOpAsDfGhJkLzXcVbNm0"
|
||||
ENV_VAR_VALUE_GENERIC_SECRET = "Tr0ub4dor3xKq9vLmZ"
|
||||
|
||||
|
||||
class Test_batch_job_definition_no_secrets:
|
||||
def test_no_job_definitions(self):
|
||||
from prowler.providers.aws.services.batch.batch_service import Batch
|
||||
|
||||
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with (
|
||||
patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=mocked_aws_provider,
|
||||
),
|
||||
patch(
|
||||
"prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets.batch_client",
|
||||
new=Batch(mocked_aws_provider),
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets import (
|
||||
batch_job_definition_no_secrets,
|
||||
)
|
||||
|
||||
check = batch_job_definition_no_secrets()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
@mock_aws
|
||||
def test_job_definition_env_var_no_secrets(self):
|
||||
batch_client = client("batch", region_name=AWS_REGION_US_EAST_1)
|
||||
|
||||
response = batch_client.register_job_definition(
|
||||
jobDefinitionName=JOB_NAME,
|
||||
type="container",
|
||||
containerProperties={
|
||||
"image": "test-image:latest",
|
||||
"memory": 128,
|
||||
"vcpus": 1,
|
||||
"environment": [
|
||||
{
|
||||
"name": ENV_VAR_NAME_NO_SECRETS,
|
||||
"value": ENV_VAR_VALUE_NO_SECRETS,
|
||||
}
|
||||
],
|
||||
},
|
||||
)
|
||||
job_arn = response["jobDefinitionArn"]
|
||||
|
||||
from prowler.providers.aws.services.batch.batch_service import Batch
|
||||
|
||||
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with (
|
||||
patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=mocked_aws_provider,
|
||||
),
|
||||
patch(
|
||||
"prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets.batch_client",
|
||||
new=Batch(mocked_aws_provider),
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets import (
|
||||
batch_job_definition_no_secrets,
|
||||
)
|
||||
|
||||
check = batch_job_definition_no_secrets()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"No secrets found in Batch job definition {JOB_NAME} with revision {JOB_REVISION}."
|
||||
)
|
||||
assert result[0].resource_id == f"{JOB_NAME}:{JOB_REVISION}"
|
||||
assert result[0].resource_arn == job_arn
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
@mock_aws
|
||||
def test_job_definition_env_var_with_secret(self):
|
||||
batch_client = client("batch", region_name=AWS_REGION_US_EAST_1)
|
||||
|
||||
response = batch_client.register_job_definition(
|
||||
jobDefinitionName=JOB_NAME,
|
||||
type="container",
|
||||
containerProperties={
|
||||
"image": "test-image:latest",
|
||||
"memory": 128,
|
||||
"vcpus": 1,
|
||||
"environment": [
|
||||
{
|
||||
"name": ENV_VAR_NAME_NO_SECRETS,
|
||||
"value": ENV_VAR_VALUE_WITH_SECRETS,
|
||||
}
|
||||
],
|
||||
},
|
||||
)
|
||||
job_arn = response["jobDefinitionArn"]
|
||||
|
||||
from prowler.providers.aws.services.batch.batch_service import Batch
|
||||
|
||||
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with (
|
||||
patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=mocked_aws_provider,
|
||||
),
|
||||
patch(
|
||||
"prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets.batch_client",
|
||||
new=Batch(mocked_aws_provider),
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets import (
|
||||
batch_job_definition_no_secrets,
|
||||
)
|
||||
|
||||
check = batch_job_definition_no_secrets()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
f"Potential secrets found in Batch job definition {JOB_NAME} with revision {JOB_REVISION}:"
|
||||
in result[0].status_extended
|
||||
)
|
||||
assert (
|
||||
"JSON Web Token (base64url-encoded) on the environment variable host"
|
||||
in result[0].status_extended
|
||||
)
|
||||
assert result[0].resource_id == f"{JOB_NAME}:{JOB_REVISION}"
|
||||
assert result[0].resource_arn == job_arn
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
@mock_aws
|
||||
def test_job_definition_env_var_with_keyword(self):
|
||||
batch_client = client("batch", region_name=AWS_REGION_US_EAST_1)
|
||||
|
||||
response = batch_client.register_job_definition(
|
||||
jobDefinitionName=JOB_NAME,
|
||||
type="container",
|
||||
containerProperties={
|
||||
"image": "test-image:latest",
|
||||
"memory": 128,
|
||||
"vcpus": 1,
|
||||
"environment": [
|
||||
{
|
||||
"name": ENV_VAR_NAME_WITH_KEYWORD,
|
||||
"value": ENV_VAR_VALUE_GENERIC_SECRET,
|
||||
}
|
||||
],
|
||||
},
|
||||
)
|
||||
job_arn = response["jobDefinitionArn"]
|
||||
|
||||
from prowler.providers.aws.services.batch.batch_service import Batch
|
||||
|
||||
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with (
|
||||
patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=mocked_aws_provider,
|
||||
),
|
||||
patch(
|
||||
"prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets.batch_client",
|
||||
new=Batch(mocked_aws_provider),
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets import (
|
||||
batch_job_definition_no_secrets,
|
||||
)
|
||||
|
||||
check = batch_job_definition_no_secrets()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
f"Potential secrets found in Batch job definition {JOB_NAME} with revision {JOB_REVISION}:"
|
||||
in result[0].status_extended
|
||||
)
|
||||
assert (
|
||||
"Generic Password on the environment variable DB_PASSWORD"
|
||||
in result[0].status_extended
|
||||
)
|
||||
assert result[0].resource_id == f"{JOB_NAME}:{JOB_REVISION}"
|
||||
assert result[0].resource_arn == job_arn
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
@mock_aws
|
||||
def test_job_definition_no_env_vars(self):
|
||||
batch_client = client("batch", region_name=AWS_REGION_US_EAST_1)
|
||||
|
||||
response = batch_client.register_job_definition(
|
||||
jobDefinitionName=JOB_NAME,
|
||||
type="container",
|
||||
containerProperties={
|
||||
"image": "test-image:latest",
|
||||
"memory": 128,
|
||||
"vcpus": 1,
|
||||
},
|
||||
)
|
||||
job_arn = response["jobDefinitionArn"]
|
||||
|
||||
from prowler.providers.aws.services.batch.batch_service import Batch
|
||||
|
||||
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with (
|
||||
patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=mocked_aws_provider,
|
||||
),
|
||||
patch(
|
||||
"prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets.batch_client",
|
||||
new=Batch(mocked_aws_provider),
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets import (
|
||||
batch_job_definition_no_secrets,
|
||||
)
|
||||
|
||||
check = batch_job_definition_no_secrets()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"No secrets found in Batch job definition {JOB_NAME} with revision {JOB_REVISION}."
|
||||
)
|
||||
assert result[0].resource_id == f"{JOB_NAME}:{JOB_REVISION}"
|
||||
assert result[0].resource_arn == job_arn
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
@mock_aws
|
||||
def test_job_definition_command_with_secret(self):
|
||||
batch_client = client("batch", region_name=AWS_REGION_US_EAST_1)
|
||||
|
||||
response = batch_client.register_job_definition(
|
||||
jobDefinitionName=JOB_NAME,
|
||||
type="container",
|
||||
containerProperties={
|
||||
"image": "test-image:latest",
|
||||
"memory": 128,
|
||||
"vcpus": 1,
|
||||
"command": [
|
||||
"python",
|
||||
"app.py",
|
||||
"--token=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.dozjgNryP4J3jVmNHl0w5N_XgL0n3I9PlFUP0THsR8U",
|
||||
],
|
||||
},
|
||||
)
|
||||
job_arn = response["jobDefinitionArn"]
|
||||
|
||||
from prowler.providers.aws.services.batch.batch_service import Batch
|
||||
|
||||
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with (
|
||||
patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=mocked_aws_provider,
|
||||
),
|
||||
patch(
|
||||
"prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets.batch_client",
|
||||
new=Batch(mocked_aws_provider),
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets import (
|
||||
batch_job_definition_no_secrets,
|
||||
)
|
||||
|
||||
check = batch_job_definition_no_secrets()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
f"Potential secrets found in Batch job definition {JOB_NAME} with revision {JOB_REVISION}:"
|
||||
in result[0].status_extended
|
||||
)
|
||||
assert "Secrets in command" in result[0].status_extended
|
||||
assert result[0].resource_id == f"{JOB_NAME}:{JOB_REVISION}"
|
||||
assert result[0].resource_arn == job_arn
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
@mock_aws
|
||||
def test_job_definition_command_no_secrets(self):
|
||||
batch_client = client("batch", region_name=AWS_REGION_US_EAST_1)
|
||||
|
||||
response = batch_client.register_job_definition(
|
||||
jobDefinitionName=JOB_NAME,
|
||||
type="container",
|
||||
containerProperties={
|
||||
"image": "test-image:latest",
|
||||
"memory": 128,
|
||||
"vcpus": 1,
|
||||
"command": ["python", "app.py"],
|
||||
},
|
||||
)
|
||||
job_arn = response["jobDefinitionArn"]
|
||||
|
||||
from prowler.providers.aws.services.batch.batch_service import Batch
|
||||
|
||||
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with (
|
||||
patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=mocked_aws_provider,
|
||||
),
|
||||
patch(
|
||||
"prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets.batch_client",
|
||||
new=Batch(mocked_aws_provider),
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets import (
|
||||
batch_job_definition_no_secrets,
|
||||
)
|
||||
|
||||
check = batch_job_definition_no_secrets()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"No secrets found in Batch job definition {JOB_NAME} with revision {JOB_REVISION}."
|
||||
)
|
||||
assert result[0].resource_id == f"{JOB_NAME}:{JOB_REVISION}"
|
||||
assert result[0].resource_arn == job_arn
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
@mock_aws
|
||||
def test_job_definition_multiple_env_vars_with_secrets(self):
|
||||
batch_client = client("batch", region_name=AWS_REGION_US_EAST_1)
|
||||
|
||||
response = batch_client.register_job_definition(
|
||||
jobDefinitionName=JOB_NAME,
|
||||
type="container",
|
||||
containerProperties={
|
||||
"image": "test-image:latest",
|
||||
"memory": 128,
|
||||
"vcpus": 1,
|
||||
"environment": [
|
||||
{
|
||||
"name": ENV_VAR_NAME_WITH_KEYWORD,
|
||||
"value": ENV_VAR_VALUE_WITH_SECRETS,
|
||||
},
|
||||
{
|
||||
"name": ENV_VAR_NAME_NO_SECRETS,
|
||||
"value": ENV_VAR_VALUE_WITH_SECRETS2,
|
||||
},
|
||||
],
|
||||
},
|
||||
)
|
||||
job_arn = response["jobDefinitionArn"]
|
||||
|
||||
from prowler.providers.aws.services.batch.batch_service import Batch
|
||||
|
||||
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with (
|
||||
patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=mocked_aws_provider,
|
||||
),
|
||||
patch(
|
||||
"prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets.batch_client",
|
||||
new=Batch(mocked_aws_provider),
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets import (
|
||||
batch_job_definition_no_secrets,
|
||||
)
|
||||
|
||||
check = batch_job_definition_no_secrets()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
f"Potential secrets found in Batch job definition {JOB_NAME} with revision {JOB_REVISION}:"
|
||||
in result[0].status_extended
|
||||
)
|
||||
assert (
|
||||
"JSON Web Token (base64url-encoded) on the environment variable DB_PASSWORD"
|
||||
in result[0].status_extended
|
||||
)
|
||||
assert (
|
||||
"Generic Password on the environment variable DB_PASSWORD"
|
||||
in result[0].status_extended
|
||||
)
|
||||
assert (
|
||||
"JSON Web Token (base64url-encoded) on the environment variable host"
|
||||
in result[0].status_extended
|
||||
)
|
||||
assert result[0].resource_id == f"{JOB_NAME}:{JOB_REVISION}"
|
||||
assert result[0].resource_arn == job_arn
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
@mock_aws
|
||||
def test_job_definition_all_env_vars_with_keyword_and_secret(self):
|
||||
batch_client = client("batch", region_name=AWS_REGION_US_EAST_1)
|
||||
|
||||
response = batch_client.register_job_definition(
|
||||
jobDefinitionName=JOB_NAME,
|
||||
type="container",
|
||||
containerProperties={
|
||||
"image": "test-image:latest",
|
||||
"memory": 128,
|
||||
"vcpus": 1,
|
||||
"environment": [
|
||||
{
|
||||
"name": ENV_VAR_NAME_WITH_KEYWORD,
|
||||
"value": ENV_VAR_VALUE_WITH_SECRETS,
|
||||
},
|
||||
{
|
||||
"name": ENV_VAR_NAME_WITH_KEYWORD2,
|
||||
"value": ENV_VAR_VALUE_GENERIC_SECRET,
|
||||
},
|
||||
],
|
||||
},
|
||||
)
|
||||
job_arn = response["jobDefinitionArn"]
|
||||
|
||||
from prowler.providers.aws.services.batch.batch_service import Batch
|
||||
|
||||
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with (
|
||||
patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=mocked_aws_provider,
|
||||
),
|
||||
patch(
|
||||
"prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets.batch_client",
|
||||
new=Batch(mocked_aws_provider),
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets import (
|
||||
batch_job_definition_no_secrets,
|
||||
)
|
||||
|
||||
check = batch_job_definition_no_secrets()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
f"Potential secrets found in Batch job definition {JOB_NAME} with revision {JOB_REVISION}:"
|
||||
in result[0].status_extended
|
||||
)
|
||||
assert (
|
||||
"JSON Web Token (base64url-encoded) on the environment variable DB_PASSWORD"
|
||||
in result[0].status_extended
|
||||
)
|
||||
assert (
|
||||
"Generic Password on the environment variable DB_PASSWORD"
|
||||
in result[0].status_extended
|
||||
)
|
||||
assert (
|
||||
"Generic Password on the environment variable DATABASE_PASSWORD"
|
||||
in result[0].status_extended
|
||||
)
|
||||
assert result[0].resource_id == f"{JOB_NAME}:{JOB_REVISION}"
|
||||
assert result[0].resource_arn == job_arn
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
def test_scan_failure_reports_manual(self):
|
||||
from prowler.lib.utils.utils import SecretsScanError
|
||||
|
||||
batch_client = mock.MagicMock()
|
||||
job_definition_arn = f"arn:aws:batch:{AWS_REGION_US_EAST_1}:123456789012:job-definition/{JOB_NAME}:1"
|
||||
batch_client.job_definitions = {
|
||||
job_definition_arn: BatchJobDefinition(
|
||||
name=JOB_NAME,
|
||||
arn=job_definition_arn,
|
||||
revision=JOB_REVISION,
|
||||
region=AWS_REGION_US_EAST_1,
|
||||
container_properties=BatchContainerProperties(
|
||||
image="test-image:latest",
|
||||
command=[],
|
||||
environment=[
|
||||
ContainerEnvVariable(name="DB_PASSWORD", value="pass-12343")
|
||||
],
|
||||
),
|
||||
)
|
||||
}
|
||||
batch_client.audit_config = {
|
||||
"secrets_ignore_patterns": [],
|
||||
"secrets_validate": False,
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets.batch_client",
|
||||
batch_client,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets.detect_secrets_scan_batch",
|
||||
side_effect=SecretsScanError("Kingfisher exited with code 1"),
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.batch.batch_job_definition_no_secrets.batch_job_definition_no_secrets import (
|
||||
batch_job_definition_no_secrets,
|
||||
)
|
||||
|
||||
check = batch_job_definition_no_secrets()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "MANUAL"
|
||||
assert "Could not scan" in result[0].status_extended
|
||||
@@ -0,0 +1,223 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
import botocore
|
||||
|
||||
from prowler.providers.aws.services.batch.batch_service import Batch
|
||||
from tests.providers.aws.utils import (
|
||||
AWS_REGION_EU_WEST_1,
|
||||
AWS_REGION_US_EAST_1,
|
||||
set_mocked_aws_provider,
|
||||
)
|
||||
|
||||
make_api_call = botocore.client.BaseClient._make_api_call
|
||||
|
||||
|
||||
def mock_make_api_call(self, operation_name, kwarg):
|
||||
if operation_name == "DescribeJobDefinitions":
|
||||
return {
|
||||
"jobDefinitions": [
|
||||
{
|
||||
"jobDefinitionName": "test-batch-job",
|
||||
"jobDefinitionArn": "arn:aws:batch:eu-west-1:123456789012:job-definition/test-batch-job:1",
|
||||
"revision": 1,
|
||||
"containerProperties": {
|
||||
"image": "test-image:latest",
|
||||
"command": ["python", "app.py"],
|
||||
"environment": [
|
||||
{"name": "DB_PASSWORD", "value": "pass-12343"},
|
||||
{"name": "APP_NAME", "value": "myapp"},
|
||||
],
|
||||
},
|
||||
}
|
||||
]
|
||||
}
|
||||
return make_api_call(self, operation_name, kwarg)
|
||||
|
||||
|
||||
def mock_generate_regional_clients(provider, service):
|
||||
regional_client = provider._session.current_session.client(
|
||||
service, region_name=AWS_REGION_EU_WEST_1
|
||||
)
|
||||
regional_client.region = AWS_REGION_EU_WEST_1
|
||||
return {AWS_REGION_EU_WEST_1: regional_client}
|
||||
|
||||
|
||||
def mock_generate_multi_region_clients(provider, service):
|
||||
eu_west_1_client = provider._session.current_session.client(
|
||||
service, region_name=AWS_REGION_EU_WEST_1
|
||||
)
|
||||
eu_west_1_client.region = AWS_REGION_EU_WEST_1
|
||||
|
||||
us_east_1_client = provider._session.current_session.client(
|
||||
service, region_name=AWS_REGION_US_EAST_1
|
||||
)
|
||||
us_east_1_client.region = AWS_REGION_US_EAST_1
|
||||
|
||||
return {
|
||||
AWS_REGION_EU_WEST_1: eu_west_1_client,
|
||||
AWS_REGION_US_EAST_1: us_east_1_client,
|
||||
}
|
||||
|
||||
|
||||
@patch(
|
||||
"prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
|
||||
new=mock_generate_regional_clients,
|
||||
)
|
||||
class Test_Batch_Service:
|
||||
def test_service(self):
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
||||
batch = Batch(aws_provider)
|
||||
assert batch.service == "batch"
|
||||
|
||||
def test_client(self):
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
||||
batch = Batch(aws_provider)
|
||||
for reg_client in batch.regional_clients.values():
|
||||
assert reg_client.__class__.__name__ == "Batch"
|
||||
|
||||
def test__get_session__(self):
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
||||
batch = Batch(aws_provider)
|
||||
assert batch.session.__class__.__name__ == "Session"
|
||||
|
||||
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
|
||||
def test_list_job_definitions(self):
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
||||
batch = Batch(aws_provider)
|
||||
|
||||
assert len(batch.job_definitions) == 1
|
||||
jd_arn = "arn:aws:batch:eu-west-1:123456789012:job-definition/test-batch-job:1"
|
||||
jd = batch.job_definitions[jd_arn]
|
||||
assert jd.name == "test-batch-job"
|
||||
assert jd.arn == jd_arn
|
||||
assert jd.revision == 1
|
||||
assert jd.region == AWS_REGION_EU_WEST_1
|
||||
|
||||
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
|
||||
def test_describe_job_definitions(self):
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
||||
batch = Batch(aws_provider)
|
||||
|
||||
assert len(batch.job_definitions) == 1
|
||||
jd = list(batch.job_definitions.values())[0]
|
||||
assert jd.name == "test-batch-job"
|
||||
assert jd.container_properties.image == "test-image:latest"
|
||||
assert jd.container_properties.command == ["python", "app.py"]
|
||||
assert len(jd.container_properties.environment) == 2
|
||||
assert jd.container_properties.environment[0].name == "DB_PASSWORD"
|
||||
assert jd.container_properties.environment[0].value == "pass-12343"
|
||||
assert jd.container_properties.environment[1].name == "APP_NAME"
|
||||
assert jd.container_properties.environment[1].value == "myapp"
|
||||
|
||||
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
|
||||
def test_no_job_definitions(self):
|
||||
def mock_make_api_call_empty(self, operation_name, kwarg):
|
||||
if operation_name == "DescribeJobDefinitions":
|
||||
return {"jobDefinitions": []}
|
||||
return make_api_call(self, operation_name, kwarg)
|
||||
|
||||
with patch(
|
||||
"botocore.client.BaseClient._make_api_call",
|
||||
new=mock_make_api_call_empty,
|
||||
):
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
||||
batch = Batch(aws_provider)
|
||||
assert len(batch.job_definitions) == 0
|
||||
|
||||
def test_job_definitions_are_loaded_for_analysis(self):
|
||||
describe_calls = []
|
||||
|
||||
def counting_make_api_call(self, operation_name, kwarg):
|
||||
if operation_name == "DescribeJobDefinitions":
|
||||
describe_calls.append(kwarg)
|
||||
return {
|
||||
"jobDefinitions": [
|
||||
{
|
||||
"jobDefinitionName": f"job-{i}",
|
||||
"jobDefinitionArn": f"arn:aws:batch:eu-west-1:123456789012:job-definition/job-{i}:{i}",
|
||||
"revision": i,
|
||||
"containerProperties": {
|
||||
"image": "test-image:latest",
|
||||
"environment": [],
|
||||
},
|
||||
}
|
||||
for i in (3, 2, 1)
|
||||
]
|
||||
}
|
||||
return make_api_call(self, operation_name, kwarg)
|
||||
|
||||
with patch(
|
||||
"botocore.client.BaseClient._make_api_call", new=counting_make_api_call
|
||||
):
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
||||
batch = Batch(aws_provider)
|
||||
|
||||
assert [jd.revision for jd in batch.job_definitions.values()] == [3, 2, 1]
|
||||
assert len(describe_calls) == 1
|
||||
assert describe_calls[0].get("status") == "ACTIVE"
|
||||
|
||||
def test_job_definition_limit_exposes_only_selected_resources(self):
|
||||
describe_calls = []
|
||||
|
||||
def counting_make_api_call(self, operation_name, kwarg):
|
||||
if operation_name == "DescribeJobDefinitions":
|
||||
describe_calls.append(kwarg)
|
||||
return {
|
||||
"jobDefinitions": [
|
||||
{
|
||||
"jobDefinitionName": f"job-{i}",
|
||||
"jobDefinitionArn": f"arn:aws:batch:eu-west-1:123456789012:job-definition/job-{i}:{i}",
|
||||
"revision": i,
|
||||
"containerProperties": {
|
||||
"image": "test-image:latest",
|
||||
"environment": [],
|
||||
},
|
||||
}
|
||||
for i in (3, 2, 1)
|
||||
]
|
||||
}
|
||||
return make_api_call(self, operation_name, kwarg)
|
||||
|
||||
with patch(
|
||||
"botocore.client.BaseClient._make_api_call", new=counting_make_api_call
|
||||
):
|
||||
aws_provider = set_mocked_aws_provider(
|
||||
[AWS_REGION_EU_WEST_1],
|
||||
audit_config={"max_batch_job_definitions": 2},
|
||||
)
|
||||
batch = Batch(aws_provider)
|
||||
|
||||
assert [jd.revision for jd in batch.job_definitions.values()] == [3, 2]
|
||||
assert len(describe_calls) == 1
|
||||
|
||||
def test_audit_resources_filters_job_definitions(self):
|
||||
def counting_make_api_call(self, operation_name, kwarg):
|
||||
if operation_name == "DescribeJobDefinitions":
|
||||
return {
|
||||
"jobDefinitions": [
|
||||
{
|
||||
"jobDefinitionName": f"job-{i}",
|
||||
"jobDefinitionArn": f"arn:aws:batch:eu-west-1:123456789012:job-definition/job-{i}:{i}",
|
||||
"revision": i,
|
||||
"containerProperties": {
|
||||
"image": "test-image:latest",
|
||||
"environment": [],
|
||||
},
|
||||
}
|
||||
for i in (1, 2)
|
||||
]
|
||||
}
|
||||
return make_api_call(self, operation_name, kwarg)
|
||||
|
||||
with patch(
|
||||
"botocore.client.BaseClient._make_api_call", new=counting_make_api_call
|
||||
):
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
||||
aws_provider._audit_resources = [
|
||||
"arn:aws:batch:eu-west-1:123456789012:job-definition/job-2:2"
|
||||
]
|
||||
batch = Batch(aws_provider)
|
||||
|
||||
assert list(batch.job_definitions.keys()) == [
|
||||
"arn:aws:batch:eu-west-1:123456789012:job-definition/job-2:2"
|
||||
]
|
||||
+1
-1
@@ -135,7 +135,7 @@ class Test_entra_password_hash_sync_enabled:
|
||||
|
||||
def test_empty_organization(self):
|
||||
entra_client = mock.MagicMock()
|
||||
entra_client.organization = []
|
||||
entra_client.organizations = []
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
|
||||
+66
@@ -0,0 +1,66 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID,
|
||||
)
|
||||
from tests.providers.m365.m365_fixtures import set_mocked_m365_provider
|
||||
|
||||
CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_password_protection_custom_banned_list_enforced.entra_password_protection_custom_banned_list_enforced"
|
||||
|
||||
|
||||
class Test_entra_password_protection_custom_banned_list_enforced:
|
||||
def _run(self, directory_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_password_protection_custom_banned_list_enforced.entra_password_protection_custom_banned_list_enforced import (
|
||||
entra_password_protection_custom_banned_list_enforced,
|
||||
)
|
||||
|
||||
entra_client.directory_settings = directory_settings
|
||||
return entra_password_protection_custom_banned_list_enforced().execute()
|
||||
|
||||
def test_template_absent(self):
|
||||
result = self._run({})
|
||||
assert len(result) == 0
|
||||
|
||||
def test_enforced_with_list(self):
|
||||
result = self._run(
|
||||
{
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID: {
|
||||
"EnableBannedPasswordCheck": "True",
|
||||
"BannedPasswordList": "contoso\nproduct",
|
||||
}
|
||||
}
|
||||
)
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
def test_enforced_but_empty(self):
|
||||
result = self._run(
|
||||
{
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID: {
|
||||
"EnableBannedPasswordCheck": "True",
|
||||
"BannedPasswordList": "",
|
||||
}
|
||||
}
|
||||
)
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_not_enforced(self):
|
||||
result = self._run(
|
||||
{
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID: {
|
||||
"EnableBannedPasswordCheck": "False",
|
||||
"BannedPasswordList": "contoso",
|
||||
}
|
||||
}
|
||||
)
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
+41
@@ -0,0 +1,41 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID,
|
||||
)
|
||||
from tests.providers.m365.m365_fixtures import set_mocked_m365_provider
|
||||
|
||||
CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_password_protection_lockout_duration_configured.entra_password_protection_lockout_duration_configured"
|
||||
|
||||
|
||||
class Test_entra_password_protection_lockout_duration_configured:
|
||||
def _run(self, directory_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_password_protection_lockout_duration_configured.entra_password_protection_lockout_duration_configured import (
|
||||
entra_password_protection_lockout_duration_configured,
|
||||
)
|
||||
|
||||
entra_client.directory_settings = directory_settings
|
||||
return entra_password_protection_lockout_duration_configured().execute()
|
||||
|
||||
def test_template_absent(self):
|
||||
assert len(self._run({})) == 0
|
||||
|
||||
def test_at_minimum(self):
|
||||
result = self._run(
|
||||
{PASSWORD_RULE_SETTINGS_TEMPLATE_ID: {"LockoutDurationInSeconds": "60"}}
|
||||
)
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
def test_below_minimum(self):
|
||||
result = self._run(
|
||||
{PASSWORD_RULE_SETTINGS_TEMPLATE_ID: {"LockoutDurationInSeconds": "30"}}
|
||||
)
|
||||
assert result[0].status == "FAIL"
|
||||
+41
@@ -0,0 +1,41 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID,
|
||||
)
|
||||
from tests.providers.m365.m365_fixtures import set_mocked_m365_provider
|
||||
|
||||
CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_password_protection_lockout_threshold_limited.entra_password_protection_lockout_threshold_limited"
|
||||
|
||||
|
||||
class Test_entra_password_protection_lockout_threshold_limited:
|
||||
def _run(self, directory_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_password_protection_lockout_threshold_limited.entra_password_protection_lockout_threshold_limited import (
|
||||
entra_password_protection_lockout_threshold_limited,
|
||||
)
|
||||
|
||||
entra_client.directory_settings = directory_settings
|
||||
return entra_password_protection_lockout_threshold_limited().execute()
|
||||
|
||||
def test_template_absent(self):
|
||||
assert len(self._run({})) == 0
|
||||
|
||||
def test_within_limit(self):
|
||||
result = self._run(
|
||||
{PASSWORD_RULE_SETTINGS_TEMPLATE_ID: {"LockoutThreshold": "10"}}
|
||||
)
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
def test_exceeds_limit(self):
|
||||
result = self._run(
|
||||
{PASSWORD_RULE_SETTINGS_TEMPLATE_ID: {"LockoutThreshold": "20"}}
|
||||
)
|
||||
assert result[0].status == "FAIL"
|
||||
+100
@@ -0,0 +1,100 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID,
|
||||
Organization,
|
||||
)
|
||||
from tests.providers.m365.m365_fixtures import set_mocked_m365_provider
|
||||
|
||||
CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_password_protection_on_premises_enforced.entra_password_protection_on_premises_enforced"
|
||||
|
||||
|
||||
class Test_entra_password_protection_on_premises_enforced:
|
||||
def _run(self, directory_settings, organizations=None):
|
||||
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_password_protection_on_premises_enforced.entra_password_protection_on_premises_enforced import (
|
||||
entra_password_protection_on_premises_enforced,
|
||||
)
|
||||
|
||||
entra_client.directory_settings = directory_settings
|
||||
entra_client.organizations = organizations or []
|
||||
return entra_password_protection_on_premises_enforced().execute()
|
||||
|
||||
def test_no_resources(self):
|
||||
result = self._run({})
|
||||
assert len(result) == 0
|
||||
|
||||
def test_enabled_and_enforced(self):
|
||||
result = self._run(
|
||||
{
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID: {
|
||||
"EnableBannedPasswordCheckOnPremises": "True",
|
||||
"BannedPasswordCheckOnPremisesMode": "Enforced",
|
||||
}
|
||||
},
|
||||
[
|
||||
Organization(
|
||||
id="org-001",
|
||||
name="Hybrid Org",
|
||||
on_premises_sync_enabled=True,
|
||||
)
|
||||
],
|
||||
)
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
def test_unknown_organizations_still_evaluates_settings(self):
|
||||
result = self._run(
|
||||
{
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID: {
|
||||
"EnableBannedPasswordCheckOnPremises": "True",
|
||||
"BannedPasswordCheckOnPremisesMode": "Enforced",
|
||||
}
|
||||
},
|
||||
[],
|
||||
)
|
||||
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
def test_audit_mode(self):
|
||||
result = self._run(
|
||||
{
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID: {
|
||||
"EnableBannedPasswordCheckOnPremises": "True",
|
||||
"BannedPasswordCheckOnPremisesMode": "Audit",
|
||||
}
|
||||
},
|
||||
[
|
||||
Organization(
|
||||
id="org-001",
|
||||
name="Hybrid Org",
|
||||
on_premises_sync_enabled=True,
|
||||
)
|
||||
],
|
||||
)
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_cloud_only_tenant_has_no_finding(self):
|
||||
result = self._run(
|
||||
{
|
||||
PASSWORD_RULE_SETTINGS_TEMPLATE_ID: {
|
||||
"EnableBannedPasswordCheckOnPremises": "False",
|
||||
"BannedPasswordCheckOnPremisesMode": "Audit",
|
||||
}
|
||||
},
|
||||
[
|
||||
Organization(
|
||||
id="org-001",
|
||||
name="Cloud Only Org",
|
||||
on_premises_sync_enabled=False,
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
assert result == []
|
||||
+45
@@ -0,0 +1,45 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
GROUP_UNIFIED_SETTINGS_TEMPLATE_ID,
|
||||
)
|
||||
from tests.providers.m365.m365_fixtures import set_mocked_m365_provider
|
||||
|
||||
CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_policy_default_user_cannot_create_m365_groups.entra_policy_default_user_cannot_create_m365_groups"
|
||||
|
||||
|
||||
class Test_entra_policy_default_user_cannot_create_m365_groups:
|
||||
def _run(self, directory_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_policy_default_user_cannot_create_m365_groups.entra_policy_default_user_cannot_create_m365_groups import (
|
||||
entra_policy_default_user_cannot_create_m365_groups,
|
||||
)
|
||||
|
||||
entra_client.directory_settings = directory_settings
|
||||
return entra_policy_default_user_cannot_create_m365_groups().execute()
|
||||
|
||||
def test_template_absent(self):
|
||||
result = self._run({})
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_group_creation_enabled(self):
|
||||
result = self._run(
|
||||
{GROUP_UNIFIED_SETTINGS_TEMPLATE_ID: {"EnableGroupCreation": "true"}}
|
||||
)
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_group_creation_disabled(self):
|
||||
result = self._run(
|
||||
{GROUP_UNIFIED_SETTINGS_TEMPLATE_ID: {"EnableGroupCreation": "false"}}
|
||||
)
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
+139
@@ -0,0 +1,139 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
AuthorizationPolicy,
|
||||
DefaultUserRolePermissions,
|
||||
)
|
||||
from tests.providers.m365.m365_fixtures import set_mocked_m365_provider
|
||||
|
||||
|
||||
class Test_entra_policy_default_user_cannot_create_security_groups:
|
||||
def test_users_can_create_security_groups(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_policy_default_user_cannot_create_security_groups.entra_policy_default_user_cannot_create_security_groups.entra_client",
|
||||
new=entra_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_policy_default_user_cannot_create_security_groups.entra_policy_default_user_cannot_create_security_groups import (
|
||||
entra_policy_default_user_cannot_create_security_groups,
|
||||
)
|
||||
|
||||
entra_client.authorization_policy = AuthorizationPolicy(
|
||||
id="authorizationPolicy",
|
||||
name="Authorization Policy",
|
||||
description="",
|
||||
default_user_role_permissions=DefaultUserRolePermissions(
|
||||
allowed_to_create_security_groups=True,
|
||||
),
|
||||
)
|
||||
|
||||
check = entra_policy_default_user_cannot_create_security_groups()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== "Non-admin users are allowed to create security groups."
|
||||
)
|
||||
assert result[0].resource_id == "authorizationPolicy"
|
||||
assert result[0].resource_name == "Authorization Policy"
|
||||
|
||||
def test_authorization_policy_none(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_policy_default_user_cannot_create_security_groups.entra_policy_default_user_cannot_create_security_groups.entra_client",
|
||||
new=entra_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_policy_default_user_cannot_create_security_groups.entra_policy_default_user_cannot_create_security_groups import (
|
||||
entra_policy_default_user_cannot_create_security_groups,
|
||||
)
|
||||
|
||||
entra_client.authorization_policy = None
|
||||
|
||||
result = entra_policy_default_user_cannot_create_security_groups().execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].resource_id == "authorizationPolicy"
|
||||
|
||||
def test_security_group_creation_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_policy_default_user_cannot_create_security_groups.entra_policy_default_user_cannot_create_security_groups.entra_client",
|
||||
new=entra_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_policy_default_user_cannot_create_security_groups.entra_policy_default_user_cannot_create_security_groups import (
|
||||
entra_policy_default_user_cannot_create_security_groups,
|
||||
)
|
||||
|
||||
entra_client.authorization_policy = AuthorizationPolicy(
|
||||
id="authorizationPolicy",
|
||||
name="Authorization Policy",
|
||||
description="",
|
||||
default_user_role_permissions=DefaultUserRolePermissions(
|
||||
allowed_to_create_security_groups=False,
|
||||
),
|
||||
)
|
||||
|
||||
check = entra_policy_default_user_cannot_create_security_groups()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== "Non-admin users are not allowed to create security groups."
|
||||
)
|
||||
|
||||
def test_unknown_security_group_creation_permission_fails(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_policy_default_user_cannot_create_security_groups.entra_policy_default_user_cannot_create_security_groups.entra_client",
|
||||
new=entra_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_policy_default_user_cannot_create_security_groups.entra_policy_default_user_cannot_create_security_groups import (
|
||||
entra_policy_default_user_cannot_create_security_groups,
|
||||
)
|
||||
|
||||
entra_client.authorization_policy = AuthorizationPolicy(
|
||||
id="authorizationPolicy",
|
||||
name="Authorization Policy",
|
||||
description="",
|
||||
default_user_role_permissions=DefaultUserRolePermissions(
|
||||
allowed_to_create_security_groups=None,
|
||||
),
|
||||
)
|
||||
|
||||
result = entra_policy_default_user_cannot_create_security_groups().execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
+64
@@ -0,0 +1,64 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.m365.services.entra.entra_service import B2BCollaborationPolicy
|
||||
from tests.providers.m365.m365_fixtures import set_mocked_m365_provider
|
||||
|
||||
CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_policy_guest_invitations_restricted_to_allowed_domains.entra_policy_guest_invitations_restricted_to_allowed_domains"
|
||||
|
||||
|
||||
class Test_entra_policy_guest_invitations_restricted_to_allowed_domains:
|
||||
def _run(self, policy):
|
||||
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_policy_guest_invitations_restricted_to_allowed_domains.entra_policy_guest_invitations_restricted_to_allowed_domains import (
|
||||
entra_policy_guest_invitations_restricted_to_allowed_domains,
|
||||
)
|
||||
|
||||
entra_client.b2b_collaboration_policy = policy
|
||||
return (
|
||||
entra_policy_guest_invitations_restricted_to_allowed_domains().execute()
|
||||
)
|
||||
|
||||
def test_no_policy(self):
|
||||
assert self._run(None) == []
|
||||
|
||||
def test_restricted(self):
|
||||
result = self._run(
|
||||
B2BCollaborationPolicy(
|
||||
invitations_restricted_to_allowed_domains=True,
|
||||
allowed_domains=["partner.com"],
|
||||
)
|
||||
)
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== "Guest invitations are restricted to an allow-list of 1 domain(s)."
|
||||
)
|
||||
|
||||
def test_not_restricted(self):
|
||||
result = self._run(
|
||||
B2BCollaborationPolicy(
|
||||
invitations_restricted_to_allowed_domains=False,
|
||||
allowed_domains=[],
|
||||
)
|
||||
)
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_restricted_with_empty_allowed_domains(self):
|
||||
result = self._run(
|
||||
B2BCollaborationPolicy(
|
||||
invitations_restricted_to_allowed_domains=True,
|
||||
allowed_domains=[],
|
||||
)
|
||||
)
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== "Guest invitations are blocked for all external domains."
|
||||
)
|
||||
@@ -1,5 +1,6 @@
|
||||
import asyncio
|
||||
import importlib
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
@@ -199,6 +200,43 @@ async def mock_entra_get_default_app_management_policy(_):
|
||||
|
||||
|
||||
class Test_Entra_Service:
|
||||
@staticmethod
|
||||
def _load_b2b_policy(invitation_policy):
|
||||
service = object.__new__(Entra)
|
||||
service.client = MagicMock()
|
||||
service.client.request_adapter.send_primitive_async = AsyncMock(
|
||||
return_value=json.dumps(
|
||||
{
|
||||
"value": [
|
||||
{
|
||||
"definition": [
|
||||
json.dumps(
|
||||
{
|
||||
"B2BManagementPolicy": {
|
||||
"InvitationsAllowedAndBlockedDomainsPolicy": invitation_policy
|
||||
}
|
||||
}
|
||||
)
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
).encode()
|
||||
)
|
||||
return asyncio.run(service._get_b2b_collaboration_policy())
|
||||
|
||||
def test_get_b2b_policy_empty_allowed_domains_is_restricted(self):
|
||||
policy = self._load_b2b_policy({"AllowedDomains": []})
|
||||
|
||||
assert policy.invitations_restricted_to_allowed_domains is True
|
||||
assert policy.allowed_domains == []
|
||||
|
||||
def test_get_b2b_policy_without_allowed_domains_is_unrestricted(self):
|
||||
policy = self._load_b2b_policy({})
|
||||
|
||||
assert policy.invitations_restricted_to_allowed_domains is False
|
||||
assert policy.allowed_domains == []
|
||||
|
||||
def test_get_client(self):
|
||||
with patch("prowler.providers.m365.lib.service.service.M365PowerShell"):
|
||||
admincenter_client = Entra(
|
||||
|
||||
@@ -25,7 +25,11 @@ import {
|
||||
hasDateOrScanFilter,
|
||||
} from "@/lib";
|
||||
import { getFindingGroupFilterOptions } from "@/lib/finding-group-filter-options";
|
||||
import { resolveFindingScanDateFilters } from "@/lib/findings-scan-filters";
|
||||
import {
|
||||
FINDING_SCAN_DATE_SOURCE_PARAM,
|
||||
parseFindingScanDateSource,
|
||||
resolveFindingScanDateFilters,
|
||||
} from "@/lib/findings-scan-filters";
|
||||
import { isCloud } from "@/lib/shared/env";
|
||||
import { ScanEntity, ScanProps } from "@/types";
|
||||
import { SearchParamsProps } from "@/types/components";
|
||||
@@ -52,6 +56,9 @@ export default async function Findings({
|
||||
const response = await getScan(scanId);
|
||||
return response?.data;
|
||||
},
|
||||
dateSource: parseFindingScanDateSource(
|
||||
resolvedSearchParams[FINDING_SCAN_DATE_SOURCE_PARAM],
|
||||
),
|
||||
});
|
||||
const resolvedFilters = applyDefaultMutedFilter(filtersWithScanDates);
|
||||
const hasHistoricalData = hasDateOrScanFilter(filtersWithScanDates);
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
Scan `View Findings` navigation now preserves the user's local date while querying findings with the scan's UTC completion date
|
||||
@@ -20,6 +20,10 @@ import { ExpandableSection } from "@/components/shadcn/expandable-section";
|
||||
import { DataTableFilterCustom } from "@/components/shadcn/table/data-table-filter-custom";
|
||||
import { useFilterBatch } from "@/hooks/use-filter-batch";
|
||||
import { getCategoryLabel, getGroupLabel } from "@/lib/categories";
|
||||
import {
|
||||
FINDING_SCAN_DATE_PROVENANCE_FILTER_KEYS,
|
||||
FINDING_SCAN_DATE_SOURCE_PARAM,
|
||||
} from "@/lib/findings-scan-filters";
|
||||
import { FILTER_FIELD, ScanEntity } from "@/types";
|
||||
import { ProviderGroup } from "@/types/components";
|
||||
import { DATA_TABLE_FILTER_MODE } from "@/types/filters";
|
||||
@@ -369,6 +373,12 @@ export const FindingsFilters = (props: FindingsFiltersProps) => {
|
||||
} = useFilterBatch({
|
||||
defaultParams: { "filter[muted]": "false" },
|
||||
exclusiveFilterGroups: [FINDING_GROUP_FILTER_KEYS],
|
||||
urlParamInvalidationRules: [
|
||||
{
|
||||
param: FINDING_SCAN_DATE_SOURCE_PARAM,
|
||||
filterKeys: FINDING_SCAN_DATE_PROVENANCE_FILTER_KEYS,
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
return (
|
||||
|
||||
@@ -314,7 +314,7 @@ describe("ScanJobsRowActions", () => {
|
||||
).not.toBeInTheDocument();
|
||||
});
|
||||
|
||||
it("links completed scans to filtered findings", async () => {
|
||||
it("links completed scans to findings with a local display date", async () => {
|
||||
// Given
|
||||
const user = userEvent.setup();
|
||||
render(
|
||||
@@ -334,9 +334,11 @@ describe("ScanJobsRowActions", () => {
|
||||
await user.click(screen.getByRole("menuitem", { name: /view findings/i }));
|
||||
|
||||
// Then
|
||||
expect(pushMock).toHaveBeenCalledWith(
|
||||
"/findings?filter[scan__in]=scan-1&filter[inserted_at]=2026-01-01&filter[status__in]=FAIL",
|
||||
);
|
||||
const findingsHref = pushMock.mock.calls[0]?.[0] as string;
|
||||
expect(findingsHref).toContain("filter[scan__in]=scan-1");
|
||||
expect(findingsHref).toContain("filter[inserted_at]=2026-01-01");
|
||||
expect(findingsHref).toContain("filter[status__in]=FAIL");
|
||||
expect(findingsHref).toContain("scanDateSource=scan-action:2026-01-01");
|
||||
});
|
||||
|
||||
it("triggers downloadScanZip with the scan id when downloading reports", async () => {
|
||||
|
||||
@@ -31,6 +31,10 @@ import {
|
||||
} from "@/components/shadcn/dropdown";
|
||||
import { buildPerScanComplianceHref } from "@/lib/compliance/compliance-tab-url";
|
||||
import { toLocalDateString } from "@/lib/date-utils";
|
||||
import {
|
||||
buildFindingScanDateSource,
|
||||
FINDING_SCAN_DATE_SOURCE_PARAM,
|
||||
} from "@/lib/findings-scan-filters";
|
||||
import { downloadScanZip } from "@/lib/helper";
|
||||
import { getScanScheduleCapability } from "@/lib/schedules";
|
||||
import { isCloud } from "@/lib/shared/env";
|
||||
@@ -93,8 +97,9 @@ export function ScanJobsRowActions({
|
||||
|
||||
const openFindings = () => {
|
||||
if (!isCompleted || !scanDate) return;
|
||||
|
||||
router.push(
|
||||
`/findings?filter[scan__in]=${scan.id}&filter[inserted_at]=${scanDate}&filter[status__in]=FAIL`,
|
||||
`/findings?filter[scan__in]=${scan.id}&filter[inserted_at]=${scanDate}&filter[status__in]=FAIL&${FINDING_SCAN_DATE_SOURCE_PARAM}=${buildFindingScanDateSource(scanDate)}`,
|
||||
);
|
||||
};
|
||||
|
||||
|
||||
@@ -84,7 +84,11 @@ const provideSurveys = (surveys: Survey[]): void => {
|
||||
});
|
||||
};
|
||||
|
||||
const renderSurvey = async () => {
|
||||
// Pre-evaluating the lazy chunk keeps findBy* queries from racing module
|
||||
// evaluation under full-suite load; guard tests that assert the chunk is never
|
||||
// touched opt out via preloadRuntime: false.
|
||||
const renderSurvey = async ({ preloadRuntime = true } = {}) => {
|
||||
if (preloadRuntime) await import("./runtime-feedback-survey");
|
||||
const { FeedbackSurvey } = await import("./feedback-survey");
|
||||
return render(<FeedbackSurvey />);
|
||||
};
|
||||
@@ -475,7 +479,7 @@ describe("FeedbackSurvey", () => {
|
||||
});
|
||||
|
||||
// When
|
||||
const view = await renderSurvey();
|
||||
const view = await renderSurvey({ preloadRuntime: false });
|
||||
|
||||
// Then - no init, no survey fetch, nothing rendered
|
||||
expect(mocks.init).not.toHaveBeenCalled();
|
||||
@@ -495,7 +499,7 @@ describe("FeedbackSurvey", () => {
|
||||
});
|
||||
|
||||
// When
|
||||
const view = await renderSurvey();
|
||||
const view = await renderSurvey({ preloadRuntime: false });
|
||||
|
||||
// Then
|
||||
expect(view.container).toBeEmptyDOMElement();
|
||||
@@ -515,7 +519,7 @@ describe("FeedbackSurvey", () => {
|
||||
provideSurveys([SURVEY_FIXTURE]);
|
||||
|
||||
// When
|
||||
const view = await renderSurvey();
|
||||
const view = await renderSurvey({ preloadRuntime: false });
|
||||
|
||||
// Then - no init, no trigger, no survey fetch, no capture, regardless of config
|
||||
expect(view.container).toBeEmptyDOMElement();
|
||||
|
||||
@@ -1,6 +1,12 @@
|
||||
import { act, renderHook } from "@testing-library/react";
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import {
|
||||
buildFindingScanDateSource,
|
||||
FINDING_SCAN_DATE_PROVENANCE_FILTER_KEYS,
|
||||
FINDING_SCAN_DATE_SOURCE_PARAM,
|
||||
} from "@/lib/findings-scan-filters";
|
||||
|
||||
// --- Mock next/navigation ---
|
||||
const mockPush = vi.fn();
|
||||
let mockSearchParamsValue = new URLSearchParams();
|
||||
@@ -432,6 +438,84 @@ describe("useFilterBatch", () => {
|
||||
expect(calledUrl).toContain("filter%5Bsearch%5D=my-search");
|
||||
expect(calledUrl).toContain("filter%5Bmuted%5D=false");
|
||||
});
|
||||
|
||||
it("should clear scan-date provenance when the displayed date changes", () => {
|
||||
// Given
|
||||
setSearchParams({
|
||||
"filter[scan__in]": "scan-1",
|
||||
"filter[inserted_at]": "2026-04-06",
|
||||
[FINDING_SCAN_DATE_SOURCE_PARAM]:
|
||||
buildFindingScanDateSource("2026-04-06"),
|
||||
expandedCheckId: "check-1",
|
||||
});
|
||||
const { result } = renderHook(() =>
|
||||
useFilterBatch({
|
||||
urlParamInvalidationRules: [
|
||||
{
|
||||
param: FINDING_SCAN_DATE_SOURCE_PARAM,
|
||||
filterKeys: FINDING_SCAN_DATE_PROVENANCE_FILTER_KEYS,
|
||||
},
|
||||
],
|
||||
}),
|
||||
);
|
||||
|
||||
act(() => {
|
||||
result.current.setPending("filter[inserted_at]", ["2026-04-05"]);
|
||||
});
|
||||
|
||||
// When
|
||||
act(() => {
|
||||
result.current.applyAll();
|
||||
});
|
||||
|
||||
// Then
|
||||
const calledUrl = new URL(
|
||||
mockPush.mock.calls[0][0],
|
||||
"https://example.com",
|
||||
);
|
||||
expect(calledUrl.searchParams.has(FINDING_SCAN_DATE_SOURCE_PARAM)).toBe(
|
||||
false,
|
||||
);
|
||||
expect(calledUrl.searchParams.get("expandedCheckId")).toBe("check-1");
|
||||
});
|
||||
|
||||
it("should preserve scan-date provenance for unrelated filter changes", () => {
|
||||
// Given
|
||||
const dateSource = buildFindingScanDateSource("2026-04-06");
|
||||
setSearchParams({
|
||||
"filter[scan__in]": "scan-1",
|
||||
"filter[inserted_at]": "2026-04-06",
|
||||
[FINDING_SCAN_DATE_SOURCE_PARAM]: dateSource,
|
||||
});
|
||||
const { result } = renderHook(() =>
|
||||
useFilterBatch({
|
||||
urlParamInvalidationRules: [
|
||||
{
|
||||
param: FINDING_SCAN_DATE_SOURCE_PARAM,
|
||||
filterKeys: FINDING_SCAN_DATE_PROVENANCE_FILTER_KEYS,
|
||||
},
|
||||
],
|
||||
}),
|
||||
);
|
||||
|
||||
act(() => {
|
||||
result.current.setPending("filter[severity__in]", ["critical"]);
|
||||
});
|
||||
|
||||
// When
|
||||
act(() => {
|
||||
result.current.applyAll();
|
||||
});
|
||||
|
||||
// Then
|
||||
const calledUrl = new URL(
|
||||
mockPush.mock.calls[0][0],
|
||||
"https://example.com",
|
||||
);
|
||||
expect(calledUrl.searchParams.get(FINDING_SCAN_DATE_SOURCE_PARAM)).toBe(
|
||||
dateSource,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
// ── discardAll ─────────────────────────────────────────────────────────────
|
||||
@@ -635,5 +719,41 @@ describe("useFilterBatch", () => {
|
||||
const calledUrl: string = mockPush.mock.calls[0][0];
|
||||
expect(calledUrl).toContain("page=1");
|
||||
});
|
||||
|
||||
it("should clear scan-date provenance without deleting unrelated URL params", () => {
|
||||
// Given
|
||||
setSearchParams({
|
||||
"filter[scan__in]": "scan-1",
|
||||
"filter[inserted_at]": "2026-04-06",
|
||||
[FINDING_SCAN_DATE_SOURCE_PARAM]:
|
||||
buildFindingScanDateSource("2026-04-06"),
|
||||
expandedCheckId: "check-1",
|
||||
});
|
||||
const { result } = renderHook(() =>
|
||||
useFilterBatch({
|
||||
urlParamInvalidationRules: [
|
||||
{
|
||||
param: FINDING_SCAN_DATE_SOURCE_PARAM,
|
||||
filterKeys: FINDING_SCAN_DATE_PROVENANCE_FILTER_KEYS,
|
||||
},
|
||||
],
|
||||
}),
|
||||
);
|
||||
|
||||
// When
|
||||
act(() => {
|
||||
result.current.clearAndApply();
|
||||
});
|
||||
|
||||
// Then
|
||||
const calledUrl = new URL(
|
||||
mockPush.mock.calls[0][0],
|
||||
"https://example.com",
|
||||
);
|
||||
expect(calledUrl.searchParams.has(FINDING_SCAN_DATE_SOURCE_PARAM)).toBe(
|
||||
false,
|
||||
);
|
||||
expect(calledUrl.searchParams.get("expandedCheckId")).toBe("check-1");
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -157,6 +157,16 @@ export interface UseFilterBatchOptions {
|
||||
* applied together.
|
||||
*/
|
||||
exclusiveFilterGroups?: string[][];
|
||||
/**
|
||||
* Non-filter URL params whose provenance becomes invalid when specific
|
||||
* URL-backed filters change.
|
||||
*/
|
||||
urlParamInvalidationRules?: UrlParamInvalidationRule[];
|
||||
}
|
||||
|
||||
export interface UrlParamInvalidationRule {
|
||||
param: string;
|
||||
filterKeys: readonly string[];
|
||||
}
|
||||
|
||||
function normalizeFilterKey(key: string): string {
|
||||
@@ -233,10 +243,27 @@ export const useFilterBatch = (
|
||||
};
|
||||
|
||||
/** Private helper — builds URLSearchParams from a pending state and pushes. */
|
||||
const buildAndPush = (nextPending: PendingFilters) => {
|
||||
const buildAndPush = (
|
||||
nextPending: PendingFilters,
|
||||
clearInvalidatedParams = false,
|
||||
) => {
|
||||
setAppliedFilters(nextPending);
|
||||
const params = new URLSearchParams(searchParams.toString());
|
||||
|
||||
options?.urlParamInvalidationRules?.forEach(({ param, filterKeys }) => {
|
||||
const isInvalidated =
|
||||
clearInvalidatedParams ||
|
||||
filterKeys.some((filterKey) => {
|
||||
const currentValue = searchParams.get(filterKey) || "";
|
||||
const nextValue = (nextPending[filterKey] ?? [])
|
||||
.filter(Boolean)
|
||||
.join(",");
|
||||
return currentValue !== nextValue;
|
||||
});
|
||||
|
||||
if (isInvalidated) params.delete(param);
|
||||
});
|
||||
|
||||
// Remove all batch-managed filter params
|
||||
Array.from(params.keys()).forEach((key) => {
|
||||
if (key.startsWith("filter[") && !EXCLUDED_FROM_BATCH.includes(key)) {
|
||||
@@ -290,7 +317,7 @@ export const useFilterBatch = (
|
||||
*/
|
||||
const clearAndApply = () => {
|
||||
setPendingFilters({});
|
||||
buildAndPush({});
|
||||
buildAndPush({}, true);
|
||||
};
|
||||
|
||||
const removeAppliedAndApply = (key: string, value?: string) => {
|
||||
|
||||
@@ -109,4 +109,56 @@ describe("resolveFindingScanDateFilters", () => {
|
||||
"filter[inserted_at__gte]": "2026-04-01",
|
||||
});
|
||||
});
|
||||
|
||||
it("replaces a scan-action local date with the scan UTC completion date", async () => {
|
||||
const result = await resolveFindingScanDateFilters({
|
||||
filters: {
|
||||
"filter[muted]": "false",
|
||||
"filter[scan__in]": "scan-1",
|
||||
"filter[inserted_at]": "2026-04-06",
|
||||
"filter[status__in]": "FAIL",
|
||||
},
|
||||
scans: [
|
||||
{
|
||||
id: "scan-1",
|
||||
attributes: {
|
||||
completed_at: "2026-04-07T00:30:00Z",
|
||||
},
|
||||
},
|
||||
],
|
||||
loadScan: vi.fn(),
|
||||
dateSource: "scan-action:2026-04-06",
|
||||
});
|
||||
|
||||
expect(result).toEqual({
|
||||
"filter[muted]": "false",
|
||||
"filter[scan__in]": "scan-1",
|
||||
"filter[inserted_at]": "2026-04-07",
|
||||
"filter[status__in]": "FAIL",
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves a manually changed date when the scan-action marker is stale", async () => {
|
||||
const result = await resolveFindingScanDateFilters({
|
||||
filters: {
|
||||
"filter[scan__in]": "scan-1",
|
||||
"filter[inserted_at]": "2026-04-05",
|
||||
},
|
||||
scans: [
|
||||
{
|
||||
id: "scan-1",
|
||||
attributes: {
|
||||
completed_at: "2026-04-07T00:30:00Z",
|
||||
},
|
||||
},
|
||||
],
|
||||
loadScan: vi.fn(),
|
||||
dateSource: "scan-action:2026-04-06",
|
||||
});
|
||||
|
||||
expect(result).toEqual({
|
||||
"filter[scan__in]": "scan-1",
|
||||
"filter[inserted_at]": "2026-04-05",
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -13,6 +13,36 @@ interface ResolveFindingScanDateFiltersOptions {
|
||||
filters: Record<string, string>;
|
||||
scans: ScanDateSource[];
|
||||
loadScan: (scanId: string) => Promise<ScanDateSource | null | undefined>;
|
||||
dateSource?: FindingScanDateSource;
|
||||
}
|
||||
|
||||
export const FINDING_SCAN_DATE_SOURCE_PARAM = "scanDateSource";
|
||||
|
||||
export const FINDING_SCAN_DATE_SOURCE = {
|
||||
SCAN_ACTION: "scan-action",
|
||||
} as const;
|
||||
|
||||
type FindingScanDateSource =
|
||||
`${typeof FINDING_SCAN_DATE_SOURCE.SCAN_ACTION}:${string}`;
|
||||
|
||||
const SCAN_ACTION_DATE_PREFIX =
|
||||
`${FINDING_SCAN_DATE_SOURCE.SCAN_ACTION}:` as const;
|
||||
|
||||
export function buildFindingScanDateSource(
|
||||
displayDate: string,
|
||||
): FindingScanDateSource {
|
||||
return `${SCAN_ACTION_DATE_PREFIX}${displayDate}`;
|
||||
}
|
||||
|
||||
export function parseFindingScanDateSource(
|
||||
value: string | string[] | undefined,
|
||||
): FindingScanDateSource | undefined {
|
||||
const candidate = Array.isArray(value) ? value[0] : value;
|
||||
|
||||
return candidate?.startsWith(SCAN_ACTION_DATE_PREFIX) &&
|
||||
candidate.length > SCAN_ACTION_DATE_PREFIX.length
|
||||
? (candidate as FindingScanDateSource)
|
||||
: undefined;
|
||||
}
|
||||
|
||||
const INSERTED_AT_FILTER_KEYS = [
|
||||
@@ -22,6 +52,12 @@ const INSERTED_AT_FILTER_KEYS = [
|
||||
"filter[inserted_at__lte]",
|
||||
] as const;
|
||||
|
||||
export const FINDING_SCAN_DATE_PROVENANCE_FILTER_KEYS = [
|
||||
"filter[scan__in]",
|
||||
"filter[scan]",
|
||||
...INSERTED_AT_FILTER_KEYS,
|
||||
] as const;
|
||||
|
||||
function getScanFilterIds(filters: Record<string, string>): string[] {
|
||||
const scanIds = filters["filter[scan__in]"] || filters["filter[scan]"] || "";
|
||||
return Array.from(new Set(scanIds.split(",").filter(Boolean)));
|
||||
@@ -64,10 +100,20 @@ export async function resolveFindingScanDateFilters({
|
||||
filters,
|
||||
scans,
|
||||
loadScan,
|
||||
dateSource,
|
||||
}: ResolveFindingScanDateFiltersOptions): Promise<Record<string, string>> {
|
||||
const scanIds = getScanFilterIds(filters);
|
||||
const scanActionDisplayDate = dateSource?.slice(
|
||||
SCAN_ACTION_DATE_PREFIX.length,
|
||||
);
|
||||
const isScanActionDate =
|
||||
Boolean(scanActionDisplayDate) &&
|
||||
filters["filter[inserted_at]"] === scanActionDisplayDate;
|
||||
|
||||
if (scanIds.length === 0 || hasInsertedAtFilter(filters)) {
|
||||
if (
|
||||
scanIds.length === 0 ||
|
||||
(hasInsertedAtFilter(filters) && !isScanActionDate)
|
||||
) {
|
||||
return filters;
|
||||
}
|
||||
|
||||
@@ -96,8 +142,16 @@ export async function resolveFindingScanDateFilters({
|
||||
return filters;
|
||||
}
|
||||
|
||||
const apiFilters = isScanActionDate
|
||||
? Object.fromEntries(
|
||||
Object.entries(filters).filter(([key]) =>
|
||||
INSERTED_AT_FILTER_KEYS.every((dateKey) => dateKey !== key),
|
||||
),
|
||||
)
|
||||
: filters;
|
||||
|
||||
return {
|
||||
...filters,
|
||||
...apiFilters,
|
||||
...dateFilters,
|
||||
};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user