chore(resource): PoC for the initial version

prowler/lib/check/models.py

@@ -415,6 +415,7 @@ class Check_Report:
     resource: dict
     resource_details: str
     resource_tags: list
+    resource_service: str
     muted: bool
 
     def __init__(self, metadata: Dict, resource: Any) -> None:
@@ -444,6 +445,7 @@ class Check_Report:
         self.resource_details = ""
         self.resource_tags = getattr(resource, "tags", []) if resource else []
         self.muted = False
+        self.resource_service = getattr(resource, "service", None)
 
 
 @dataclass

prowler/lib/outputs/finding.py

@@ -48,6 +48,7 @@ class Finding(BaseModel):
     resource_name: str
     resource_details: str
     resource_tags: dict = Field(default_factory=dict)
+    resource_service: str
     partition: Optional[str] = None
     region: str
     compliance: dict
@@ -133,6 +134,7 @@ class Finding(BaseModel):
         try:
             output_data["provider"] = provider.type
             output_data["resource_metadata"] = check_output.resource
+            output_data["resource_service"] = check_output.resource_service
 
             if provider.type == "aws":
                 output_data["account_uid"] = get_nested_attribute(

prowler/lib/resource/resource.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel
+
+
+class Resource(BaseModel):
+    """
+    Represents a generic resource.
+
+    Attributes:
+        service (str): The name of the service associated with the resource.
+    """
+
+    service: str

prowler/providers/aws/lib/resource/resource.py

@@ -0,0 +1,49 @@
+import re
+from typing import List, Optional
+
+from pydantic import Field, root_validator, validator
+
+from prowler.lib.resource.resource import Resource
+
+_ARN_PATTERN = re.compile(
+    r"^arn:(?P<partition>[^:]*):(?P<service>[^:]*):"
+    r"(?P<region>[^:]*):(?P<account>[^:]*):(?P<resource_id>.+)$"
+)
+
+
+class AWSResource(Resource):
+    """
+    Represents an AWS resource with its associated attributes.
+    Attributes:
+        arn (str): The Amazon Resource Name (ARN) uniquely identifying the resource.
+        id (str): The unique identifier of the resource.
+    """
+
+    arn: str
+    id: Optional[str] = None
+    name: Optional[str] = None
+    tags: List[str] = Field(default_factory=list)
+    region: str
+
+    @root_validator(pre=True)
+    def populate_from_arn(cls, values):
+        arn = values.get("arn")
+        if arn:
+            match = _ARN_PATTERN.match(arn)
+            if not match:
+                raise ValueError(f"Invalid ARN: {arn!r}")
+            # Only overwrite if not provided explicitly
+            values.setdefault("service", match.group("service"))
+            values.setdefault("region", match.group("region"))
+            # Extract the last part after the '/' if it exists
+            resource_id = match.group("resource_id")
+            values.setdefault(
+                "id", resource_id.split("/")[-1] if "/" in resource_id else resource_id
+            )
+        return values
+
+    @validator("region")
+    def region_must_be_valid(cls, v):
+        if not v:
+            raise ValueError("region cannot be empty")
+        return v

prowler/providers/aws/services/accessanalyzer/accessanalyzer_service.py

@@ -5,6 +5,7 @@ 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.lib.resource.resource import AWSResource
 from prowler.providers.aws.lib.service.service import AWSService
 
 
@@ -127,11 +128,8 @@ class Finding(BaseModel):
     status: str = ""
 
 
-class Analyzer(BaseModel):
-    arn: str
-    name: str
+class Analyzer(AWSResource):
     status: str
     findings: list[Finding] = []
     tags: Optional[list] = []
     type: str
-    region: str

tests/lib/resource/resource_test.py

@@ -0,0 +1,8 @@
+from prowler.lib.resource.resource import Resource
+
+
+class ResourceTest:
+    def test_resource(self):
+        resource = Resource(service="test_service")
+        assert resource.service == "test_service"
+        assert isinstance(resource, Resource)

tests/providers/aws/lib/resource/resource_test.py

@@ -0,0 +1,97 @@
+import pytest
+from pydantic import ValidationError
+
+# Adjust the import path as needed for your project layout
+from prowler.lib.resource.resource import Resource
+from prowler.providers.aws.lib.resource.resource import _ARN_PATTERN, AWSResource
+
+# Sample ARNs for testing
+VALID_S3_ARN = "arn:aws:s3:eu-central-1:123456789012:bucket/my-bucket"
+VALID_LAMBDA_ARN = "arn:aws:lambda:us-east-1:123456789012:function:my-function"
+INVALID_ARN = "arn:aws::123456:badformat"
+
+
+class TestAWSResource:
+    def test_arn_regex_matches(self):
+        """Ensure the regex itself captures the expected groups."""
+        m = _ARN_PATTERN.match(VALID_S3_ARN)
+        assert m
+        assert m.group("service") == "s3"
+        assert m.group("region") == "eu-central-1"
+        assert m.group("resource_id") == "bucket/my-bucket"
+
+    @pytest.mark.parametrize(
+        "arn,expected_service,expected_region,expected_id",
+        [
+            (VALID_S3_ARN, "s3", "eu-central-1", "bucket/my-bucket"),
+            (VALID_LAMBDA_ARN, "lambda", "us-east-1", "function:my-function"),
+        ],
+    )
+    def test_populate_from_arn_defaults(
+        self, arn, expected_service, expected_region, expected_id
+    ):
+        """Fields service, region, and id should be auto‐filled from the ARN."""
+        resource = AWSResource(arn=arn)
+        assert isinstance(resource, Resource)
+        assert resource.service == expected_service
+        assert resource.region == expected_region
+        assert resource.id == expected_id
+
+    def test_override_service_and_region_and_id(self):
+        """Explicitly provided values should override ARN‐derived defaults."""
+        resource = AWSResource(
+            arn=VALID_S3_ARN,
+            service="override-service",
+            region="override-region",
+            id="override-id",
+        )
+        assert resource.service == "override-service"
+        assert resource.region == "override-region"
+        assert resource.id == "override-id"
+
+    def test_missing_arn_raises_error(self):
+        """If no ARN is given, an error is raised."""
+
+        with pytest.raises(ValidationError) as excinfo:
+            AWSResource(
+                service="manual-service", region="manual-region", id="manual-id"
+            )
+        assert (
+            "1 validation error for AWSResource\narn\n  field required (type=value_error.missing)"
+            in str(excinfo.value)
+        )
+
+    def test_empty_region_raises_error(self):
+        """An explicitly empty region (or an ARN with empty region) should error."""
+        # Empty string region
+        with pytest.raises(ValidationError) as excinfo:
+            AWSResource(arn=None, service="s3", region="")
+        assert "region cannot be empty" in str(excinfo.value)
+
+        # ARN with missing region segment
+        bad_arn = "arn:aws:s3::123456789012:bucket/x"
+        with pytest.raises(ValidationError) as excinfo2:
+            AWSResource(arn=bad_arn)
+        # root_validator will run first but region validator will catch empty region
+        assert "region cannot be empty" in str(excinfo2.value)
+
+    def test_invalid_arn_raises_value_error(self):
+        """Malformed ARN should trigger ValueError from the root_validator."""
+        with pytest.raises(ValidationError) as excinfo:
+            AWSResource(arn=INVALID_ARN)
+        # unwrap to see the inner ValueError message
+        err = excinfo.value.errors()[0]
+        assert "Invalid ARN" in err["msg"]
+
+    def test_tags_default_and_mutability(self):
+        """Ensure tags default to an empty list and are independent per instance."""
+        resource_1 = AWSResource(arn=VALID_S3_ARN)
+        resource_2 = AWSResource(arn=VALID_S3_ARN)
+        assert resource_1.tags == [] and resource_2.tags == []
+        resource_1.tags.append("tag1")
+        assert resource_2.tags == []
+
+    def test_name_field_passthrough(self):
+        """Name should be accepted and stored unchanged."""
+        res = AWSResource(arn=VALID_S3_ARN, name="my-name")
+        assert res.name == "my-name"

tests/providers/aws/services/accessanalyzer/accessanalyzer_enabled/accessanalyzer_enabled_test.py

@@ -72,10 +72,16 @@ class Test_accessanalyzer_enabled:
                 result[0].status_extended
                 == f"IAM Access Analyzer in account {AWS_ACCOUNT_NUMBER} is not enabled."
             )
