feat(compliance): FedRAMP 20x Class C FRR + AWS checks (#12808)

This commit is contained in:
Pedro Martín
2026-09-14 16:35:05 +02:00
committed by GitHub
parent 1b228d590b
commit 3860cd3dce
45 changed files with 5100 additions and 8 deletions
+2
View File
@@ -138,6 +138,8 @@ config_aws = {
"organizations_enabled_regions": [],
"organizations_trusted_delegated_administrators": [],
"ecr_repository_vulnerability_minimum_severity": "MEDIUM",
"inspector2_max_days_since_last_scan": 3,
"inspector2_active_finding_max_age_days": 192,
"verify_premium_support_plans": True,
"threat_detection_privilege_escalation_threshold": 0.2,
"threat_detection_privilege_escalation_minutes": 1440,
+8
View File
@@ -139,6 +139,14 @@ aws:
# MEDIUM
ecr_repository_vulnerability_minimum_severity: "MEDIUM"
# AWS Inspector2
# aws.inspector2_coverage_recently_scanned
# Maximum days since Inspector2 last scanned an actively covered resource
inspector2_max_days_since_last_scan: 3
# aws.inspector2_active_findings_within_max_age
# Maximum days an Inspector2 finding can stay active since it was first observed
inspector2_active_finding_max_age_days: 192
# AWS Trusted Advisor
# aws.trustedadvisor_premium_support_plan_subscribed
verify_premium_support_plans: True
@@ -0,0 +1,156 @@
from unittest import mock
from boto3 import client, resource
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_REGION_EU_WEST_1,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
CHECK_MODULE = "prowler.providers.aws.services.elbv2.elbv2_listener_fips_tls_enabled.elbv2_listener_fips_tls_enabled"
FIPS_POLICY = "ELBSecurityPolicy-TLS13-1-2-FIPS-2023-04"
NON_FIPS_POLICY = "ELBSecurityPolicy-TLS13-1-2-2021-06"
def create_application_load_balancer():
conn = client("elbv2", region_name=AWS_REGION_EU_WEST_1)
ec2 = resource("ec2", region_name=AWS_REGION_EU_WEST_1)
security_group = ec2.create_security_group(
GroupName="a-security-group", Description="First One"
)
vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
subnet1 = ec2.create_subnet(
VpcId=vpc.id,
CidrBlock="172.28.7.192/26",
AvailabilityZone=f"{AWS_REGION_EU_WEST_1}a",
)
subnet2 = ec2.create_subnet(
VpcId=vpc.id,
CidrBlock="172.28.7.0/26",
AvailabilityZone=f"{AWS_REGION_EU_WEST_1}b",
)
lb = conn.create_load_balancer(
Name="my-lb",
Subnets=[subnet1.id, subnet2.id],
SecurityGroups=[security_group.id],
Scheme="internal",
Type="application",
)["LoadBalancers"][0]
target_group_arn = conn.create_target_group(
Name="a-target", Protocol="HTTP", Port=8080, VpcId=vpc.id
)["TargetGroups"][0]["TargetGroupArn"]
return conn, lb, target_group_arn
def create_listener(conn, lb, target_group_arn, protocol, port, ssl_policy=None):
listener_args = {
"LoadBalancerArn": lb["LoadBalancerArn"],
"Protocol": protocol,
"Port": port,
"DefaultActions": [{"Type": "forward", "TargetGroupArn": target_group_arn}],
}
if ssl_policy:
listener_args["SslPolicy"] = ssl_policy
return conn.create_listener(**listener_args)["Listeners"][0]
def execute_check(service=None):
from prowler.providers.aws.services.elbv2.elbv2_service import ELBv2
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1],
create_default_organization=False,
)
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
),
mock.patch(f"{CHECK_MODULE}.elbv2_client", new=service or ELBv2(aws_provider)),
):
from prowler.providers.aws.services.elbv2.elbv2_listener_fips_tls_enabled.elbv2_listener_fips_tls_enabled import (
elbv2_listener_fips_tls_enabled,
)
return elbv2_listener_fips_tls_enabled().execute()
class Test_elbv2_listener_fips_tls_enabled:
@mock_aws
def test_no_load_balancers(self):
assert execute_check() == []
@mock_aws
def test_http_listener_only(self):
conn, lb, target_group_arn = create_application_load_balancer()
create_listener(conn, lb, target_group_arn, "HTTP", 80)
result = execute_check()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].status_extended == "ELBv2 my-lb has no HTTPS/TLS listeners."
@mock_aws
def test_fips_policy(self):
conn, lb, target_group_arn = create_application_load_balancer()
create_listener(conn, lb, target_group_arn, "HTTPS", 443, FIPS_POLICY)
result = execute_check()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "ELBv2 my-lb has all HTTPS/TLS listeners using a FIPS TLS security policy."
)
assert result[0].resource_id == "my-lb"
assert result[0].resource_arn == lb["LoadBalancerArn"]
assert result[0].region == AWS_REGION_EU_WEST_1
@mock_aws
def test_non_fips_policy(self):
conn, lb, target_group_arn = create_application_load_balancer()
listener = create_listener(
conn, lb, target_group_arn, "HTTPS", 443, NON_FIPS_POLICY
)
result = execute_check()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"ELBv2 my-lb has HTTPS/TLS listeners without a FIPS TLS security policy: HTTPS:443 ({listener['ListenerArn']}) uses {NON_FIPS_POLICY}."
)
@mock_aws
def test_mixed_listeners(self):
conn, lb, target_group_arn = create_application_load_balancer()
create_listener(conn, lb, target_group_arn, "HTTPS", 443, FIPS_POLICY)
create_listener(conn, lb, target_group_arn, "HTTPS", 8443, NON_FIPS_POLICY)
result = execute_check()
assert len(result) == 1
assert result[0].status == "FAIL"
assert NON_FIPS_POLICY in result[0].status_extended
assert FIPS_POLICY not in result[0].status_extended
@mock_aws
def test_listener_discovery_failed(self):
from prowler.providers.aws.services.elbv2.elbv2_service import ELBv2
conn, lb, target_group_arn = create_application_load_balancer()
create_listener(conn, lb, target_group_arn, "HTTPS", 443, NON_FIPS_POLICY)
service = ELBv2(
set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1],
create_default_organization=False,
)
)
service.loadbalancersv2[lb["LoadBalancerArn"]].listener_discovery_failed = True
assert execute_check(service) == []
@@ -0,0 +1,136 @@
from datetime import datetime, timedelta, timezone
from unittest import mock
from prowler.providers.aws.services.inspector2.inspector2_service import (
Finding,
Inspector,
KnownExploitedVulnerability,
)
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
set_mocked_aws_provider,
)
INSPECTOR_ARN = (
f"arn:aws:inspector2:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:inspector2"
)
FINDING_ARN = f"arn:aws:inspector2:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:finding/0e436649379db5f327e3cf5bb4421d76"
KEV_ID = "CVE-2024-3400"
CHECK_MODULE = "prowler.providers.aws.services.inspector2.inspector2_active_findings_kev_within_due_date.inspector2_active_findings_kev_within_due_date"
def build_inspector(findings=None, status="ENABLED"):
return Inspector(
id="Inspector2",
arn=INSPECTOR_ARN,
region=AWS_REGION_EU_WEST_1,
status=status,
ec2_status="ENABLED",
ecr_status="ENABLED",
lambda_status="ENABLED",
lambda_code_status="ENABLED",
findings=findings,
)
def build_finding(vulnerability_id=KEV_ID):
return Finding(
arn=FINDING_ARN,
type="PACKAGE_VULNERABILITY",
severity="CRITICAL",
first_observed_at=datetime.now(timezone.utc),
vulnerability_id=vulnerability_id,
resource_ids=["i-0123456789abcdef0"],
)
def build_kev(date_due):
return KnownExploitedVulnerability(
id=KEV_ID,
date_added=datetime(2024, 4, 12, tzinfo=timezone.utc),
date_due=date_due,
)
def execute_check(inspectors, known_exploited=None, lookup_failed=None):
inspector2_client = mock.MagicMock()
inspector2_client.inspectors = inspectors
inspector2_client.known_exploited_vulnerabilities = known_exploited or {}
inspector2_client.vulnerability_lookup_failed = lookup_failed or set()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider([AWS_REGION_EU_WEST_1]),
),
mock.patch(f"{CHECK_MODULE}.inspector2_client", new=inspector2_client),
):
from prowler.providers.aws.services.inspector2.inspector2_active_findings_kev_within_due_date.inspector2_active_findings_kev_within_due_date import (
inspector2_active_findings_kev_within_due_date,
)
return inspector2_active_findings_kev_within_due_date().execute()
class Test_inspector2_active_findings_kev_within_due_date:
def test_no_resources(self):
assert execute_check([]) == []
def test_inspector_disabled(self):
assert execute_check([build_inspector(findings=[], status="DISABLED")]) == []
def test_kev_past_due_date(self):
result = execute_check(
[build_inspector(findings=[build_finding()])],
known_exploited={
KEV_ID: build_kev(datetime(2024, 4, 19, tzinfo=timezone.utc))
},
)
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Inspector2 has active findings in region {AWS_REGION_EU_WEST_1} for CISA Known Exploited Vulnerabilities past their remediation due date: {KEV_ID} (due 2024-04-19)."
)
assert result[0].resource_id == "Inspector2"
assert result[0].resource_arn == INSPECTOR_ARN
assert result[0].region == AWS_REGION_EU_WEST_1
def test_kev_within_due_date(self):
result = execute_check(
[build_inspector(findings=[build_finding()])],
known_exploited={
KEV_ID: build_kev(datetime.now(timezone.utc) + timedelta(days=7))
},
)
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Inspector2 has no active findings in region {AWS_REGION_EU_WEST_1} for CISA Known Exploited Vulnerabilities past their remediation due date."
)
def test_no_kev_findings(self):
result = execute_check(
[build_inspector(findings=[build_finding("CVE-2022-40897")])]
)
assert len(result) == 1
assert result[0].status == "PASS"
def test_kev_status_not_verified(self):
result = execute_check(
[build_inspector(findings=[build_finding()])],
lookup_failed={KEV_ID},
)
assert len(result) == 1
assert result[0].status == "MANUAL"
def test_findings_not_retrieved(self):
result = execute_check([build_inspector(findings=None)])
assert len(result) == 1
assert result[0].status == "MANUAL"
@@ -0,0 +1,149 @@
from datetime import datetime, timezone
from unittest import mock
from prowler.providers.aws.services.inspector2.inspector2_service import (
Finding,
Inspector,
KnownExploitedVulnerability,
)
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
set_mocked_aws_provider,
)
INSPECTOR_ARN = (
f"arn:aws:inspector2:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:inspector2"
)
FINDING_ARN = f"arn:aws:inspector2:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:finding/0e436649379db5f327e3cf5bb4421d76"
KEV_ID = "CVE-2024-3400"
CHECK_MODULE = "prowler.providers.aws.services.inspector2.inspector2_active_findings_no_known_exploited_vulnerabilities.inspector2_active_findings_no_known_exploited_vulnerabilities"
def build_inspector(findings=None, status="ENABLED"):
return Inspector(
id="Inspector2",
arn=INSPECTOR_ARN,
region=AWS_REGION_EU_WEST_1,
status=status,
ec2_status="ENABLED",
ecr_status="ENABLED",
lambda_status="ENABLED",
lambda_code_status="ENABLED",
findings=findings,
)
def build_finding(vulnerability_id):
return Finding(
arn=FINDING_ARN,
type="PACKAGE_VULNERABILITY",
severity="CRITICAL",
first_observed_at=datetime.now(timezone.utc),
vulnerability_id=vulnerability_id,
resource_ids=["i-0123456789abcdef0"],
)
def build_kev(vulnerability_id):
return KnownExploitedVulnerability(
id=vulnerability_id,
date_added=datetime(2024, 4, 12, tzinfo=timezone.utc),
date_due=datetime(2024, 4, 19, tzinfo=timezone.utc),
)
def execute_check(inspectors, known_exploited=None, lookup_failed=None):
inspector2_client = mock.MagicMock()
inspector2_client.inspectors = inspectors
inspector2_client.known_exploited_vulnerabilities = known_exploited or {}
inspector2_client.vulnerability_lookup_failed = lookup_failed or set()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider([AWS_REGION_EU_WEST_1]),
),
mock.patch(f"{CHECK_MODULE}.inspector2_client", new=inspector2_client),
):
from prowler.providers.aws.services.inspector2.inspector2_active_findings_no_known_exploited_vulnerabilities.inspector2_active_findings_no_known_exploited_vulnerabilities import (
inspector2_active_findings_no_known_exploited_vulnerabilities,
)
return inspector2_active_findings_no_known_exploited_vulnerabilities().execute()
class Test_inspector2_active_findings_no_known_exploited_vulnerabilities:
def test_no_resources(self):
assert execute_check([]) == []
def test_inspector_disabled(self):
assert execute_check([build_inspector(findings=[], status="DISABLED")]) == []
def test_no_kev_findings(self):
result = execute_check(
[build_inspector(findings=[build_finding("CVE-2022-40897")])]
)
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Inspector2 has no active findings in region {AWS_REGION_EU_WEST_1} for CISA Known Exploited Vulnerabilities."
)
assert result[0].resource_id == "Inspector2"
assert result[0].resource_arn == INSPECTOR_ARN
assert result[0].region == AWS_REGION_EU_WEST_1
def test_kev_finding(self):
result = execute_check(
[
build_inspector(
findings=[build_finding(KEV_ID), build_finding("CVE-2022-40897")]
)
],
known_exploited={KEV_ID: build_kev(KEV_ID)},
)
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Inspector2 has active findings in region {AWS_REGION_EU_WEST_1} for CISA Known Exploited Vulnerabilities: {KEV_ID}."
)
def test_many_kev_findings_are_truncated(self):
vulnerability_ids = [f"CVE-2024-{number:04d}" for number in range(1, 13)]
result = execute_check(
[
build_inspector(
findings=[build_finding(vid) for vid in vulnerability_ids]
)
],
known_exploited={vid: build_kev(vid) for vid in vulnerability_ids},
)
assert result[0].status == "FAIL"
assert result[0].status_extended.endswith("CVE-2024-0010 and 2 more.")
def test_kev_status_not_verified(self):
result = execute_check(
[build_inspector(findings=[build_finding(KEV_ID)])],
lookup_failed={KEV_ID},
)
assert len(result) == 1
assert result[0].status == "MANUAL"
assert (
result[0].status_extended
== f"Inspector2 could not verify the CISA Known Exploited Vulnerabilities status of {KEV_ID} in region {AWS_REGION_EU_WEST_1}; verify the inspector2:BatchGetFindingDetails permission."
)
def test_findings_not_retrieved(self):
result = execute_check([build_inspector(findings=None)])
assert len(result) == 1
assert result[0].status == "MANUAL"
assert (
result[0].status_extended
== f"Inspector2 findings could not be retrieved in region {AWS_REGION_EU_WEST_1}; verify the inspector2:ListFindings permission."
)
@@ -0,0 +1,140 @@
from datetime import datetime, timedelta, timezone
from unittest import mock
from prowler.providers.aws.services.inspector2.inspector2_service import (
Finding,
Inspector,
)
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
set_mocked_aws_provider,
)
INSPECTOR_ARN = (
f"arn:aws:inspector2:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:inspector2"
)
FINDING_ARN = f"arn:aws:inspector2:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:finding/0e436649379db5f327e3cf5bb4421d76"
CHECK_MODULE = "prowler.providers.aws.services.inspector2.inspector2_active_findings_within_max_age.inspector2_active_findings_within_max_age"
def build_inspector(findings=None, status="ENABLED"):
return Inspector(
id="Inspector2",
arn=INSPECTOR_ARN,
region=AWS_REGION_EU_WEST_1,
status=status,
ec2_status="ENABLED",
ecr_status="ENABLED",
lambda_status="ENABLED",
lambda_code_status="ENABLED",
findings=findings,
)
def build_finding(age_days):
first_observed_at = (
datetime.now(timezone.utc) - timedelta(days=age_days, hours=1)
if age_days is not None
else None
)
return Finding(
arn=FINDING_ARN,
type="PACKAGE_VULNERABILITY",
severity="HIGH",
first_observed_at=first_observed_at,
vulnerability_id="CVE-2022-40897",
resource_ids=["i-0123456789abcdef0"],
)
def execute_check(inspectors, audit_config=None):
inspector2_client = mock.MagicMock()
inspector2_client.inspectors = inspectors
inspector2_client.audit_config = audit_config or {}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider([AWS_REGION_EU_WEST_1]),
),
mock.patch(f"{CHECK_MODULE}.inspector2_client", new=inspector2_client),
):
from prowler.providers.aws.services.inspector2.inspector2_active_findings_within_max_age.inspector2_active_findings_within_max_age import (
inspector2_active_findings_within_max_age,
)
return inspector2_active_findings_within_max_age().execute()
class Test_inspector2_active_findings_within_max_age:
def test_no_resources(self):
assert execute_check([]) == []
def test_inspector_disabled(self):
assert execute_check([build_inspector(findings=[], status="DISABLED")]) == []
def test_no_active_findings(self):
result = execute_check([build_inspector(findings=[])])
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Inspector2 has no active findings in region {AWS_REGION_EU_WEST_1} first observed more than 192 days ago."
)
assert result[0].resource_id == "Inspector2"
assert result[0].resource_arn == INSPECTOR_ARN
assert result[0].region == AWS_REGION_EU_WEST_1
def test_recent_findings(self):
result = execute_check([build_inspector(findings=[build_finding(30)])])
assert len(result) == 1
assert result[0].status == "PASS"
def test_stale_findings(self):
result = execute_check(
[
build_inspector(
findings=[
build_finding(30),
build_finding(200),
build_finding(400),
]
)
]
)
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Inspector2 has 2 active findings in region {AWS_REGION_EU_WEST_1} first observed more than 192 days ago, the oldest 400 days ago."
)
def test_finding_just_over_max_age(self):
result = execute_check([build_inspector(findings=[build_finding(192)])])
assert len(result) == 1
assert result[0].status == "FAIL"
def test_custom_max_age(self):
result = execute_check(
[build_inspector(findings=[build_finding(30)])],
audit_config={"inspector2_active_finding_max_age_days": 14},
)
assert len(result) == 1
assert result[0].status == "FAIL"
def test_finding_without_first_observed_date_is_ignored(self):
result = execute_check([build_inspector(findings=[build_finding(None)])])
assert len(result) == 1
assert result[0].status == "PASS"
def test_findings_not_retrieved(self):
result = execute_check([build_inspector(findings=None)])
assert len(result) == 1
assert result[0].status == "MANUAL"
@@ -0,0 +1,170 @@
from datetime import datetime, timedelta, timezone
from unittest import mock
from prowler.providers.aws.services.inspector2.inspector2_service import (
CoveredResource,
Inspector,
)
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
set_mocked_aws_provider,
)
INSPECTOR_ARN = (
f"arn:aws:inspector2:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:inspector2"
)
INSTANCE_ID = "i-0123456789abcdef0"
INSTANCE_ARN = (
f"arn:aws:ec2:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:instance/{INSTANCE_ID}"
)
CHECK_MODULE = "prowler.providers.aws.services.inspector2.inspector2_coverage_recently_scanned.inspector2_coverage_recently_scanned"
def build_inspector(coverage=None, status="ENABLED"):
return Inspector(
id="Inspector2",
arn=INSPECTOR_ARN,
region=AWS_REGION_EU_WEST_1,
status=status,
ec2_status="ENABLED",
ecr_status="ENABLED",
lambda_status="ENABLED",
lambda_code_status="ENABLED",
coverage=coverage,
)
def build_instance(
days_since_scan=None, scan_status_code="ACTIVE", scan_status_reason="SUCCESSFUL"
):
last_scanned_at = (
datetime.now(timezone.utc) - timedelta(days=days_since_scan, hours=1)
if days_since_scan is not None
else None
)
return CoveredResource(
id=INSTANCE_ID,
arn=INSTANCE_ARN,
region=AWS_REGION_EU_WEST_1,
resource_type="AWS_EC2_INSTANCE",
scan_type="PACKAGE",
scan_status_code=scan_status_code,
scan_status_reason=scan_status_reason,
last_scanned_at=last_scanned_at,
)
def execute_check(inspectors, audit_config=None):
inspector2_client = mock.MagicMock()
inspector2_client.inspectors = inspectors
inspector2_client.audit_config = audit_config or {}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider([AWS_REGION_EU_WEST_1]),
),
mock.patch(f"{CHECK_MODULE}.inspector2_client", new=inspector2_client),
):
from prowler.providers.aws.services.inspector2.inspector2_coverage_recently_scanned.inspector2_coverage_recently_scanned import (
inspector2_coverage_recently_scanned,
)
return inspector2_coverage_recently_scanned().execute()
class Test_inspector2_coverage_recently_scanned:
def test_no_resources(self):
assert execute_check([]) == []
def test_inspector_disabled(self):
assert execute_check([build_inspector(status="DISABLED", coverage=[])]) == []
def test_recently_scanned_resource(self):
result = execute_check([build_inspector(coverage=[build_instance(1)])])
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"AWS_EC2_INSTANCE {INSTANCE_ID} was last scanned by Inspector2 within the last 3 days."
)
assert result[0].resource_id == INSTANCE_ID
assert result[0].resource_arn == INSTANCE_ARN
assert result[0].region == AWS_REGION_EU_WEST_1
def test_stale_resource(self):
result = execute_check([build_inspector(coverage=[build_instance(10)])])
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"AWS_EC2_INSTANCE {INSTANCE_ID} was last scanned by Inspector2 more than 3 days ago."
)
def test_resource_scanned_just_over_max_days(self):
result = execute_check([build_inspector(coverage=[build_instance(3)])])
assert len(result) == 1
assert result[0].status == "FAIL"
def test_custom_max_days(self):
result = execute_check(
[build_inspector(coverage=[build_instance(10)])],
audit_config={"inspector2_max_days_since_last_scan": 14},
)
assert len(result) == 1
assert result[0].status == "PASS"
def test_resource_without_recorded_scan(self):
result = execute_check([build_inspector(coverage=[build_instance(None)])])
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"AWS_EC2_INSTANCE {INSTANCE_ID} has no recorded Inspector2 scan."
)
def test_pending_initial_scan_is_skipped(self):
assert (
execute_check(
[
build_inspector(
coverage=[
build_instance(
None, scan_status_reason="PENDING_INITIAL_SCAN"
)
]
)
]
)
== []
)
def test_inactive_resource_is_skipped(self):
assert (
execute_check(
[
build_inspector(
coverage=[
build_instance(
10,
scan_status_code="INACTIVE",
scan_status_reason="NO_INVENTORY",
)
]
)
]
)
== []
)
def test_coverage_not_retrieved(self):
result = execute_check([build_inspector(coverage=None)])
assert len(result) == 1
assert result[0].status == "MANUAL"
assert result[0].resource_arn == INSPECTOR_ARN
@@ -0,0 +1,132 @@
from datetime import datetime, timezone
from unittest import mock
from prowler.providers.aws.services.inspector2.inspector2_service import (
CoveredResource,
Inspector,
)
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
set_mocked_aws_provider,
)
INSPECTOR_ARN = (
f"arn:aws:inspector2:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:inspector2"
)
INSTANCE_ID = "i-0123456789abcdef0"
INSTANCE_ARN = (
f"arn:aws:ec2:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:instance/{INSTANCE_ID}"
)
CHECK_MODULE = "prowler.providers.aws.services.inspector2.inspector2_coverage_scan_status_active.inspector2_coverage_scan_status_active"
def build_inspector(status="ENABLED", coverage=None):
return Inspector(
id="Inspector2",
arn=INSPECTOR_ARN,
region=AWS_REGION_EU_WEST_1,
status=status,
ec2_status="ENABLED",
ecr_status="ENABLED",
lambda_status="ENABLED",
lambda_code_status="ENABLED",
coverage=coverage,
)
def build_instance(scan_status_code, scan_status_reason):
return CoveredResource(
id=INSTANCE_ID,
arn=INSTANCE_ARN,
region=AWS_REGION_EU_WEST_1,
resource_type="AWS_EC2_INSTANCE",
scan_type="PACKAGE",
scan_status_code=scan_status_code,
scan_status_reason=scan_status_reason,
last_scanned_at=datetime.now(timezone.utc),
)
def execute_check(inspectors):
inspector2_client = mock.MagicMock()
inspector2_client.inspectors = inspectors
inspector2_client.audit_config = {}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider([AWS_REGION_EU_WEST_1]),
),
mock.patch(f"{CHECK_MODULE}.inspector2_client", new=inspector2_client),
):
from prowler.providers.aws.services.inspector2.inspector2_coverage_scan_status_active.inspector2_coverage_scan_status_active import (
inspector2_coverage_scan_status_active,
)
return inspector2_coverage_scan_status_active().execute()
class Test_inspector2_coverage_scan_status_active:
def test_no_resources(self):
assert execute_check([]) == []
def test_inspector_disabled(self):
assert execute_check([build_inspector(status="DISABLED", coverage=[])]) == []
def test_no_covered_resources(self):
assert execute_check([build_inspector(coverage=[])]) == []
def test_active_resource(self):
result = execute_check(
[build_inspector(coverage=[build_instance("ACTIVE", "SUCCESSFUL")])]
)
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Inspector2 is actively scanning AWS_EC2_INSTANCE {INSTANCE_ID}."
)
assert result[0].resource_id == INSTANCE_ID
assert result[0].resource_arn == INSTANCE_ARN
assert result[0].region == AWS_REGION_EU_WEST_1
def test_inactive_resource(self):
result = execute_check(
[
build_inspector(
coverage=[build_instance("INACTIVE", "UNMANAGED_EC2_INSTANCE")]
)
]
)
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Inspector2 is not scanning AWS_EC2_INSTANCE {INSTANCE_ID}: UNMANAGED_EC2_INSTANCE."
)
assert result[0].resource_id == INSTANCE_ID
def test_not_applicable_resource_is_skipped(self):
assert (
execute_check(
[
build_inspector(
coverage=[build_instance("INACTIVE", "EC2_INSTANCE_STOPPED")]
)
]
)
== []
)
def test_coverage_not_retrieved(self):
result = execute_check([build_inspector(coverage=None)])
assert len(result) == 1
assert result[0].status == "MANUAL"
assert (
result[0].status_extended
== f"Inspector2 coverage could not be retrieved in region {AWS_REGION_EU_WEST_1}; verify the inspector2:ListCoverage permission."
)
assert result[0].resource_arn == INSPECTOR_ARN
@@ -1,18 +1,30 @@
from datetime import datetime
from datetime import datetime, timezone
from unittest.mock import patch
import botocore
from botocore.exceptions import ClientError
from prowler.providers.aws.services.inspector2.inspector2_service import Inspector2
from prowler.providers.aws.services.inspector2.inspector2_service import (
Finding,
Inspector,
Inspector2,
)
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
FINDING_ARN = (
"arn:aws:inspector2:us-east-1:123456789012:finding/0e436649379db5f327e3cf5bb4421d76"
)
VULNERABILITY_ID = "CVE-2022-40897"
INSTANCE_ID = "i-0123456789abcdef0"
FIRST_OBSERVED_AT = datetime(2024, 1, 1, tzinfo=timezone.utc)
LAST_SCANNED_AT = datetime(2024, 6, 1, tzinfo=timezone.utc)
KEV_DATE_ADDED = datetime(2024, 4, 12, tzinfo=timezone.utc)
KEV_DATE_DUE = datetime(2024, 5, 3, tzinfo=timezone.utc)
# Mocking Calls
make_api_call = botocore.client.BaseClient._make_api_call
@@ -64,16 +76,83 @@ def mock_make_api_call(self, operation_name, kwargs):
"description": "Finding Description",
"severity": "MEDIUM",
"status": "ACTIVE",
"title": "CVE-2022-40897 - setuptools",
"title": f"{VULNERABILITY_ID} - setuptools",
"type": "PACKAGE_VULNERABILITY",
"firstObservedAt": FIRST_OBSERVED_AT,
"updatedAt": datetime(2024, 1, 1),
"packageVulnerabilityDetails": {
"vulnerabilityId": VULNERABILITY_ID
},
"resources": [{"id": INSTANCE_ID, "type": "AWS_EC2_INSTANCE"}],
}
]
}
if operation_name == "ListCoverage":
return {
"coveredResources": [
{
"resourceId": INSTANCE_ID,
"resourceType": "AWS_EC2_INSTANCE",
"accountId": AWS_ACCOUNT_NUMBER,
"scanType": "PACKAGE",
"scanStatus": {"statusCode": "ACTIVE", "reason": "SUCCESSFUL"},
"lastScannedAt": LAST_SCANNED_AT,
}
]
}
if operation_name == "BatchGetFindingDetails":
return {
"findingDetails": [
{
"findingArn": FINDING_ARN,
"cisaData": {
"dateAdded": KEV_DATE_ADDED,
"dateDue": KEV_DATE_DUE,
},
}
],
"errors": [],
}
return make_api_call(self, operation_name, kwargs)
def mock_make_api_call_finding_details_denied(self, operation_name, kwargs):
if operation_name == "BatchGetFindingDetails":
raise ClientError(
{"Error": {"Code": "AccessDeniedException", "Message": "denied"}},
operation_name,
)
return mock_make_api_call(self, operation_name, kwargs)
def mock_finding_details_error(error_code):
def _mock(self, operation_name, kwargs):
if operation_name == "BatchGetFindingDetails":
return {
"findingDetails": [],
"errors": [
{
"findingArn": FINDING_ARN,
"errorCode": error_code,
"errorMessage": "error",
}
],
}
return mock_make_api_call(self, operation_name, kwargs)
return _mock
def mock_make_api_call_list_denied(self, operation_name, kwargs):
if operation_name in ("ListFindings", "ListCoverage"):
raise ClientError(
{"Error": {"Code": "AccessDeniedException", "Message": "denied"}},
operation_name,
)
return mock_make_api_call(self, operation_name, kwargs)
def mock_generate_regional_clients(provider, service):
regional_client = provider._session.current_session.client(
service, region_name=AWS_REGION_EU_WEST_1
@@ -82,6 +161,29 @@ def mock_generate_regional_clients(provider, service):
return {AWS_REGION_EU_WEST_1: regional_client}
def build_inspector(region, vulnerability_ids):
return Inspector(
id="Inspector2",
arn=f"arn:aws:inspector2:{region}:{AWS_ACCOUNT_NUMBER}:inspector2",
region=region,
status="ENABLED",
ec2_status="ENABLED",
ecr_status="ENABLED",
lambda_status="ENABLED",
lambda_code_status="ENABLED",
findings=[
Finding(
arn=f"arn:aws:inspector2:{region}:{AWS_ACCOUNT_NUMBER}:finding/{index}",
type="PACKAGE_VULNERABILITY",
severity="HIGH",
first_observed_at=FIRST_OBSERVED_AT,
vulnerability_id=vulnerability_id,
)
for index, vulnerability_id in enumerate(vulnerability_ids)
],
)
# Patch every AWS call using Boto3 and generate_regional_clients to have 1 client
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
@patch(
@@ -118,3 +220,115 @@ class Test_Inspector2_Service:
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
inspector2 = Inspector2(aws_provider)
assert inspector2.inspectors[0].active_findings
def test_list_findings(self):
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
inspector2 = Inspector2(aws_provider)
findings = inspector2.inspectors[0].findings
assert len(findings) == 1
assert findings[0].arn == FINDING_ARN
assert findings[0].type == "PACKAGE_VULNERABILITY"
assert findings[0].severity == "MEDIUM"
assert findings[0].first_observed_at == FIRST_OBSERVED_AT
assert findings[0].vulnerability_id == VULNERABILITY_ID
assert findings[0].resource_ids == [INSTANCE_ID]
def test_list_coverage(self):
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
inspector2 = Inspector2(aws_provider)
coverage = inspector2.inspectors[0].coverage
assert len(coverage) == 1
assert coverage[0].id == INSTANCE_ID
assert (
coverage[0].arn
== f"arn:aws:ec2:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:instance/{INSTANCE_ID}"
)
assert coverage[0].region == AWS_REGION_EU_WEST_1
assert coverage[0].resource_type == "AWS_EC2_INSTANCE"
assert coverage[0].scan_type == "PACKAGE"
assert coverage[0].scan_status_code == "ACTIVE"
assert coverage[0].scan_status_reason == "SUCCESSFUL"
assert coverage[0].last_scanned_at == LAST_SCANNED_AT
def test_list_coverage_keeps_audited_resources(self):
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
aws_provider._audit_resources = [
f"arn:aws:ec2:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:instance/{INSTANCE_ID}"
]
inspector2 = Inspector2(aws_provider)
assert len(inspector2.inspectors[0].coverage) == 1
def test_list_coverage_skips_non_audited_resources(self):
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
aws_provider._audit_resources = [
f"arn:aws:ec2:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:instance/i-0fedcba9876543210"
]
inspector2 = Inspector2(aws_provider)
assert inspector2.inspectors[0].coverage == []
def test_batch_get_finding_details(self):
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
inspector2 = Inspector2(aws_provider)
known_exploited = inspector2.known_exploited_vulnerabilities[VULNERABILITY_ID]
assert known_exploited.id == VULNERABILITY_ID
assert known_exploited.date_added == KEV_DATE_ADDED
assert known_exploited.date_due == KEV_DATE_DUE
assert inspector2.vulnerability_lookup_failed == set()
def test_batch_get_finding_details_denied(self):
with patch(
"botocore.client.BaseClient._make_api_call",
new=mock_make_api_call_finding_details_denied,
):
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
inspector2 = Inspector2(aws_provider)
assert inspector2.known_exploited_vulnerabilities == {}
assert inspector2.vulnerability_lookup_failed == {VULNERABILITY_ID}
def test_finding_details_not_found_is_not_a_lookup_failure(self):
with patch(
"botocore.client.BaseClient._make_api_call",
new=mock_finding_details_error("FINDING_DETAILS_NOT_FOUND"),
):
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
inspector2 = Inspector2(aws_provider)
assert inspector2.known_exploited_vulnerabilities == {}
assert inspector2.vulnerability_lookup_failed == set()
def test_finding_details_error_is_a_lookup_failure(self):
with patch(
"botocore.client.BaseClient._make_api_call",
new=mock_finding_details_error("INTERNAL_ERROR"),
):
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
inspector2 = Inspector2(aws_provider)
assert inspector2.known_exploited_vulnerabilities == {}
assert inspector2.vulnerability_lookup_failed == {VULNERABILITY_ID}
def test_list_findings_and_coverage_denied(self):
with patch(
"botocore.client.BaseClient._make_api_call",
new=mock_make_api_call_list_denied,
):
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
inspector2 = Inspector2(aws_provider)
assert inspector2.inspectors[0].findings is None
assert inspector2.inspectors[0].coverage is None
assert inspector2.known_exploited_vulnerabilities == {}
def test_finding_detail_batches_use_one_finding_per_cve(self):
vulnerability_ids = [f"CVE-2024-{number:04d}" for number in range(25)]
batches = Inspector2._get_finding_detail_batches(
[
build_inspector(
AWS_REGION_EU_WEST_1,
vulnerability_ids + vulnerability_ids[:5] + ["GHSA-xxxx-yyyy-zzzz"],
),
build_inspector(AWS_REGION_US_EAST_1, vulnerability_ids[:3]),
]
)
assert [region for region, _ in batches] == [AWS_REGION_EU_WEST_1] * 3
assert [len(findings) for _, findings in batches] == [10, 10, 5]
assert sorted(cve for _, findings in batches for _, cve in findings) == sorted(
vulnerability_ids
)
@@ -0,0 +1,111 @@
from unittest import mock
from unittest.mock import patch
import botocore
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
SERVER_ID = "s-01234567890abcdef"
SERVER_ARN = (
f"arn:aws:transfer:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:server/{SERVER_ID}"
)
CHECK_MODULE = "prowler.providers.aws.services.transfer.transfer_server_fips_security_policy_enabled.transfer_server_fips_security_policy_enabled"
make_api_call = botocore.client.BaseClient._make_api_call
def mock_server_with_policy(security_policy_name):
def _mock(self, operation_name, kwarg):
if operation_name == "ListServers":
return {"Servers": [{"Arn": SERVER_ARN, "ServerId": SERVER_ID}]}
if operation_name == "DescribeServer":
return {
"Server": {
"Arn": SERVER_ARN,
"ServerId": SERVER_ID,
"Protocols": ["SFTP"],
"SecurityPolicyName": security_policy_name,
}
}
return make_api_call(self, operation_name, kwarg)
return _mock
def execute_check():
from prowler.providers.aws.services.transfer.transfer_service import Transfer
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
),
mock.patch(f"{CHECK_MODULE}.transfer_client", new=Transfer(aws_provider)),
):
from prowler.providers.aws.services.transfer.transfer_server_fips_security_policy_enabled.transfer_server_fips_security_policy_enabled import (
transfer_server_fips_security_policy_enabled,
)
return transfer_server_fips_security_policy_enabled().execute()
class Test_transfer_server_fips_security_policy_enabled:
@mock_aws
def test_no_servers(self):
assert execute_check() == []
@patch(
"botocore.client.BaseClient._make_api_call",
new=mock_server_with_policy("TransferSecurityPolicy-FIPS-2025-03"),
)
@mock_aws
def test_fips_policy(self):
result = execute_check()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Transfer Server {SERVER_ID} uses FIPS security policy TransferSecurityPolicy-FIPS-2025-03."
)
assert result[0].resource_id == SERVER_ID
assert result[0].resource_arn == SERVER_ARN
assert result[0].region == AWS_REGION_US_EAST_1
@patch(
"botocore.client.BaseClient._make_api_call",
new=mock_server_with_policy("TransferSecurityPolicy-2024-01"),
)
@mock_aws
def test_non_fips_policy(self):
result = execute_check()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Transfer Server {SERVER_ID} uses security policy TransferSecurityPolicy-2024-01, which is not a FIPS security policy."
)
@patch(
"botocore.client.BaseClient._make_api_call",
new=mock_server_with_policy(""),
)
@mock_aws
def test_policy_not_retrieved(self):
result = execute_check()
assert len(result) == 1
assert result[0].status == "MANUAL"
assert (
result[0].status_extended
== f"Transfer Server security policies could not be retrieved for {SERVER_ID}; verify the transfer:DescribeServer permission."
)
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].region == AWS_REGION_US_EAST_1