Merge remote-tracking branch 'origin/master' into HEAD

This commit is contained in:
Hugo P.Brito committed 2026-04-08 13:25:22 +01:00
commit db6828ec28
79 files changed
+2901 -180

No files matched your search

+63 -1
View File
@@ -1,8 +1,14 @@
import logging
from unittest.mock import patch
import pytest
from prowler.lib.cli.redact import REDACTED_VALUE, get_sensitive_arguments, redact_argv
from prowler.lib.cli.redact import (
REDACTED_VALUE,
get_sensitive_arguments,
redact_argv,
warn_sensitive_argument_values,
)
@pytest.fixture
@@ -87,6 +93,62 @@ class TestRedactArgv:
assert redact_argv(argv) == "aws --region=us-east-1"
class TestWarnSensitiveArgumentValues:
def test_no_warning_without_sensitive_flags(self, caplog, mock_sensitive_args):
with caplog.at_level(logging.WARNING):
warn_sensitive_argument_values(["aws", "--region", "eu-west-1"])
assert caplog.text == ""
def test_no_warning_flag_without_value(self, caplog, mock_sensitive_args):
with caplog.at_level(logging.WARNING):
warn_sensitive_argument_values(["github", "--personal-access-token"])
assert caplog.text == ""
def test_no_warning_flag_followed_by_another_flag(
self, caplog, mock_sensitive_args
):
with caplog.at_level(logging.WARNING):
warn_sensitive_argument_values(
["github", "--personal-access-token", "--region", "eu-west-1"]
)
assert caplog.text == ""
def test_warning_flag_with_value(self, caplog, mock_sensitive_args):
with caplog.at_level(logging.WARNING):
warn_sensitive_argument_values(
["github", "--personal-access-token", "ghp_secret"]
)
assert "--personal-access-token" in caplog.text
assert "not recommended" in caplog.text
def test_warning_flag_with_equals_syntax(self, caplog, mock_sensitive_args):
with caplog.at_level(logging.WARNING):
warn_sensitive_argument_values(["aws", "--shodan=key123"])
assert "--shodan" in caplog.text
assert "not recommended" in caplog.text
def test_warning_multiple_flags(self, caplog, mock_sensitive_args):
with caplog.at_level(logging.WARNING):
warn_sensitive_argument_values(
[
"github",
"--personal-access-token",
"ghp_secret",
"--shodan",
"key",
]
)
assert "--personal-access-token" in caplog.text
assert "--shodan" in caplog.text
def test_no_color_output(self, caplog, mock_sensitive_args):
with caplog.at_level(logging.WARNING):
warn_sensitive_argument_values(["--no-color", "aws", "--shodan", "key123"])
assert "not recommended" in caplog.text
# Should not contain ANSI escape codes
assert "\033[" not in caplog.text
class TestGetSensitiveArguments:
def test_discovers_known_sensitive_arguments(self):
"""Integration test: verify the discovery mechanism finds flags from provider modules."""
@@ -78,7 +78,9 @@ class TestAWSService:
def test_AWSService_non_global_service_uses_profile_region(self):
"""Non-global services should use the profile region when available."""
service_name = "s3"
provider = set_mocked_aws_provider(profile_region=AWS_REGION_EU_WEST_1)
provider = set_mocked_aws_provider(
audited_regions=[], profile_region=AWS_REGION_EU_WEST_1
)
service = AWSService(service_name, provider)
assert service.region == AWS_REGION_EU_WEST_1
@@ -312,7 +312,9 @@ class Test_awslambda_function_not_publicly_accessible:
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider(),
return_value=set_mocked_aws_provider(
audited_regions=[AWS_REGION_EU_WEST_1]
),
),
mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible.awslambda_client",
@@ -552,7 +554,9 @@ class Test_awslambda_function_not_publicly_accessible:
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider(),
return_value=set_mocked_aws_provider(
audited_regions=[AWS_REGION_EU_WEST_1]
),
),
mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible.awslambda_client",
@@ -615,7 +619,9 @@ class Test_awslambda_function_not_publicly_accessible:
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider(),
return_value=set_mocked_aws_provider(
audited_regions=[AWS_REGION_EU_WEST_1]
),
),
mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible.awslambda_client",
@@ -690,7 +696,9 @@ class Test_awslambda_function_not_publicly_accessible:
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider(),
return_value=set_mocked_aws_provider(
audited_regions=[AWS_REGION_EU_WEST_1]
),
),
mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible.awslambda_client",
@@ -54,6 +54,9 @@ def mock_make_api_call(self, operation_name, kwarg):
}
if operation_name == "ListPackageVersions":
assert (
kwarg.get("maxResults") == 1
), "list_package_versions must pass maxResults=1 to avoid fetching all versions"
return {
"defaultDisplayVersion": "latest",
"format": "pypi",
@@ -204,3 +207,102 @@ class Test_CodeArtifact_Service:
.latest_version.origin.origin_type
== OriginInformationValues.INTERNAL
)
def mock_make_api_call_no_namespace(self, operation_name, kwarg):
"""Mock for packages without a namespace to exercise the else branch"""
if operation_name == "ListRepositories":
return {
"repositories": [
{
"name": "test-repository",
"administratorAccount": AWS_ACCOUNT_NUMBER,
"domainName": "test-domain",
"domainOwner": AWS_ACCOUNT_NUMBER,
"arn": TEST_REPOSITORY_ARN,
"description": "test description",
},
]
}
if operation_name == "ListPackages":
return {
"packages": [
{
"format": "pypi",
"package": "test-package-no-ns",
"originConfiguration": {
"restrictions": {
"publish": "ALLOW",
"upstream": "BLOCK",
}
},
},
],
}
if operation_name == "ListPackageVersions":
assert (
kwarg.get("maxResults") == 1
), "list_package_versions must pass maxResults=1 to avoid fetching all versions"
assert (
"namespace" not in kwarg
), "namespace should not be passed when package has no namespace"
return {
"defaultDisplayVersion": "1.0.0",
"format": "pypi",
"package": "test-package-no-ns",
"versions": [
{
"version": "1.0.0",
"revision": "abc123",
"status": "Published",
"origin": {
"domainEntryPoint": {
"repositoryName": "test-repository",
"externalConnectionName": "",
},
"originType": "EXTERNAL",
},
},
],
}
if operation_name == "ListTagsForResource":
return {"tags": []}
return make_api_call(self, operation_name, kwarg)
@patch(
"botocore.client.BaseClient._make_api_call",
new=mock_make_api_call_no_namespace,
)
@patch(
"prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
new=mock_generate_regional_clients,
)
class Test_CodeArtifact_Service_No_Namespace:
def test_list_packages_no_namespace(self):
codeartifact = CodeArtifact(
set_mocked_aws_provider([AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1])
)
assert len(codeartifact.repositories[TEST_REPOSITORY_ARN].packages) == 1
package = codeartifact.repositories[TEST_REPOSITORY_ARN].packages[0]
assert package.name == "test-package-no-ns"
assert package.namespace is None
assert package.format == "pypi"
assert (
package.origin_configuration.restrictions.publish == RestrictionValues.ALLOW
)
assert (
package.origin_configuration.restrictions.upstream
== RestrictionValues.BLOCK
)
assert package.latest_version.version == "1.0.0"
assert package.latest_version.status == LatestPackageVersionStatus.Published
assert (
package.latest_version.origin.origin_type
== OriginInformationValues.EXTERNAL
)
@@ -139,7 +139,7 @@ class Test_Codebuild_Service:
)
@mock_aws
def test_codebuild_service(self):
codebuild = Codebuild(set_mocked_aws_provider())
codebuild = Codebuild(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
assert codebuild.session.__class__.__name__ == "Session"
assert codebuild.service == "codebuild"
@@ -76,7 +76,7 @@ class Test_CodePipeline_Service:
)
@mock_aws
def test_codepipeline_service(self):
codepipeline = CodePipeline(set_mocked_aws_provider())
codepipeline = CodePipeline(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
assert codepipeline.session.__class__.__name__ == "Session"
assert codepipeline.service == "codepipeline"
@@ -106,27 +106,27 @@ def mock_generate_regional_clients(provider, service):
class Test_DataSync_Service:
# Test DataSync Service initialization
def test_service(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
datasync = DataSync(aws_provider)
assert datasync.service == "datasync"
# Test DataSync clients creation
def test_client(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
datasync = DataSync(aws_provider)
for reg_client in datasync.regional_clients.values():
assert reg_client.__class__.__name__ == "DataSync"
# Test DataSync session
def test__get_session__(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
datasync = DataSync(aws_provider)
assert datasync.session.__class__.__name__ == "Session"
# Test listing DataSync tasks
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_list_tasks(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
datasync = DataSync(aws_provider)
task_arn = "arn:aws:datasync:eu-west-1:123456789012:task/task-12345678901234567"
@@ -142,7 +142,7 @@ class Test_DataSync_Service:
# Test generic exception in list_tasks
def test_list_tasks_generic_exception(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
# Mock the regional client's list_tasks method specifically
mock_client = MagicMock()
@@ -155,7 +155,7 @@ class Test_DataSync_Service:
# Test describing DataSync tasks with various exceptions
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_describe_tasks_with_exceptions(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
datasync = DataSync(aws_provider)
# Check all tasks were processed despite exceptions
@@ -183,7 +183,7 @@ class Test_DataSync_Service:
# Test listing task tags with various exceptions
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_list_task_tags_with_exceptions(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
datasync = DataSync(aws_provider)
tasks_by_name = {task.name: task for task in datasync.tasks.values()}
@@ -170,20 +170,20 @@ def mock_generate_regional_clients(provider, service):
class Test_ECR_Service:
# Test ECR Service
def test_service(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecr = ECR(aws_provider)
assert ecr.service == "ecr"
# Test ECR client
def test_client(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecr = ECR(aws_provider)
for regional_client in ecr.regional_clients.values():
assert regional_client.__class__.__name__ == "ECR"
# Test ECR session
def test_get_session(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecr = ECR(aws_provider)
assert ecr.session.__class__.__name__ == "Session"
@@ -198,7 +198,7 @@ class Test_ECR_Service:
{"Key": "test", "Value": "test"},
],
)
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecr = ECR(aws_provider)
assert len(ecr.registries) == 1
@@ -226,7 +226,7 @@ class Test_ECR_Service:
imageScanningConfiguration={"scanOnPush": True},
imageTagMutability="IMMUTABLE",
)
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecr = ECR(aws_provider)
assert len(ecr.registries) == 1
assert len(ecr.registries[AWS_REGION_EU_WEST_1].repositories) == 1
@@ -255,7 +255,7 @@ class Test_ECR_Service:
imageScanningConfiguration={"scanOnPush": True},
imageTagMutability="IMMUTABLE",
)
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecr = ECR(aws_provider)
assert len(ecr.registries) == 1
assert len(ecr.registries[AWS_REGION_EU_WEST_1].repositories) == 1
@@ -273,7 +273,7 @@ class Test_ECR_Service:
repositoryName=repo_name,
imageScanningConfiguration={"scanOnPush": True},
)
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecr = ECR(aws_provider)
assert len(ecr.registries) == 1
@@ -366,7 +366,7 @@ class Test_ECR_Service:
# Test get ECR Registries Scanning Configuration
@mock_aws
def test_get_registry_scanning_configuration(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecr = ECR(aws_provider)
assert len(ecr.registries) == 1
assert ecr.registries[AWS_REGION_EU_WEST_1].id == AWS_ACCOUNT_NUMBER
@@ -122,27 +122,27 @@ def mock_generate_regional_clients(provider, service):
class Test_ECS_Service:
# Test ECS Service
def test_service(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecs = ECS(aws_provider)
assert ecs.service == "ecs"
# Test ECS client
def test_client(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecs = ECS(aws_provider)
for reg_client in ecs.regional_clients.values():
assert reg_client.__class__.__name__ == "ECS"
# Test ECS session
def test__get_session__(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecs = ECS(aws_provider)
assert ecs.session.__class__.__name__ == "Session"
# Test list ECS task definitions
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_list_task_definitions(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecs = ECS(aws_provider)
task_arn = "arn:aws:ecs:eu-west-1:123456789012:task-definition/test_cluster_1/test_ecs_task:1"
@@ -156,7 +156,7 @@ class Test_ECS_Service:
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
# Test describe ECS task definitions
def test_describe_task_definitions(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecs = ECS(aws_provider)
task_arn = "arn:aws:ecs:eu-west-1:123456789012:task-definition/test_cluster_1/test_ecs_task:1"
@@ -204,7 +204,7 @@ class Test_ECS_Service:
# Test list ECS clusters
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_list_clusters(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecs = ECS(aws_provider)
cluster_arn1 = "arn:aws:ecs:eu-west-1:123456789012:cluster/test_cluster_1"
@@ -217,7 +217,7 @@ class Test_ECS_Service:
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
# Test describe ECS clusters
def test_describe_clusters(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecs = ECS(aws_provider)
cluster_arn1 = "arn:aws:ecs:eu-west-1:123456789012:cluster/test_cluster_1"
@@ -237,7 +237,7 @@ class Test_ECS_Service:
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
# Test describe ECS services
def test_describe_services(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
ecs = ECS(aws_provider)
service_arn = (
@@ -93,18 +93,18 @@ def mock_generate_regional_clients(provider, service):
class Test_EFS:
# Test EFS Session
def test__get_session__(self):
access_analyzer = EFS(set_mocked_aws_provider())
access_analyzer = EFS(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
assert access_analyzer.session.__class__.__name__ == "Session"
# Test EFS Service
def test__get_service__(self):
access_analyzer = EFS(set_mocked_aws_provider())
access_analyzer = EFS(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
assert access_analyzer.service == "efs"
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
# Test EFS describe file systems
def test_describe_file_systems(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
efs = EFS(aws_provider)
efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{FILE_SYSTEM_ID}"
assert len(efs.filesystems) == 1
@@ -119,7 +119,7 @@ class Test_EFS:
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
# Test EFS describe file systems policies
def test_describe_file_system_policies(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
efs = EFS(aws_provider)
efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{FILE_SYSTEM_ID}"
assert len(efs.filesystems) == 1
@@ -131,7 +131,7 @@ class Test_EFS:
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
# Test EFS describe mount targets
def test_describe_mount_targets(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
efs = EFS(aws_provider)
assert len(efs.filesystems) == 1
efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{FILE_SYSTEM_ID}"
@@ -144,7 +144,7 @@ class Test_EFS:
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
# Test EFS describe access points
def test_describe_access_points(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
efs = EFS(aws_provider)
assert len(efs.filesystems) == 1
efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{FILE_SYSTEM_ID}"
@@ -31,20 +31,20 @@ def mock_generate_regional_clients(provider, service):
class Test_EKS_Service:
# Test EKS Service
def test_service(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
eks = EKS(aws_provider)
assert eks.service == "eks"
# Test EKS client
def test_client(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
eks = EKS(aws_provider)
for reg_client in eks.regional_clients.values():
assert reg_client.__class__.__name__ == "EKS"
# Test EKS session
def test__get_session__(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
eks = EKS(aws_provider)
assert eks.session.__class__.__name__ == "Session"
@@ -73,7 +73,7 @@ class Test_EKS_Service:
roleArn=f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/eks-service-role-AWSServiceRoleForAmazonEKS-J7ONKE3BQ4PI",
tags={"test": "test"},
)
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
eks = EKS(aws_provider)
assert len(eks.clusters) == 1
assert eks.clusters[0].name == cluster_name
@@ -126,7 +126,7 @@ class Test_EKS_Service:
},
],
)
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
eks = EKS(aws_provider)
assert len(eks.clusters) == 1
assert eks.clusters[0].name == cluster_name
@@ -59,7 +59,9 @@ class Test_ElasticBeanstalk_Service:
# Test ElasticBeanstalk Client
@mock_aws
def test_get_client(self):
elasticbeanstalk = ElasticBeanstalk(set_mocked_aws_provider())
elasticbeanstalk = ElasticBeanstalk(
set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
)
assert (
elasticbeanstalk.regional_clients[AWS_REGION_EU_WEST_1].__class__.__name__
== "ElasticBeanstalk"
@@ -68,13 +70,17 @@ class Test_ElasticBeanstalk_Service:
# Test ElasticBeanstalk Session
@mock_aws
def test__get_session__(self):
elasticbeanstalk = ElasticBeanstalk(set_mocked_aws_provider())
elasticbeanstalk = ElasticBeanstalk(
set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
)
assert elasticbeanstalk.session.__class__.__name__ == "Session"
# Test ElasticBeanstalk Service
@mock_aws
def test__get_service__(self):
elasticbeanstalk = ElasticBeanstalk(set_mocked_aws_provider())
elasticbeanstalk = ElasticBeanstalk(
set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
)
assert elasticbeanstalk.service == "elasticbeanstalk"
# Test _describe_environments
@@ -90,7 +96,9 @@ class Test_ElasticBeanstalk_Service:
EnvironmentName="test-env",
)
# ElasticBeanstalk Class
elasticbeanstalk = ElasticBeanstalk(set_mocked_aws_provider())
elasticbeanstalk = ElasticBeanstalk(
set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
)
assert len(elasticbeanstalk.environments) == 1
assert (
@@ -125,7 +133,9 @@ class Test_ElasticBeanstalk_Service:
EnvironmentName="test-env",
)
# ElasticBeanstalk Class
elasticbeanstalk = ElasticBeanstalk(set_mocked_aws_provider())
elasticbeanstalk = ElasticBeanstalk(
set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
)
assert (
elasticbeanstalk.environments[
environment["EnvironmentArn"]
@@ -158,7 +168,9 @@ class Test_ElasticBeanstalk_Service:
Tags=[{"Key": "test-key", "Value": "test-value"}],
)
# ElasticBeanstalk Class
elasticbeanstalk = ElasticBeanstalk(set_mocked_aws_provider())
elasticbeanstalk = ElasticBeanstalk(
set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
)
assert elasticbeanstalk.environments[environment["EnvironmentArn"]].tags == [
{"Key": "test-key", "Value": "test-value"}
]
@@ -91,7 +91,11 @@ class Test_emr_cluster_publicly_accesible:
),
mock.patch(
"prowler.providers.aws.services.emr.emr_cluster_publicly_accesible.emr_cluster_publicly_accesible.ec2_client",
new=EC2(set_mocked_aws_provider(create_default_organization=False)),
new=EC2(
set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1], create_default_organization=False
)
),
),
):
# Test Check
@@ -161,7 +165,11 @@ class Test_emr_cluster_publicly_accesible:
),
mock.patch(
"prowler.providers.aws.services.emr.emr_cluster_publicly_accesible.emr_cluster_publicly_accesible.ec2_client",
new=EC2(set_mocked_aws_provider(create_default_organization=False)),
new=EC2(
set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1], create_default_organization=False
)
),
),
):
# Test Check
@@ -248,7 +256,11 @@ class Test_emr_cluster_publicly_accesible:
),
mock.patch(
"prowler.providers.aws.services.emr.emr_cluster_publicly_accesible.emr_cluster_publicly_accesible.ec2_client",
new=EC2(set_mocked_aws_provider(create_default_organization=False)),
new=EC2(
set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1], create_default_organization=False
)
),
),
):
# Test Check
@@ -338,7 +350,11 @@ class Test_emr_cluster_publicly_accesible:
),
mock.patch(
"prowler.providers.aws.services.emr.emr_cluster_publicly_accesible.emr_cluster_publicly_accesible.ec2_client",
new=EC2(set_mocked_aws_provider(create_default_organization=False)),
new=EC2(
set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1], create_default_organization=False
)
),
),
):
# Test Check
@@ -425,7 +441,11 @@ class Test_emr_cluster_publicly_accesible:
),
mock.patch(
"prowler.providers.aws.services.emr.emr_cluster_publicly_accesible.emr_cluster_publicly_accesible.ec2_client",
new=EC2(set_mocked_aws_provider(create_default_organization=False)),
new=EC2(
set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1], create_default_organization=False
)
),
),
):
# Test Check
@@ -53,19 +53,19 @@ class Test_EMR_Service:
# Test EMR Client
@mock_aws
def test_get_client(self):
emr = EMR(set_mocked_aws_provider())
emr = EMR(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
assert emr.regional_clients[AWS_REGION_EU_WEST_1].__class__.__name__ == "EMR"
# Test EMR Session
@mock_aws
def test__get_session__(self):
emr = EMR(set_mocked_aws_provider())
emr = EMR(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
assert emr.session.__class__.__name__ == "Session"
# Test EMR Service
@mock_aws
def test__get_service__(self):
emr = EMR(set_mocked_aws_provider())
emr = EMR(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
assert emr.service == "emr"
# Test _list_clusters and _describe_cluster
@@ -93,7 +93,7 @@ class Test_EMR_Service:
)
cluster_id = emr_client.run_job_flow(**run_job_flow_args)["JobFlowId"]
# EMR Class
emr = EMR(set_mocked_aws_provider())
emr = EMR(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
assert len(emr.clusters) == 1
assert emr.clusters[cluster_id].id == cluster_id
@@ -115,7 +115,7 @@ class Test_EMR_Service:
@mock_aws
def test_get_block_public_access_configuration(self):
emr = EMR(set_mocked_aws_provider())
emr = EMR(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
assert len(emr.block_public_access_configuration) == 1
assert emr.block_public_access_configuration[
@@ -55,27 +55,27 @@ class Test_GlobalAccelerator_Service:
# Test GlobalAccelerator Service
def test_service(self):
# GlobalAccelerator client for this test class
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_US_WEST_2])
globalaccelerator = GlobalAccelerator(aws_provider)
assert globalaccelerator.service == "globalaccelerator"
# Test GlobalAccelerator Client
def test_client(self):
# GlobalAccelerator client for this test class
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_US_WEST_2])
globalaccelerator = GlobalAccelerator(aws_provider)
assert globalaccelerator.client.__class__.__name__ == "GlobalAccelerator"
# Test GlobalAccelerator Session
def test__get_session__(self):
# GlobalAccelerator client for this test class
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_US_WEST_2])
globalaccelerator = GlobalAccelerator(aws_provider)
assert globalaccelerator.session.__class__.__name__ == "Session"
def test_list_accelerators(self):
# GlobalAccelerator client for this test class
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_US_WEST_2])
globalaccelerator = GlobalAccelerator(aws_provider)
accelerator_name = "TestAccelerator"
@@ -99,7 +99,7 @@ class Test_GlobalAccelerator_Service:
def test_list_tags(self):
# GlobalAccelerator client for this test class
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_US_WEST_2])
globalaccelerator = GlobalAccelerator(aws_provider)
assert len(globalaccelerator.accelerators) == 1
@@ -39,7 +39,7 @@ def mock_make_api_call_members_managers(self, operation_name, api_params):
class Test_guardduty_centrally_managed:
@mock_aws
def test_no_detectors(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -67,7 +67,7 @@ class Test_guardduty_centrally_managed:
detector_id = guardduty_client.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -112,7 +112,7 @@ class Test_guardduty_centrally_managed:
detector_id = guardduty_client.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -156,7 +156,7 @@ class Test_guardduty_centrally_managed:
detector_id = guardduty_client.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -64,7 +64,7 @@ class Test_guardduty_delegated_admin_enabled_all_regions:
@mock_aws
def test_no_detectors(self):
"""Test when no GuardDuty detectors exist."""
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -102,7 +102,7 @@ class Test_guardduty_delegated_admin_enabled_all_regions:
guardduty_client_boto = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
detector_id = guardduty_client_boto.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -148,7 +148,7 @@ class Test_guardduty_delegated_admin_enabled_all_regions:
guardduty_client_boto = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
detector_id = guardduty_client_boto.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -194,7 +194,7 @@ class Test_guardduty_delegated_admin_enabled_all_regions:
guardduty_client_boto = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
detector_id = guardduty_client_boto.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -44,7 +44,7 @@ def mock_make_api_call(self, operation_name, kwarg):
class Test_guardduty_ec2_malware_protection_enabled:
def test_no_detectors(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -74,7 +74,7 @@ class Test_guardduty_ec2_malware_protection_enabled:
guardduty_client.create_detector(Enable=False)
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -112,7 +112,7 @@ class Test_guardduty_ec2_malware_protection_enabled:
},
)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -161,7 +161,7 @@ class Test_guardduty_ec2_malware_protection_enabled:
},
)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -12,7 +12,7 @@ from tests.providers.aws.utils import (
class Test_guardduty_eks_audit_log_enabled:
def test_no_detectors(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -42,7 +42,7 @@ class Test_guardduty_eks_audit_log_enabled:
guardduty_client.create_detector(Enable=False)
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -74,7 +74,7 @@ class Test_guardduty_eks_audit_log_enabled:
Enable=True, DataSources={"Kubernetes": {"AuditLogs": {"Enable": True}}}
)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -118,7 +118,7 @@ class Test_guardduty_eks_audit_log_enabled:
Enable=True, DataSources={"Kubernetes": {"AuditLogs": {"Enable": False}}}
)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -6,6 +6,7 @@ from moto import mock_aws
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
@@ -13,7 +14,7 @@ from tests.providers.aws.utils import (
class Test_guardduty_is_enabled:
@mock_aws
def test_no_detectors(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -43,7 +44,7 @@ class Test_guardduty_is_enabled:
detector_id = guardduty_client.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -85,7 +86,7 @@ class Test_guardduty_is_enabled:
detector_id = guardduty_client.create_detector(Enable=False)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -131,7 +132,7 @@ class Test_guardduty_is_enabled:
detector_id = guardduty_client.create_detector(Enable=False)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -177,7 +178,9 @@ class Test_guardduty_is_enabled:
detector_id = guardduty_client.create_detector(Enable=False)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider(
[AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
)
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -15,7 +15,7 @@ orig = botocore.client.BaseClient._make_api_call
class Test_guardduty_lambda_protection_enabled:
def test_no_detectors(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -45,7 +45,7 @@ class Test_guardduty_lambda_protection_enabled:
guardduty_client.create_detector(Enable=False)
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -78,7 +78,7 @@ class Test_guardduty_lambda_protection_enabled:
Features=[{"Name": "LAMBDA_NETWORK_LOGS", "Status": "ENABLED"}],
)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -123,7 +123,7 @@ class Test_guardduty_lambda_protection_enabled:
Features=[{"Name": "LAMBDA_NETWORK_LOGS", "Status": "DISABLED"}],
)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -28,7 +28,7 @@ def mock_make_api_call(self, operation_name, kwarg):
class Test_guardduty_no_high_severity_findings:
@mock_aws
def test_no_detectors(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -56,7 +56,7 @@ class Test_guardduty_no_high_severity_findings:
detector_id = guardduty_client.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -97,7 +97,7 @@ class Test_guardduty_no_high_severity_findings:
detector_id = guardduty_client.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
@@ -66,20 +66,20 @@ def mock_generate_regional_clients(provider, service):
class Test_GuardDuty_Service:
# Test GuardDuty Service
def test_service(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
assert guardduty.service == "guardduty"
# Test GuardDuty client
def test_client(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
for reg_client in guardduty.regional_clients.values():
assert reg_client.__class__.__name__ == "GuardDuty"
# Test GuardDuty session
def test__get_session__(self):
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
assert guardduty.session.__class__.__name__ == "Session"
@@ -89,7 +89,7 @@ class Test_GuardDuty_Service:
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
response = guardduty_client.create_detector(Enable=True, Tags={"test": "test"})
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
assert len(guardduty.detectors) == 1
@@ -121,7 +121,7 @@ class Test_GuardDuty_Service:
],
)
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
assert len(guardduty.detectors) == 1
@@ -149,7 +149,7 @@ class Test_GuardDuty_Service:
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
response = guardduty_client.create_detector(Enable=True)
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
assert len(guardduty.detectors) == 1
@@ -170,7 +170,7 @@ class Test_GuardDuty_Service:
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
response = guardduty_client.create_detector(Enable=True)
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
assert len(guardduty.detectors) == 1
@@ -192,7 +192,7 @@ class Test_GuardDuty_Service:
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
response = guardduty_client.create_detector(Enable=True)
aws_provider = set_mocked_aws_provider()
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
assert len(guardduty.detectors) == 1
@@ -1413,6 +1413,115 @@ class Test_Policy:
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_aws_CalledVia_str(self):
condition_statement = {
"StringEquals": {"aws:CalledVia": "cloudformation.amazonaws.com"}
}
assert is_condition_block_restrictive(
condition_statement,
TRUSTED_AWS_ACCOUNT_NUMBER,
is_cross_account_allowed=True,
)
def test_condition_parser_string_equals_aws_CalledViaFirst_str(self):
condition_statement = {
"StringEquals": {"aws:CalledViaFirst": "cloudformation.amazonaws.com"}
}
assert is_condition_block_restrictive(
condition_statement,
TRUSTED_AWS_ACCOUNT_NUMBER,
is_cross_account_allowed=True,
)
def test_condition_parser_string_equals_aws_CalledViaLast_str(self):
condition_statement = {
"StringEquals": {"aws:CalledViaLast": "glue.amazonaws.com"}
}
assert is_condition_block_restrictive(
condition_statement,
TRUSTED_AWS_ACCOUNT_NUMBER,
is_cross_account_allowed=True,
)
def test_condition_parser_string_like_aws_CalledVia_str(self):
condition_statement = {"StringLike": {"aws:CalledVia": "*.amazonaws.com"}}
assert is_condition_block_restrictive(
condition_statement,
TRUSTED_AWS_ACCOUNT_NUMBER,
is_cross_account_allowed=True,
)
def test_condition_parser_string_equals_kms_CallerAccount_str(self):
condition_statement = {
"StringEquals": {"kms:CallerAccount": TRUSTED_AWS_ACCOUNT_NUMBER}
}
assert is_condition_block_restrictive(
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_kms_CallerAccount_str_not_valid(self):
condition_statement = {
"StringEquals": {"kms:CallerAccount": NON_TRUSTED_AWS_ACCOUNT_NUMBER}
}
assert not is_condition_block_restrictive(
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_kms_CallerAccount_list(self):
condition_statement = {
"StringEquals": {"kms:CallerAccount": [TRUSTED_AWS_ACCOUNT_NUMBER]}
}
assert is_condition_block_restrictive(
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_kms_CallerAccount_list_not_valid(self):
condition_statement = {
"StringEquals": {
"kms:CallerAccount": [
TRUSTED_AWS_ACCOUNT_NUMBER,
NON_TRUSTED_AWS_ACCOUNT_NUMBER,
]
}
}
assert not is_condition_block_restrictive(
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_kms_ViaService_str(self):
condition_statement = {
"StringEquals": {"kms:ViaService": "glue.eu-central-1.amazonaws.com"}
}
assert is_condition_block_restrictive(
condition_statement,
TRUSTED_AWS_ACCOUNT_NUMBER,
is_cross_account_allowed=True,
)
def test_condition_parser_string_like_kms_CallerAccount_str(self):
condition_statement = {
"StringLike": {"kms:CallerAccount": TRUSTED_AWS_ACCOUNT_NUMBER}
}
assert is_condition_block_restrictive(
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_like_kms_CallerAccount_str_not_valid(self):
condition_statement = {
"StringLike": {"kms:CallerAccount": NON_TRUSTED_AWS_ACCOUNT_NUMBER}
}
assert not is_condition_block_restrictive(
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_like_kms_ViaService_str(self):
condition_statement = {"StringLike": {"kms:ViaService": "glue.*.amazonaws.com"}}
assert is_condition_block_restrictive(
condition_statement,
TRUSTED_AWS_ACCOUNT_NUMBER,
is_cross_account_allowed=True,
)
def test_condition_parser_two_lists_unrestrictive(self):
condition_statement = {
"StringLike": {
@@ -2357,6 +2466,71 @@ class Test_Policy:
trusted_ips=["1.2.3.4", "5.6.7.8"],
)
def test_is_policy_public_kms_caller_account_and_via_service(self):
policy = {
"Version": "2008-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": "*"},
"Action": [
"kms:Encrypt",
"kms:Decrypt",
"kms:ReEncrypt*",
"kms:GenerateDataKey*",
"kms:CreateGrant",
"kms:DescribeKey",
],
"Resource": "*",
"Condition": {
"StringEquals": {
"kms:ViaService": "glue.eu-central-1.amazonaws.com",
"kms:CallerAccount": TRUSTED_AWS_ACCOUNT_NUMBER,
}
},
},
],
}
assert not is_policy_public(policy, TRUSTED_AWS_ACCOUNT_NUMBER)
def test_is_policy_public_kms_caller_account_only(self):
policy = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": "*"},
"Action": ["kms:Decrypt"],
"Resource": "*",
"Condition": {
"StringEquals": {
"kms:CallerAccount": TRUSTED_AWS_ACCOUNT_NUMBER,
}
},
},
],
}
assert not is_policy_public(policy, TRUSTED_AWS_ACCOUNT_NUMBER)
def test_is_policy_public_kms_via_service_without_account_restriction(self):
policy = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": "*"},
"Action": ["kms:Decrypt"],
"Resource": "*",
"Condition": {
"StringEquals": {
"kms:ViaService": "glue.eu-central-1.amazonaws.com",
}
},
},
],
}
assert not is_policy_public(policy, TRUSTED_AWS_ACCOUNT_NUMBER)
def test_check_admin_access(self):
policy = {
"Version": "2012-10-17",
@@ -104,26 +104,26 @@ def mock_generate_regional_clients(provider, service):
class TestOpenSearchServiceService:
# Test OpenSearchService Service
def test_service(self):
aws_provider = set_mocked_aws_provider([])
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
opensearch = OpenSearchService(aws_provider)
assert opensearch.service == "opensearch"
# Test OpenSearchService_ client
def test_client(self):
aws_provider = set_mocked_aws_provider([])
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
opensearch = OpenSearchService(aws_provider)
for reg_client in opensearch.regional_clients.values():
assert reg_client.__class__.__name__ == "OpenSearchService"
# Test OpenSearchService session
def test__get_session__(self):
aws_provider = set_mocked_aws_provider([])
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
opensearch = OpenSearchService(aws_provider)
assert opensearch.session.__class__.__name__ == "Session"
# Test OpenSearchService list domains names
def test_list_domain_names(self):
aws_provider = set_mocked_aws_provider([])
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
opensearch = OpenSearchService(aws_provider)
assert len(opensearch.opensearch_domains) == 1
assert opensearch.opensearch_domains[domain_arn].name == test_domain_name
@@ -132,7 +132,7 @@ class TestOpenSearchServiceService:
# Test OpenSearchService describe domain
@mock_aws
def test_describe_domain(self):
aws_provider = set_mocked_aws_provider([])
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
opensearch = OpenSearchService(aws_provider)
assert len(opensearch.opensearch_domains) == 1
assert opensearch.opensearch_domains[domain_arn].name == test_domain_name
@@ -237,7 +237,7 @@ class TestOpenSearchServiceService:
"botocore.client.BaseClient._make_api_call",
new=mock_make_api_call_missing_fields,
):
aws_provider = set_mocked_aws_provider([])
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
opensearch = OpenSearchService(aws_provider)
# Should not crash even with missing optional fields
@@ -248,6 +248,7 @@ class Test_rds_instance_no_public_access:
PubliclyAccessible=True,
VpcSecurityGroupIds=[default_sg_id],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
from prowler.providers.aws.services.rds.rds_service import RDS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
@@ -256,9 +257,15 @@ class Test_rds_instance_no_public_access:
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access.rds_client",
new=RDS(aws_provider),
with (
mock.patch(
"prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access.rds_client",
new=RDS(aws_provider),
),
mock.patch(
"prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access.ec2_client",
new=EC2(aws_provider),
),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access import (
+4 -2
View File
@@ -96,7 +96,7 @@ ADMINISTRATOR_ROLE_ASSUME_ROLE_POLICY = {
# This here causes to call this function mocking the AWS calls
@mock_aws
def set_mocked_aws_provider(
audited_regions: list[str] = [],
audited_regions: list[str] = [AWS_REGION_US_EAST_1],
audited_account: str = AWS_ACCOUNT_NUMBER,
audited_account_arn: str = AWS_ACCOUNT_ARN,
audited_partition: str = AWS_COMMERCIAL_PARTITION,
@@ -143,7 +143,9 @@ def set_mocked_aws_provider(
# Mock Configiration
provider._scan_unused_services = scan_unused_services
provider._enabled_regions = (
enabled_regions if enabled_regions else set(audited_regions)
enabled_regions
if enabled_regions is not None
else (set(audited_regions) if audited_regions else None)
)
# TODO: we can create the organizations metadata here with moto
provider._organizations_metadata = None
@@ -43,6 +43,39 @@ USER_3 = {
}
# Role data for Directory API role tests
SUPER_ADMIN_ROLE_ID = "13801188331880449"
SEED_ADMIN_ROLE_ID = "13801188331880451"
GROUPS_ADMIN_ROLE_ID = "13801188331880450"
ROLE_SUPER_ADMIN = {
"roleId": SUPER_ADMIN_ROLE_ID,
"roleName": "Super Admin",
"roleDescription": "Super Admin",
"isSystemRole": True,
"isSuperAdminRole": True,
}
# Google automatically assigns _SEED_ADMIN_ROLE to the first account that
# created the domain. It is a super-admin-capable system role with a
# different name, so it must also be excluded when counting "extra" roles.
ROLE_SEED_ADMIN = {
"roleId": SEED_ADMIN_ROLE_ID,
"roleName": "_SEED_ADMIN_ROLE",
"roleDescription": "Super Admin",
"isSystemRole": True,
"isSuperAdminRole": True,
}
ROLE_GROUPS_ADMIN = {
"roleId": GROUPS_ADMIN_ROLE_ID,
"roleName": "_GROUPS_ADMIN_ROLE",
"roleDescription": "Groups Administrator",
"isSystemRole": True,
"isSuperAdminRole": False,
}
def set_mocked_googleworkspace_provider(
identity: GoogleWorkspaceIdentityInfo = GoogleWorkspaceIdentityInfo(
domain=DOMAIN,
@@ -0,0 +1,130 @@
from unittest.mock import patch
from prowler.providers.googleworkspace.services.calendar.calendar_service import (
CalendarPolicies,
)
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
class TestCalendarExternalInvitationsWarning:
def test_pass_warnings_enabled(self):
"""Test PASS when external invitation warnings are enabled"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_external_invitations_warning.calendar_external_invitations_warning.calendar_client"
) as mock_calendar_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_invitations_warning.calendar_external_invitations_warning import (
calendar_external_invitations_warning,
)
mock_calendar_client.provider = mock_provider
mock_calendar_client.policies_fetched = True
mock_calendar_client.policies = CalendarPolicies(
external_invitations_warning=True
)
check = calendar_external_invitations_warning()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "enabled" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].customer_id == CUSTOMER_ID
def test_fail_warnings_disabled(self):
"""Test FAIL when external invitation warnings are disabled"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_external_invitations_warning.calendar_external_invitations_warning.calendar_client"
) as mock_calendar_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_invitations_warning.calendar_external_invitations_warning import (
calendar_external_invitations_warning,
)
mock_calendar_client.provider = mock_provider
mock_calendar_client.policies_fetched = True
mock_calendar_client.policies = CalendarPolicies(
external_invitations_warning=False
)
check = calendar_external_invitations_warning()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "disabled" in findings[0].status_extended
def test_fail_no_policy_set(self):
"""Test FAIL when no explicit policy is set (None) but fetch succeeded"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_external_invitations_warning.calendar_external_invitations_warning.calendar_client"
) as mock_calendar_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_invitations_warning.calendar_external_invitations_warning import (
calendar_external_invitations_warning,
)
mock_calendar_client.provider = mock_provider
mock_calendar_client.policies_fetched = True
mock_calendar_client.policies = CalendarPolicies(
external_invitations_warning=None
)
check = calendar_external_invitations_warning()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_external_invitations_warning.calendar_external_invitations_warning.calendar_client"
) as mock_calendar_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_invitations_warning.calendar_external_invitations_warning import (
calendar_external_invitations_warning,
)
mock_calendar_client.provider = mock_provider
mock_calendar_client.policies_fetched = False
mock_calendar_client.policies = CalendarPolicies()
check = calendar_external_invitations_warning()
findings = check.execute()
assert len(findings) == 0
@@ -0,0 +1,161 @@
from unittest.mock import patch
from prowler.providers.googleworkspace.services.calendar.calendar_service import (
CalendarPolicies,
)
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
class TestCalendarExternalSharingPrimaryCalendar:
def test_pass_free_busy_only(self):
"""Test PASS when external sharing is restricted to free/busy only"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_primary_calendar.calendar_external_sharing_primary_calendar.calendar_client"
) as mock_calendar_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_primary_calendar.calendar_external_sharing_primary_calendar import (
calendar_external_sharing_primary_calendar,
)
mock_calendar_client.provider = mock_provider
mock_calendar_client.policies_fetched = True
mock_calendar_client.policies = CalendarPolicies(
primary_calendar_external_sharing="EXTERNAL_FREE_BUSY_ONLY"
)
check = calendar_external_sharing_primary_calendar()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "free/busy information only" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].customer_id == CUSTOMER_ID
def test_fail_read_only(self):
"""Test FAIL when external sharing allows read-only access"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_primary_calendar.calendar_external_sharing_primary_calendar.calendar_client"
) as mock_calendar_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_primary_calendar.calendar_external_sharing_primary_calendar import (
calendar_external_sharing_primary_calendar,
)
mock_calendar_client.provider = mock_provider
mock_calendar_client.policies_fetched = True
mock_calendar_client.policies = CalendarPolicies(
primary_calendar_external_sharing="EXTERNAL_ALL_INFO_READ_ONLY"
)
check = calendar_external_sharing_primary_calendar()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "EXTERNAL_ALL_INFO_READ_ONLY" in findings[0].status_extended
assert "free/busy information only" in findings[0].status_extended
def test_fail_read_write(self):
"""Test FAIL when external sharing allows read-write access"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_primary_calendar.calendar_external_sharing_primary_calendar.calendar_client"
) as mock_calendar_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_primary_calendar.calendar_external_sharing_primary_calendar import (
calendar_external_sharing_primary_calendar,
)
mock_calendar_client.provider = mock_provider
mock_calendar_client.policies_fetched = True
mock_calendar_client.policies = CalendarPolicies(
primary_calendar_external_sharing="EXTERNAL_ALL_INFO_READ_WRITE"
)
check = calendar_external_sharing_primary_calendar()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "EXTERNAL_ALL_INFO_READ_WRITE" in findings[0].status_extended
def test_fail_no_policy_set(self):
"""Test FAIL when no explicit policy is set (None) but fetch succeeded"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_primary_calendar.calendar_external_sharing_primary_calendar.calendar_client"
) as mock_calendar_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_primary_calendar.calendar_external_sharing_primary_calendar import (
calendar_external_sharing_primary_calendar,
)
mock_calendar_client.provider = mock_provider
mock_calendar_client.policies_fetched = True
mock_calendar_client.policies = CalendarPolicies(
primary_calendar_external_sharing=None
)
check = calendar_external_sharing_primary_calendar()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_primary_calendar.calendar_external_sharing_primary_calendar.calendar_client"
) as mock_calendar_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_primary_calendar.calendar_external_sharing_primary_calendar import (
calendar_external_sharing_primary_calendar,
)
mock_calendar_client.provider = mock_provider
mock_calendar_client.policies_fetched = False
mock_calendar_client.policies = CalendarPolicies()
check = calendar_external_sharing_primary_calendar()
findings = check.execute()
assert len(findings) == 0
@@ -0,0 +1,161 @@
from unittest.mock import patch
from prowler.providers.googleworkspace.services.calendar.calendar_service import (
CalendarPolicies,
)
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
class TestCalendarExternalSharingSecondaryCalendar:
def test_pass_free_busy_only(self):
"""Test PASS when external sharing is restricted to free/busy only"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_secondary_calendar.calendar_external_sharing_secondary_calendar.calendar_client"
) as mock_calendar_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_secondary_calendar.calendar_external_sharing_secondary_calendar import (
calendar_external_sharing_secondary_calendar,
)
mock_calendar_client.provider = mock_provider
mock_calendar_client.policies_fetched = True
mock_calendar_client.policies = CalendarPolicies(
secondary_calendar_external_sharing="EXTERNAL_FREE_BUSY_ONLY"
)
check = calendar_external_sharing_secondary_calendar()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "free/busy information only" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].customer_id == CUSTOMER_ID
def test_fail_read_only(self):
"""Test FAIL when external sharing allows read-only access"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_secondary_calendar.calendar_external_sharing_secondary_calendar.calendar_client"
) as mock_calendar_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_secondary_calendar.calendar_external_sharing_secondary_calendar import (
calendar_external_sharing_secondary_calendar,
)
mock_calendar_client.provider = mock_provider
mock_calendar_client.policies_fetched = True
mock_calendar_client.policies = CalendarPolicies(
secondary_calendar_external_sharing="EXTERNAL_ALL_INFO_READ_ONLY"
)
check = calendar_external_sharing_secondary_calendar()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "EXTERNAL_ALL_INFO_READ_ONLY" in findings[0].status_extended
assert "free/busy information only" in findings[0].status_extended
def test_fail_read_write_manage(self):
"""Test FAIL when external sharing allows read-write-manage access"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_secondary_calendar.calendar_external_sharing_secondary_calendar.calendar_client"
) as mock_calendar_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_secondary_calendar.calendar_external_sharing_secondary_calendar import (
calendar_external_sharing_secondary_calendar,
)
mock_calendar_client.provider = mock_provider
mock_calendar_client.policies_fetched = True
mock_calendar_client.policies = CalendarPolicies(
secondary_calendar_external_sharing="EXTERNAL_ALL_INFO_READ_WRITE_MANAGE"
)
check = calendar_external_sharing_secondary_calendar()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "EXTERNAL_ALL_INFO_READ_WRITE_MANAGE" in findings[0].status_extended
def test_fail_no_policy_set(self):
"""Test FAIL when no explicit policy is set (None) but fetch succeeded"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_secondary_calendar.calendar_external_sharing_secondary_calendar.calendar_client"
) as mock_calendar_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_secondary_calendar.calendar_external_sharing_secondary_calendar import (
calendar_external_sharing_secondary_calendar,
)
mock_calendar_client.provider = mock_provider
mock_calendar_client.policies_fetched = True
mock_calendar_client.policies = CalendarPolicies(
secondary_calendar_external_sharing=None
)
check = calendar_external_sharing_secondary_calendar()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_secondary_calendar.calendar_external_sharing_secondary_calendar.calendar_client"
) as mock_calendar_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_secondary_calendar.calendar_external_sharing_secondary_calendar import (
calendar_external_sharing_secondary_calendar,
)
mock_calendar_client.provider = mock_provider
mock_calendar_client.policies_fetched = False
mock_calendar_client.policies = CalendarPolicies()
check = calendar_external_sharing_secondary_calendar()
findings = check.execute()
assert len(findings) == 0
@@ -0,0 +1,231 @@
from unittest.mock import MagicMock, patch
from tests.providers.googleworkspace.googleworkspace_fixtures import (
set_mocked_googleworkspace_provider,
)
class TestCalendarService:
def test_calendar_fetch_policies_all_settings(self):
"""Test fetching all 3 calendar policy settings from Cloud Identity API"""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_credentials = MagicMock()
mock_session = MagicMock()
mock_session.credentials = mock_credentials
mock_provider.session = mock_session
mock_service = MagicMock()
mock_policies_list = MagicMock()
# Mock the actual Cloud Identity Policy API v1 response shape:
# - "type" (not "name"), prefixed with "settings/"
# - inner value field names are camelCase
mock_policies_list.execute.return_value = {
"policies": [
{
"setting": {
"type": "settings/calendar.primary_calendar_max_allowed_external_sharing",
"value": {
"maxAllowedExternalSharing": "EXTERNAL_FREE_BUSY_ONLY"
},
}
},
{
"setting": {
"type": "settings/calendar.secondary_calendar_max_allowed_external_sharing",
"value": {
"maxAllowedExternalSharing": "EXTERNAL_ALL_INFO_READ_ONLY"
},
}
},
{
"setting": {
"type": "settings/calendar.external_invitations",
"value": {"warnOnInvite": True},
}
},
]
}
mock_service.policies().list.return_value = mock_policies_list
mock_service.policies().list_next.return_value = None
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.calendar.calendar_service import (
Calendar,
)
calendar = Calendar(mock_provider)
assert calendar.policies_fetched is True
assert (
calendar.policies.primary_calendar_external_sharing
== "EXTERNAL_FREE_BUSY_ONLY"
)
assert (
calendar.policies.secondary_calendar_external_sharing
== "EXTERNAL_ALL_INFO_READ_ONLY"
)
assert calendar.policies.external_invitations_warning is True
def test_calendar_fetch_policies_empty_response(self):
"""Test handling empty policies response"""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_session = MagicMock()
mock_session.credentials = MagicMock()
mock_provider.session = mock_session
mock_service = MagicMock()
mock_policies_list = MagicMock()
mock_policies_list.execute.return_value = {"policies": []}
mock_service.policies().list.return_value = mock_policies_list
mock_service.policies().list_next.return_value = None
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.calendar.calendar_service import (
Calendar,
)
calendar = Calendar(mock_provider)
assert calendar.policies_fetched is True
assert calendar.policies.primary_calendar_external_sharing is None
assert calendar.policies.secondary_calendar_external_sharing is None
assert calendar.policies.external_invitations_warning is None
def test_calendar_fetch_policies_api_error(self):
"""Test handling of API errors during policy fetch"""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_session = MagicMock()
mock_session.credentials = MagicMock()
mock_provider.session = mock_session
mock_service = MagicMock()
mock_service.policies().list.side_effect = Exception("API Error")
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.calendar.calendar_service import (
Calendar,
)
calendar = Calendar(mock_provider)
assert calendar.policies_fetched is False
assert calendar.policies.primary_calendar_external_sharing is None
assert calendar.policies.secondary_calendar_external_sharing is None
assert calendar.policies.external_invitations_warning is None
def test_calendar_fetch_policies_build_service_returns_none(self):
"""Test early return when _build_service fails to construct the client"""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_session = MagicMock()
mock_session.credentials = MagicMock()
mock_provider.session = mock_session
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_service.GoogleWorkspaceService._build_service",
return_value=None,
),
):
from prowler.providers.googleworkspace.services.calendar.calendar_service import (
Calendar,
)
calendar = Calendar(mock_provider)
assert calendar.policies_fetched is False
assert calendar.policies.primary_calendar_external_sharing is None
assert calendar.policies.secondary_calendar_external_sharing is None
assert calendar.policies.external_invitations_warning is None
def test_calendar_fetch_policies_execute_raises(self):
"""Test inner except handler when request.execute() raises during pagination"""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_session = MagicMock()
mock_session.credentials = MagicMock()
mock_provider.session = mock_session
mock_service = MagicMock()
mock_request = MagicMock()
mock_request.execute.side_effect = Exception("Execute failed")
mock_service.policies().list.return_value = mock_request
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.calendar.calendar_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.calendar.calendar_service import (
Calendar,
)
calendar = Calendar(mock_provider)
assert calendar.policies_fetched is False
assert calendar.policies.primary_calendar_external_sharing is None
assert calendar.policies.secondary_calendar_external_sharing is None
assert calendar.policies.external_invitations_warning is None
def test_calendar_policies_model(self):
"""Test CalendarPolicies Pydantic model"""
from prowler.providers.googleworkspace.services.calendar.calendar_service import (
CalendarPolicies,
)
policies = CalendarPolicies(
primary_calendar_external_sharing="EXTERNAL_FREE_BUSY_ONLY",
secondary_calendar_external_sharing="EXTERNAL_ALL_INFO_READ_WRITE",
external_invitations_warning=True,
)
assert policies.primary_calendar_external_sharing == "EXTERNAL_FREE_BUSY_ONLY"
assert (
policies.secondary_calendar_external_sharing
== "EXTERNAL_ALL_INFO_READ_WRITE"
)
assert policies.external_invitations_warning is True
@@ -1,6 +1,12 @@
from unittest.mock import MagicMock, patch
from tests.providers.googleworkspace.googleworkspace_fixtures import (
GROUPS_ADMIN_ROLE_ID,
ROLE_GROUPS_ADMIN,
ROLE_SEED_ADMIN,
ROLE_SUPER_ADMIN,
SEED_ADMIN_ROLE_ID,
SUPER_ADMIN_ROLE_ID,
USER_1,
USER_2,
USER_3,
@@ -25,6 +31,24 @@ class TestDirectoryService:
mock_service.users().list.return_value = mock_users_list
mock_service.users().list_next.return_value = None
# Mock roles response
mock_roles_list = MagicMock()
mock_roles_list.execute.return_value = {
"items": [ROLE_SUPER_ADMIN, ROLE_GROUPS_ADMIN]
}
mock_service.roles().list.return_value = mock_roles_list
mock_service.roles().list_next.return_value = None
mock_ra = MagicMock()
mock_ra.execute.return_value = {
"items": [
{"assignedTo": "user1-id", "roleId": SUPER_ADMIN_ROLE_ID},
{"assignedTo": "user2-id", "roleId": SUPER_ADMIN_ROLE_ID},
]
}
mock_service.roleAssignments().list.return_value = mock_ra
mock_service.roleAssignments().list_next.return_value = None
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -67,6 +91,17 @@ class TestDirectoryService:
mock_service.users().list.return_value = mock_users_list
mock_service.users().list_next.return_value = None
# Mock roles response
mock_roles_list = MagicMock()
mock_roles_list.execute.return_value = {"items": []}
mock_service.roles().list.return_value = mock_roles_list
mock_service.roles().list_next.return_value = None
mock_ra = MagicMock()
mock_ra.execute.return_value = {"items": []}
mock_service.roleAssignments().list.return_value = mock_ra
mock_service.roleAssignments().list_next.return_value = None
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -97,6 +132,16 @@ class TestDirectoryService:
mock_service = MagicMock()
mock_service.users().list.side_effect = Exception("API Error")
mock_roles_list = MagicMock()
mock_roles_list.execute.return_value = {"items": []}
mock_service.roles().list.return_value = mock_roles_list
mock_service.roles().list_next.return_value = None
mock_ra = MagicMock()
mock_ra.execute.return_value = {"items": []}
mock_service.roleAssignments().list.return_value = mock_ra
mock_service.roleAssignments().list_next.return_value = None
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -130,3 +175,193 @@ class TestDirectoryService:
assert user.id == "test-id"
assert user.email == "test@test-company.com"
assert user.is_admin is True
assert user.role_assignments == []
def test_directory_list_roles(self):
"""Test that _list_roles correctly builds a roleId-to-roleName mapping"""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_session = MagicMock()
mock_session.credentials = MagicMock()
mock_provider.session = mock_session
mock_service = MagicMock()
# Mock empty users
mock_users_list = MagicMock()
mock_users_list.execute.return_value = {"users": []}
mock_service.users().list.return_value = mock_users_list
mock_service.users().list_next.return_value = None
# Mock roles response
mock_roles_list = MagicMock()
mock_roles_list.execute.return_value = {
"items": [ROLE_SUPER_ADMIN, ROLE_GROUPS_ADMIN]
}
mock_service.roles().list.return_value = mock_roles_list
mock_service.roles().list_next.return_value = None
mock_ra = MagicMock()
mock_ra.execute.return_value = {"items": []}
mock_service.roleAssignments().list.return_value = mock_ra
mock_service.roleAssignments().list_next.return_value = None
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.directory.directory_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.directory.directory_service import (
Directory,
)
directory = Directory(mock_provider)
super_admin_role = directory._roles[SUPER_ADMIN_ROLE_ID]
assert super_admin_role.name == "Super Admin"
assert super_admin_role.description == "Super Admin"
assert super_admin_role.is_super_admin_role is True
groups_admin_role = directory._roles[GROUPS_ADMIN_ROLE_ID]
assert groups_admin_role.name == "_GROUPS_ADMIN_ROLE"
assert groups_admin_role.description == "Groups Administrator"
assert groups_admin_role.is_super_admin_role is False
def test_directory_role_assignments_populated(self):
"""Test that role assignments are fetched and resolved for super admins"""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_session = MagicMock()
mock_session.credentials = MagicMock()
mock_provider.session = mock_session
mock_service = MagicMock()
# Mock users - one super admin
mock_users_list = MagicMock()
mock_users_list.execute.return_value = {"users": [USER_1]}
mock_service.users().list.return_value = mock_users_list
mock_service.users().list_next.return_value = None
# Mock roles
mock_roles_list = MagicMock()
mock_roles_list.execute.return_value = {
"items": [ROLE_SUPER_ADMIN, ROLE_GROUPS_ADMIN]
}
mock_service.roles().list.return_value = mock_roles_list
mock_service.roles().list_next.return_value = None
mock_ra = MagicMock()
mock_ra.execute.return_value = {
"items": [
{"assignedTo": "user1-id", "roleId": SUPER_ADMIN_ROLE_ID},
{"assignedTo": "user1-id", "roleId": GROUPS_ADMIN_ROLE_ID},
]
}
mock_service.roleAssignments().list.return_value = mock_ra
mock_service.roleAssignments().list_next.return_value = None
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.directory.directory_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.directory.directory_service import (
Directory,
)
directory = Directory(mock_provider)
user = directory.users["user1-id"]
role_names = [r.name for r in user.role_assignments]
role_descriptions = [r.description for r in user.role_assignments]
assert "Super Admin" in role_names
assert "_GROUPS_ADMIN_ROLE" in role_names
assert "Groups Administrator" in role_descriptions
assert len(user.role_assignments) == 2
assert user.is_admin is True
def test_directory_second_super_admin_detected_via_role_assignments(self):
"""Regression: a second super admin whose users.list().isAdmin still
reads False (e.g. API propagation lag, or only holding
_SEED_ADMIN_ROLE) must still be recognised as a super admin through
the Role Assignments API, AND any extra non-super-admin roles they
hold must be surfaced on their User object."""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_session = MagicMock()
mock_session.credentials = MagicMock()
mock_provider.session = mock_session
mock_service = MagicMock()
stale_user_1 = {
"id": "user1-id",
"primaryEmail": "admin1@test-company.com",
"isAdmin": False,
}
stale_user_2 = {
"id": "user2-id",
"primaryEmail": "admin2@test-company.com",
"isAdmin": False,
}
mock_users_list = MagicMock()
mock_users_list.execute.return_value = {"users": [stale_user_1, stale_user_2]}
mock_service.users().list.return_value = mock_users_list
mock_service.users().list_next.return_value = None
mock_roles_list = MagicMock()
mock_roles_list.execute.return_value = {
"items": [ROLE_SUPER_ADMIN, ROLE_SEED_ADMIN, ROLE_GROUPS_ADMIN]
}
mock_service.roles().list.return_value = mock_roles_list
mock_service.roles().list_next.return_value = None
mock_ra = MagicMock()
mock_ra.execute.return_value = {
"items": [
{"assignedTo": "user1-id", "roleId": SEED_ADMIN_ROLE_ID},
{"assignedTo": "user2-id", "roleId": SUPER_ADMIN_ROLE_ID},
{"assignedTo": "user2-id", "roleId": GROUPS_ADMIN_ROLE_ID},
]
}
mock_service.roleAssignments().list.return_value = mock_ra
mock_service.roleAssignments().list_next.return_value = None
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.directory.directory_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.directory.directory_service import (
Directory,
)
directory = Directory(mock_provider)
user1 = directory.users["user1-id"]
user2 = directory.users["user2-id"]
assert user1.is_admin is True
assert user2.is_admin is True
assert [r.name for r in user1.role_assignments] == ["_SEED_ADMIN_ROLE"]
user2_role_names = {r.name for r in user2.role_assignments}
assert user2_role_names == {"Super Admin", "_GROUPS_ADMIN_ROLE"}
@@ -0,0 +1,446 @@
from unittest.mock import patch
from prowler.providers.googleworkspace.services.directory.directory_service import (
Role,
User,
)
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
SUPER_ADMIN_ROLE = Role(
id="13801188331880449",
name="Super Admin",
description="Super Admin",
is_super_admin_role=True,
)
SEED_ADMIN_ROLE = Role(
id="13801188331880451",
name="_SEED_ADMIN_ROLE",
description="Super Admin",
is_super_admin_role=True,
)
GROUPS_ADMIN_ROLE = Role(
id="13801188331880450",
name="_GROUPS_ADMIN_ROLE",
description="Groups Administrator",
is_super_admin_role=False,
)
USER_MANAGEMENT_ADMIN_ROLE = Role(
id="13801188331880452",
name="_USER_MANAGEMENT_ADMIN_ROLE",
description="User Management Administrator",
is_super_admin_role=False,
)
CUSTOM_ROLE_NO_DESCRIPTION = Role(
id="13801188331880453",
name="custom-helpdesk-role",
description="",
is_super_admin_role=False,
)
class TestDirectorySuperAdminOnlyAdminRoles:
def test_pass_super_admins_only_super_admin_role(self):
"""Test PASS when super admins have only the Super Admin role"""
users = {
"admin1-id": User(
id="admin1-id",
email="admin1@test-company.com",
is_admin=True,
role_assignments=[SUPER_ADMIN_ROLE],
),
"admin2-id": User(
id="admin2-id",
email="admin2@test-company.com",
is_admin=True,
role_assignments=[SUPER_ADMIN_ROLE],
),
"user1-id": User(
id="user1-id",
email="user@test-company.com",
is_admin=False,
role_assignments=[],
),
}
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles.directory_client"
) as mock_directory_client,
):
from prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles import (
directory_super_admin_only_admin_roles,
)
mock_directory_client.users = users
mock_directory_client.provider = mock_provider
check = directory_super_admin_only_admin_roles()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "used only for super admin activities" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].customer_id == CUSTOMER_ID
def test_pass_super_admin_with_seed_admin_role(self):
"""Test PASS when a super admin only holds _SEED_ADMIN_ROLE.
_SEED_ADMIN_ROLE is auto-assigned by Google to the original domain
creator and has isSuperAdminRole=True, so it must not count as an
"extra" role.
"""
users = {
"admin1-id": User(
id="admin1-id",
email="playground@prowler.cloud",
is_admin=True,
role_assignments=[SEED_ADMIN_ROLE],
),
}
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles.directory_client"
) as mock_directory_client,
):
from prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles import (
directory_super_admin_only_admin_roles,
)
mock_directory_client.users = users
mock_directory_client.provider = mock_provider
check = directory_super_admin_only_admin_roles()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "_SEED_ADMIN_ROLE" not in findings[0].status_extended
def test_pass_super_admin_with_both_super_admin_and_seed_admin(self):
"""Test PASS when admin holds both Super Admin and _SEED_ADMIN_ROLE"""
users = {
"admin1-id": User(
id="admin1-id",
email="playground@prowler.cloud",
is_admin=True,
role_assignments=[SUPER_ADMIN_ROLE, SEED_ADMIN_ROLE],
),
}
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles.directory_client"
) as mock_directory_client,
):
from prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles import (
directory_super_admin_only_admin_roles,
)
mock_directory_client.users = users
mock_directory_client.provider = mock_provider
check = directory_super_admin_only_admin_roles()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
def test_fail_super_admin_with_additional_roles(self):
"""Test FAIL when a super admin also has additional admin roles"""
users = {
"admin1-id": User(
id="admin1-id",
email="admin1@test-company.com",
is_admin=True,
role_assignments=[SUPER_ADMIN_ROLE, GROUPS_ADMIN_ROLE],
),
"user1-id": User(
id="user1-id",
email="user@test-company.com",
is_admin=False,
role_assignments=[],
),
}
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles.directory_client"
) as mock_directory_client,
):
from prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles import (
directory_super_admin_only_admin_roles,
)
mock_directory_client.users = users
mock_directory_client.provider = mock_provider
check = directory_super_admin_only_admin_roles()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "admin1@test-company.com" in findings[0].status_extended
assert "Groups Administrator" in findings[0].status_extended
assert "_GROUPS_ADMIN_ROLE" not in findings[0].status_extended
assert "used only for super admin activities" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].customer_id == CUSTOMER_ID
def test_fail_seed_admin_with_additional_roles(self):
"""Test FAIL when a _SEED_ADMIN_ROLE holder also has extra roles"""
users = {
"admin1-id": User(
id="admin1-id",
email="playground@prowler.cloud",
is_admin=True,
role_assignments=[SEED_ADMIN_ROLE, GROUPS_ADMIN_ROLE],
),
}
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles.directory_client"
) as mock_directory_client,
):
from prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles import (
directory_super_admin_only_admin_roles,
)
mock_directory_client.users = users
mock_directory_client.provider = mock_provider
check = directory_super_admin_only_admin_roles()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "playground@prowler.cloud" in findings[0].status_extended
assert "Groups Administrator" in findings[0].status_extended
assert "_GROUPS_ADMIN_ROLE" not in findings[0].status_extended
assert "_SEED_ADMIN_ROLE" not in findings[0].status_extended
def test_fail_multiple_super_admins_with_extra_roles(self):
"""Test FAIL lists all super admins that have additional roles"""
users = {
"admin1-id": User(
id="admin1-id",
email="admin1@test-company.com",
is_admin=True,
role_assignments=[SUPER_ADMIN_ROLE, GROUPS_ADMIN_ROLE],
),
"admin2-id": User(
id="admin2-id",
email="admin2@test-company.com",
is_admin=True,
role_assignments=[SUPER_ADMIN_ROLE, USER_MANAGEMENT_ADMIN_ROLE],
),
"admin3-id": User(
id="admin3-id",
email="admin3@test-company.com",
is_admin=True,
role_assignments=[SUPER_ADMIN_ROLE],
),
}
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles.directory_client"
) as mock_directory_client,
):
from prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles import (
directory_super_admin_only_admin_roles,
)
mock_directory_client.users = users
mock_directory_client.provider = mock_provider
check = directory_super_admin_only_admin_roles()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "admin1@test-company.com" in findings[0].status_extended
assert "admin2@test-company.com" in findings[0].status_extended
assert "admin3@test-company.com" not in findings[0].status_extended
def test_no_findings_when_no_users(self):
"""Test no findings when there are no users"""
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles.directory_client"
) as mock_directory_client,
):
from prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles import (
directory_super_admin_only_admin_roles,
)
mock_directory_client.users = {}
mock_directory_client.provider = mock_provider
check = directory_super_admin_only_admin_roles()
findings = check.execute()
assert len(findings) == 0
def test_non_super_admin_with_roles_not_flagged(self):
"""Test that users who are not super admins are ignored even if they have roles"""
users = {
"admin1-id": User(
id="admin1-id",
email="admin1@test-company.com",
is_admin=True,
role_assignments=[SUPER_ADMIN_ROLE],
),
"delegated1-id": User(
id="delegated1-id",
email="delegated@test-company.com",
is_admin=False,
role_assignments=[GROUPS_ADMIN_ROLE],
),
}
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles.directory_client"
) as mock_directory_client,
):
from prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles import (
directory_super_admin_only_admin_roles,
)
mock_directory_client.users = users
mock_directory_client.provider = mock_provider
check = directory_super_admin_only_admin_roles()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "delegated@test-company.com" not in findings[0].status_extended
def test_pass_super_admin_with_empty_role_assignments(self):
"""Test PASS when super admin has no role assignments (edge case)"""
users = {
"admin1-id": User(
id="admin1-id",
email="admin1@test-company.com",
is_admin=True,
role_assignments=[],
),
}
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles.directory_client"
) as mock_directory_client,
):
from prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles import (
directory_super_admin_only_admin_roles,
)
mock_directory_client.users = users
mock_directory_client.provider = mock_provider
check = directory_super_admin_only_admin_roles()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
def test_fail_custom_role_without_description_falls_back_to_name(self):
"""A custom role with an empty description should be displayed
using its name as a fall-back, so the FAIL message is never blank
for users that genuinely hold extra roles."""
users = {
"admin1-id": User(
id="admin1-id",
email="admin1@test-company.com",
is_admin=True,
role_assignments=[SUPER_ADMIN_ROLE, CUSTOM_ROLE_NO_DESCRIPTION],
),
}
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles.directory_client"
) as mock_directory_client,
):
from prowler.providers.googleworkspace.services.directory.directory_super_admin_only_admin_roles.directory_super_admin_only_admin_roles import (
directory_super_admin_only_admin_roles,
)
mock_directory_client.users = users
mock_directory_client.provider = mock_provider
check = directory_super_admin_only_admin_roles()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "custom-helpdesk-role" in findings[0].status_extended