+            # Review this values too
             assert result[0].resource_id == "123456789012"
+            # Review the following line, maybe it should be:
+            # "arn:aws:access-analyzer::123456789012:root"
+            # "arn:aws:access-analyzer::123456789012:account"
             assert result[0].resource_arn == "arn:aws:iam::123456789012:root"
             assert result[0].region == AWS_REGION_1
             assert result[0].resource_tags == []
+            assert result[0].resource_service == "access-analyzer"
+            # assert result[0].resource == ""
 
     def test_one_analyzer_not_available_muted(self):
         # Include analyzers to check
@@ -125,6 +131,8 @@ class Test_accessanalyzer_enabled:
             assert result[0].resource_arn == "arn:aws:iam::123456789012:root"
             assert result[0].region == AWS_REGION_1
             assert result[0].resource_tags == []
+            assert result[0].resource_service == "access-analyzer"
+            # assert result[0].resource == ""
 
     def test_two_analyzers(self):
         accessanalyzer_client = mock.MagicMock
@@ -196,6 +204,8 @@ class Test_accessanalyzer_enabled:
             assert result[1].resource_arn == ACCESS_ANALYZER_ARN
             assert result[1].resource_tags == []
             assert result[1].region == AWS_REGION_2
+            assert result[0].resource_service == "access-analyzer"
+            # assert result[0].resource == ""
 
     def test_one_active_analyzer(self):
         accessanalyzer_client = mock.MagicMock
