feat(access-control): detect cross-account async destinations and alias traffic shadowing

Two native-feature abuse paths went unchecked. First, async-invoke
destinations: an attacker who can call lambda:PutFunctionEventInvokeConfig
points a function's OnSuccess/OnFailure destination at an SQS queue, SNS topic,
Lambda, or EventBridge bus in their own account and exfiltrates every async
invocation result or payload, and the config keeps firing after their access is
revoked. Second, alias traffic shadowing: a weighted alias
(RoutingConfig.AdditionalVersionWeights) quietly routes a fraction of
invocations to a second, attacker-published version while the primary version
still looks clean, a stealthy persistence and backdoor technique.

check_destinations reads the function event-invoke config, parses each
destination ARN, compares its account against the scanning account, and reports
has_external_destination plus external_destinations. A CRITICAL
external_account_destination finding is raised and the function score is reduced
by 20. check_aliases lists aliases and flags any with an additional-version
routing config, reporting shadowed_aliases; a MEDIUM alias_traffic_shadowing
finding is raised and the score reduced by 5. Same-account destinations and
plain (unweighted) aliases are unaffected.

Adds five tests across the two checks. Bumps the check count to 21 and the
version to 1.0.2.

Author: TarekCheikh

Dockerfile

@@ -2,7 +2,7 @@ FROM python:3.11-slim-bookworm
 
 LABEL maintainer="Toc Consulting <[email protected]>"
 LABEL description="AWS Lambda security scanner with multi-framework compliance mapping"
-LABEL version="1.0.1"
+LABEL version="1.0.2"
 
 ENV PYTHONDONTWRITEBYTECODE=1
 ENV PYTHONUNBUFFERED=1

README.md

@@ -12,7 +12,7 @@
   <a href="https://aws.amazon.com/lambda/"><img src="https://img.shields.io/badge/AWS-Lambda-orange.svg" alt="AWS"></a>
 </p>
 
-A comprehensive AWS Lambda security scanner with 19 security checks across 5 categories and compliance mapping for 10 frameworks (81 controls). Features multi-threaded scanning, secret detection in environment variables, and interactive HTML dashboards.
+A comprehensive AWS Lambda security scanner with 21 security checks across 5 categories and compliance mapping for 10 frameworks (81 controls). Features multi-threaded scanning, secret detection in environment variables, and interactive HTML dashboards.
 
 <p align="center">
   <img src="https://raw.githubusercontent.com/TocConsulting/lambda-security-scanner/main/assets/demo.gif" alt="Lambda Security Scanner demo: secrets, public URLs, IAM, and multi-framework compliance" width="100%">
@@ -22,7 +22,7 @@ A comprehensive AWS Lambda security scanner with 19 security checks across 5 cat
 
 ### **Comprehensive Security Analysis**
 - **Function Configuration**: Deprecated runtime detection, timeout tuning, environment variable secret scanning, ephemeral storage, external layers, X-Ray tracing, dead letter queues
-- **Access Control**: Resource policy public access, function URL authentication, CORS wildcard origins, overly permissive execution roles, shared role detection
+- **Access Control**: Resource policy public access, function URL authentication, CORS wildcard origins, overly permissive execution roles, shared role detection, cross-account async-invoke destinations, alias traffic shadowing
 - **Network Security**: VPC configuration, multi-AZ deployment, unrestricted security group egress
 - **Logging & Monitoring**: CloudWatch log group validation, log retention policies, reserved concurrency
 - **Code & Supply Chain**: Code signing configuration, event source mapping failure destinations
@@ -139,7 +139,7 @@ lambda-security-scanner security -f json -q
 
 ## Security Checks
 
-### 19 Checks Across 5 Categories
+### 21 Checks Across 5 Categories
 
 | ID  | Check                                    | Severity          | Category              |
 |-----|------------------------------------------|-------------------|-----------------------|
@@ -155,6 +155,8 @@ lambda-security-scanner security -f json -q
 | B.3 | Function URL CORS allows all origins     | HIGH              | Access Control        |
 | B.4 | Overly permissive execution role         | CRITICAL/HIGH     | Access Control        |
 | B.5 | Shared execution role                    | HIGH              | Access Control        |
