chore(awslambda): Enhance function public access check called from other resource

prowler/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible.metadata.json

@@ -1,14 +1,14 @@
 {
   "Provider": "aws",
   "CheckID": "awslambda_function_not_publicly_accessible",
-  "CheckTitle": "Check if Lambda functions have resource-based policy set as Public.",
+  "CheckTitle": "Check if Lambda functions is publicly accessible through Internet",
   "CheckType": [],
   "ServiceName": "lambda",
   "SubServiceName": "",
   "ResourceIdTemplate": "arn:partition:lambda:region:account-id:function/function-name",
   "Severity": "critical",
   "ResourceType": "AwsLambdaFunction",
-  "Description": "Check if Lambda functions have resource-based policy set as Public.",
+  "Description": "Check if Lambda functions is publicly accessible through Internet via ALB and function's permission.",
   "Risk": "Publicly accessible services could expose sensitive data to bad actors.",
   "RelatedUrl": "https://docs.aws.amazon.com/lambda/latest/dg/access-control-resource-based.html",
   "Remediation": {

prowler/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible.py

@@ -1,5 +1,6 @@
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.awslambda.awslambda_client import awslambda_client
+from prowler.providers.aws.services.elbv2.elbv2_client import elbv2_client
 
 
 class awslambda_function_not_publicly_accessible(Check):
@@ -13,30 +14,68 @@
             report.resource_tags = function.tags
 
             report.status = "PASS"
-            report.status_extended = f"Lambda function {function.name} has a policy resource-based policy not public."
+            report.status_extended = (
+                f"Lambda function {function.name} is not publicly accessible."
+            )
 
-            public_access = False
-            if function.policy:
+            # 1. Check if the function is associated with a public load balancer
+            target_group_redirects_alb_to_function = ""
+
+            for tg_arn, target_group in elbv2_client.target_groups.items():
+                if (
+                    target_group.target_type == "lambda"
+                ) and function.arn in target_group.targets:
+                    for lb_arn in target_group.load_balancer_arns:
+                        lb = elbv2_client.loadbalancersv2.get(
+                            lb_arn,
+                            None,  # Bug: If prowler is filtering by ARN, this will fail because loadbalancersv2 are filtered
+                        )
+                        if lb and lb.scheme == "internet-facing":
+                            target_group_redirects_alb_to_function = tg_arn
+                            break
+
+            public_policy = False
+
+            # 2. Check if the function policy allows public access
+            if target_group_redirects_alb_to_function and function.policy:
                 for statement in function.policy["Statement"]:
                     # Only check allow statements
-                    if statement["Effect"] == "Allow":
-                        if (
+                    if (
+                        statement["Effect"] == "Allow"
+                        and (
                             "*" in statement["Principal"]
-                            or (
-                                "AWS" in statement["Principal"]
-                                and "*" in statement["Principal"]["AWS"]
-                            )
-                            or (
-                                "CanonicalUser" in statement["Principal"]
-                                and "*" in statement["Principal"]["CanonicalUser"]
-                            )
-                        ):
-                            public_access = True
+                            or "*" in statement["Principal"].get("AWS", "")
+                            or "*" in statement["Principal"].get("CanonicalUser", "")
+                            or "elasticloadbalancing.amazonaws.com"
+                            in statement["Principal"].get("Service", "")
+                        )
+                        and (
+                            "*" in statement["Action"]
+                            or "lambda:InvokeFunction" in statement["Action"]
+                        )
+                        and (
+                            "*" in statement["Resource"]
+                            or function.arn in statement["Resource"]
+                        )
+                    ):
+                        if statement.get("Condition", {}):
+                            if (
+                                statement["Condition"]
+                                .get("ArnLike", {})
+                                .get("AWS:SourceArn", "")
+                                == target_group_redirects_alb_to_function
+                            ):
+                                public_policy = True
+                                break
+                        else:
+                            public_policy = True
                             break
 
-            if public_access:
+            if public_policy:
                 report.status = "FAIL"
-                report.status_extended = f"Lambda function {function.name} has a policy resource-based policy with public access."
+                report.status_extended = (
+                    f"Lambda function {function.name} is publicly accessible."
+                )
 
             findings.append(report)
 

prowler/providers/aws/services/awslambda/awslambda_service.py

@@ -14,7 +14,6 @@
 from prowler.providers.aws.lib.service.service import AWSService
 
 
-################## Lambda
 class Lambda(AWSService):
     def __init__(self, provider):
         # Call AWSService's __init__

prowler/providers/aws/services/elbv2/elbv2_service.py

@@ -27,6 +27,8 @@ def __init__(self, provider):
             ],
         )
         self.__threading_call__(self._describe_tags, self.loadbalancersv2.items())
+        self.target_groups = {}
+        self.__threading_call__(self._describe_target_groups)
 
     def _describe_load_balancers(self, regional_client):
         logger.info("ELBv2 - Describing load balancers...")
@@ -174,6 +176,35 @@ def _describe_tags(self, load_balancer):
                 f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
+    def _describe_target_groups(self, regional_client):
+        logger.info("ELBv2 - Describing target groups...")
+        try:
+            describe_target_groups_paginator = regional_client.get_paginator(
+                "describe_target_groups"
+            )
+            for page in describe_target_groups_paginator.paginate():
+                for target_group in page["TargetGroups"]:
+                    targets = {
+                        (tg["Target"]["Id"], tg["Target"].get("Port", 0))
+                        for tg in regional_client.describe_target_health(
+                            TargetGroupArn=target_group["TargetGroupArn"]
+                        )["TargetHealthDescriptions"]
+                    }
+
+                    self.target_groups[target_group["TargetGroupArn"]] = TargetGroup(
+                        name=target_group.get("TargetGroupName", ""),
+                        protocol=target_group.get("Protocol", ""),
+                        port=target_group.get("Port", 0),
+                        vpc_id=target_group.get("VpcId", ""),
+                        target_type=target_group.get("TargetType", ""),
+                        targets=dict(targets),
+                        load_balancer_arns=target_group.get("LoadBalancerArns", []),
+                    )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
 
 class ListenerRule(BaseModel):
     arn: str
@@ -201,3 +232,14 @@ class LoadBalancerv2(BaseModel):
     listeners: Dict[str, Listenerv2] = {}
     scheme: Optional[str]
     tags: Optional[list] = []
+
+
+class TargetGroup(BaseModel):
+    name: str
+    protocol: str
+    port: int
+    vpc_id: str
+    target_type: str
+    # Key: ID, Value: Port (only available for IP targets)
+    targets: Dict[str, int] = []
+    load_balancer_arns: list[str]