fix(aws): filter VPC endpoint services by audited account to prevent AccessDenied errors (#10152)

Co-authored-by: Copilot <198982749+Copilot@users.noreply.github.com>
Co-authored-by: jfagoagas <16007882+jfagoagas@users.noreply.github.com>

Author: jfagoagas

prowler/CHANGELOG.md

@@ -32,10 +32,6 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - CIS 6.0 for the AWS provider [(#10127)](https://github.com/prowler-cloud/prowler/pull/10127)
 - OpenStack provider multiple regions support [(#10135)](https://github.com/prowler-cloud/prowler/pull/10135)
 
-### 🐞 Fixed
-
-- Standardize resource_id values across Azure checks to use actual Azure resource IDs and prevent duplicate resource entries [(#9994)](https://github.com/prowler-cloud/prowler/pull/9994)
-
 ### 🔄 Changed
 
 - Update Azure Monitor service metadata to new format [(#9622)](https://github.com/prowler-cloud/prowler/pull/9622)
@@ -61,6 +57,8 @@ All notable changes to the **Prowler SDK** are documented in this file.
 ### 🐞 Fixed
 
 - Update AWS checks metadata URLs to replace deprecated Trend Micro CloudOne Conformity (EOL July 2026) with Vision One and remove docs.prowler.com references [(#10068)](https://github.com/prowler-cloud/prowler/pull/10068)
+- Standardize resource_id values across Azure checks to use actual Azure resource IDs and prevent duplicate resource entries [(#9994)](https://github.com/prowler-cloud/prowler/pull/9994)
+- VPC endpoint service collection filtering third-party services that caused AccessDenied errors on `DescribeVpcEndpointServicePermissions` [(#10152)](https://github.com/prowler-cloud/prowler/pull/10152)
 
 ### 🔐 Security
 

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

@@ -264,7 +264,10 @@ def _describe_vpc_endpoint_services(self, regional_client):
             for page in describe_vpc_endpoint_services_paginator.paginate():
                 for endpoint in page["ServiceDetails"]:
                     try:
-                        if endpoint["Owner"] != "amazon":
+                        # Only collect endpoint services owned by the audited account.
+                        # The API returns ALL available services in the region,
+                        # including Amazon and third-party ones we can't inspect.
+                        if endpoint["Owner"] == self.audited_account:
                             arn = f"arn:{self.audited_partition}:ec2:{regional_client.region}:{self.audited_account}:vpc-endpoint-service/{endpoint['ServiceId']}"
                             if not self.audit_resources or (
                                 is_resource_filtered(arn, self.audit_resources)
@@ -303,9 +306,13 @@ def _describe_vpc_endpoint_service_permissions(self):
                     ]:
                         service.allowed_principals.append(principal["Principal"])
                 except ClientError as error:
-                    if (
-                        error.response["Error"]["Code"]
-                        == "InvalidVpcEndpointServiceId.NotFound"
+                    # AccessDenied/UnauthorizedOperation can occur if a
+                    # non-owned service slips through or permissions change
+                    # between collection and this call.
+                    if error.response["Error"]["Code"] in (
+                        "InvalidVpcEndpointServiceId.NotFound",
+                        "AccessDenied",
+                        "UnauthorizedOperation",
                     ):
                         logger.warning(
                             f"{service.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"

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

@@ -3,6 +3,7 @@
 import botocore
 import mock
 from boto3 import client, resource
+from botocore.exceptions import ClientError
 from moto import mock_aws
 
 from prowler.providers.aws.services.vpc.vpc_service import VPC, Route
@@ -13,9 +14,125 @@
     set_mocked_aws_provider,
 )
 
+THIRD_PARTY_ACCOUNT = "178579023202"
+
 make_api_call = botocore.client.BaseClient._make_api_call
 
 
+def mock_make_api_call_endpoint_services(self, operation_name, kwarg):
+    """Mock that returns VPC endpoint services from mixed owners:
+    audited account, amazon, and a third-party account."""
+    if operation_name == "DescribeVpcEndpointServices":
+        return {
+            "ServiceDetails": [
+                {
+                    "ServiceId": "vpce-svc-owned123",
+                    "ServiceName": "com.amazonaws.vpce.us-east-1.vpce-svc-owned123",
+                    "ServiceType": [{"ServiceType": "Interface"}],
+                    "Owner": AWS_ACCOUNT_NUMBER,
+                    "Tags": [{"Key": "Name", "Value": "owned-service"}],
+                },
+                {
+                    "ServiceId": "vpce-svc-amazon456",
+                    "ServiceName": "com.amazonaws.us-east-1.s3",
+                    "ServiceType": [{"ServiceType": "Gateway"}],
+                    "Owner": "amazon",
+                    "Tags": [],
+                },
+                {
+                    "ServiceId": "vpce-svc-thirdparty789",
+                    "ServiceName": "com.amazonaws.vpce.us-east-1.vpce-svc-thirdparty789",
+                    "ServiceType": [{"ServiceType": "Interface"}],
+                    "Owner": THIRD_PARTY_ACCOUNT,
+                    "Tags": [],
+                },
+            ],
+            "ServiceNames": [],
+        }
+    if operation_name == "DescribeVpcEndpointServicePermissions":
+        return {"AllowedPrincipals": []}
+    return make_api_call(self, operation_name, kwarg)
+
+
+def mock_make_api_call_endpoint_services_access_denied(self, operation_name, kwarg):
+    """Mock where DescribeVpcEndpointServicePermissions raises AccessDenied."""
+    if operation_name == "DescribeVpcEndpointServices":
+        return {
+            "ServiceDetails": [
+                {
+                    "ServiceId": "vpce-svc-owned123",
+                    "ServiceName": "com.amazonaws.vpce.us-east-1.vpce-svc-owned123",
+                    "ServiceType": [{"ServiceType": "Interface"}],
+                    "Owner": AWS_ACCOUNT_NUMBER,
+                    "Tags": [],
+                },
+            ],
+            "ServiceNames": [],
+        }
+    if operation_name == "DescribeVpcEndpointServicePermissions":
+        raise ClientError(
+            {"Error": {"Code": "AccessDenied", "Message": "Access denied"}},
+            operation_name,
+        )
+    return make_api_call(self, operation_name, kwarg)
+
+
+def mock_make_api_call_endpoint_services_unauthorized(self, operation_name, kwarg):
+    """Mock where DescribeVpcEndpointServicePermissions raises UnauthorizedOperation."""
+    if operation_name == "DescribeVpcEndpointServices":
+        return {
+            "ServiceDetails": [
+                {
+                    "ServiceId": "vpce-svc-owned123",
+                    "ServiceName": "com.amazonaws.vpce.us-east-1.vpce-svc-owned123",
+                    "ServiceType": [{"ServiceType": "Interface"}],
+                    "Owner": AWS_ACCOUNT_NUMBER,
+                    "Tags": [],
+                },
+            ],
+            "ServiceNames": [],
+        }
+    if operation_name == "DescribeVpcEndpointServicePermissions":
+        raise ClientError(
+            {
+                "Error": {
+                    "Code": "UnauthorizedOperation",
+                    "Message": "Unauthorized",
+                }
+            },
+            operation_name,
+        )
+    return make_api_call(self, operation_name, kwarg)
+
+
+def mock_make_api_call_endpoint_services_not_found(self, operation_name, kwarg):
+    """Mock where DescribeVpcEndpointServicePermissions raises InvalidVpcEndpointServiceId.NotFound."""
+    if operation_name == "DescribeVpcEndpointServices":
+        return {
+            "ServiceDetails": [
+                {
+                    "ServiceId": "vpce-svc-owned123",
+                    "ServiceName": "com.amazonaws.vpce.us-east-1.vpce-svc-owned123",
+                    "ServiceType": [{"ServiceType": "Interface"}],
+                    "Owner": AWS_ACCOUNT_NUMBER,
+                    "Tags": [],
+                },
+            ],
+            "ServiceNames": [],
+        }
+    if operation_name == "DescribeVpcEndpointServicePermissions":
+        raise ClientError(
+            {
+                "Error": {
+                    "Code": "InvalidVpcEndpointServiceId.NotFound",
+                    "Message": "Service not found",
+                }
+            },
+            operation_name,
+        )
+    return make_api_call(self, operation_name, kwarg)
+
+
 def mock_make_api_call(self, operation_name, kwarg):
     if operation_name == "DescribeVpnConnections":
         return {
@@ -477,3 +594,67 @@ def test_describe_vpn_connections(self):
         assert vpn_conn.region == AWS_REGION_US_EAST_1
         assert vpn_conn.arn == vpn_arn
         assert len(vpn_conn.tunnels) == 2
+
+    # Test VPC Endpoint Services filters out third-party and Amazon-owned services
+    @mock.patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_endpoint_services,
+    )
+    def test_describe_vpc_endpoint_services_filters_third_party(self):
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        vpc = VPC(aws_provider)
+
+        # Only the service owned by the audited account should be collected
+        assert len(vpc.vpc_endpoint_services) == 1
+        assert vpc.vpc_endpoint_services[0].id == "vpce-svc-owned123"
+        assert vpc.vpc_endpoint_services[0].owner_id == AWS_ACCOUNT_NUMBER
+        assert vpc.vpc_endpoint_services[0].service == (
+            "com.amazonaws.vpce.us-east-1.vpce-svc-owned123"
+        )
+        assert vpc.vpc_endpoint_services[0].region == AWS_REGION_US_EAST_1
+        # Third-party service (178579023202) must NOT be in the list
+        for svc in vpc.vpc_endpoint_services:
+            assert svc.owner_id != THIRD_PARTY_ACCOUNT
+            assert svc.owner_id != "amazon"
+
+    # Test that AccessDenied in DescribeVpcEndpointServicePermissions is handled gracefully
+    @mock.patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_endpoint_services_access_denied,
+    )
+    def test_describe_vpc_endpoint_service_permissions_access_denied(self):
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        vpc = VPC(aws_provider)
+
+        assert len(vpc.vpc_endpoint_services) == 1
+        assert vpc.vpc_endpoint_services[0].id == "vpce-svc-owned123"
+        # allowed_principals must remain empty when AccessDenied is raised
+        assert vpc.vpc_endpoint_services[0].allowed_principals == []
+
+    # Test that UnauthorizedOperation in DescribeVpcEndpointServicePermissions is handled gracefully
+    @mock.patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_endpoint_services_unauthorized,
+    )
+    def test_describe_vpc_endpoint_service_permissions_unauthorized(self):
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        vpc = VPC(aws_provider)
+
+        assert len(vpc.vpc_endpoint_services) == 1
+        assert vpc.vpc_endpoint_services[0].id == "vpce-svc-owned123"
+        # allowed_principals must remain empty when UnauthorizedOperation is raised
+        assert vpc.vpc_endpoint_services[0].allowed_principals == []
+
+    # Test that InvalidVpcEndpointServiceId.NotFound in DescribeVpcEndpointServicePermissions is handled gracefully
+    @mock.patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_endpoint_services_not_found,
+    )
+    def test_describe_vpc_endpoint_service_permissions_not_found(self):
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        vpc = VPC(aws_provider)
+
+        assert len(vpc.vpc_endpoint_services) == 1
+        assert vpc.vpc_endpoint_services[0].id == "vpce-svc-owned123"
+        # allowed_principals must remain empty when service is not found
+        assert vpc.vpc_endpoint_services[0].allowed_principals == []