+| B.6 | Async-invoke destination to external account | CRITICAL      | Access Control        |
+| B.7 | Alias traffic shadowing (weighted alias) | MEDIUM            | Access Control        |
 | C.1 | No VPC configuration                     | LOW               | Network Security      |
 | C.2 | VPC single AZ                            | MEDIUM            | Network Security      |
 | C.3 | Unrestricted SG egress                   | MEDIUM            | Network Security      |
@@ -287,6 +289,8 @@ docker run --rm \
             "lambda:GetCodeSigningConfig",
             "lambda:GetFunctionConcurrency",
             "lambda:ListEventSourceMappings",
+            "lambda:GetFunctionEventInvokeConfig",
+            "lambda:ListAliases",
             "iam:ListAttachedRolePolicies",
             "iam:GetPolicy",
             "iam:GetPolicyVersion",
@@ -395,7 +399,7 @@ lambda_security_scanner/
 ├── checks/                     # Security check modules
 │   ├── base.py                 # BaseChecker (session factory, error handling)
 │   ├── function_config.py      # A.1-A.7: Runtime, secrets, layers, tracing
-│   ├── access_control.py       # B.1-B.5: Policies, URLs, roles
+│   ├── access_control.py       # B.1-B.7: Policies, URLs, roles, destinations, aliases
 │   ├── network_security.py     # C.1-C.3: VPC, AZ, security groups
 │   ├── logging_monitoring.py   # D.1-D.2: Log groups, concurrency
 │   └── code_security.py        # E.1-E.2: Code signing, ESM

lambda_security_scanner/__init__.py

@@ -1,7 +1,7 @@
 """Lambda Security Scanner - Comprehensive AWS Lambda security
 auditing tool with multi-framework compliance mapping."""
 
-__version__ = "1.0.1"
+__version__ = "1.0.2"
 __author__ = "Toc Consulting"
 __email__ = "[email protected]"
 

lambda_security_scanner/checks/access_control.py

@@ -626,3 +626,126 @@ def check_shared_role(
             "shared_count": count,
             "role_arn": role_arn,
         }
+
+    def check_destinations(
+        self, function_name: str, region: str, account_id: str
+    ) -> Dict:
+        """B.6: Check async-invoke destinations for external accounts.
+
+        An attacker who can call PutFunctionEventInvokeConfig points a
+        function's OnSuccess/OnFailure destination (SQS, SNS, Lambda, or
+        EventBridge) at a target in their own account, exfiltrating every
+        async invocation result or payload. The configuration keeps firing
+        after the attacker's access is revoked.
+
+        Args:
+            function_name: Lambda function name or ARN.
+            region: AWS region name.
+            account_id: AWS account ID of the scanning account.
+
+        Returns:
+            Dict with has_destinations, destinations,
+            has_external_destination, and external_destinations.
+        """
+        client = self.get_client("lambda", region)
+        try:
+            response = client.get_function_event_invoke_config(
+                FunctionName=function_name
+            )
+        except ClientError as e:
+            error_code = e.response.get("Error", {}).get(
+                "Code", "Unknown"
+            )
+            if error_code == "ResourceNotFoundException":
+                return {
+                    "has_destinations": False,
+                    "destinations": [],
+                    "has_external_destination": False,
+                    "external_destinations": [],
+                }
+            return self.handle_client_error(
+                e,
+                {
+                    "has_destinations": False,
+                    "destinations": [],
+                    "has_external_destination": False,
+                    "external_destinations": [],
+                },
+            )
+
+        dest_config = response.get("DestinationConfig", {})
+        targets: List[str] = []
+        for key in ("OnSuccess", "OnFailure"):
+            arn = dest_config.get(key, {}).get("Destination")
+            if arn:
+                targets.append(arn)
+
+        external: List[str] = []
+        for arn in targets:
+            # arn:aws:<service>:<region>:<account-id>:<resource>
+            parts = arn.split(":")
+            if len(parts) >= 5:
+                target_account = parts[4]
+                if target_account and target_account != account_id:
+                    external.append(arn)
+
+        return {
+            "has_destinations": len(targets) > 0,
+            "destinations": targets,
+            "has_external_destination": len(external) > 0,
+            "external_destinations": external,
+        }
+
+    def check_aliases(
+        self, function_name: str, region: str
+    ) -> Dict:
+        """B.7: Check for alias traffic shadowing.
+
+        A weighted alias (RoutingConfig.AdditionalVersionWeights) sends a
+        fraction of invocations to a second version. An attacker who can
+        publish a version and call UpdateAlias uses this to quietly serve
+        backdoored code to a slice of traffic while the primary version still
+        looks clean, a stealthy persistence technique. Aliases with a routing
+        config are surfaced for review.
+
+        Args:
+            function_name: Lambda function name or ARN.
+            region: AWS region name.
+
+        Returns:
+            Dict with alias_count, shadowed_aliases, and
+            has_shadowed_alias.
+        """
+        client = self.get_client("lambda", region)
+        try:
+            response = client.list_aliases(
+                FunctionName=function_name
+            )
+        except ClientError as e:
+            return self.handle_client_error(
+                e,
+                {
+                    "alias_count": 0,
+                    "shadowed_aliases": [],
+                    "has_shadowed_alias": False,
+                },
+            )
+
+        aliases = response.get("Aliases", [])
+        shadowed: List[Dict] = []
+        for alias in aliases:
+            weights = alias.get("RoutingConfig", {}).get(
+                "AdditionalVersionWeights", {}
+            )
+            if weights:
+                shadowed.append({
+                    "name": alias.get("Name"),
+                    "primary_version": alias.get("FunctionVersion"),
+                    "additional_versions": weights,
+                })
+
+        return {
+            "alias_count": len(aliases),
+            "shadowed_aliases": shadowed,
+            "has_shadowed_alias": len(shadowed) > 0,
+        }

