[aws][feat] Add additional policies to the cloudwatch (#2216)

Author: 1101-1

plugins/aws/fix_plugin_aws/resource/apigateway.py

@@ -11,7 +11,7 @@
 
 from fixlib.baseresources import EdgeType, ModelReference
 from fixlib.graph import Graph
-from fixlib.json_bender import Bender, S, Bend, bend
+from fixlib.json_bender import Bender, S, Bend, ParseJson, Sorted, bend
 from fixlib.types import Json
 
 service_name = "apigateway"
@@ -496,7 +496,7 @@ class AwsApiGatewayRestApi(ApiGatewayTaggable, AwsResource):
         "api_minimum_compression_size": S("minimumCompressionSize"),
         "api_key_source": S("apiKeySource"),
         "api_endpoint_configuration": S("endpointConfiguration") >> Bend(AwsApiGatewayEndpointConfiguration.mapping),
-        "api_policy": S("policy"),
+        "api_policy": S("policy") >> ParseJson() >> Sorted(sort_list=True),
         "api_disable_execute_api_endpoint": S("disableExecuteApiEndpoint"),
     }
     description: Optional[str] = field(default=None)
@@ -506,7 +506,7 @@ class AwsApiGatewayRestApi(ApiGatewayTaggable, AwsResource):
     api_minimum_compression_size: Optional[int] = field(default=None)
     api_key_source: Optional[str] = field(default=None)
     api_endpoint_configuration: Optional[AwsApiGatewayEndpointConfiguration] = field(default=None)
-    api_policy: Optional[str] = field(default=None)
+    api_policy: Optional[Json] = field(default=None)
     api_disable_execute_api_endpoint: Optional[bool] = field(default=None)
 
     @classmethod

plugins/aws/fix_plugin_aws/resource/bedrock.py

@@ -764,7 +764,10 @@ class AwsBedrockEvaluationJob(BedrockTaggable, BaseAIJob, AwsResource):
         "successors": {"default": [AwsS3Bucket.kind, AwsKmsKey.kind]},
     }
     api_spec: ClassVar[AwsApiSpec] = AwsApiSpec(
-        "bedrock", "list-evaluation-jobs", "jobSummaries", expected_errors=["AccessDeniedException"]
+        "bedrock",
+        "list-evaluation-jobs",
+        "jobSummaries",
+        expected_errors=["AccessDeniedException"],
     )
     mapping: ClassVar[Dict[str, Bender]] = {
         "id": S("jobArn"),

plugins/aws/fix_plugin_aws/resource/cloudwatch.py

@@ -4,6 +4,7 @@
 from datetime import datetime, timedelta
 from typing import Callable, ClassVar, Dict, List, Optional, Type, Tuple, TypeVar, Any, Union
 from concurrent.futures import as_completed
+from json import loads as json_loads
 
 from attr import define, field, frozen
 
@@ -14,7 +15,7 @@
 from fixlib.baseresources import MetricName, MetricUnit, ModelReference, BaseResource, StatName
 from fixlib.durations import duration_str
 from fixlib.graph import Graph
-from fixlib.json import from_json
+from fixlib.json import from_json, sort_json
 from fixlib.json_bender import S, Bend, Bender, ForallBend, bend, F, SecondsFromEpochToDatetime
 from fixlib.types import Json
 from fixlib.utils import chunks
@@ -362,11 +363,75 @@ class AwsCloudwatchLogGroup(LogsTaggable, AwsResource):
     group_metric_filter_count: Optional[int] = field(default=None, metadata=dict(ignore_history=True))
     group_stored_bytes: Optional[int] = field(default=None, metadata=dict(ignore_history=True))
     group_data_protection_status: Optional[str] = field(default=None)
+    group_policy: Optional[Json] = field(default=None)
 
     def connect_in_graph(self, builder: GraphBuilder, source: Json) -> None:
         if kms_key_id := source.get("kmsKeyId"):
             builder.dependant_node(self, clazz=AwsKmsKey, id=AwsKmsKey.normalise_id(kms_key_id))
 
+    @classmethod
+    def collect(cls: Type[AwsResource], json: List[Json], builder: GraphBuilder) -> None:
+        def add_log_group_policy(group: AwsCloudwatchLogGroup) -> None:
+            def is_arn_match(resource_arn: str, target_arn: str) -> bool:
+                if resource_arn == target_arn:
+                    return True
+                if resource_arn.endswith(":*"):
+                    return target_arn.startswith(resource_arn[:-1])
+                return False
+
+            def parse_resource_arn(resource: Any) -> List[str]:
+                if isinstance(resource, str):
+                    return [resource.split(":log-stream:")[0]]
+                elif isinstance(resource, list):
+                    return [arn.split(":log-stream:")[0] for arn in resource]
+                return []
+
+            def process_log_group_policies(raw_policies: List[Dict[str, Any]], target_group_arn: str) -> Dict[str, Any]:
+                associated_policies = {}
+                for policy in raw_policies:
+                    policy_name = policy.get("policyName", "Unknown")
+                    policy_document = json_loads(policy.get("policyDocument", "{}"))
+                    policy_statements = policy_document.get("Statement", [])
+
+                    if not isinstance(policy_statements, list):
+                        policy_statements = [policy_statements]
+
+                    for statement in policy_statements:
+                        statement_resources = statement.get("Resource")
+                        log_group_arns = parse_resource_arn(statement_resources)
+
+                        if any(is_arn_match(arn, target_group_arn) for arn in log_group_arns):
+                            # If a match is found, associate the policy and move to the next policy
+                            associated_policies[policy_name] = policy
+                            break
+
+                return associated_policies
+
+            with builder.suppress(f"{service_name}.describe-resource-policies"):
+                if raw_policies := builder.client.list(
+                    "logs",
+                    "describe-resource-policies",
+                    "resourcePolicies",
+                    expected_errors=["ResourceNotFoundException"],
+                ):
+                    if not group.arn:
+                        return
+                    associated_policies = process_log_group_policies(raw_policies, group.arn)
+                    if associated_policies:
+                        group.group_policy = sort_json(associated_policies, sort_list=True)
+
+        for js in json:
+            if instance := cls.from_api(js, builder):
+                builder.add_node(instance, js)
+                builder.submit_work(service_name, add_log_group_policy, instance)
+
+    @classmethod
+    def called_collect_apis(cls) -> List[AwsApiSpec]:
+        return [
+            cls.api_spec,
+            AwsApiSpec(service_name, "describe-resource-policies"),
+        ]
+
     @classmethod
     def called_mutator_apis(cls) -> List[AwsApiSpec]:
         return super().called_mutator_apis() + [AwsApiSpec("logs", "delete-log-group")]

plugins/aws/fix_plugin_aws/resource/s3.py

@@ -198,6 +198,7 @@ def called_collect_apis(cls) -> List[AwsApiSpec]:
             AwsApiSpec(service_name, "get-bucket-acl"),
             AwsApiSpec(service_name, "get-bucket-logging"),
             AwsApiSpec(service_name, "get-bucket-location"),
+            AwsApiSpec(service_name, "get-bucket-lifecycle-configuration"),
         ]
 
     @classmethod

