feat(wellarchitected): add WellArchitected service and check (#2461)

README.md

@@ -39,7 +39,7 @@ It contains hundreds of controls covering CIS, NIST 800, NIST CSF, CISA, RBI, Fe
 
 | Provider | Checks | Services | [Compliance Frameworks](https://docs.prowler.cloud/en/latest/tutorials/compliance/) | [Categories](https://docs.prowler.cloud/en/latest/tutorials/misc/#categories) |
 |---|---|---|---|---|
-| AWS | 281 | 54 -> `prowler aws --list-services` | 21 -> `prowler aws --list-compliance` | 6 -> `prowler aws --list-categories` |
+| AWS | 282 | 55 -> `prowler aws --list-services` | 21 -> `prowler aws --list-compliance` | 5 -> `prowler aws --list-categories` |
 | GCP | 59 | 10 -> `prowler gcp --list-services` | CIS soon | 0 -> `prowler gcp --list-categories`|
 | Azure | 20 | 3 -> `prowler azure --list-services` | CIS soon | 1 -> `prowler azure --list-categories` |
 | Kubernetes | Planned | - | - | - |

permissions/prowler-additions-policy.json

@@ -32,7 +32,8 @@
         "ssm:GetDocument",
         "ssm-incidents:List*",
         "support:Describe*",
-        "tag:GetTagKeys"
+        "tag:GetTagKeys",
+        "wellarchitected:List*"
       ],
       "Resource": "*",
       "Effect": "Allow",
@@ -44,7 +45,7 @@
         "apigateway:GET"
       ],
       "Resource": [
-        "arn:aws:apigateway:*::/restapis/*", 
+        "arn:aws:apigateway:*::/restapis/*",
         "arn:aws:apigateway:*::/apis/*"
       ]
     }

prowler/compliance/aws/aws_well_architected_framework_security_pillar_aws.json

@@ -138,7 +138,9 @@
           "ImplementationGuidanceUrl": "https://docs.aws.amazon.com/wellarchitected/latest/security-pillar/sec_securely_operate_threat_model.html#implementation-guidance."
         }
       ],
-      "Checks": []
+      "Checks": [
+        "wellarchitected_workload_no_high_or_medium_risks"
+      ]
     },
     {
       "Id": "SEC01-BP08",

prowler/providers/aws/aws_regions_by_service.json

@@ -9675,6 +9675,35 @@
         "aws-us-gov": []
       }
     },
+    "wellarchitected": {
+      "regions": {
+        "aws": [
+          "ap-east-1",
+          "ap-northeast-1",
+          "ap-northeast-2",
+          "ap-south-1",
+          "ap-southeast-1",
+          "ap-southeast-2",
+          "ca-central-1",
+          "eu-central-1",
+          "eu-north-1",
+          "eu-west-1",
+          "eu-west-2",
+          "eu-west-3",
+          "me-south-1",
+          "sa-east-1",
+          "us-east-1",
+          "us-east-2",
+          "us-west-1",
+          "us-west-2"
+        ],
+        "aws-cn": [],
+        "aws-us-gov": [
+          "us-gov-east-1",
+          "us-gov-west-1"
+        ]
+      }
+    },
     "wellarchitectedtool": {
       "regions": {
         "aws": [
@@ -9851,4 +9880,4 @@
       }
     }
   }
-}
+}

prowler/providers/aws/services/wellarchitected/wellarchitected_client.py

@@ -0,0 +1,6 @@
+from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
+from prowler.providers.aws.services.wellarchitected.wellarchitected_service import (
+    WellArchitected,
+)
+
+wellarchitected_client = WellArchitected(current_audit_info)

prowler/providers/aws/services/wellarchitected/wellarchitected_service.py