lambda_security_scanner/scanner.py

@@ -167,7 +167,7 @@ def scan_function(
         progress=None,
         task=None,
     ) -> Dict[str, Any]:
-        """Run all 19 checks for a single function."""
+        """Run all 21 checks for a single function."""
         func_name = func_config.get(
             "FunctionName", "unknown"
         )
@@ -218,7 +218,7 @@ def scan_function(
                 )
             )
 
-            # B. Access Control (B.1-B.5)
+            # B. Access Control (B.1-B.7)
             checks["resource_policy"] = (
                 self.access_checker.check_resource_policy(
                     func_name, self.region
@@ -244,6 +244,16 @@ def scan_function(
                     role_arn, all_role_arns
                 )
             )
+            checks["destinations"] = (
+                self.access_checker.check_destinations(
+                    func_name, self.region, self.account_id
+                )
+            )
+            checks["aliases"] = (
+                self.access_checker.check_aliases(
+                    func_name, self.region
+                )
+            )
 
             # C. Network Security (C.1-C.3)
             checks["vpc_config"] = (
@@ -524,6 +534,31 @@ def add(severity, issue_type, desc, rec):
                 "Create unique roles per function",
             )
 
+        # B.6 External-account async destination
+        dest = checks.get("destinations", {})
+        if dest.get("has_external_destination"):
+            add(
+                "CRITICAL", "external_account_destination",
+                "Async-invoke destination points to an external "
+                "account: "
+                f"{dest.get('external_destinations', [])}",
+                "Remove the cross-account destination and restrict "
+                "lambda:PutFunctionEventInvokeConfig; allow-list any "
+                "intended account",
+            )
+
+        # B.7 Alias traffic shadowing
+        alias = checks.get("aliases", {})
+        if alias.get("has_shadowed_alias"):
+            add(
+                "MEDIUM", "alias_traffic_shadowing",
+                "Alias splits traffic to additional versions "
+                "(possible shadow version): "
+                f"{alias.get('shadowed_aliases', [])}",
+                "Verify the additional versions are intended and "
+                "restrict lambda:UpdateAlias / lambda:PublishVersion",
+            )
+
         # C.1 VPC
         if not checks.get("vpc_config", {}).get("in_vpc"):
             add(

lambda_security_scanner/utils.py

@@ -104,6 +104,14 @@ def get(check_name, key, default=False):
     if get("shared_role", "is_shared"):
         score -= 10
 
+    # B.6: Async-invoke destination to an external account (CRITICAL)
+    if get("destinations", "has_external_destination"):
+        score -= 20
+
+    # B.7: Alias splits traffic to a shadow version (MEDIUM)
+    if get("aliases", "has_shadowed_alias"):
+        score -= 5
+
     # A.6: Tracing not enabled (observability hygiene, LOW)
     if not get("tracing", "enabled"):
         score -= 2

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "lambda-security-scanner"
-version = "1.0.1"
+version = "1.0.2"
 description = "A comprehensive AWS Lambda security scanner with multi-framework compliance mapping"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_access_control.py

@@ -454,3 +454,101 @@ def test_shared_role(self):
         )
         self.assertTrue(result["is_shared"])
         self.assertEqual(result["shared_count"], 2)
+
+
+class TestCheckDestinations(TestCase):
+    """B.6 - Async-invoke destination external-account detection."""
+
+    def setUp(self):
+        self.mock_client = Mock()
+        mock_session = Mock()
+        mock_session.client.return_value = self.mock_client
+        self.checker = AccessControlChecker(lambda: mock_session)
+
+    def test_no_event_invoke_config(self):
+        self.mock_client.get_function_event_invoke_config.side_effect = (
+            _resource_not_found("GetFunctionEventInvokeConfig")
+        )
+        result = self.checker.check_destinations(
+            "fn", "us-east-1", "111111111111"
+        )
+        self.assertFalse(result["has_destinations"])
+        self.assertFalse(result["has_external_destination"])
+
+    def test_external_account_destination(self):
+        self.mock_client.get_function_event_invoke_config.return_value = {
+            "DestinationConfig": {
+                "OnSuccess": {
+                    "Destination":
+                    "arn:aws:sqs:us-east-1:999999999999:attacker-q"
+                },
+                "OnFailure": {
+                    "Destination":
+                    "arn:aws:sqs:us-east-1:111111111111:dlq"
+                },
+            }
+        }
+        result = self.checker.check_destinations(
+            "fn", "us-east-1", "111111111111"
+        )
+        self.assertTrue(result["has_destinations"])
+        self.assertTrue(result["has_external_destination"])
+        self.assertEqual(len(result["external_destinations"]), 1)
+        self.assertIn(
+            "999999999999", result["external_destinations"][0]
+        )
+
+    def test_same_account_destination_not_flagged(self):
+        self.mock_client.get_function_event_invoke_config.return_value = {
+            "DestinationConfig": {
+                "OnSuccess": {
+                    "Destination":
+                    "arn:aws:sns:us-east-1:111111111111:ok"
+                }
+            }
+        }
+        result = self.checker.check_destinations(
+            "fn", "us-east-1", "111111111111"
+        )
+        self.assertTrue(result["has_destinations"])
+        self.assertFalse(result["has_external_destination"])
+
+
+class TestCheckAliases(TestCase):
+    """B.7 - Alias traffic-shadowing detection."""
+
+    def setUp(self):
+        self.mock_client = Mock()
+        mock_session = Mock()
+        mock_session.client.return_value = self.mock_client
+        self.checker = AccessControlChecker(lambda: mock_session)
+
+    def test_no_routing_config_not_flagged(self):
+        self.mock_client.list_aliases.return_value = {
+            "Aliases": [
+                {"Name": "prod", "FunctionVersion": "7"}
+            ]
+        }
+        result = self.checker.check_aliases("fn", "us-east-1")
+        self.assertEqual(result["alias_count"], 1)
+        self.assertFalse(result["has_shadowed_alias"])
+
+    def test_weighted_alias_flagged(self):
+        self.mock_client.list_aliases.return_value = {
+            "Aliases": [{
+                "Name": "prod",
+                "FunctionVersion": "7",
+                "RoutingConfig": {
+                    "AdditionalVersionWeights": {"8": 0.1}
+                },
+            }]
+        }
+        result = self.checker.check_aliases("fn", "us-east-1")
+        self.assertTrue(result["has_shadowed_alias"])
+        self.assertEqual(
+            result["shadowed_aliases"][0]["primary_version"], "7"
+        )
+        self.assertIn(
+            "8",
+            result["shadowed_aliases"][0]["additional_versions"],
+        )