plugins/aws/test/resources/apigateway_test.py

@@ -8,7 +8,7 @@
 
 
 def test_rest_apis() -> None:
-    api, builder = round_trip_for(AwsApiGatewayRestApi)
+    api, builder = round_trip_for(AwsApiGatewayRestApi, "api_policy")
     assert len(builder.resources_of(AwsApiGatewayRestApi)) == 1
     assert len(api.tags) == 1
     assert api.arn == "arn:aws:apigateway:eu-central-1::/restapis/2lsd9i45ub"
@@ -22,7 +22,7 @@ def test_rest_apis() -> None:
 
 
 def test_api_tagging() -> None:
-    api, builder = round_trip_for(AwsApiGatewayRestApi)
+    api, builder = round_trip_for(AwsApiGatewayRestApi, "api_policy")
 
     def validate_update_args(**kwargs: Any) -> None:
         assert kwargs["action"] == "tag-resource"
@@ -42,7 +42,7 @@ def validate_delete_args(**kwargs: Any) -> None:
 
 
 def test_delete_api() -> None:
-    api, _ = round_trip_for(AwsApiGatewayRestApi)
+    api, _ = round_trip_for(AwsApiGatewayRestApi, "api_policy")
 
     def validate_delete_args(**kwargs: Any) -> Any:
         assert kwargs["action"] == "delete-rest-api"

plugins/aws/test/resources/cloudwatch_test.py

@@ -26,13 +26,12 @@ def test_alarms() -> None:
 
 
 def test_log_groups() -> None:
-    round_trip_for(AwsCloudwatchLogGroup)
+    round_trip_for(AwsCloudwatchLogGroup, "group_policy")
 
 
 def test_metrics_filter() -> None:
-    first, aws_builder = round_trip_for(AwsCloudwatchMetricFilter)
+    first, aws_builder = round_trip_for(AwsCloudwatchMetricFilter, collect_also=[AwsCloudwatchLogGroup])
     # test connection to log group
-    AwsCloudwatchLogGroup.collect_resources(aws_builder)
     first.connect_in_graph(aws_builder, aws_builder.graph.nodes(data=True)[first]["source"])
     assert len(aws_builder.edges_of(AwsCloudwatchLogGroup, AwsCloudwatchMetricFilter)) == 1