@@ -0,0 +1,81 @@
+import threading
+from typing import Optional
+
+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
+
+
+################################ WellArchitected
+class WellArchitected:
+    def __init__(self, audit_info):
+        self.service = "wellarchitected"
+        self.session = audit_info.audit_session
+        self.audit_resources = audit_info.audit_resources
+        self.regional_clients = generate_regional_clients(self.service, audit_info)
+        self.workloads = []
+        self.__threading_call__(self.__list_workloads__)
+        self.__list_tags_for_resource__()
+
+    def __get_session__(self):
+        return self.session
+
+    def __threading_call__(self, call):
+        threads = []
+        for regional_client in self.regional_clients.values():
+            threads.append(threading.Thread(target=call, args=(regional_client,)))
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+    def __list_workloads__(self, regional_client):
+        logger.info("WellArchitected - Listing Workloads...")
+        try:
+            for workload in regional_client.list_workloads()["WorkloadSummaries"]:
+                if not self.audit_resources or (
+                    is_resource_filtered(workload["WorkloadArn"], self.audit_resources)
+                ):
+                    self.workloads.append(
+                        Workload(
+                            id=workload["WorkloadId"],
+                            arn=workload["WorkloadArn"],
+                            name=workload["WorkloadName"],
+                            region=regional_client.region,
+                            lenses=workload["Lenses"],
+                            improvement_status=workload["ImprovementStatus"],
+                            risks=workload["RiskCounts"],
+                        )
+                    )
+
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def __list_tags_for_resource__(self):
+        logger.info("WellArchitected - Listing Tags...")
+        try:
+            for workload in self.workloads:
+                regional_client = self.regional_clients[workload.region]
+                response = regional_client.list_tags_for_resource(
+                    WorkloadArn=workload.arn
+                )["Tags"]
+                workload.tags = [response]
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+
+class Workload(BaseModel):
+    id: str
+    arn: str
+    name: str
+    region: str
+    lenses: list
+    improvement_status: str
+    risks: dict
+    tags: Optional[list] = []

prowler/providers/aws/services/wellarchitected/wellarchitected_workload_no_high_or_medium_risks/wellarchitected_workload_no_high_or_medium_risks.metadata.json

