mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-11-09 04:38:49 +00:00
StylusFrost
204870f48a
_get_ep_compliance_dirs collapsed compliance directories into a dict keyed by provider, so when two or more dynamically-registered packages shipped compliance for the same provider, only one survived and the others were silently dropped from discovery, default output modes and the stdout table. Accumulate the directories per provider and iterate them all in get_available_compliance_frameworks, matching the behaviour Compliance.get_bulk already had. Built-in precedence and filename-stem dedup are preserved.
634 lines
22 KiB
Python
634 lines
22 KiB
Python
import logging
|
|
import os
|
|
import pathlib
|
|
from unittest import mock
|
|
|
|
from requests import Response
|
|
|
|
from prowler.config.config import (
|
|
Provider,
|
|
check_current_version,
|
|
get_available_compliance_frameworks,
|
|
load_and_validate_config_file,
|
|
load_and_validate_fixer_config_file,
|
|
)
|
|
|
|
MOCK_PROWLER_VERSION = "3.3.0"
|
|
MOCK_OLD_PROWLER_VERSION = "0.0.0"
|
|
MOCK_PROWLER_MASTER_VERSION = "3.4.0"
|
|
|
|
|
|
def test_provider_enum_includes_stackit_for_global_discovery():
|
|
providers = [provider.value for provider in Provider]
|
|
|
|
assert "stackit" in providers
|
|
|
|
|
|
def mock_prowler_get_latest_release(_, **_kwargs):
|
|
"""Mock requests.get() to get the Prowler latest release"""
|
|
response = Response()
|
|
response._content = b'[{"name":"3.3.0"}]'
|
|
return response
|
|
|
|
|
|
old_config_aws = {
|
|
"shodan_api_key": None,
|
|
"max_security_group_rules": 50,
|
|
"max_ec2_instance_age_in_days": 180,
|
|
"ec2_allowed_interface_types": ["api_gateway_managed", "vpc_endpoint"],
|
|
"ec2_allowed_instance_owners": ["amazon-elb"],
|
|
"trusted_account_ids": [],
|
|
"trusted_ips": [],
|
|
"log_group_retention_days": 365,
|
|
"max_idle_disconnect_timeout_in_seconds": 600,
|
|
"max_disconnect_timeout_in_seconds": 300,
|
|
"max_session_duration_seconds": 36000,
|
|
"obsolete_lambda_runtimes": [
|
|
"java8",
|
|
"go1.x",
|
|
"provided",
|
|
"python3.6",
|
|
"python2.7",
|
|
"python3.7",
|
|
"nodejs4.3",
|
|
"nodejs4.3-edge",
|
|
"nodejs6.10",
|
|
"nodejs",
|
|
"nodejs8.10",
|
|
"nodejs10.x",
|
|
"nodejs12.x",
|
|
"nodejs14.x",
|
|
"dotnet5.0",
|
|
"dotnetcore1.0",
|
|
"dotnetcore2.0",
|
|
"dotnetcore2.1",
|
|
"dotnetcore3.1",
|
|
"ruby2.5",
|
|
"ruby2.7",
|
|
],
|
|
"organizations_enabled_regions": [],
|
|
"organizations_trusted_delegated_administrators": [],
|
|
"check_rds_instance_replicas": False,
|
|
"days_to_expire_threshold": 7,
|
|
"eks_required_log_types": [
|
|
"api",
|
|
"audit",
|
|
"authenticator",
|
|
"controllerManager",
|
|
"scheduler",
|
|
],
|
|
}
|
|
config_aws = {
|
|
"mute_non_default_regions": False,
|
|
"max_unused_access_keys_days": 45,
|
|
"max_console_access_days": 45,
|
|
"max_unused_sagemaker_access_days": 90,
|
|
"shodan_api_key": None,
|
|
"max_security_group_rules": 50,
|
|
"max_ec2_instance_age_in_days": 180,
|
|
"ec2_allowed_interface_types": ["api_gateway_managed", "vpc_endpoint"],
|
|
"ec2_allowed_instance_owners": ["amazon-elb"],
|
|
"ec2_high_risk_ports": [
|
|
25,
|
|
110,
|
|
135,
|
|
143,
|
|
445,
|
|
3000,
|
|
4333,
|
|
5000,
|
|
5500,
|
|
8080,
|
|
8088,
|
|
],
|
|
"fargate_linux_latest_version": "1.4.0",
|
|
"fargate_windows_latest_version": "1.0.0",
|
|
"trusted_account_ids": [],
|
|
"trusted_ips": [],
|
|
"log_group_retention_days": 365,
|
|
"max_idle_disconnect_timeout_in_seconds": 600,
|
|
"max_disconnect_timeout_in_seconds": 300,
|
|
"max_session_duration_seconds": 36000,
|
|
"obsolete_lambda_runtimes": [
|
|
"java8",
|
|
"go1.x",
|
|
"provided",
|
|
"python3.6",
|
|
"python2.7",
|
|
"python3.7",
|
|
"nodejs4.3",
|
|
"nodejs4.3-edge",
|
|
"nodejs6.10",
|
|
"nodejs",
|
|
"nodejs8.10",
|
|
"nodejs10.x",
|
|
"nodejs12.x",
|
|
"nodejs14.x",
|
|
"dotnet5.0",
|
|
"dotnetcore1.0",
|
|
"dotnetcore2.0",
|
|
"dotnetcore2.1",
|
|
"dotnetcore3.1",
|
|
"ruby2.5",
|
|
"ruby2.7",
|
|
],
|
|
"lambda_min_azs": 2,
|
|
"organizations_enabled_regions": [],
|
|
"organizations_trusted_delegated_administrators": [],
|
|
"ecr_repository_vulnerability_minimum_severity": "MEDIUM",
|
|
"verify_premium_support_plans": True,
|
|
"threat_detection_privilege_escalation_threshold": 0.2,
|
|
"threat_detection_privilege_escalation_minutes": 1440,
|
|
"threat_detection_privilege_escalation_actions": [
|
|
"AddPermission",
|
|
"AddRoleToInstanceProfile",
|
|
"AddUserToGroup",
|
|
"AssociateAccessPolicy",
|
|
"AssumeRole",
|
|
"AttachGroupPolicy",
|
|
"AttachRolePolicy",
|
|
"AttachUserPolicy",
|
|
"ChangePassword",
|
|
"CreateAccessEntry",
|
|
"CreateAccessKey",
|
|
"CreateDevEndpoint",
|
|
"CreateEventSourceMapping",
|
|
"CreateFunction",
|
|
"CreateGroup",
|
|
"CreateJob",
|
|
"CreateKeyPair",
|
|
"CreateLoginProfile",
|
|
"CreatePipeline",
|
|
"CreatePolicyVersion",
|
|
"CreateRole",
|
|
"CreateStack",
|
|
"DeleteRolePermissionsBoundary",
|
|
"DeleteRolePolicy",
|
|
"DeleteUserPermissionsBoundary",
|
|
"DeleteUserPolicy",
|
|
"DetachRolePolicy",
|
|
"DetachUserPolicy",
|
|
"GetCredentialsForIdentity",
|
|
"GetId",
|
|
"GetPolicyVersion",
|
|
"GetUserPolicy",
|
|
"Invoke",
|
|
"ModifyInstanceAttribute",
|
|
"PassRole",
|
|
"PutGroupPolicy",
|
|
"PutPipelineDefinition",
|
|
"PutRolePermissionsBoundary",
|
|
"PutRolePolicy",
|
|
"PutUserPermissionsBoundary",
|
|
"PutUserPolicy",
|
|
"ReplaceIamInstanceProfileAssociation",
|
|
"RunInstances",
|
|
"SetDefaultPolicyVersion",
|
|
"UpdateAccessKey",
|
|
"UpdateAssumeRolePolicy",
|
|
"UpdateDevEndpoint",
|
|
"UpdateEventSourceMapping",
|
|
"UpdateFunctionCode",
|
|
"UpdateJob",
|
|
"UpdateLoginProfile",
|
|
],
|
|
"threat_detection_enumeration_threshold": 0.3,
|
|
"threat_detection_enumeration_minutes": 1440,
|
|
"threat_detection_enumeration_actions": [
|
|
"DescribeAccessEntry",
|
|
"DescribeAccountAttributes",
|
|
"DescribeAvailabilityZones",
|
|
"DescribeBundleTasks",
|
|
"DescribeCarrierGateways",
|
|
"DescribeClientVpnRoutes",
|
|
"DescribeCluster",
|
|
"DescribeDhcpOptions",
|
|
"DescribeFlowLogs",
|
|
"DescribeImages",
|
|
"DescribeInstanceAttribute",
|
|
"DescribeInstanceInformation",
|
|
"DescribeInstanceTypes",
|
|
"DescribeInstances",
|
|
"DescribeInstances",
|
|
"DescribeKeyPairs",
|
|
"DescribeLogGroups",
|
|
"DescribeLogStreams",
|
|
"DescribeOrganization",
|
|
"DescribeRegions",
|
|
"DescribeSecurityGroups",
|
|
"DescribeSnapshotAttribute",
|
|
"DescribeSnapshotTierStatus",
|
|
"DescribeSubscriptionFilters",
|
|
"DescribeTransitGatewayMulticastDomains",
|
|
"DescribeVolumes",
|
|
"DescribeVolumesModifications",
|
|
"DescribeVpcEndpointConnectionNotifications",
|
|
"DescribeVpcs",
|
|
"GetAccount",
|
|
"GetAccountAuthorizationDetails",
|
|
"GetAccountSendingEnabled",
|
|
"GetBucketAcl",
|
|
"GetBucketLogging",
|
|
"GetBucketPolicy",
|
|
"GetBucketReplication",
|
|
"GetBucketVersioning",
|
|
"GetCallerIdentity",
|
|
"GetCertificate",
|
|
"GetConsoleScreenshot",
|
|
"GetCostAndUsage",
|
|
"GetDetector",
|
|
"GetEbsDefaultKmsKeyId",
|
|
"GetEbsEncryptionByDefault",
|
|
"GetFindings",
|
|
"GetFlowLogsIntegrationTemplate",
|
|
"GetIdentityVerificationAttributes",
|
|
"GetInstances",
|
|
"GetIntrospectionSchema",
|
|
"GetLaunchTemplateData",
|
|
"GetLaunchTemplateData",
|
|
"GetLogRecord",
|
|
"GetParameters",
|
|
"GetPolicyVersion",
|
|
"GetPublicAccessBlock",
|
|
"GetQueryResults",
|
|
"GetRegions",
|
|
"GetSMSAttributes",
|
|
"GetSMSSandboxAccountStatus",
|
|
"GetSendQuota",
|
|
"GetTransitGatewayRouteTableAssociations",
|
|
"GetUserPolicy",
|
|
"HeadObject",
|
|
"ListAccessKeys",
|
|
"ListAccounts",
|
|
"ListAllMyBuckets",
|
|
"ListAssociatedAccessPolicies",
|
|
"ListAttachedUserPolicies",
|
|
"ListClusters",
|
|
"ListDetectors",
|
|
"ListDomains",
|
|
"ListFindings",
|
|
"ListHostedZones",
|
|
"ListIPSets",
|
|
"ListIdentities",
|
|
"ListInstanceProfiles",
|
|
"ListObjects",
|
|
"ListOrganizationalUnitsForParent",
|
|
"ListOriginationNumbers",
|
|
"ListPolicyVersions",
|
|
"ListRoles",
|
|
"ListRoles",
|
|
"ListRules",
|
|
"ListServiceQuotas",
|
|
"ListSubscriptions",
|
|
"ListTargetsByRule",
|
|
"ListTopics",
|
|
"ListUsers",
|
|
"LookupEvents",
|
|
"Search",
|
|
],
|
|
"threat_detection_llm_jacking_threshold": 0.4,
|
|
"threat_detection_llm_jacking_minutes": 1440,
|
|
"threat_detection_llm_jacking_actions": [
|
|
"PutUseCaseForModelAccess",
|
|
"PutFoundationModelEntitlement",
|
|
"PutModelInvocationLoggingConfiguration",
|
|
"CreateFoundationModelAgreement",
|
|
"InvokeModel",
|
|
"InvokeModelWithResponseStream",
|
|
"GetUseCaseForModelAccess",
|
|
"GetModelInvocationLoggingConfiguration",
|
|
"GetFoundationModelAvailability",
|
|
"ListFoundationModelAgreementOffers",
|
|
"ListFoundationModels",
|
|
"ListProvisionedModelThroughputs",
|
|
],
|
|
"check_rds_instance_replicas": False,
|
|
"days_to_expire_threshold": 7,
|
|
"insecure_key_algorithms": [
|
|
"RSA-1024",
|
|
"P-192",
|
|
],
|
|
"eks_required_log_types": [
|
|
"api",
|
|
"audit",
|
|
"authenticator",
|
|
"controllerManager",
|
|
"scheduler",
|
|
],
|
|
"eks_cluster_oldest_version_supported": "1.28",
|
|
"excluded_sensitive_environment_variables": [],
|
|
"minimum_snapshot_retention_period": 7,
|
|
"elb_min_azs": 2,
|
|
"elbv2_min_azs": 2,
|
|
"secrets_ignore_patterns": [],
|
|
"max_days_secret_unused": 90,
|
|
"max_days_secret_unrotated": 90,
|
|
}
|
|
|
|
config_azure = {
|
|
"shodan_api_key": None,
|
|
"php_latest_version": "8.2",
|
|
"python_latest_version": "3.12",
|
|
"java_latest_version": "17",
|
|
"recommended_minimal_tls_versions": ["1.2", "1.3"],
|
|
"desired_vm_sku_sizes": [
|
|
"Standard_A8_v2",
|
|
"Standard_DS3_v2",
|
|
"Standard_D4s_v3",
|
|
],
|
|
"defender_attack_path_minimal_risk_level": "High",
|
|
}
|
|
|
|
config_gcp = {
|
|
"shodan_api_key": None,
|
|
"mig_min_zones": 2,
|
|
"max_unused_account_days": 30,
|
|
}
|
|
|
|
config_kubernetes = {
|
|
"audit_log_maxbackup": 10,
|
|
"audit_log_maxsize": 100,
|
|
"audit_log_maxage": 30,
|
|
"apiserver_strong_ciphers": [
|
|
"TLS_AES_128_GCM_SHA256",
|
|
"TLS_AES_256_GCM_SHA384",
|
|
"TLS_CHACHA20_POLY1305_SHA256",
|
|
],
|
|
"kubelet_strong_ciphers": [
|
|
"TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256",
|
|
"TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256",
|
|
"TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305",
|
|
"TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384",
|
|
"TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305",
|
|
"TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384",
|
|
"TLS_RSA_WITH_AES_256_GCM_SHA384",
|
|
"TLS_RSA_WITH_AES_128_GCM_SHA256",
|
|
],
|
|
}
|
|
|
|
|
|
class Test_Config:
|
|
@mock.patch(
|
|
"prowler.config.config.requests.get", new=mock_prowler_get_latest_release
|
|
)
|
|
@mock.patch("prowler.config.config.prowler_version", new=MOCK_PROWLER_VERSION)
|
|
def test_check_current_version_with_latest(self):
|
|
assert (
|
|
check_current_version()
|
|
== f"Prowler {MOCK_PROWLER_VERSION} (You are running the latest version, yay!)"
|
|
)
|
|
|
|
@mock.patch(
|
|
"prowler.config.config.requests.get", new=mock_prowler_get_latest_release
|
|
)
|
|
@mock.patch("prowler.config.config.prowler_version", new=MOCK_OLD_PROWLER_VERSION)
|
|
def test_check_current_version_with_old(self):
|
|
assert (
|
|
check_current_version()
|
|
== f"Prowler {MOCK_OLD_PROWLER_VERSION} (latest is {MOCK_PROWLER_VERSION}, upgrade for the latest features)"
|
|
)
|
|
|
|
@mock.patch(
|
|
"prowler.config.config.requests.get", new=mock_prowler_get_latest_release
|
|
)
|
|
@mock.patch(
|
|
"prowler.config.config.prowler_version", new=MOCK_PROWLER_MASTER_VERSION
|
|
)
|
|
def test_check_current_version_with_master_version(self):
|
|
assert (
|
|
check_current_version()
|
|
== f"Prowler {MOCK_PROWLER_MASTER_VERSION} (You are running the latest version, yay!)"
|
|
)
|
|
|
|
def test_get_available_compliance_frameworks(self):
|
|
compliance_frameworks = [
|
|
"csa_ccm_4.0",
|
|
"cisa_aws",
|
|
"soc2_aws",
|
|
"cis_1.4_aws",
|
|
"cis_1.5_aws",
|
|
"mitre_attack_aws",
|
|
"gdpr_aws",
|
|
"aws_foundational_security_best_practices_aws",
|
|
"iso27001_2013_aws",
|
|
"hipaa_aws",
|
|
"cis_2.0_aws",
|
|
"gxp_21_cfr_part_11_aws",
|
|
"aws_well_architected_framework_security_pillar_aws",
|
|
"gxp_eu_annex_11_aws",
|
|
"nist_800_171_revision_2_aws",
|
|
"nist_800_53_revision_4_aws",
|
|
"nist_800_53_revision_5_aws",
|
|
"ens_rd2022_aws",
|
|
"nist_csf_1.1_aws",
|
|
"aws_well_architected_framework_reliability_pillar_aws",
|
|
"aws_audit_manager_control_tower_guardrails_aws",
|
|
"rbi_cyber_security_framework_aws",
|
|
"ffiec_aws",
|
|
"pci_3.2.1_aws",
|
|
"fedramp_moderate_revision_4_aws",
|
|
"fedramp_low_revision_4_aws",
|
|
"cis_2.0_gcp",
|
|
"cis_1.8_kubernetes",
|
|
"kisa_isms-p_2023_aws",
|
|
"kisa_isms-p_2023-korean_aws",
|
|
]
|
|
assert (
|
|
get_available_compliance_frameworks().sort() == compliance_frameworks.sort()
|
|
)
|
|
|
|
def test_get_available_compliance_frameworks_filters_universal_by_provider(self):
|
|
aws_frameworks = get_available_compliance_frameworks("aws")
|
|
kubernetes_frameworks = get_available_compliance_frameworks("kubernetes")
|
|
|
|
assert "csa_ccm_4.0" in aws_frameworks
|
|
assert "csa_ccm_4.0" not in kubernetes_frameworks
|
|
|
|
def test_get_available_compliance_frameworks_no_provider_includes_universals(self):
|
|
"""Regression test for the variable shadowing bug.
|
|
|
|
Previously, the inner ``for provider in providers`` loop shadowed
|
|
the outer ``provider`` parameter. When called without a provider,
|
|
the post-loop ``if provider:`` branch wrongly applied
|
|
``framework.supports_provider(<last provider iterated>)`` and
|
|
excluded universal frameworks from the result.
|
|
|
|
Result: the parser-level ``available_compliance_frameworks``
|
|
constant was missing universal frameworks like ``csa_ccm_4.0``,
|
|
which made ``--compliance csa_ccm_4.0`` reject the choice.
|
|
"""
|
|
all_frameworks = get_available_compliance_frameworks()
|
|
assert "csa_ccm_4.0" in all_frameworks
|
|
|
|
def test_get_available_compliance_frameworks_does_not_mutate_provider_param(self):
|
|
"""Calling with a specific provider must not affect a subsequent
|
|
call without provider. Validates that the loop variable rename
|
|
prevents leaking state between calls."""
|
|
# Force an iteration over multiple providers first
|
|
get_available_compliance_frameworks("kubernetes")
|
|
# Then a no-provider call must still include universals supported
|
|
# by ANY provider (not filtered by some leaked value)
|
|
all_frameworks = get_available_compliance_frameworks()
|
|
assert "csa_ccm_4.0" in all_frameworks
|
|
|
|
@mock.patch("prowler.config.config._get_ep_compliance_dirs")
|
|
def test_get_available_compliance_frameworks_dedupes_ep_collisions_with_builtins(
|
|
self, mock_dirs
|
|
):
|
|
"""Entry-point compliance frameworks that collide with a built-in
|
|
name must appear only once in the available frameworks list.
|
|
Built-in wins silently — same policy as the universal frameworks
|
|
loop and as Compliance.get_bulk."""
|
|
import json
|
|
import tempfile
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# cis_2.0_aws ships as a built-in under prowler/compliance/aws/
|
|
json_path = os.path.join(tmpdir, "cis_2.0_aws.json")
|
|
with open(json_path, "w") as f:
|
|
json.dump({"Framework": "CIS", "Provider": "aws"}, f)
|
|
|
|
mock_dirs.return_value = {"aws": [tmpdir]}
|
|
|
|
frameworks = get_available_compliance_frameworks("aws")
|
|
|
|
assert frameworks.count("cis_2.0_aws") == 1, (
|
|
f"Expected cis_2.0_aws to appear exactly once, got "
|
|
f"{frameworks.count('cis_2.0_aws')} occurrences in: {frameworks}"
|
|
)
|
|
|
|
@mock.patch("prowler.config.config._get_ep_compliance_dirs")
|
|
def test_get_available_compliance_frameworks_merges_multiple_ep_dirs_same_provider(
|
|
self, mock_dirs
|
|
):
|
|
"""Frameworks from every package contributing the same provider must
|
|
surface, not just the last directory discovered."""
|
|
import json
|
|
import tempfile
|
|
|
|
with (
|
|
tempfile.TemporaryDirectory() as pkg_a,
|
|
tempfile.TemporaryDirectory() as pkg_b,
|
|
):
|
|
with open(os.path.join(pkg_a, "cis_1.0_template.json"), "w") as f:
|
|
json.dump({"Framework": "CIS", "Provider": "template"}, f)
|
|
with open(os.path.join(pkg_b, "nis2_1.0_template.json"), "w") as f:
|
|
json.dump({"Framework": "NIS2", "Provider": "template"}, f)
|
|
|
|
# Two packages register `prowler.compliance` with the same name.
|
|
mock_dirs.return_value = {"template": [pkg_a, pkg_b]}
|
|
|
|
frameworks = get_available_compliance_frameworks("template")
|
|
|
|
assert "cis_1.0_template" in frameworks
|
|
assert "nis2_1.0_template" in frameworks
|
|
|
|
def test_load_and_validate_config_file_aws(self):
|
|
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
|
|
config_test_file = f"{path}/fixtures/config.yaml"
|
|
provider = "aws"
|
|
assert load_and_validate_config_file(provider, config_test_file) == config_aws
|
|
|
|
def test_load_and_validate_config_file_gcp(self):
|
|
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
|
|
config_test_file = f"{path}/fixtures/config.yaml"
|
|
provider = "gcp"
|
|
|
|
assert load_and_validate_config_file(provider, config_test_file) == config_gcp
|
|
|
|
def test_load_and_validate_config_file_kubernetes(self):
|
|
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
|
|
config_test_file = f"{path}/fixtures/config.yaml"
|
|
provider = "kubernetes"
|
|
assert (
|
|
load_and_validate_config_file(provider, config_test_file)
|
|
== config_kubernetes
|
|
)
|
|
|
|
def test_load_and_validate_config_file_azure(self):
|
|
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
|
|
config_test_file = f"{path}/fixtures/config.yaml"
|
|
provider = "azure"
|
|
|
|
assert load_and_validate_config_file(provider, config_test_file) == config_azure
|
|
|
|
def test_load_and_validate_config_file_old_format(self):
|
|
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
|
|
config_test_file = f"{path}/fixtures/config_old.yaml"
|
|
assert load_and_validate_config_file("aws", config_test_file) == old_config_aws
|
|
assert load_and_validate_config_file("gcp", config_test_file) == {}
|
|
assert load_and_validate_config_file("azure", config_test_file) == {}
|
|
assert load_and_validate_config_file("kubernetes", config_test_file) == {}
|
|
|
|
def test_load_and_validate_config_file_namespaced_non_listed_provider(self):
|
|
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
|
|
config_test_file = f"{path}/fixtures/config_namespaced_external.yaml"
|
|
# github is a built-in not in the legacy hardcoded list; namespaced format must unwrap it.
|
|
assert load_and_validate_config_file("github", config_test_file) == {
|
|
"token": "abc",
|
|
"org": "prowler-cloud",
|
|
}
|
|
|
|
def test_load_and_validate_config_file_namespaced_external_provider(self):
|
|
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
|
|
config_test_file = f"{path}/fixtures/config_namespaced_external.yaml"
|
|
# External plug-in provider: namespaced format must unwrap its block.
|
|
assert load_and_validate_config_file("custom_plugin", config_test_file) == {
|
|
"setting": "value",
|
|
"nested": {"key": 42},
|
|
}
|
|
|
|
def test_load_and_validate_config_file_namespaced_missing_provider(self):
|
|
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
|
|
config_test_file = f"{path}/fixtures/config_namespaced_external.yaml"
|
|
# Provider with no section in a namespaced file must return empty config,
|
|
# not the full file (prevents cross-provider config leakage).
|
|
assert load_and_validate_config_file("aws", config_test_file) == {}
|
|
assert load_and_validate_config_file("gcp", config_test_file) == {}
|
|
|
|
def test_load_and_validate_config_file_invalid_config_file_path(self, caplog):
|
|
provider = "aws"
|
|
config_file_path = "invalid/path/to/fixer_config.yaml"
|
|
|
|
with caplog.at_level(logging.ERROR):
|
|
result = load_and_validate_config_file(provider, config_file_path)
|
|
assert "FileNotFoundError" in caplog.text
|
|
assert result == {}
|
|
|
|
def test_load_and_validate_fixer_config_aws(self):
|
|
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
|
|
config_test_file = f"{path}/fixtures/fixer_config.yaml"
|
|
provider = "aws"
|
|
|
|
assert load_and_validate_fixer_config_file(provider, config_test_file)
|
|
|
|
def test_load_and_validate_fixer_config_gcp(self):
|
|
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
|
|
config_test_file = f"{path}/fixtures/fixer_config.yaml"
|
|
provider = "gcp"
|
|
|
|
assert load_and_validate_fixer_config_file(provider, config_test_file) == {}
|
|
|
|
def test_load_and_validate_fixer_config_kubernetes(self):
|
|
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
|
|
config_test_file = f"{path}/fixtures/fixer_config.yaml"
|
|
provider = "kubernetes"
|
|
|
|
assert load_and_validate_fixer_config_file(provider, config_test_file) == {}
|
|
|
|
def test_load_and_validate_fixer_config_azure(self):
|
|
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
|
|
config_test_file = f"{path}/fixtures/fixer_config.yaml"
|
|
provider = "azure"
|
|
|
|
assert load_and_validate_fixer_config_file(provider, config_test_file) == {}
|
|
|
|
def test_load_and_validate_fixer_config_invalid_fixer_config_path(self, caplog):
|
|
provider = "aws"
|
|
fixer_config_path = "invalid/path/to/fixer_config.yaml"
|
|
|
|
with caplog.at_level(logging.ERROR):
|
|
result = load_and_validate_fixer_config_file(provider, fixer_config_path)
|
|
assert "FileNotFoundError" in caplog.text
|
|
assert result == {}
|