[aws][feat]: Add more policies to collect (#2202)

Co-authored-by: Matthias Veit <aquamatthias@users.noreply.github.com>

Author: 1101-1

plugins/aws/fix_plugin_aws/resource/backup.py

@@ -1,6 +1,7 @@
 import logging
 from datetime import datetime
 from typing import Any, ClassVar, Dict, Optional, List, Type
+from json import loads as json_loads
 
 from attrs import define, field
 
@@ -18,6 +19,7 @@
 from fixlib.graph import Graph
 from fixlib.json_bender import F, Bender, S, ForallBend, Bend
 from fixlib.types import Json
+from fixlib.json import sort_json
 
 log = logging.getLogger("fix.plugins.aws")
 service_name = "backup"
@@ -345,13 +347,15 @@ class AwsBackupVault(BackupResourceTaggable, AwsResource):
     min_retention_days: Optional[int] = field(default=None, metadata={"description": "The Backup Vault Lock setting that specifies the minimum retention period that the vault retains its recovery points. If this parameter is not specified, Vault Lock does not enforce a minimum retention period. If specified, any backup or copy job to the vault must have a lifecycle policy with a retention period equal to or longer than the minimum retention period. If the job's retention period is shorter than that minimum retention period, then the vault fails the backup or copy job, and you should either modify your lifecycle settings or use a different vault. Recovery points already stored in the vault prior to Vault Lock are not affected."})  # fmt: skip
     max_retention_days: Optional[int] = field(default=None, metadata={"description": "The Backup Vault Lock setting that specifies the maximum retention period that the vault retains its recovery points. If this parameter is not specified, Vault Lock does not enforce a maximum retention period on the recovery points in the vault (allowing indefinite storage). If specified, any backup or copy job to the vault must have a lifecycle policy with a retention period equal to or shorter than the maximum retention period. If the job's retention period is longer than that maximum retention period, then the vault fails the backup or copy job, and you should either modify your lifecycle settings or use a different vault. Recovery points already stored in the vault prior to Vault Lock are not affected."})  # fmt: skip
     lock_date: Optional[datetime] = field(default=None, metadata={"description": "The date and time when Backup Vault Lock configuration becomes immutable, meaning it cannot be changed or deleted. If you applied Vault Lock to your vault without specifying a lock date, you can change your Vault Lock settings, or delete Vault Lock from the vault entirely, at any time. This value is in Unix format, Coordinated Universal Time (UTC), and accurate to milliseconds. For example, the value 1516925490.087 represents Friday, January 26, 2018 12:11:30.087 AM."})  # fmt: skip
+    vault_policy: Optional[Json] = field(default=None)
 
     @classmethod
     def called_collect_apis(cls) -> List[AwsApiSpec]:
         return [
             cls.api_spec,
             AwsApiSpec(service_name, "list-tags"),
             AwsApiSpec(service_name, "list-recovery-points-by-backup-vault"),
+            AwsApiSpec(service_name, "get-backup-vault-access-policy"),
         ]
 
     @classmethod
@@ -394,11 +398,23 @@ def add_tags(backup_plan: AwsBackupVault) -> None:
                 for tag in tags:
                     backup_plan.tags.update(tag)
 
+        def add_vault_policy(vault: AwsBackupVault) -> None:
+            with builder.suppress(f"{service_name}.get-backup-vault-access-policy"):
+                if raw_policy := builder.client.get(
+                    service_name,
+                    "get-backup-vault-access-policy",
+                    "Policy",
+                    BackupVaultName=vault.name,
+                    expected_errors=["ResourceNotFoundException"],
+                ):
+                    vault.vault_policy = sort_json(json_loads(raw_policy), sort_list=True)  # type: ignore
+
         for js in json:
             if instance := cls.from_api(js, builder):
                 builder.add_node(instance, js)
                 builder.submit_work(service_name, collect_recovery_points, instance)
                 builder.submit_work(service_name, add_tags, instance)
+                builder.submit_work(service_name, add_vault_policy, instance)
 
 
 @define(eq=False, slots=False)

plugins/aws/fix_plugin_aws/resource/dynamodb.py

@@ -1,6 +1,7 @@
 from datetime import datetime
 from typing import ClassVar, Dict, List, Optional, Type, Any
 from attrs import define, field
+from json import loads as json_loads
 
 from fix_plugin_aws.aws_client import AwsClient
 from fix_plugin_aws.resource.base import AwsApiSpec, AwsResource, GraphBuilder, parse_json
@@ -11,6 +12,7 @@
 from fixlib.graph import Graph
 from fixlib.json_bender import S, Bend, Bender, ForallBend, bend
 from fixlib.types import Json
+from fixlib.json import sort_json
 
 service_name = "dynamodb"
 
@@ -418,6 +420,7 @@ class AwsDynamoDbTable(DynamoDbTaggable, AwsResource):
     dynamodb_archival_summary: Optional[AwsDynamoDbArchivalSummary] = field(default=None)
     dynamodb_table_class_summary: Optional[AwsDynamoDbTableClassSummary] = field(default=None)
     dynamodb_continuous_backup: Optional[AwsDynamoDbContinuousBackup] = field(default=None)
+    dynamodb_policy: Optional[Json] = field(default=None)
 
     @classmethod
     def called_collect_apis(cls) -> List[AwsApiSpec]:
@@ -426,10 +429,20 @@ def called_collect_apis(cls) -> List[AwsApiSpec]:
             AwsApiSpec(service_name, "describe-table"),
             AwsApiSpec(service_name, "list-tags-of-resource"),
             AwsApiSpec(service_name, "describe-continuous-backups"),
+            AwsApiSpec(service_name, "get-resource-policy"),
         ]
 
     @classmethod
     def collect(cls: Type[AwsResource], json: List[Json], builder: GraphBuilder) -> None:
