From 9afe7408cdf152dc0f4aa4b0bd83ccc080085f2c Mon Sep 17 00:00:00 2001 From: Gabriel Soltz Date: Fri, 28 Apr 2023 11:47:55 +0200 Subject: [PATCH] feat(FMS): New Service FMS and Check fms_accounts_compliant (#2259) Co-authored-by: Pepe Fagoaga Co-authored-by: Nacho Rivera --- .../providers/aws/services/fms/__init__.py | 0 .../providers/aws/services/fms/fms_client.py | 4 + .../fms/fms_policy_compliant/__init__.py | 0 .../fms_policy_compliant.metadata.json | 30 ++++ .../fms_policy_compliant.py | 29 ++++ .../providers/aws/services/fms/fms_service.py | 108 ++++++++++++ .../fms_policy_compliant_test.py | 164 ++++++++++++++++++ .../aws/services/fms/fms_service_test.py | 123 +++++++++++++ 8 files changed, 458 insertions(+) create mode 100644 prowler/providers/aws/services/fms/__init__.py create mode 100644 prowler/providers/aws/services/fms/fms_client.py create mode 100644 prowler/providers/aws/services/fms/fms_policy_compliant/__init__.py create mode 100644 prowler/providers/aws/services/fms/fms_policy_compliant/fms_policy_compliant.metadata.json create mode 100644 prowler/providers/aws/services/fms/fms_policy_compliant/fms_policy_compliant.py create mode 100644 prowler/providers/aws/services/fms/fms_service.py create mode 100644 tests/providers/aws/services/fms/fms_policy_compliant/fms_policy_compliant_test.py create mode 100644 tests/providers/aws/services/fms/fms_service_test.py diff --git a/prowler/providers/aws/services/fms/__init__.py b/prowler/providers/aws/services/fms/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/prowler/providers/aws/services/fms/fms_client.py b/prowler/providers/aws/services/fms/fms_client.py new file mode 100644 index 00000000..0bd41b0c --- /dev/null +++ b/prowler/providers/aws/services/fms/fms_client.py @@ -0,0 +1,4 @@ +from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info +from prowler.providers.aws.services.fms.fms_service import FMS + +fms_client = FMS(current_audit_info) diff --git a/prowler/providers/aws/services/fms/fms_policy_compliant/__init__.py b/prowler/providers/aws/services/fms/fms_policy_compliant/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/prowler/providers/aws/services/fms/fms_policy_compliant/fms_policy_compliant.metadata.json b/prowler/providers/aws/services/fms/fms_policy_compliant/fms_policy_compliant.metadata.json new file mode 100644 index 00000000..35dd57fb --- /dev/null +++ b/prowler/providers/aws/services/fms/fms_policy_compliant/fms_policy_compliant.metadata.json @@ -0,0 +1,30 @@ +{ + "Provider": "aws", + "CheckID": "fms_policy_compliant", + "CheckTitle": "Ensure that all FMS policies inside an admin account are compliant", + "CheckType": [], + "ServiceName": "fms", + "SubServiceName": "", + "ResourceIdTemplate": "arn:aws:fms:region:account-id:policy/policy", + "Severity": "medium", + "ResourceType": "Other", + "Description": "This check ensures all FMS policies inside an admin account are compliant", + "Risk": "If FMS policies are not compliant, means there are resources unprotected by the policies", + "RelatedUrl": "https://docs.aws.amazon.com/waf/latest/developerguide/getting-started-fms-intro.html", + "Remediation": { + "Code": { + "CLI": "aws fms list-policies", + "NativeIaC": "", + "Other": "", + "Terraform": "" + }, + "Recommendation": { + "Text": "Ensure FMS is enabled and all the policies are compliant across your AWS accounts", + "Url": "" + } + }, + "Categories": [], + "DependsOn": [], + "RelatedTo": [], + "Notes": "" +} diff --git a/prowler/providers/aws/services/fms/fms_policy_compliant/fms_policy_compliant.py b/prowler/providers/aws/services/fms/fms_policy_compliant/fms_policy_compliant.py new file mode 100644 index 00000000..f7244333 --- /dev/null +++ b/prowler/providers/aws/services/fms/fms_policy_compliant/fms_policy_compliant.py @@ -0,0 +1,29 @@ +from prowler.lib.check.models import Check, Check_Report_AWS +from prowler.providers.aws.services.fms.fms_client import fms_client + + +class fms_policy_compliant(Check): + def execute(self): + findings = [] + if fms_client.fms_admin_account: + report = Check_Report_AWS(self.metadata()) + report.resource_id = "FMS" + report.resource_arn = "" + report.region = fms_client.region + report.status = "PASS" + report.status_extended = "FMS enabled with all compliant accounts" + non_compliant_policy = False + for policy in fms_client.fms_policies: + for policy_to_account in policy.compliance_status: + if policy_to_account.status == "NON_COMPLIANT": + report.status = "FAIL" + report.status_extended = f"FMS with non-compliant policy {policy.name} for account {policy_to_account.account_id}" + report.resource_id = policy.id + report.resource_arn = policy.arn + non_compliant_policy = True + break + if non_compliant_policy: + break + + findings.append(report) + return findings diff --git a/prowler/providers/aws/services/fms/fms_service.py b/prowler/providers/aws/services/fms/fms_service.py new file mode 100644 index 00000000..d23a99ad --- /dev/null +++ b/prowler/providers/aws/services/fms/fms_service.py @@ -0,0 +1,108 @@ +from botocore.client import ClientError +from pydantic import BaseModel + +from prowler.lib.logger import logger +from prowler.lib.scan_filters.scan_filters import is_resource_filtered +from prowler.providers.aws.aws_provider import generate_regional_clients + + +################## FMS +class FMS: + def __init__(self, audit_info): + self.service = "fms" + self.session = audit_info.audit_session + self.audited_account = audit_info.audited_account + self.audited_partition = audit_info.audited_partition + self.audit_resources = audit_info.audit_resources + global_client = generate_regional_clients( + self.service, audit_info, global_service=True + ) + self.client = list(global_client.values())[0] + self.region = self.client.region + self.fms_admin_account = True + self.fms_policies = [] + self.__list_policies__() + self.__list_compliance_status__() + + def __list_policies__(self): + logger.info("FMS - Listing Policies...") + try: + list_policies_paginator = self.client.get_paginator("list_policies") + try: + for page in list_policies_paginator.paginate(): + for fms_policy in page["PolicyList"]: + if not self.audit_resources or ( + is_resource_filtered( + fms_policy["PolicyArn"], self.audit_resources + ) + ): + self.fms_policies.append( + Policy( + arn=fms_policy.get("PolicyArn"), + id=fms_policy.get("PolicyId"), + name=fms_policy.get("PolicyName"), + resource_type=fms_policy.get("ResourceType"), + service_type=fms_policy.get("SecurityServiceType"), + remediation_enabled=fms_policy.get( + "RemediationEnabled" + ), + delete_unused_managed_resources=fms_policy.get( + "DeleteUnusedFMManagedResources" + ), + ) + ) + except ClientError as error: + if error.response["Error"]["Code"] == "AccessDeniedException": + if ( + "No default admin could be found for account" + in error.response["Error"]["Message"] + ): + # FMS is not enabled in this account + self.fms_admin_account = False + except Exception as error: + logger.error( + f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}" + ) + + def __list_compliance_status__(self): + logger.info("FMS - Listing Policies...") + try: + for fms_policy in self.fms_policies: + list_compliance_status_paginator = self.client.get_paginator( + "list_compliance_status" + ) + for page in list_compliance_status_paginator.paginate( + PolicyId=fms_policy.id + ): + for fms_compliance_status in page["PolicyComplianceStatusList"]: + fms_policy.compliance_status.append( + PolicyAccountComplianceStatus( + account_id=fms_compliance_status.get("MemberAccount"), + policy_id=fms_compliance_status.get("PolicyId"), + status=fms_compliance_status.get("EvaluationResults")[ + 0 + ].get("ComplianceStatus"), + ) + ) + + except Exception as error: + logger.error( + f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}" + ) + + +class PolicyAccountComplianceStatus(BaseModel): + account_id: str + policy_id: str + status: str + + +class Policy(BaseModel): + arn: str + id: str + name: str + resource_type: str + service_type: str + remediation_enabled: bool + delete_unused_managed_resources: bool + compliance_status: list[PolicyAccountComplianceStatus] = [] diff --git a/tests/providers/aws/services/fms/fms_policy_compliant/fms_policy_compliant_test.py b/tests/providers/aws/services/fms/fms_policy_compliant/fms_policy_compliant_test.py new file mode 100644 index 00000000..1a1592aa --- /dev/null +++ b/tests/providers/aws/services/fms/fms_policy_compliant/fms_policy_compliant_test.py @@ -0,0 +1,164 @@ +from unittest import mock + +from prowler.providers.aws.services.fms.fms_service import ( + Policy, + PolicyAccountComplianceStatus, +) + +AWS_REGION = "us-east-1" + + +class Test_fms_policy_compliant: + def test_fms_not_admin(self): + fms_client = mock.MagicMock + fms_client.region = AWS_REGION + fms_client.fms_admin_account = False + with mock.patch( + "prowler.providers.aws.services.fms.fms_service.FMS", + new=fms_client, + ): + # Test Check + from prowler.providers.aws.services.fms.fms_policy_compliant.fms_policy_compliant import ( + fms_policy_compliant, + ) + + check = fms_policy_compliant() + result = check.execute() + + assert len(result) == 0 + + def test_fms_admin_with_non_compliant_policies(self): + fms_client = mock.MagicMock + fms_client.region = AWS_REGION + fms_client.fms_admin_account = True + fms_client.fms_policies = [ + Policy( + arn="arn:aws:fms:us-east-1:12345678901", + id="12345678901", + name="test", + resource_type="AWS::EC2::Instance", + service_type="WAF", + remediation_enabled=True, + delete_unused_managed_resources=True, + compliance_status=[ + PolicyAccountComplianceStatus( + account_id="12345678901", + policy_id="12345678901", + status="NON_COMPLIANT", + ) + ], + ) + ] + with mock.patch( + "prowler.providers.aws.services.fms.fms_service.FMS", + new=fms_client, + ): + # Test Check + from prowler.providers.aws.services.fms.fms_policy_compliant.fms_policy_compliant import ( + fms_policy_compliant, + ) + + check = fms_policy_compliant() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"FMS with non-compliant policy {fms_client.fms_policies[0].name} for account {fms_client.fms_policies[0].compliance_status[0].account_id}" + ) + assert result[0].resource_id == "12345678901" + assert result[0].resource_arn == "arn:aws:fms:us-east-1:12345678901" + assert result[0].region == AWS_REGION + + def test_fms_admin_with_compliant_policies(self): + fms_client = mock.MagicMock + fms_client.region = AWS_REGION + fms_client.fms_admin_account = True + fms_client.fms_policies = [ + Policy( + arn="arn:aws:fms:us-east-1:12345678901", + id="12345678901", + name="test", + resource_type="AWS::EC2::Instance", + service_type="WAF", + remediation_enabled=True, + delete_unused_managed_resources=True, + compliance_status=[ + PolicyAccountComplianceStatus( + account_id="12345678901", + policy_id="12345678901", + status="COMPLIANT", + ) + ], + ) + ] + with mock.patch( + "prowler.providers.aws.services.fms.fms_service.FMS", + new=fms_client, + ): + # Test Check + from prowler.providers.aws.services.fms.fms_policy_compliant.fms_policy_compliant import ( + fms_policy_compliant, + ) + + check = fms_policy_compliant() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended == "FMS enabled with all compliant accounts" + ) + assert result[0].resource_id == "FMS" + assert result[0].resource_arn == "" + assert result[0].region == AWS_REGION + + def test_fms_admin_with_non_and_compliant_policies(self): + fms_client = mock.MagicMock + fms_client.region = AWS_REGION + fms_client.fms_admin_account = True + fms_client.fms_policies = [ + Policy( + arn="arn:aws:fms:us-east-1:12345678901", + id="12345678901", + name="test", + resource_type="AWS::EC2::Instance", + service_type="WAF", + remediation_enabled=True, + delete_unused_managed_resources=True, + compliance_status=[ + PolicyAccountComplianceStatus( + account_id="12345678901", + policy_id="12345678901", + status="COMPLIANT", + ), + PolicyAccountComplianceStatus( + account_id="12345678901", + policy_id="12345678901", + status="NON_COMPLIANT", + ), + ], + ) + ] + with mock.patch( + "prowler.providers.aws.services.fms.fms_service.FMS", + new=fms_client, + ): + # Test Check + from prowler.providers.aws.services.fms.fms_policy_compliant.fms_policy_compliant import ( + fms_policy_compliant, + ) + + check = fms_policy_compliant() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"FMS with non-compliant policy {fms_client.fms_policies[0].name} for account {fms_client.fms_policies[0].compliance_status[0].account_id}" + ) + assert result[0].resource_id == "12345678901" + assert result[0].resource_arn == "arn:aws:fms:us-east-1:12345678901" + assert result[0].region == AWS_REGION diff --git a/tests/providers/aws/services/fms/fms_service_test.py b/tests/providers/aws/services/fms/fms_service_test.py new file mode 100644 index 00000000..9947cccd --- /dev/null +++ b/tests/providers/aws/services/fms/fms_service_test.py @@ -0,0 +1,123 @@ +from datetime import datetime +from unittest.mock import patch + +import botocore +from boto3 import session + +from prowler.providers.aws.lib.audit_info.audit_info import AWS_Audit_Info +from prowler.providers.aws.services.fms.fms_service import FMS + +# Mock Test Region +AWS_REGION = "us-east-1" +POLICY_ARN = "arn:aws:fms:us-east-1:123456789012:policy/MyFMSManagedPolicy" +POLICY_ID = "12345678-1234-1234-1234-123456789012" +POLICY_NAME = "MyFMSManagedPolicy" +RESOURCE_TYPE = "AWS::EC2::Instance" +SERVICE_TYPE = "WAF" +REMEDIATION_ENABLED = True +DELETE_UNUSED_MANAGED_RESOURCES = True + +# Mocking Calls +make_api_call = botocore.client.BaseClient._make_api_call + + +def mock_make_api_call(self, operation_name, kwargs): + """We have to mock every AWS API call using Boto3""" + if operation_name == "ListPolicies": + return { + "PolicyList": [ + { + "DeleteUnusedFMManagedResources": DELETE_UNUSED_MANAGED_RESOURCES, + "PolicyArn": POLICY_ARN, + "PolicyId": POLICY_ID, + "PolicyName": POLICY_NAME, + "RemediationEnabled": REMEDIATION_ENABLED, + "ResourceType": RESOURCE_TYPE, + "SecurityServiceType": SERVICE_TYPE, + } + ] + } + if operation_name == "ListComplianceStatus": + return { + "PolicyComplianceStatusList": [ + { + "EvaluationResults": [ + { + "ComplianceStatus": "COMPLIANT", + "EvaluationLimitExceeded": False, + "ViolatorCount": 10, + } + ], + "IssueInfoMap": {"string": "test"}, + "LastUpdated": datetime(2024, 1, 1), + "MemberAccount": "123456789012", + "PolicyId": POLICY_ID, + "PolicyName": POLICY_NAME, + "PolicyOwner": "123456789011", + } + ] + } + + return make_api_call(self, operation_name, kwargs) + + +# Patch every AWS call using Boto3 +@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) +class Test_FMS_Service: + + # Mocked Audit Info + def set_mocked_audit_info(self): + audit_info = AWS_Audit_Info( + session_config=None, + original_session=None, + audit_session=session.Session( + profile_name=None, + botocore_session=None, + ), + audited_account=None, + audited_user_id=None, + audited_partition="aws", + audited_identity_arn=None, + profile=None, + profile_region=None, + credentials=None, + assumed_role_info=None, + audited_regions=None, + organizations_metadata=None, + audit_resources=None, + ) + return audit_info + + def test__get_client__(self): + audit_info = self.set_mocked_audit_info() + fms = FMS(audit_info) + assert fms.client.__class__.__name__ == "FMS" + + def test__get_service__(self): + audit_info = self.set_mocked_audit_info() + fms = FMS(audit_info) + assert fms.service == "fms" + + def test__list_policies__(self): + audit_info = self.set_mocked_audit_info() + fms = FMS(audit_info) + assert len(fms.fms_policies) == 1 + assert fms.fms_admin_account is True + assert fms.fms_policies[0].arn == POLICY_ARN + assert fms.fms_policies[0].id == POLICY_ID + assert fms.fms_policies[0].name == POLICY_NAME + assert fms.fms_policies[0].resource_type == RESOURCE_TYPE + assert fms.fms_policies[0].service_type == SERVICE_TYPE + assert fms.fms_policies[0].remediation_enabled == REMEDIATION_ENABLED + assert ( + fms.fms_policies[0].delete_unused_managed_resources + == DELETE_UNUSED_MANAGED_RESOURCES + ) + + def test__list_compliance_status__(self): + audit_info = self.set_mocked_audit_info() + fms = FMS(audit_info) + assert len(fms.fms_policies) == 1 + assert fms.fms_policies[0].compliance_status[0].status == "COMPLIANT" + assert fms.fms_policies[0].compliance_status[0].account_id == "123456789012" + assert fms.fms_policies[0].compliance_status[0].policy_id == POLICY_ID