chore(aws): add resource metadata to services from t to w (#6546)

prowler/providers/aws/services/transfer/transfer_server_in_transit_encryption_enabled/transfer_server_in_transit_encryption_enabled.py

@@ -21,11 +21,9 @@ class transfer_server_in_transit_encryption_enabled(Check):
         """
         findings = []
         for server in transfer_client.servers.values():
-            report = Check_Report_AWS(self.metadata())
-            report.region = server.region
-            report.resource_id = server.id
-            report.resource_arn = server.arn
-            report.resource_tags = server.tags
+            report = Check_Report_AWS(
+                metadata=self.metadata(), resource_metadata=server
+            )
             report.status = "PASS"
             report.status_extended = (
                 f"Transfer Server {server.id} does have encryption in transit enabled."

prowler/providers/aws/services/trustedadvisor/trustedadvisor_errors_and_warnings/trustedadvisor_errors_and_warnings.py

@@ -14,17 +14,19 @@ class trustedadvisor_errors_and_warnings(Check):
                         if (
                             check.status != "not_available"
                         ):  # avoid not_available checks since there are no resources that apply
-                            report = Check_Report_AWS(self.metadata())
-                            report.region = check.region
-                            report.resource_id = check.id
-                            report.resource_arn = check.arn
+                            report = Check_Report_AWS(
+                                metadata=self.metadata(), resource_metadata=check
+                            )
                             report.status = "FAIL"
                             report.status_extended = f"Trusted Advisor check {check.name} is in state {check.status}."
                             if check.status == "ok":
                                 report.status = "PASS"
                             findings.append(report)
             else:
-                report = Check_Report_AWS(self.metadata())
+                report = Check_Report_AWS(
+                    metadata=self.metadata(),
+                    resource_metadata=trustedadvisor_client.checks,
+                )
                 report.status = "MANUAL"
                 report.status_extended = "Amazon Web Services Premium Support Subscription is required to use this service."
                 report.resource_id = trustedadvisor_client.audited_account

prowler/providers/aws/services/trustedadvisor/trustedadvisor_premium_support_plan_subscribed/trustedadvisor_premium_support_plan_subscribed.py

@@ -13,7 +13,10 @@ class trustedadvisor_premium_support_plan_subscribed(Check):
                 "verify_premium_support_plans", True
             )
         ):
-            report = Check_Report_AWS(self.metadata())
+            report = Check_Report_AWS(
+                metadata=self.metadata(),
+                resource_metadata=trustedadvisor_client.premium_support,
+            )
             report.status = "FAIL"
             report.status_extended = (
                 "Amazon Web Services Premium Support Plan isn't subscribed."

prowler/providers/aws/services/vpc/vpc_different_regions/vpc_different_regions.py

@@ -11,7 +11,9 @@ class vpc_different_regions(Check):
                 if not vpc.default:
                     vpc_regions.add(vpc.region)
 
-            report = Check_Report_AWS(self.metadata())
+            report = Check_Report_AWS(
+                metadata=self.metadata(), resource_metadata=vpc_client.vpcs
+            )
             report.region = vpc_client.region
             report.resource_id = vpc_client.audited_account
             report.resource_arn = vpc_client.vpc_arn_template

prowler/providers/aws/services/vpc/vpc_endpoint_connections_trust_boundaries/vpc_endpoint_connections_trust_boundaries.py

@@ -25,11 +25,9 @@ class vpc_endpoint_connections_trust_boundaries(Check):
                         break
                     if "*" == statement["Principal"]:
                         access_from_trusted_accounts = False
-                        report = Check_Report_AWS(self.metadata())
-                        report.region = endpoint.region
-                        report.resource_id = endpoint.id
-                        report.resource_arn = endpoint.arn
-                        report.resource_tags = endpoint.tags
+                        report = Check_Report_AWS(
+                            metadata=self.metadata(), resource_metadata=endpoint
+                        )
 
                         if "Condition" in statement:
                             for account_id in trusted_account_ids:
@@ -62,11 +60,9 @@ class vpc_endpoint_connections_trust_boundaries(Check):
                             # If the principal is not an AWS principal, we don't need to check it since it could be a service or a federated principal
                             principals = []
                         for principal_arn in principals:
-                            report = Check_Report_AWS(self.metadata())
-                            report.region = endpoint.region
-                            report.resource_id = endpoint.id
-                            report.resource_arn = endpoint.arn
-                            report.resource_tags = endpoint.tags
+                            report = Check_Report_AWS(
+                                metadata=self.metadata(), resource_metadata=endpoint
+                            )
 
                             if principal_arn == "*":
                                 access_from_trusted_accounts = False

prowler/providers/aws/services/vpc/vpc_endpoint_for_ec2_enabled/vpc_endpoint_for_ec2_enabled.py

@@ -7,13 +7,11 @@ class vpc_endpoint_for_ec2_enabled(Check):
         findings = []
         for vpc_id, vpc in vpc_client.vpcs.items():
             if vpc_client.provider.scan_unused_services or vpc.in_use:
-                report = Check_Report_AWS(self.metadata())
-                report.region = vpc.region
-                report.resource_tags = vpc.tags
+                report = Check_Report_AWS(
+                    metadata=self.metadata(), resource_metadata=vpc
+                )
                 report.status = "FAIL"
                 report.status_extended = f"VPC {vpc.id} has no EC2 endpoint."
-                report.resource_id = vpc.id
-                report.resource_arn = vpc.arn
                 for endpoint in vpc_client.vpc_endpoints:
                     if endpoint.vpc_id == vpc_id and "ec2" in endpoint.service_name:
                         report.status = "PASS"

prowler/providers/aws/services/vpc/vpc_endpoint_multi_az_enabled/vpc_endpoint_multi_az_enabled.py

@@ -7,11 +7,9 @@ class vpc_endpoint_multi_az_enabled(Check):
         findings = []
         for endpoint in vpc_client.vpc_endpoints:
             if endpoint.vpc_id in vpc_client.vpcs and endpoint.type == "Interface":
-                report = Check_Report_AWS(self.metadata())
-                report.region = endpoint.region
-                report.resource_tags = endpoint.tags
-                report.resource_id = endpoint.id
-                report.resource_arn = endpoint.arn
+                report = Check_Report_AWS(
+                    metadata=self.metadata(), resource_metadata=endpoint
+                )
                 report.status = "FAIL"
                 report.status_extended = f"VPC Endpoint {endpoint.id} in VPC {endpoint.vpc_id} has subnets in different AZs."
                 if len(endpoint.subnet_ids) > 1:

prowler/providers/aws/services/vpc/vpc_endpoint_services_allowed_principals_trust_boundaries/vpc_endpoint_services_allowed_principals_trust_boundaries.py

@@ -10,11 +10,9 @@ class vpc_endpoint_services_allowed_principals_trust_boundaries(Check):
         # Get trusted account_ids from prowler.config.yaml
         trusted_account_ids = vpc_client.audit_config.get("trusted_account_ids", [])
         for service in vpc_client.vpc_endpoint_services:
-            report = Check_Report_AWS(self.metadata())
-            report.region = service.region
-            report.resource_id = service.id
-            report.resource_arn = service.arn
-            report.resource_tags = service.tags
+            report = Check_Report_AWS(
+                metadata=self.metadata(), resource_metadata=service
+            )
 
             if not service.allowed_principals:
                 report.status = "PASS"

prowler/providers/aws/services/vpc/vpc_flow_logs_enabled/vpc_flow_logs_enabled.py

@@ -7,11 +7,9 @@ class vpc_flow_logs_enabled(Check):
         findings = []
         for vpc in vpc_client.vpcs.values():
             if vpc_client.provider.scan_unused_services or vpc.in_use:
-                report = Check_Report_AWS(self.metadata())
-                report.region = vpc.region
-                report.resource_tags = vpc.tags
-                report.resource_id = vpc.id
-                report.resource_arn = vpc.arn
+                report = Check_Report_AWS(
+                    metadata=self.metadata(), resource_metadata=vpc
+                )
                 report.status = "FAIL"
                 report.status_extended = (
                     f"VPC {vpc.name if vpc.name else vpc.id} Flow logs are disabled."

prowler/providers/aws/services/vpc/vpc_peering_routing_tables_with_least_privilege/vpc_peering_routing_tables_with_least_privilege.py

@@ -6,11 +6,7 @@ class vpc_peering_routing_tables_with_least_privilege(Check):
     def execute(self):
         findings = []
         for peer in vpc_client.vpc_peering_connections:
-            report = Check_Report_AWS(self.metadata())
-            report.region = peer.region
-            report.resource_tags = peer.tags
-            report.resource_id = peer.id
-            report.resource_arn = peer.arn
+            report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=peer)
             report.status = "PASS"
             report.status_extended = (
                 f"VPC Peering Connection {peer.id} comply with least privilege access."

prowler/providers/aws/services/vpc/vpc_service.py

@@ -422,6 +422,7 @@ class VPC(AWSService):
                         )
                     self.vpn_connections[arn] = VpnConnection(
                         id=vpn_connection["VpnConnectionId"],
+                        arn=arn,
                         tunnels=tunnels,
                         region=regional_client.region,
                         tags=vpn_connection.get("Tags"),
@@ -510,6 +511,7 @@ class VpnTunnel(BaseModel):
 
 class VpnConnection(BaseModel):
     id: str
+    arn: str
     tunnels: list[VpnTunnel]
     region: str
     tags: Optional[list] = []

prowler/providers/aws/services/vpc/vpc_subnet_different_az/vpc_subnet_different_az.py

@@ -7,15 +7,13 @@ class vpc_subnet_different_az(Check):
         findings = []
         for vpc in vpc_client.vpcs.values():
             if vpc_client.provider.scan_unused_services or vpc.in_use:
-                report = Check_Report_AWS(self.metadata())
-                report.region = vpc.region
-                report.resource_tags = vpc.tags
+                report = Check_Report_AWS(
+                    metadata=self.metadata(), resource_metadata=vpc
+                )
                 report.status = "FAIL"
                 report.status_extended = (
                     f"VPC {vpc.name if vpc.name else vpc.id} has no subnets."
                 )
-                report.resource_id = vpc.id
-                report.resource_arn = vpc.arn
                 if vpc.subnets:
                     availability_zone = None
                     for subnet in vpc.subnets:

prowler/providers/aws/services/vpc/vpc_subnet_no_public_ip_by_default/vpc_subnet_no_public_ip_by_default.py

@@ -9,11 +9,9 @@ class vpc_subnet_no_public_ip_by_default(Check):
             for subnet in vpc.subnets:
                 # Check if ignoring flag is set and if the VPC Subnet is in use
                 if vpc_client.provider.scan_unused_services or subnet.in_use:
-                    report = Check_Report_AWS(self.metadata())
-                    report.region = subnet.region
-                    report.resource_tags = subnet.tags
-                    report.resource_id = subnet.id
-                    report.resource_arn = subnet.arn
+                    report = Check_Report_AWS(
+                        metadata=self.metadata(), resource_metadata=subnet
+                    )
                     if subnet.mapPublicIpOnLaunch:
                         report.status = "FAIL"
                         report.status_extended = f"VPC subnet {subnet.name if subnet.name else subnet.id} assigns public IP by default."

prowler/providers/aws/services/vpc/vpc_subnet_separate_private_public/vpc_subnet_separate_private_public.py

@@ -7,15 +7,13 @@ class vpc_subnet_separate_private_public(Check):
         findings = []
         for vpc in vpc_client.vpcs.values():
             if vpc_client.provider.scan_unused_services or vpc.in_use:
-                report = Check_Report_AWS(self.metadata())
-                report.region = vpc.region
-                report.resource_tags = vpc.tags
+                report = Check_Report_AWS(
+                    metadata=self.metadata(), resource_metadata=vpc
+                )
                 report.status = "FAIL"
                 report.status_extended = (
                     f"VPC {vpc.name if vpc.name else vpc.id} has no subnets."
                 )
-                report.resource_id = vpc.id
-                report.resource_arn = vpc.arn
                 if vpc.subnets:
                     public = False
                     private = False

prowler/providers/aws/services/vpc/vpc_vpn_connection_tunnels_up/vpc_vpn_connection_tunnels_up.py

@@ -5,12 +5,10 @@ from prowler.providers.aws.services.vpc.vpc_client import vpc_client
 class vpc_vpn_connection_tunnels_up(Check):
     def execute(self):
         findings = []
-        for vpn_arn, vpn_connection in vpc_client.vpn_connections.items():
-            report = Check_Report_AWS(self.metadata())
-            report.region = vpn_connection.region
-            report.resource_id = vpn_connection.id
-            report.resource_arn = vpn_arn
-            report.resource_tags = vpn_connection.tags
+        for vpn_connection in vpc_client.vpn_connections.values():
+            report = Check_Report_AWS(
+                metadata=self.metadata(), resource_metadata=vpn_connection
+            )
 
             if (
                 vpn_connection.tunnels[0].status != "UP"

prowler/providers/aws/services/waf/waf_global_rule_with_conditions/waf_global_rule_with_conditions.py

@@ -6,11 +6,7 @@ class waf_global_rule_with_conditions(Check):
     def execute(self):
         findings = []
         for rule in waf_client.rules.values():
-            report = Check_Report_AWS(self.metadata())
-            report.region = rule.region
-            report.resource_id = rule.id
-            report.resource_arn = rule.arn
-            report.resource_tags = rule.tags
+            report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=rule)
             report.status = "FAIL"
             report.status_extended = (
                 f"AWS WAF Global Rule {rule.name} does not have any conditions."

prowler/providers/aws/services/waf/waf_global_rulegroup_not_empty/waf_global_rulegroup_not_empty.py

@@ -6,11 +6,9 @@ class waf_global_rulegroup_not_empty(Check):
     def execute(self):
         findings = []
         for rule_group in waf_client.rule_groups.values():
-            report = Check_Report_AWS(self.metadata())
-            report.region = rule_group.region
-            report.resource_id = rule_group.id
-            report.resource_arn = rule_group.arn
-            report.resource_tags = rule_group.tags
+            report = Check_Report_AWS(
+                metadata=self.metadata(), resource_metadata=rule_group
+            )
             report.status = "FAIL"
             report.status_extended = (
                 f"AWS WAF Global Rule Group {rule_group.name} does not have any rules."

prowler/providers/aws/services/waf/waf_global_webacl_logging_enabled/waf_global_webacl_logging_enabled.py

@@ -6,11 +6,7 @@ class waf_global_webacl_logging_enabled(Check):
     def execute(self):
         findings = []
         for acl in waf_client.web_acls.values():
-            report = Check_Report_AWS(self.metadata())
-            report.region = acl.region
-            report.resource_id = acl.id
-            report.resource_arn = acl.arn
-            report.resource_tags = acl.tags
+            report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=acl)
             report.status = "FAIL"
             report.status_extended = (
                 f"AWS WAF Global Web ACL {acl.name} does not have logging enabled."

prowler/providers/aws/services/waf/waf_global_webacl_with_rules/waf_global_webacl_with_rules.py

@@ -6,11 +6,7 @@ class waf_global_webacl_with_rules(Check):
     def execute(self):
         findings = []
         for acl in waf_client.web_acls.values():
-            report = Check_Report_AWS(self.metadata())
-            report.region = acl.region
-            report.resource_id = acl.id
-            report.resource_arn = acl.arn
-            report.resource_tags = acl.tags
+            report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=acl)
             report.status = "FAIL"
             report.status_extended = f"AWS WAF Global Web ACL {acl.name} does not have any rules or rule groups."
 

prowler/providers/aws/services/waf/waf_regional_rule_with_conditions/waf_regional_rule_with_conditions.py

@@ -6,11 +6,7 @@ class waf_regional_rule_with_conditions(Check):
     def execute(self):
         findings = []
         for rule in wafregional_client.rules.values():
-            report = Check_Report_AWS(self.metadata())
-            report.region = rule.region
-            report.resource_id = rule.id
-            report.resource_arn = rule.arn
-            report.resource_tags = rule.tags
+            report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=rule)
             report.status = "FAIL"
             report.status_extended = (
                 f"AWS WAF Regional Rule {rule.name} does not have any conditions."

prowler/providers/aws/services/waf/waf_regional_rulegroup_not_empty/waf_regional_rulegroup_not_empty.py

@@ -6,11 +6,9 @@ class waf_regional_rulegroup_not_empty(Check):
     def execute(self):
         findings = []
         for rule_group in wafregional_client.rule_groups.values():
-            report = Check_Report_AWS(self.metadata())
-            report.region = rule_group.region
-            report.resource_id = rule_group.id
-            report.resource_arn = rule_group.arn
-            report.resource_tags = rule_group.tags
+            report = Check_Report_AWS(
+                metadata=self.metadata(), resource_metadata=rule_group
+            )
             report.status = "FAIL"
             report.status_extended = f"AWS WAF Regional Rule Group {rule_group.name} does not have any rules."
 

prowler/providers/aws/services/waf/waf_regional_webacl_with_rules/waf_regional_webacl_with_rules.py

@@ -6,11 +6,7 @@ class waf_regional_webacl_with_rules(Check):
     def execute(self):
         findings = []
         for acl in wafregional_client.web_acls.values():
-            report = Check_Report_AWS(self.metadata())
-            report.region = acl.region
-            report.resource_id = acl.id
-            report.resource_arn = acl.arn
-            report.resource_tags = acl.tags
+            report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=acl)
             report.status = "FAIL"
             report.status_extended = f"AWS WAF Regional Web ACL {acl.name} does not have any rules or rule groups."
 

prowler/providers/aws/services/wafv2/wafv2_webacl_logging_enabled/wafv2_webacl_logging_enabled.py

@@ -6,11 +6,9 @@ class wafv2_webacl_logging_enabled(Check):
     def execute(self):
         findings = []
         for web_acl in wafv2_client.web_acls.values():
-            report = Check_Report_AWS(self.metadata())
-            report.region = web_acl.region
-            report.resource_id = web_acl.id
-            report.resource_arn = web_acl.arn
-            report.resource_tags = web_acl.tags
+            report = Check_Report_AWS(
+                metadata=self.metadata(), resource_metadata=web_acl
+            )
 
             if web_acl.logging_enabled:
                 report.status = "PASS"

prowler/providers/aws/services/wafv2/wafv2_webacl_rule_logging_enabled/wafv2_webacl_rule_logging_enabled.py

@@ -6,11 +6,9 @@ class wafv2_webacl_rule_logging_enabled(Check):
     def execute(self):
         findings = []
         for web_acl in wafv2_client.web_acls.values():
-            report = Check_Report_AWS(self.metadata())
-            report.region = web_acl.region
-            report.resource_id = web_acl.id
-            report.resource_arn = web_acl.arn
-            report.resource_tags = web_acl.tags
+            report = Check_Report_AWS(
+                metadata=self.metadata(), resource_metadata=web_acl
+            )
 
             if web_acl.rules or web_acl.rule_groups:
                 report.status = "PASS"

prowler/providers/aws/services/wafv2/wafv2_webacl_with_rules/wafv2_webacl_with_rules.py

@@ -6,11 +6,9 @@ class wafv2_webacl_with_rules(Check):
     def execute(self):
         findings = []
         for web_acl in wafv2_client.web_acls.values():
-            report = Check_Report_AWS(self.metadata())
-            report.region = web_acl.region
-            report.resource_id = web_acl.id
-            report.resource_arn = web_acl.arn
-            report.resource_tags = web_acl.tags
+            report = Check_Report_AWS(
+                metadata=self.metadata(), resource_metadata=web_acl
+            )
             report.status = "FAIL"
             report.status_extended = f"AWS WAFv2 Web ACL {web_acl.name} does not have any rules or rule groups attached."
 

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

@@ -8,11 +8,9 @@ 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 = Check_Report_AWS(
+                metadata=self.metadata(), resource_metadata=workload
+            )
             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:

prowler/providers/aws/services/workspaces/workspaces_volume_encryption_enabled/workspaces_volume_encryption_enabled.py

@@ -8,11 +8,9 @@ class workspaces_volume_encryption_enabled(Check):
     def execute(self):
         findings = []
         for workspace in workspaces_client.workspaces:
-            report = Check_Report_AWS(self.metadata())
-            report.region = workspace.region
-            report.resource_id = workspace.id
-            report.resource_arn = workspace.arn
-            report.resource_tags = workspace.tags
+            report = Check_Report_AWS(
+                metadata=self.metadata(), resource_metadata=workspace
+            )
             report.status = "PASS"
             report.status_extended = f"WorkSpaces workspace {workspace.id} root and user volumes are encrypted."
             if not workspace.user_volume_encryption_enabled:

prowler/providers/aws/services/workspaces/workspaces_vpc_2private_1public_subnets_nat/workspaces_vpc_2private_1public_subnets_nat.py

@@ -9,11 +9,9 @@ class workspaces_vpc_2private_1public_subnets_nat(Check):
     def execute(self):
         findings = []
         for workspace in workspaces_client.workspaces:
-            report = Check_Report_AWS(self.metadata())
-            report.region = workspace.region
-            report.resource_id = workspace.id
-            report.resource_arn = workspace.arn
-            report.resource_tags = workspace.tags
+            report = Check_Report_AWS(
+                metadata=self.metadata(), resource_metadata=workspace
+            )
             report.status = "PASS"
             report.status_extended = f"Workspace {workspace.id} is in a private subnet within a VPC which has 1 public subnet 2 private subnets with a NAT Gateway attached."
             vpc_object = None

tests/providers/aws/services/vpc/vpc_service_test.py

@@ -475,4 +475,5 @@ class Test_VPC_Service:
         vpn_conn = vpc.vpn_connections[vpn_arn]
         assert vpn_conn.id == "vpn-1234567890abcdef0"
         assert vpn_conn.region == AWS_REGION_US_EAST_1
+        assert vpn_conn.arn == vpn_arn
         assert len(vpn_conn.tunnels) == 2