+        def add_dynamodb_policy(table: AwsDynamoDbTable) -> None:
+            with builder.suppress(f"{service_name}.get-bucket-policy"):
+                if raw_policy := builder.client.get(
+                    service_name,
+                    "get-resource-policy",
+                    "Policy",
+                    ResourceArn=table.arn,
+                ):
+                    table.dynamodb_policy = sort_json(json_loads(raw_policy), sort_list=True)  # type: ignore
 
         def add_backup_description(table: AwsDynamoDbTable) -> None:
             if continuous_backup := builder.client.get(
@@ -446,6 +459,7 @@ def add_instance(table: str) -> None:
                     builder.add_node(instance, table_description)
                     builder.submit_work(service_name, add_tags, instance)
                     builder.submit_work(service_name, add_backup_description, instance)
+                    builder.submit_work(service_name, add_dynamodb_policy, instance)
 
         def add_tags(table: AwsDynamoDbTable) -> None:
             tags = builder.client.list(service_name, "list-tags-of-resource", "Tags", ResourceArn=table.arn)
@@ -515,17 +529,30 @@ class AwsDynamoDbGlobalTable(DynamoDbTaggable, AwsResource):
     arn: Optional[str] = field(default=None)
     dynamodb_replication_group: List[AwsDynamoDbReplicaDescription] = field(factory=list)
     dynamodb_global_table_status: Optional[str] = field(default=None)
+    dynamodb_policy: Optional[Json] = field(default=None)
 
     @classmethod
     def called_collect_apis(cls) -> List[AwsApiSpec]:
         return [
             cls.api_spec,
             AwsApiSpec(service_name, "describe-global-table"),
             AwsApiSpec(service_name, "list-tags-of-resource"),
+            AwsApiSpec(service_name, "get-resource-policy"),
         ]
 
     @classmethod
     def collect(cls: Type[AwsResource], json: List[Json], builder: GraphBuilder) -> None:
+        def add_dynamodb_policy(table: AwsDynamoDbGlobalTable) -> None:
+            with builder.suppress(f"{service_name}.get-bucket-policy"):
+                if raw_policy := builder.client.get(
+                    service_name,
+                    "get-resource-policy",
+                    "Policy",
+                    ResourceArn=table.arn,
+                    expected_errors=["PolicyNotFoundException"],
+                ):
+                    table.dynamodb_policy = sort_json(json_loads(raw_policy), sort_list=True)  # type: ignore
+
         def add_instance(table: Dict[str, str]) -> None:
             table_description = builder.client.get(
                 service_name,
@@ -537,6 +564,7 @@ def add_instance(table: Dict[str, str]) -> None:
                 if instance := cls.from_api(table_description, builder):
                     builder.add_node(instance, table_description)
                     builder.submit_work(service_name, add_tags, instance)
+                    builder.submit_work(service_name, add_dynamodb_policy, instance)
 
         def add_tags(table: AwsDynamoDbGlobalTable) -> None:
             tags = builder.client.list(service_name, "list-tags-of-resource", "Tags", ResourceArn=table.arn)

plugins/aws/fix_plugin_aws/resource/ecr.py

@@ -1,6 +1,7 @@
 import json
 import logging
 from typing import ClassVar, Dict, Optional, List, Type, Any
+from json import loads as json_loads
 
 from attrs import define, field
 from boto3.exceptions import Boto3Error
@@ -53,9 +54,21 @@ class AwsEcrRepository(AwsResource):
     encryption_configuration: Optional[AwsEcrEncryptionConfiguration] = field(default=None, metadata={"description": "The encryption configuration for the repository. This determines how the contents of your repository are encrypted at rest."})  # fmt: skip
     repository_visibility: Optional[str] = field(default=None, metadata={"description": "The repository is either public or private."})  # fmt: skip
     lifecycle_policy: Optional[Json] = field(default=None, metadata={"description": "The repository lifecycle policy."})  # fmt: skip
+    repository_policy: Optional[Json] = field(default=None, metadata={"description": "The repository policy."})  # fmt: skip
 
     @classmethod
     def collect_resources(cls, builder: GraphBuilder) -> None:
+        def add_repository_policy(repository: AwsEcrRepository) -> None:
+            with builder.suppress(f"{service_name}.get-repository-policy"):
+                if raw_policy := builder.client.get(
+                    service_name,
+                    "get-repository-policy",
+                    "policyText",
+                    repositoryName=repository.name,
+                    expected_errors=["RepositoryPolicyNotFoundException", "RepositoryNotFoundException"],
+                ):
+                    repository.repository_policy = sort_json(json_loads(raw_policy), sort_list=True)  # type: ignore
+
         def fetch_lifecycle_policy(repository: AwsEcrRepository) -> None:
             with builder.suppress(f"{service_name}.get-lifecycle-policy"):
                 if policy := builder.client.get(
@@ -79,8 +92,9 @@ def collect(visibility: str, spec: AwsApiSpec) -> None:
                 for js in items:
                     if instance := cls.from_api(js, builder):
                         instance.repository_visibility = visibility
-                        builder.submit_work(service_name, fetch_lifecycle_policy, instance)
                         builder.add_node(instance, js)
+                        builder.submit_work(service_name, fetch_lifecycle_policy, instance)
+                        builder.submit_work(service_name, add_repository_policy, instance)
             except Boto3Error as e:
                 msg = f"Error while collecting {cls.__name__} in region {builder.region.name}: {e}"
                 builder.core_feedback.error(msg, log)
@@ -98,7 +112,12 @@ def collect(visibility: str, spec: AwsApiSpec) -> None:
 
     @classmethod
     def called_collect_apis(cls) -> List[AwsApiSpec]:
-        return [cls.api_spec, cls.public_spec, AwsApiSpec("ecr", "get-lifecycle-policy", None)]
+        return [
+            cls.api_spec,
+            cls.public_spec,
+            AwsApiSpec(service_name, "get-lifecycle-policy"),
+            AwsApiSpec(service_name, "get-repository-policy"),
+        ]
 
 
 # @define(eq=False, slots=False)

plugins/aws/fix_plugin_aws/resource/efs.py

@@ -149,10 +149,11 @@ def fetch_file_system_policy(fs: AwsEfsFileSystem) -> None:
                 if policy := builder.client.get(
                     service_name,
                     "describe-file-system-policy",
+                    "Policy",
                     FileSystemId=fs.id,
-                    expected_errors=["PolicyNotFound"],
+                    expected_errors=["PolicyNotFound", "FileSystemNotFound"],
                 ):
-                    fs.file_system_policy = sort_json(json.loads(policy["Policy"]), sort_list=True)
+                    fs.file_system_policy = sort_json(json.loads(policy), sort_list=True)  # type: ignore
 
         for js in js_list:
             if instance := cls.from_api(js, builder):

plugins/aws/fix_plugin_aws/resource/kinesis.py

@@ -1,4 +1,5 @@
 from typing import ClassVar, Dict, Optional, List, Any
+from json import loads as json_loads
 
 from attrs import define, field
 
@@ -11,6 +12,7 @@
 from fixlib.graph import Graph
 from fixlib.json_bender import Bender, S, Bend, bend, ForallBend
 from fixlib.types import Json
+from fixlib.json import sort_json
 from typing import Type
 
 service_name = "kinesis"
@@ -132,17 +134,30 @@ class AwsKinesisStream(AwsResource):
     kinesis_enhanced_monitoring: List[AwsKinesisEnhancedMetrics] = field(factory=list)
     kinesis_encryption_type: Optional[str] = field(default=None)
     kinesis_key_id: Optional[str] = field(default=None)
+    kinesis_policy: Optional[Json] = field(default=None)
 
     @classmethod
     def called_collect_apis(cls) -> List[AwsApiSpec]:
         return [
             cls.api_spec,
             AwsApiSpec(service_name, "describe-stream"),
             AwsApiSpec(service_name, "list-tags-for-stream"),
+            AwsApiSpec(service_name, "get-resource-policy"),
         ]
 
     @classmethod
     def collect(cls: Type[AwsResource], json: List[Json], builder: GraphBuilder) -> None:
+        def add_kinesis_policy(kinesis: AwsKinesisStream) -> None:
+            with builder.suppress(f"{service_name}.get-resource-policy"):
+                if raw_policy := builder.client.get(
+                    service_name,
+                    "get-resource-policy",
+                    "Policy",
+                    ResourceARN=kinesis.arn,
+                    expected_errors=["AccessDeniedException"],
+                ):
+                    kinesis.kinesis_policy = sort_json(json_loads(raw_policy), sort_list=True)  # type: ignore
+
         def add_instance(stream_name: str) -> None:
             # this call is paginated and will return a list
             stream_descriptions = builder.client.list(
@@ -156,6 +171,7 @@ def add_instance(stream_name: str) -> None:
                 if stream := AwsKinesisStream.from_api(js, builder):
                     builder.add_node(stream, js)
                     builder.submit_work(service_name, add_tags, stream)
+                    builder.submit_work(service_name, add_kinesis_policy, stream)
 
         def add_tags(stream: AwsKinesisStream) -> None:
             tags = builder.client.list(stream.api_spec.service, "list-tags-for-stream", "Tags", StreamName=stream.name)

plugins/aws/fix_plugin_aws/resource/sns.py

@@ -2,7 +2,6 @@
 from typing import ClassVar, Dict, List, Optional, Type, Any
 from attrs import define, field
 
-
 from fix_plugin_aws.aws_client import AwsClient
 from fix_plugin_aws.resource.base import AwsApiSpec, AwsResource, GraphBuilder
 from fix_plugin_aws.resource.cloudwatch import AwsCloudwatchQuery, normalizer_factory

plugins/aws/test/resources/backup_test.py

@@ -23,7 +23,7 @@ def test_backup_plans() -> None:
 
 
 def test_backup_vaults() -> None:
-    round_trip_for(AwsBackupVault)
+    round_trip_for(AwsBackupVault, "vault_policy")
 
 
 def test_backup_recovery_points() -> None:

plugins/aws/test/resources/dynamodb_test.py

@@ -7,13 +7,13 @@
 
 
 def test_tables() -> None:
-    first, builder = round_trip_for(AwsDynamoDbTable)
+    first, builder = round_trip_for(AwsDynamoDbTable, "dynamodb_policy")
     assert len(builder.resources_of(AwsDynamoDbTable)) == 1
     assert len(first.tags) == 1
 
 
 def test_tagging_tables() -> None:
-    table, _ = round_trip_for(AwsDynamoDbTable)
+    table, _ = round_trip_for(AwsDynamoDbTable, "dynamodb_policy")
 
     def validate_update_args(**kwargs: Any) -> Any:
         if kwargs["action"] == "list-tags-of-resource":
@@ -37,7 +37,7 @@ def validate_delete_args(**kwargs: Any) -> Any:
 
 
 def test_delete_tables() -> None:
-    table, _ = round_trip_for(AwsDynamoDbTable)
+    table, _ = round_trip_for(AwsDynamoDbTable, "dynamodb_policy")
 
     def validate_delete_args(**kwargs: Any) -> Any:
         assert kwargs["action"] == "delete-table"
@@ -48,12 +48,12 @@ def validate_delete_args(**kwargs: Any) -> Any:
 
 
 def test_global_tables() -> None:
-    first, builder = round_trip_for(AwsDynamoDbGlobalTable)
+    first, builder = round_trip_for(AwsDynamoDbGlobalTable, "dynamodb_policy")
     assert len(builder.resources_of(AwsDynamoDbGlobalTable)) == 1
 
 
 def test_tagging_global_tables() -> None:
-    table, _ = round_trip_for(AwsDynamoDbGlobalTable)
+    table, _ = round_trip_for(AwsDynamoDbGlobalTable, "dynamodb_policy")
 
     def validate_update_args(**kwargs: Any) -> Any:
         if kwargs["action"] == "list-tags-of-resource":
@@ -77,7 +77,7 @@ def validate_delete_args(**kwargs: Any) -> Any:
 
 
 def test_delete_global_tables() -> None:
-    table, _ = round_trip_for(AwsDynamoDbGlobalTable)
+    table, _ = round_trip_for(AwsDynamoDbGlobalTable, "dynamodb_policy")
 
     def validate_delete_args(**kwargs: Any) -> Any:
         assert kwargs["action"] == "delete-table"

plugins/aws/test/resources/ecr_test.py

@@ -3,5 +3,5 @@
 
 
 def test_ecr_repositories() -> None:
-    first, builder = round_trip_for(AwsEcrRepository)
+    first, builder = round_trip_for(AwsEcrRepository, "repository_policy")
     assert len(builder.resources_of(AwsEcrRepository)) == 3

plugins/aws/test/resources/kinesis_test.py

@@ -1,19 +1,19 @@
 from fix_plugin_aws.resource.kinesis import AwsKinesisStream
+from fix_plugin_aws.aws_client import AwsClient
 from fixlib.graph import Graph
 from test.resources import round_trip_for
 from typing import Any, cast
 from types import SimpleNamespace
-from fix_plugin_aws.aws_client import AwsClient
 
 
 def test_kinesis_stream() -> None:
-    res, builder = round_trip_for(AwsKinesisStream)
+    res, builder = round_trip_for(AwsKinesisStream, "kinesis_policy")
     assert len(builder.resources_of(AwsKinesisStream)) == 1
     assert len(res.tags) == 1
 
 
 def test_tagging() -> None:
-    stream, _ = round_trip_for(AwsKinesisStream)
+    stream, _ = round_trip_for(AwsKinesisStream, "kinesis_policy")
 
     def validate_update_args(**kwargs: Any) -> None:
         assert kwargs["action"] == "add-tags-to-stream"
@@ -33,7 +33,7 @@ def validate_delete_args(**kwargs: Any) -> None:
 
 
 def test_deletion() -> None:
-    stream, _ = round_trip_for(AwsKinesisStream)
+    stream, _ = round_trip_for(AwsKinesisStream, "kinesis_policy")
 
     def validate_delete_args(**kwargs: Any) -> None:
         assert kwargs["action"] == "delete-stream"