@@ -233,3 +243,5 @@ class Test_accessanalyzer_enabled:
             assert result[0].resource_arn == ACCESS_ANALYZER_ARN
             assert result[0].resource_tags == []
             assert result[0].region == AWS_REGION_2
+            assert result[0].resource_service == "access-analyzer"
+            # assert result[0].resource == ""

tests/providers/aws/services/accessanalyzer/accessanalyzer_service_test.py

@@ -14,6 +14,8 @@ from tests.providers.aws.utils import (
 # Mocking Access Analyzer Calls
 make_api_call = botocore.client.BaseClient._make_api_call
 
+ANALYZER_ARN = "arn:aws:access-analyzer:eu-west-1:111122223333:analyzer/TestAnalyzer"
+
 
 def mock_make_api_call(self, operation_name, kwarg):
     """
@@ -27,8 +29,8 @@ def mock_make_api_call(self, operation_name, kwarg):
         return {
             "analyzers": [
                 {
-                    "arn": "ARN",
-                    "name": "Test Analyzer",
+                    "arn": ANALYZER_ARN,
+                    "name": "TestAnalyzer",
                     "status": "ACTIVE",
                     "findings": 0,
                     "tags": {"test": "test"},
@@ -98,8 +100,9 @@ class Test_AccessAnalyzer_Service:
             set_mocked_aws_provider([AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1])
         )
         assert len(access_analyzer.analyzers) == 1
-        assert access_analyzer.analyzers[0].arn == "ARN"
-        assert access_analyzer.analyzers[0].name == "Test Analyzer"
+        assert access_analyzer.analyzers[0].arn == ANALYZER_ARN
+        assert access_analyzer.analyzers[0].name == "TestAnalyzer"
+        assert access_analyzer.analyzers[0].service == "access-analyzer"
         assert access_analyzer.analyzers[0].status == "ACTIVE"
         assert access_analyzer.analyzers[0].tags == [{"test": "test"}]
         assert access_analyzer.analyzers[0].type == "ACCOUNT"