feat(FMS): New Service FMS and Check fms_accounts_compliant (#2259)

Co-authored-by: Pepe Fagoaga <pepe@verica.io> Co-authored-by: Nacho Rivera <nacho@verica.io>
2026-02-11 07:15:15 +00:00 · 2023-04-28 11:47:55 +02:00
parent 5dc2347a25
commit 9afe7408cd
8 changed files with 458 additions and 0 deletions
--- 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
@@ -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
--- a/tests/providers/aws/services/fms/fms_service_test.py
+++ 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