@@ -0,0 +1,30 @@
+{
+  "Provider": "aws",
+  "CheckID": "wellarchitected_workload_no_high_or_medium_risks",
+  "CheckTitle": "Check for medium and high risks identified in workloads defined in the AWS Well-Architected Tool.",
+  "CheckType": [],
+  "ServiceName": "wellarchitected",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:aws:wellarchitected:region:account-id:workload/workload-id",
+  "Severity": "medium",
+  "ResourceType": "Other",
+  "Description": "The Well-Architected Tool uses the AWS Well-Architected Framework to compare your cloud workloads against best practices across five architectural pillars: security, reliability, performance efficiency, operational excellence, and cost optimization",
+  "Risk": "A given workload can have medium and/or high risks that have been identified based on answers provided to the questions in the Well-Architected Tool. These issues are architectural and operational choices that are not aligned with the best practices from the Well-Architected Framework",
+  "RelatedUrl": "https://aws.amazon.com/architecture/well-architected/",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/aws/WellArchitected/findings.html",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "With the AWS Well-Architected Tool tool, you can analyze your workloads using a consistent process, pinpoint any medium or high-risk issues, and identify the next steps that must be taken for improvement",
+      "Url": "https://aws.amazon.com/architecture/well-architected/"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/aws/services/wellarchitected/wellarchitected_workload_no_high_or_medium_risks/wellarchitected_workload_no_high_or_medium_risks.py

@@ -0,0 +1,23 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.wellarchitected.wellarchitected_client import (
+    wellarchitected_client,
+)
+
+
+class wellarchitected_workload_no_high_or_medium_risks(Check):
+    def execute(self):
+        findings = []
+        for workload in wellarchitected_client.workloads:
+            report = Check_Report_AWS(self.metadata())
+            report.region = workload.region
+            report.resource_id = workload.id
+            report.resource_arn = workload.arn
+            report.resource_tags = workload.tags
+            report.status = "PASS"
+            report.status_extended = f"Well Architected workload {workload.name} does not contain high or medium risks"
+            if "HIGH" in workload.risks or "MEDIUM" in workload.risks:
+                report.status = "FAIL"
+                report.status_extended = f"Well Architected workload {workload.name} contains {workload.risks.get('HIGH',0)} high and {workload.risks.get('MEDIUM',0)} medium risks"
+
+            findings.append(report)
+        return findings

tests/providers/aws/services/wellarchitected/wellarchitected_service_test.py

@@ -0,0 +1,123 @@
+from unittest.mock import patch
+from uuid import uuid4
+
+import botocore
+from boto3 import session
+
+from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
+from prowler.providers.aws.services.wellarchitected.wellarchitected_service import (
+    WellArchitected,
+)
+
+AWS_ACCOUNT_NUMBER = "123456789012"
+AWS_REGION = "eu-west-1"
+
+
+workload_id = str(uuid4())
+
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call(self, operation_name, kwarg):
+    if operation_name == "ListWorkloads":
+        return {
+            "WorkloadSummaries": [
+                {
+                    "WorkloadId": workload_id,
+                    "WorkloadArn": f"arn:aws:wellarchitected:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:workload/{workload_id}",
+                    "WorkloadName": "test",
+                    "Owner": AWS_ACCOUNT_NUMBER,
+                    "UpdatedAt": "2023-06-07T15:40:24+02:00",
+                    "Lenses": ["wellarchitected", "serverless", "softwareasaservice"],
+                    "RiskCounts": {"UNANSWERED": 56, "NOT_APPLICABLE": 4, "HIGH": 10},
+                    "ImprovementStatus": "NOT_APPLICABLE",
+                },
+            ]
+        }
+    if operation_name == "ListTagsForResource":
+        return {
+            "Tags": {"Key": "test", "Value": "test"},
+        }
+    return make_api_call(self, operation_name, kwarg)
+
+
+def mock_generate_regional_clients(service, audit_info):
+    regional_client = audit_info.audit_session.client(service, region_name=AWS_REGION)
+    regional_client.region = AWS_REGION
+    return {AWS_REGION: regional_client}
+
+
+@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
+@patch(
+    "prowler.providers.aws.services.wellarchitected.wellarchitected_service.generate_regional_clients",
+    new=mock_generate_regional_clients,
+)
+class Test_WellArchitected_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=AWS_ACCOUNT_NUMBER,
+            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
+
+    # Test WellArchitected Service
+    def test_service(self):
+        audit_info = self.set_mocked_audit_info()
+        wellarchitected = WellArchitected(audit_info)
+        assert wellarchitected.service == "wellarchitected"
+
+    # Test WellArchitected client
+    def test_client(self):
+        audit_info = self.set_mocked_audit_info()
+        wellarchitected = WellArchitected(audit_info)
+        for reg_client in wellarchitected.regional_clients.values():
+            assert reg_client.__class__.__name__ == "WellArchitected"
+
+    # Test WellArchitected session
+    def test__get_session__(self):
+        audit_info = self.set_mocked_audit_info()
+        wellarchitected = WellArchitected(audit_info)
+        assert wellarchitected.session.__class__.__name__ == "Session"
+
+    # Test WellArchitected list workloads
+    def test__list_workloads__(self):
+        audit_info = self.set_mocked_audit_info()
+        wellarchitected = WellArchitected(audit_info)
+        assert len(wellarchitected.workloads) == 1
+        assert wellarchitected.workloads[0].id == workload_id
+        assert (
+            wellarchitected.workloads[0].arn
+            == f"arn:aws:wellarchitected:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:workload/{workload_id}"
+        )
+        assert wellarchitected.workloads[0].name == "test"
+        assert wellarchitected.workloads[0].region == AWS_REGION
+        assert wellarchitected.workloads[0].tags == [
+            {"Key": "test", "Value": "test"},
+        ]
+        assert wellarchitected.workloads[0].lenses == [
+            "wellarchitected",
+            "serverless",
+            "softwareasaservice",
+        ]
+        assert wellarchitected.workloads[0].improvement_status == "NOT_APPLICABLE"
+        assert wellarchitected.workloads[0].risks == {
+            "UNANSWERED": 56,
+            "NOT_APPLICABLE": 4,
+            "HIGH": 10,
+        }

tests/providers/aws/services/wellarchitected/wellarchitected_workload_no_high_or_medium_risks/wellarchitected_workload_no_high_or_medium_risks_test.py

@@ -0,0 +1,154 @@
+from re import search
+from unittest import mock
+from uuid import uuid4
+
+from prowler.providers.aws.services.wellarchitected.wellarchitected_service import (
+    Workload,
+)
+
+AWS_REGION = "eu-west-1"
+AWS_ACCOUNT_NUMBER = "123456789012"
+
+workload_id = str(uuid4())
+
+
+class Test_wellarchitected_workload_no_high_or_medium_risks:
+    def test_no_wellarchitected(self):
+        wellarchitected_client = mock.MagicMock
+        wellarchitected_client.workloads = []
+        with mock.patch(
+            "prowler.providers.aws.services.wellarchitected.wellarchitected_service.WellArchitected",
+            wellarchitected_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.wellarchitected.wellarchitected_client.wellarchitected_client",
+            wellarchitected_client,
+        ):
+            from prowler.providers.aws.services.wellarchitected.wellarchitected_workload_no_high_or_medium_risks.wellarchitected_workload_no_high_or_medium_risks import (
+                wellarchitected_workload_no_high_or_medium_risks,
+            )
+
+            check = wellarchitected_workload_no_high_or_medium_risks()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_wellarchitected_no_risks(self):
+        wellarchitected_client = mock.MagicMock
+        wellarchitected_client.workloads = []
+        wellarchitected_client.workloads.append(
+            Workload(
+                id=workload_id,
+                arn=f"arn:aws:wellarchitected:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:workload/{workload_id}",
+                name="test",
+                lenses=["wellarchitected", "serverless", "softwareasaservice"],
+                improvement_status="NOT_APPLICABLE",
+                risks={},
+                region=AWS_REGION,
+            )
+        )
+        with mock.patch(
+            "prowler.providers.aws.services.wellarchitected.wellarchitected_service.WellArchitected",
+            wellarchitected_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.wellarchitected.wellarchitected_client.wellarchitected_client",
+            wellarchitected_client,
+        ):
+            from prowler.providers.aws.services.wellarchitected.wellarchitected_workload_no_high_or_medium_risks.wellarchitected_workload_no_high_or_medium_risks import (
+                wellarchitected_workload_no_high_or_medium_risks,
+            )
+
+            check = wellarchitected_workload_no_high_or_medium_risks()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert search(
+                "does not contain high or medium risks", result[0].status_extended
+            )
+            assert result[0].resource_id == workload_id
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:wellarchitected:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:workload/{workload_id}"
+            )
+
+    def test_wellarchitected_no_high_medium_risks(self):
+        wellarchitected_client = mock.MagicMock
+        wellarchitected_client.workloads = []
+        wellarchitected_client.workloads.append(
+            Workload(
+                id=workload_id,
+                arn=f"arn:aws:wellarchitected:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:workload/{workload_id}",
+                name="test",
+                lenses=["wellarchitected", "serverless", "softwareasaservice"],
+                improvement_status="NOT_APPLICABLE",
+                risks={
+                    "UNANSWERED": 56,
+                    "NOT_APPLICABLE": 4,
+                },
+                region=AWS_REGION,
+            )
+        )
+        with mock.patch(
+            "prowler.providers.aws.services.wellarchitected.wellarchitected_service.WellArchitected",
+            wellarchitected_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.wellarchitected.wellarchitected_client.wellarchitected_client",
+            wellarchitected_client,
+        ):
+            from prowler.providers.aws.services.wellarchitected.wellarchitected_workload_no_high_or_medium_risks.wellarchitected_workload_no_high_or_medium_risks import (
+                wellarchitected_workload_no_high_or_medium_risks,
+            )
+
+            check = wellarchitected_workload_no_high_or_medium_risks()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert search(
+                "does not contain high or medium risks", result[0].status_extended
+            )
+            assert result[0].resource_id == workload_id
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:wellarchitected:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:workload/{workload_id}"
+            )
+
+    def test_wellarchitected_with_high_medium_risks(self):
+        wellarchitected_client = mock.MagicMock
+        wellarchitected_client.workloads = []
+        wellarchitected_client.workloads.append(
+            Workload(
+                id=workload_id,
+                arn=f"arn:aws:wellarchitected:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:workload/{workload_id}",
+                name="test",
+                lenses=["wellarchitected", "serverless", "softwareasaservice"],
+                improvement_status="NOT_APPLICABLE",
+                risks={
+                    "UNANSWERED": 56,
+                    "NOT_APPLICABLE": 4,
+                    "HIGH": 10,
+                    "MEDIUM": 20,
+                },
+                region=AWS_REGION,
+            )
+        )
+        with mock.patch(
+            "prowler.providers.aws.services.wellarchitected.wellarchitected_service.WellArchitected",
+            wellarchitected_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.wellarchitected.wellarchitected_client.wellarchitected_client",
+            wellarchitected_client,
+        ):
+            from prowler.providers.aws.services.wellarchitected.wellarchitected_workload_no_high_or_medium_risks.wellarchitected_workload_no_high_or_medium_risks import (
+                wellarchitected_workload_no_high_or_medium_risks,
+            )
+
+            check = wellarchitected_workload_no_high_or_medium_risks()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert search(
+                "contains 10 high and 20 medium risks", result[0].status_extended
+            )
+            assert result[0].resource_id == workload_id
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:wellarchitected:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:workload/{workload_id}"
+            )

util/update_aws_services_regions.py

@@ -54,6 +54,10 @@ regions_by_service["services"]["opensearch"] = regions_by_service["services"]["e
 regions_by_service["services"]["elbv2"] = regions_by_service["services"]["elb"]
 # wafv2 --> waf
 regions_by_service["services"]["wafv2"] = regions_by_service["services"]["waf"]
+# wellarchitected --> wellarchitectedtool
+regions_by_service["services"]["wellarchitected"] = regions_by_service["services"][
+    "wellarchitectedtool"
+]
 
 # Write to file
 parsed_matrix_regions_aws = f"{os.path.dirname(os.path.realpath(__name__))}/prowler/providers/aws/aws_regions_by_service.json"