mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-31 07:44:21 +00:00
Merge branch 'master' into PROWLER-1511-okta-stig-signon-service-global-session-policy-checks-sdk
This commit is contained in:
34 files changed
+1301
-114
No files matched your search
+54
-7
@@ -93,6 +93,8 @@ class Test_s3_bucket_shadow_resource_vulnerability:
|
||||
|
||||
@mock_aws
|
||||
def test_bucket_not_predictable(self):
|
||||
# A bucket whose name does not match any predictable service pattern
|
||||
# is not a shadow resource and must not produce any finding.
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
aws_provider.identity.identity_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
|
||||
|
||||
@@ -113,6 +115,55 @@ class Test_s3_bucket_shadow_resource_vulnerability:
|
||||
s3_client.provider = aws_provider
|
||||
s3_client._head_bucket = mock.MagicMock(return_value=False)
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=aws_provider,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.s3.s3_bucket_shadow_resource_vulnerability.s3_bucket_shadow_resource_vulnerability.s3_client",
|
||||
new=s3_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.s3.s3_bucket_shadow_resource_vulnerability.s3_bucket_shadow_resource_vulnerability import (
|
||||
s3_bucket_shadow_resource_vulnerability,
|
||||
)
|
||||
|
||||
check = s3_bucket_shadow_resource_vulnerability()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 0
|
||||
|
||||
@mock_aws
|
||||
def test_only_predictable_bucket_reported_among_many(self):
|
||||
# With a mix of buckets, only the one matching a predictable pattern
|
||||
# must produce a finding; the rest must be silent.
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
aws_provider.identity.identity_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
|
||||
|
||||
predictable_bucket = f"sagemaker-{AWS_REGION_US_EAST_1}-{AWS_ACCOUNT_NUMBER}"
|
||||
plain_buckets = [
|
||||
"config-bucket-data",
|
||||
"my-app-data-bucket",
|
||||
"guardduty-findings-store",
|
||||
]
|
||||
|
||||
s3_client = mock.MagicMock()
|
||||
s3_client.audited_canonical_id = AWS_ACCOUNT_NUMBER
|
||||
s3_client.audited_partition = "aws"
|
||||
s3_client.buckets = {
|
||||
name: Bucket(
|
||||
name=name,
|
||||
arn=f"arn:aws:s3:::{name}",
|
||||
region=AWS_REGION_US_EAST_1,
|
||||
owner_id=AWS_ACCOUNT_NUMBER,
|
||||
tags=[],
|
||||
)
|
||||
for name in [predictable_bucket, *plain_buckets]
|
||||
}
|
||||
s3_client.provider = aws_provider
|
||||
s3_client._head_bucket = mock.MagicMock(return_value=False)
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
@@ -132,14 +183,10 @@ class Test_s3_bucket_shadow_resource_vulnerability:
|
||||
|
||||
assert len(result) == 1
|
||||
report = result[0]
|
||||
|
||||
# Test all report attributes
|
||||
assert report.status == "PASS"
|
||||
assert report.region == AWS_REGION_US_EAST_1
|
||||
assert report.resource_id == bucket_name
|
||||
assert report.resource_arn == f"arn:aws:s3:::{bucket_name}"
|
||||
assert report.resource_tags == [{"Key": "Project", "Value": "test-project"}]
|
||||
assert "is not a known shadow resource" in report.status_extended
|
||||
assert report.resource_id == predictable_bucket
|
||||
assert "SageMaker" in report.status_extended
|
||||
assert "is correctly owned by the audited account" in report.status_extended
|
||||
|
||||
@mock_aws
|
||||
def test_shadow_resource_in_other_account(self):
|
||||
|
||||
+197
@@ -264,3 +264,200 @@ class Test_sqlserver_tde_encrypted_with_cmk:
|
||||
assert result[0].resource_name == sql_server_name
|
||||
assert result[0].resource_id == sql_server_id
|
||||
assert result[0].location == "location"
|
||||
|
||||
def test_sql_servers_master_database_disabled_user_database_enabled(self):
|
||||
# System "master" database always reports TDE Disabled in Azure SQL
|
||||
# and is not customer-controllable. It must not fail a server whose
|
||||
# user databases are correctly encrypted with CMK (PROWLER-1760).
|
||||
sqlserver_client = mock.MagicMock
|
||||
sqlserver_client.subscriptions = {
|
||||
AZURE_SUBSCRIPTION_ID: AZURE_SUBSCRIPTION_NAME
|
||||
}
|
||||
sql_server_name = "SQL Server Name"
|
||||
sql_server_id = str(uuid4())
|
||||
master_database = Database(
|
||||
id="master_id",
|
||||
name="master",
|
||||
type="type",
|
||||
location="location",
|
||||
managed_by="managed_by",
|
||||
tde_encryption=TransparentDataEncryption(status="Disabled"),
|
||||
)
|
||||
user_database = Database(
|
||||
id="user_id",
|
||||
name="DynamicBudgets_Intacct",
|
||||
type="type",
|
||||
location="location",
|
||||
managed_by="managed_by",
|
||||
tde_encryption=TransparentDataEncryption(status="Enabled"),
|
||||
)
|
||||
sqlserver_client.sql_servers = {
|
||||
AZURE_SUBSCRIPTION_ID: [
|
||||
Server(
|
||||
id=sql_server_id,
|
||||
name=sql_server_name,
|
||||
location="location",
|
||||
public_network_access="",
|
||||
minimal_tls_version="",
|
||||
administrators=None,
|
||||
auditing_policies=None,
|
||||
firewall_rules=None,
|
||||
databases=[master_database, user_database],
|
||||
encryption_protector=EncryptionProtector(
|
||||
server_key_type="AzureKeyVault"
|
||||
),
|
||||
)
|
||||
]
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_azure_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.azure.services.sqlserver.sqlserver_tde_encrypted_with_cmk.sqlserver_tde_encrypted_with_cmk.sqlserver_client",
|
||||
new=sqlserver_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.azure.services.sqlserver.sqlserver_tde_encrypted_with_cmk.sqlserver_tde_encrypted_with_cmk import (
|
||||
sqlserver_tde_encrypted_with_cmk,
|
||||
)
|
||||
|
||||
check = sqlserver_tde_encrypted_with_cmk()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"SQL Server {sql_server_name} from subscription {AZURE_SUBSCRIPTION_DISPLAY} has TDE enabled with CMK."
|
||||
)
|
||||
assert result[0].subscription == AZURE_SUBSCRIPTION_ID
|
||||
assert result[0].resource_name == sql_server_name
|
||||
assert result[0].resource_id == sql_server_id
|
||||
assert result[0].location == "location"
|
||||
|
||||
def test_sql_servers_only_master_database(self):
|
||||
# A server whose only database is the system "master" has no user
|
||||
# databases to evaluate, so it must not produce a finding.
|
||||
sqlserver_client = mock.MagicMock
|
||||
sqlserver_client.subscriptions = {
|
||||
AZURE_SUBSCRIPTION_ID: AZURE_SUBSCRIPTION_NAME
|
||||
}
|
||||
sql_server_name = "SQL Server Name"
|
||||
sql_server_id = str(uuid4())
|
||||
master_database = Database(
|
||||
id="master_id",
|
||||
name="MASTER",
|
||||
type="type",
|
||||
location="location",
|
||||
managed_by="managed_by",
|
||||
tde_encryption=TransparentDataEncryption(status="Disabled"),
|
||||
)
|
||||
sqlserver_client.sql_servers = {
|
||||
AZURE_SUBSCRIPTION_ID: [
|
||||
Server(
|
||||
id=sql_server_id,
|
||||
name=sql_server_name,
|
||||
location="location",
|
||||
public_network_access="",
|
||||
minimal_tls_version="",
|
||||
administrators=None,
|
||||
auditing_policies=None,
|
||||
firewall_rules=None,
|
||||
databases=[master_database],
|
||||
encryption_protector=EncryptionProtector(
|
||||
server_key_type="AzureKeyVault"
|
||||
),
|
||||
)
|
||||
]
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_azure_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.azure.services.sqlserver.sqlserver_tde_encrypted_with_cmk.sqlserver_tde_encrypted_with_cmk.sqlserver_client",
|
||||
new=sqlserver_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.azure.services.sqlserver.sqlserver_tde_encrypted_with_cmk.sqlserver_tde_encrypted_with_cmk import (
|
||||
sqlserver_tde_encrypted_with_cmk,
|
||||
)
|
||||
|
||||
check = sqlserver_tde_encrypted_with_cmk()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_sql_servers_master_disabled_user_database_disabled(self):
|
||||
# Filtering out "master" must not mask a genuinely failing user
|
||||
# database: a disabled user DB still fails even with CMK.
|
||||
sqlserver_client = mock.MagicMock
|
||||
sqlserver_client.subscriptions = {
|
||||
AZURE_SUBSCRIPTION_ID: AZURE_SUBSCRIPTION_NAME
|
||||
}
|
||||
sql_server_name = "SQL Server Name"
|
||||
sql_server_id = str(uuid4())
|
||||
master_database = Database(
|
||||
id="master_id",
|
||||
name="master",
|
||||
type="type",
|
||||
location="location",
|
||||
managed_by="managed_by",
|
||||
tde_encryption=TransparentDataEncryption(status="Disabled"),
|
||||
)
|
||||
user_database = Database(
|
||||
id="user_id",
|
||||
name="DynamicBudgets_Intacct",
|
||||
type="type",
|
||||
location="location",
|
||||
managed_by="managed_by",
|
||||
tde_encryption=TransparentDataEncryption(status="Disabled"),
|
||||
)
|
||||
sqlserver_client.sql_servers = {
|
||||
AZURE_SUBSCRIPTION_ID: [
|
||||
Server(
|
||||
id=sql_server_id,
|
||||
name=sql_server_name,
|
||||
location="location",
|
||||
public_network_access="",
|
||||
minimal_tls_version="",
|
||||
administrators=None,
|
||||
auditing_policies=None,
|
||||
firewall_rules=None,
|
||||
databases=[master_database, user_database],
|
||||
encryption_protector=EncryptionProtector(
|
||||
server_key_type="AzureKeyVault"
|
||||
),
|
||||
)
|
||||
]
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_azure_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.azure.services.sqlserver.sqlserver_tde_encrypted_with_cmk.sqlserver_tde_encrypted_with_cmk.sqlserver_client",
|
||||
new=sqlserver_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.azure.services.sqlserver.sqlserver_tde_encrypted_with_cmk.sqlserver_tde_encrypted_with_cmk import (
|
||||
sqlserver_tde_encrypted_with_cmk,
|
||||
)
|
||||
|
||||
check = sqlserver_tde_encrypted_with_cmk()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"SQL Server {sql_server_name} from subscription {AZURE_SUBSCRIPTION_DISPLAY} has TDE disabled with CMK."
|
||||
)
|
||||
assert result[0].subscription == AZURE_SUBSCRIPTION_ID
|
||||
assert result[0].resource_name == sql_server_name
|
||||
assert result[0].resource_id == sql_server_id
|
||||
assert result[0].location == "location"
|
||||
@@ -41,7 +41,7 @@ def set_mocked_gcp_provider(
|
||||
return provider
|
||||
|
||||
|
||||
def mock_api_client(GCPService, service, api_version, _):
|
||||
def mock_api_client(_GCPService, service, _api_version, _):
|
||||
client = MagicMock()
|
||||
|
||||
mock_api_projects_calls(client)
|
||||
@@ -703,6 +703,9 @@ def mock_api_instances_calls(client: MagicMock, service: str):
|
||||
"databaseVersion": "MYSQL_5_7",
|
||||
"region": "us-central1",
|
||||
"ipAddresses": [{"type": "PRIMARY", "ipAddress": "66.66.66.66"}],
|
||||
"diskEncryptionConfiguration": {
|
||||
"kmsKeyName": "projects/123/locations/us-central1/keyRings/keyring1/cryptoKeys/key1"
|
||||
},
|
||||
"settings": {
|
||||
"ipConfiguration": {
|
||||
"requireSsl": True,
|
||||
@@ -1323,6 +1326,7 @@ def mock_api_images_calls(client: MagicMock):
|
||||
client.images().list_next.return_value = None
|
||||
|
||||
def mock_get_image_iam_policy(project, resource):
|
||||
del project
|
||||
return_value = MagicMock()
|
||||
if resource == "test-image-1":
|
||||
return_value.execute.return_value = {
|
||||
|
||||
+277
@@ -0,0 +1,277 @@
|
||||
from unittest import mock
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from tests.providers.gcp.gcp_fixtures import (
|
||||
GCP_EU1_LOCATION,
|
||||
GCP_PROJECT_ID,
|
||||
mock_is_api_active,
|
||||
set_mocked_gcp_provider,
|
||||
)
|
||||
|
||||
|
||||
class Test_cloudsql_instance_cmek_encryption_enabled:
|
||||
def test_no_instances(self):
|
||||
cloudsql_client = mock.MagicMock()
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_gcp_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.gcp.services.cloudsql.cloudsql_instance_cmek_encryption_enabled.cloudsql_instance_cmek_encryption_enabled.cloudsql_client",
|
||||
new=cloudsql_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.gcp.services.cloudsql.cloudsql_instance_cmek_encryption_enabled.cloudsql_instance_cmek_encryption_enabled import (
|
||||
cloudsql_instance_cmek_encryption_enabled,
|
||||
)
|
||||
|
||||
cloudsql_client.instances = []
|
||||
check = cloudsql_instance_cmek_encryption_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_instance_cmek_enabled(self):
|
||||
cloudsql_client = mock.MagicMock()
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_gcp_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.gcp.services.cloudsql.cloudsql_instance_cmek_encryption_enabled.cloudsql_instance_cmek_encryption_enabled.cloudsql_client",
|
||||
new=cloudsql_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.gcp.services.cloudsql.cloudsql_instance_cmek_encryption_enabled.cloudsql_instance_cmek_encryption_enabled import (
|
||||
cloudsql_instance_cmek_encryption_enabled,
|
||||
)
|
||||
from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
|
||||
Instance,
|
||||
)
|
||||
|
||||
cloudsql_client.instances = [
|
||||
Instance(
|
||||
name="db-cmek",
|
||||
version="POSTGRES_15",
|
||||
ip_addresses=[],
|
||||
region=GCP_EU1_LOCATION,
|
||||
public_ip=False,
|
||||
require_ssl=False,
|
||||
ssl_mode="ENCRYPTED_ONLY",
|
||||
automated_backups=True,
|
||||
authorized_networks=[],
|
||||
flags=[],
|
||||
project_id=GCP_PROJECT_ID,
|
||||
cmek_key_name="projects/123456789012/locations/europe-west1/keyRings/my-ring/cryptoKeys/my-key",
|
||||
)
|
||||
]
|
||||
check = cloudsql_instance_cmek_encryption_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert result[0].resource_id == "db-cmek"
|
||||
assert result[0].location == GCP_EU1_LOCATION
|
||||
assert result[0].project_id == GCP_PROJECT_ID
|
||||
|
||||
def test_instance_cmek_not_configured(self):
|
||||
cloudsql_client = mock.MagicMock()
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_gcp_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.gcp.services.cloudsql.cloudsql_instance_cmek_encryption_enabled.cloudsql_instance_cmek_encryption_enabled.cloudsql_client",
|
||||
new=cloudsql_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.gcp.services.cloudsql.cloudsql_instance_cmek_encryption_enabled.cloudsql_instance_cmek_encryption_enabled import (
|
||||
cloudsql_instance_cmek_encryption_enabled,
|
||||
)
|
||||
from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
|
||||
Instance,
|
||||
)
|
||||
|
||||
cloudsql_client.instances = [
|
||||
Instance(
|
||||
name="db-google-managed",
|
||||
version="POSTGRES_15",
|
||||
ip_addresses=[],
|
||||
region=GCP_EU1_LOCATION,
|
||||
public_ip=False,
|
||||
require_ssl=False,
|
||||
ssl_mode="ENCRYPTED_ONLY",
|
||||
automated_backups=True,
|
||||
authorized_networks=[],
|
||||
flags=[],
|
||||
project_id=GCP_PROJECT_ID,
|
||||
cmek_key_name=None,
|
||||
)
|
||||
]
|
||||
check = cloudsql_instance_cmek_encryption_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].resource_id == "db-google-managed"
|
||||
assert result[0].location == GCP_EU1_LOCATION
|
||||
assert result[0].project_id == GCP_PROJECT_ID
|
||||
|
||||
def test_instance_cmek_empty_string(self):
|
||||
cloudsql_client = mock.MagicMock()
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_gcp_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.gcp.services.cloudsql.cloudsql_instance_cmek_encryption_enabled.cloudsql_instance_cmek_encryption_enabled.cloudsql_client",
|
||||
new=cloudsql_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.gcp.services.cloudsql.cloudsql_instance_cmek_encryption_enabled.cloudsql_instance_cmek_encryption_enabled import (
|
||||
cloudsql_instance_cmek_encryption_enabled,
|
||||
)
|
||||
from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
|
||||
Instance,
|
||||
)
|
||||
|
||||
cloudsql_client.instances = [
|
||||
Instance(
|
||||
name="db-empty-key",
|
||||
version="POSTGRES_15",
|
||||
ip_addresses=[],
|
||||
region=GCP_EU1_LOCATION,
|
||||
public_ip=False,
|
||||
require_ssl=False,
|
||||
ssl_mode="ENCRYPTED_ONLY",
|
||||
automated_backups=True,
|
||||
authorized_networks=[],
|
||||
flags=[],
|
||||
project_id=GCP_PROJECT_ID,
|
||||
cmek_key_name="",
|
||||
)
|
||||
]
|
||||
check = cloudsql_instance_cmek_encryption_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].resource_id == "db-empty-key"
|
||||
|
||||
def test_unsupported_instance_type_skipped(self):
|
||||
cloudsql_client = mock.MagicMock()
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_gcp_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.gcp.services.cloudsql.cloudsql_instance_cmek_encryption_enabled.cloudsql_instance_cmek_encryption_enabled.cloudsql_client",
|
||||
new=cloudsql_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.gcp.services.cloudsql.cloudsql_instance_cmek_encryption_enabled.cloudsql_instance_cmek_encryption_enabled import (
|
||||
cloudsql_instance_cmek_encryption_enabled,
|
||||
)
|
||||
from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
|
||||
Instance,
|
||||
)
|
||||
|
||||
cloudsql_client.instances = [
|
||||
Instance(
|
||||
name="external-primary",
|
||||
version="MYSQL_8_0",
|
||||
ip_addresses=[],
|
||||
region=GCP_EU1_LOCATION,
|
||||
public_ip=False,
|
||||
require_ssl=False,
|
||||
ssl_mode="ENCRYPTED_ONLY",
|
||||
automated_backups=False,
|
||||
authorized_networks=[],
|
||||
flags=[],
|
||||
project_id=GCP_PROJECT_ID,
|
||||
instance_type="ON_PREMISES_INSTANCE",
|
||||
cmek_key_name=None,
|
||||
),
|
||||
Instance(
|
||||
name="db-cmek",
|
||||
version="POSTGRES_15",
|
||||
ip_addresses=[],
|
||||
region=GCP_EU1_LOCATION,
|
||||
public_ip=False,
|
||||
require_ssl=False,
|
||||
ssl_mode="ENCRYPTED_ONLY",
|
||||
automated_backups=True,
|
||||
authorized_networks=[],
|
||||
flags=[],
|
||||
project_id=GCP_PROJECT_ID,
|
||||
instance_type="CLOUD_SQL_INSTANCE",
|
||||
cmek_key_name="projects/p/locations/europe-west1/keyRings/r/cryptoKeys/k",
|
||||
),
|
||||
]
|
||||
check = cloudsql_instance_cmek_encryption_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].resource_id == "db-cmek"
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
def test_service_parser_missing_disk_encryption(self):
|
||||
"""Exercise the real service parser path when diskEncryptionConfiguration is absent."""
|
||||
|
||||
def mock_api_client_without_disk_encryption(*_args, **_kwargs):
|
||||
client = MagicMock()
|
||||
client.instances().list().execute.return_value = {
|
||||
"items": [
|
||||
{
|
||||
"name": "db-no-encryption-config",
|
||||
"databaseVersion": "POSTGRES_14",
|
||||
"region": "us-central1",
|
||||
"ipAddresses": [],
|
||||
"settings": {
|
||||
"ipConfiguration": {"requireSsl": True},
|
||||
"backupConfiguration": {"enabled": True},
|
||||
"databaseFlags": [],
|
||||
},
|
||||
}
|
||||
]
|
||||
}
|
||||
client.instances().list_next.return_value = None
|
||||
return client
|
||||
|
||||
with (
|
||||
patch(
|
||||
"prowler.providers.gcp.lib.service.service.GCPService.__is_api_active__",
|
||||
new=mock_is_api_active,
|
||||
),
|
||||
patch(
|
||||
"prowler.providers.gcp.lib.service.service.GCPService.__generate_client__",
|
||||
new=mock_api_client_without_disk_encryption,
|
||||
),
|
||||
patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_gcp_provider(project_ids=[GCP_PROJECT_ID]),
|
||||
),
|
||||
):
|
||||
from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
|
||||
CloudSQL,
|
||||
)
|
||||
|
||||
cloudsql_client = CloudSQL(
|
||||
set_mocked_gcp_provider(project_ids=[GCP_PROJECT_ID])
|
||||
)
|
||||
assert len(cloudsql_client.instances) == 1
|
||||
assert cloudsql_client.instances[0].cmek_key_name is None
|
||||
|
||||
with patch(
|
||||
"prowler.providers.gcp.services.cloudsql.cloudsql_instance_cmek_encryption_enabled.cloudsql_instance_cmek_encryption_enabled.cloudsql_client",
|
||||
new=cloudsql_client,
|
||||
):
|
||||
from prowler.providers.gcp.services.cloudsql.cloudsql_instance_cmek_encryption_enabled.cloudsql_instance_cmek_encryption_enabled import (
|
||||
cloudsql_instance_cmek_encryption_enabled,
|
||||
)
|
||||
|
||||
check = cloudsql_instance_cmek_encryption_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].resource_id == "db-no-encryption-config"
|
||||
@@ -43,6 +43,11 @@ class TestCloudSQLService:
|
||||
{"value": "test"}
|
||||
]
|
||||
assert cloudsql_client.instances[0].flags == []
|
||||
assert cloudsql_client.instances[0].instance_type == "CLOUD_SQL_INSTANCE"
|
||||
assert (
|
||||
cloudsql_client.instances[0].cmek_key_name
|
||||
== "projects/123/locations/us-central1/keyRings/keyring1/cryptoKeys/key1"
|
||||
)
|
||||
assert cloudsql_client.instances[0].project_id == GCP_PROJECT_ID
|
||||
|
||||
assert cloudsql_client.instances[1].name == "instance2"
|
||||
@@ -62,12 +67,14 @@ class TestCloudSQLService:
|
||||
{"value": "test"}
|
||||
]
|
||||
assert cloudsql_client.instances[1].flags == []
|
||||
assert cloudsql_client.instances[1].instance_type == "CLOUD_SQL_INSTANCE"
|
||||
assert cloudsql_client.instances[1].cmek_key_name is None
|
||||
assert cloudsql_client.instances[1].project_id == GCP_PROJECT_ID
|
||||
|
||||
def test_instances_without_backup_configuration(self):
|
||||
"""Test that CloudSQL service handles instances without backupConfiguration field"""
|
||||
|
||||
def mock_api_client_without_backup_config(*args, **kwargs):
|
||||
def mock_api_client_without_backup_config(*_args, **_kwargs):
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
client = MagicMock()
|
||||
@@ -119,7 +126,7 @@ class TestCloudSQLService:
|
||||
def test_instances_with_empty_backup_configuration(self):
|
||||
"""Test that CloudSQL service handles instances with empty backupConfiguration"""
|
||||
|
||||
def mock_api_client_with_empty_backup_config(*args, **kwargs):
|
||||
def mock_api_client_with_empty_backup_config(*_args, **_kwargs):
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
client = MagicMock()
|
||||
@@ -170,7 +177,7 @@ class TestCloudSQLService:
|
||||
def test_instances_without_settings_fields(self):
|
||||
"""Test that CloudSQL service handles instances with minimal settings"""
|
||||
|
||||
def mock_api_client_with_minimal_settings(*args, **kwargs):
|
||||
def mock_api_client_with_minimal_settings(*_args, **_kwargs):
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
client = MagicMock()
|
||||
|
||||
+282
@@ -0,0 +1,282 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from unittest import mock
|
||||
from uuid import uuid4
|
||||
|
||||
from prowler.providers.m365.services.entra.entra_service import (
|
||||
AppRegistration,
|
||||
PasswordCredential,
|
||||
)
|
||||
from tests.providers.m365.m365_fixtures import DOMAIN, set_mocked_m365_provider
|
||||
|
||||
|
||||
class Test_entra_app_registration_client_secret_unused:
|
||||
def test_no_app_registrations(self):
|
||||
"""No app registrations in tenant: no findings."""
|
||||
entra_client = mock.MagicMock
|
||||
entra_client.audited_tenant = "audited_tenant"
|
||||
entra_client.audited_domain = DOMAIN
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.m365.services.entra.entra_app_registration_client_secret_unused.entra_app_registration_client_secret_unused.entra_client",
|
||||
new=entra_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_app_registration_client_secret_unused.entra_app_registration_client_secret_unused import (
|
||||
entra_app_registration_client_secret_unused,
|
||||
)
|
||||
|
||||
entra_client.app_registrations = {}
|
||||
|
||||
check = entra_app_registration_client_secret_unused()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 0
|
||||
|
||||
def test_app_no_password_credentials(self):
|
||||
"""App with no password credentials: expected PASS."""
|
||||
app_id = str(uuid4())
|
||||
app_name = "Test App Clean"
|
||||
entra_client = mock.MagicMock
|
||||
entra_client.audited_tenant = "audited_tenant"
|
||||
entra_client.audited_domain = DOMAIN
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.m365.services.entra.entra_app_registration_client_secret_unused.entra_app_registration_client_secret_unused.entra_client",
|
||||
new=entra_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_app_registration_client_secret_unused.entra_app_registration_client_secret_unused import (
|
||||
entra_app_registration_client_secret_unused,
|
||||
)
|
||||
|
||||
entra_client.app_registrations = {
|
||||
app_id: AppRegistration(
|
||||
id=app_id,
|
||||
app_id=str(uuid4()),
|
||||
name=app_name,
|
||||
password_credentials=[],
|
||||
)
|
||||
}
|
||||
|
||||
check = entra_app_registration_client_secret_unused()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"App registration {app_name} does not use password credentials."
|
||||
)
|
||||
assert result[0].resource_name == app_name
|
||||
assert result[0].resource_id == app_id
|
||||
|
||||
def test_app_with_one_password_credential(self):
|
||||
"""App with one password credential: expected FAIL."""
|
||||
app_id = str(uuid4())
|
||||
app_name = "Test App With Secret"
|
||||
entra_client = mock.MagicMock
|
||||
entra_client.audited_tenant = "audited_tenant"
|
||||
entra_client.audited_domain = DOMAIN
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.m365.services.entra.entra_app_registration_client_secret_unused.entra_app_registration_client_secret_unused.entra_client",
|
||||
new=entra_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_app_registration_client_secret_unused.entra_app_registration_client_secret_unused import (
|
||||
entra_app_registration_client_secret_unused,
|
||||
)
|
||||
|
||||
entra_client.app_registrations = {
|
||||
app_id: AppRegistration(
|
||||
id=app_id,
|
||||
app_id=str(uuid4()),
|
||||
name=app_name,
|
||||
password_credentials=[
|
||||
PasswordCredential(
|
||||
key_id=str(uuid4()),
|
||||
display_name="My Secret",
|
||||
start_date_time="2024-01-01T00:00:00Z",
|
||||
end_date_time="2025-01-01T00:00:00Z",
|
||||
),
|
||||
],
|
||||
)
|
||||
}
|
||||
|
||||
check = entra_app_registration_client_secret_unused()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "1 password credential(s)" in result[0].status_extended
|
||||
assert "My Secret" in result[0].status_extended
|
||||
assert result[0].resource_name == app_name
|
||||
assert result[0].resource_id == app_id
|
||||
|
||||
def test_app_with_expired_password_credential_still_fails(self):
|
||||
"""App with an expired password credential: still expected FAIL."""
|
||||
app_id = str(uuid4())
|
||||
app_name = "Legacy App"
|
||||
entra_client = mock.MagicMock
|
||||
entra_client.audited_tenant = "audited_tenant"
|
||||
entra_client.audited_domain = DOMAIN
|
||||
|
||||
expired = datetime.now(timezone.utc) - timedelta(days=30)
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.m365.services.entra.entra_app_registration_client_secret_unused.entra_app_registration_client_secret_unused.entra_client",
|
||||
new=entra_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_app_registration_client_secret_unused.entra_app_registration_client_secret_unused import (
|
||||
entra_app_registration_client_secret_unused,
|
||||
)
|
||||
|
||||
entra_client.app_registrations = {
|
||||
app_id: AppRegistration(
|
||||
id=app_id,
|
||||
app_id=str(uuid4()),
|
||||
name=app_name,
|
||||
password_credentials=[
|
||||
PasswordCredential(
|
||||
key_id=str(uuid4()),
|
||||
display_name="old-secret",
|
||||
end_date_time=expired,
|
||||
),
|
||||
],
|
||||
)
|
||||
}
|
||||
|
||||
check = entra_app_registration_client_secret_unused()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "1 password credential(s)" in result[0].status_extended
|
||||
assert result[0].resource_name == app_name
|
||||
|
||||
def test_app_with_multiple_password_credentials(self):
|
||||
"""App with multiple password credentials: expected FAIL."""
|
||||
app_id = str(uuid4())
|
||||
app_name = "Test App Multiple Secrets"
|
||||
entra_client = mock.MagicMock
|
||||
entra_client.audited_tenant = "audited_tenant"
|
||||
entra_client.audited_domain = DOMAIN
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.m365.services.entra.entra_app_registration_client_secret_unused.entra_app_registration_client_secret_unused.entra_client",
|
||||
new=entra_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_app_registration_client_secret_unused.entra_app_registration_client_secret_unused import (
|
||||
entra_app_registration_client_secret_unused,
|
||||
)
|
||||
|
||||
entra_client.app_registrations = {
|
||||
app_id: AppRegistration(
|
||||
id=app_id,
|
||||
app_id=str(uuid4()),
|
||||
name=app_name,
|
||||
password_credentials=[
|
||||
PasswordCredential(
|
||||
key_id=str(uuid4()),
|
||||
display_name="Secret 1",
|
||||
end_date_time="2025-06-01T00:00:00Z",
|
||||
),
|
||||
PasswordCredential(
|
||||
key_id=str(uuid4()),
|
||||
display_name="Secret 2",
|
||||
end_date_time="2025-12-01T00:00:00Z",
|
||||
),
|
||||
],
|
||||
)
|
||||
}
|
||||
|
||||
check = entra_app_registration_client_secret_unused()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "2 password credential(s)" in result[0].status_extended
|
||||
assert result[0].resource_name == app_name
|
||||
|
||||
def test_multiple_apps_mixed(self):
|
||||
"""Multiple apps: one clean, one with secrets."""
|
||||
app_id_pass = str(uuid4())
|
||||
app_name_pass = "Clean App"
|
||||
app_id_fail = str(uuid4())
|
||||
app_name_fail = "App With Secret"
|
||||
entra_client = mock.MagicMock
|
||||
entra_client.audited_tenant = "audited_tenant"
|
||||
entra_client.audited_domain = DOMAIN
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_m365_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.m365.services.entra.entra_app_registration_client_secret_unused.entra_app_registration_client_secret_unused.entra_client",
|
||||
new=entra_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.m365.services.entra.entra_app_registration_client_secret_unused.entra_app_registration_client_secret_unused import (
|
||||
entra_app_registration_client_secret_unused,
|
||||
)
|
||||
|
||||
entra_client.app_registrations = {
|
||||
app_id_pass: AppRegistration(
|
||||
id=app_id_pass,
|
||||
app_id=str(uuid4()),
|
||||
name=app_name_pass,
|
||||
password_credentials=[],
|
||||
),
|
||||
app_id_fail: AppRegistration(
|
||||
id=app_id_fail,
|
||||
app_id=str(uuid4()),
|
||||
name=app_name_fail,
|
||||
password_credentials=[
|
||||
PasswordCredential(
|
||||
key_id=str(uuid4()),
|
||||
display_name="Legacy Secret",
|
||||
),
|
||||
],
|
||||
),
|
||||
}
|
||||
|
||||
check = entra_app_registration_client_secret_unused()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 2
|
||||
|
||||
result_pass = next(r for r in result if r.resource_id == app_id_pass)
|
||||
result_fail = next(r for r in result if r.resource_id == app_id_fail)
|
||||
|
||||
assert result_pass.status == "PASS"
|
||||
assert result_fail.status == "FAIL"
|
||||
assert "Legacy Secret" in result_fail.status_extended
|
||||
Reference in new issue
Block a user