データベースセキュリティ — RDS・Aurora・DynamoDB・Redshift

はじめに

データベースはあらゆるアプリケーションの心臓部。その保護は単なるセキュリティ対策ではなく、事業継続のための必須戦略です。本講義では、AWS三大DBサービス（RDS/Aurora、DynamoDB、Redshift）と補助サービス（ElastiCache）のセキュリティを、試験頻出かつ実務レベルで掘り下げます。

このセクションで学ぶもの：

• 暗号化戦略（保存時・転送時・スナップショット）
• IAM認証とトークンベース認証
• 監査・ロギング・コンプライアンス
• ネットワーク分離とアクセス制御
• よくある失敗パターンと回避方法

RDS 暗号化アーキテクチャ

DynamoDB 細粒度アクセス制御（LeadingKey）

---

第1部：RDS / Aurora セキュリティ

1-1. IAM認証（トークンベース認証）

仕組みの核

RDS IAM認証は、DB接続のたびにAWS STS（Security Token Service）で短期トークンを生成し、それをパスワード代わりに使う方式です。

ユーザー/アプリケーション
    ↓
STS.GetAuthorizationToken() を呼び出し
    ↓
15分有効のトークンを取得
    ↓
RDS に接続（ユーザー名@トークン）
    ↓
RDS が IAM 認証を検証 ✓
    ↓
接続確立

メリット

設定手順

Step 1: IAMポリシー作成

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "rds-db:connect",
      "Resource": "arn:aws:rds:us-east-1:123456789012:db:mydb"
    }
  ]
}

Step 2: RDS IAM 認証有効化

aws rds modify-db-instance \
  --db-instance-identifier mydb \
  --enable-iam-database-authentication \
  --apply-immediately

Step 3: IBMユーザー作成（PostgreSQL例）

CREATE USER app_user WITH LOGIN;
GRANT rds_iam TO app_user;

Step 4: クライアントから接続

import boto3
import pymysql
from datetime import datetime

rds_client = boto3.client('rds', region_name='us-east-1')

# トークン生成（15分有効）
token = rds_client.generate_db_auth_token(
    DBHostname='mydb.xxxxx.us-east-1.rds.amazonaws.com',
    Port=3306,
    DBUser='app_user',
    Region='us-east-1'
)

# MySQL接続（パスワードの代わりにトークン）
connection = pymysql.connect(
    host='mydb.xxxxx.us-east-1.rds.amazonaws.com',
    user='app_user',
    password=token,  # ← トークンがパスワード
    database='mydb'
)

試験で狙われるポイント

• 有効期限は15分（AWS docs: exactly 15 minutes）
• トークンにはIAMロール/ユーザー情報が含まれ、RDSがIAM検証
• CloudTrailでrds-db:connectアクションが記録される
• マルチAZレプリカでも同じトークンが有効（全AZで検証）

---

1-2. 暗号化戦略

暗号化の三層

RDS暗号化は「保存時」「転送時」「バックアップ」の三層で構成：

保存時暗号化（At-Rest Encryption）

KMS統合：

# 暗号化有効化（新規インスタンス作成時）
aws rds create-db-instance \
  --db-instance-identifier encrypted-mysql \
  --db-instance-class db.t3.micro \
  --engine mysql \
  --master-username admin \
  --master-user-password MySecurePass123! \
  --allocated-storage 100 \
  --storage-encrypted \
  --kms-key-id arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012 \
  --region us-east-1

キーの選択肢：

カスタマー管理キーの例：

# KMS キー作成
KEY_ID=$(aws kms create-key \
  --description "RDS encryption key" \
  --region us-east-1 \
  --query 'Key.KeyId' \
  --output text)

# RDS が使用できるようにキーポリシーを調整
aws kms put-key-policy \
  --key-id $KEY_ID \
  --policy-name default \
  --policy '{...RDS service principal を許可...}'

# RDS インスタンス作成
aws rds create-db-instance \
  --db-instance-identifier mydb \
  --engine postgres \
  --kms-key-id $KEY_ID \
  --storage-encrypted

転送時暗号化（In-Transit Encryption）

RDS はデフォルトで SSL/TLS を使用。だが認証を厳格にするには：

# MySQL 5.7+ では REQUIRE SSL を強制
mysql> ALTER USER 'app_user'@'%' REQUIRE SSL;

# PostgreSQL では SSL を強制
postgres=> ALTER USER app_user WITH PASSWORD 'xxx';
postgres=> ALTER ROLE app_user VALID UNTIL 'infinity';

クライアント側：

import pymysql

connection = pymysql.connect(
    host='mydb.xxxxx.us-east-1.rds.amazonaws.com',
    user='app_user',
    password='password',
    ssl_ca='/path/to/rds-ca-2019-root.pem',  # RDS CA証明書
    ssl_verify_cert=True,
    ssl_verify_identity=True
)

暗号化されていないDBを暗号化する

重要：RDS では「既存インスタンスに暗号化を後付けできない」

必ずこの手順を実施：

既存DBインスタンス（未暗号化）
    ↓
スナップショット作成
    ↓
スナップショットを暗号化コピー
    ↓
暗号化スナップショットから新インスタンス作成
    ↓
アプリケーション切り替え
    ↓
既存インスタンス削除

実装例：

# Step 1: スナップショット作成
aws rds create-db-snapshot \
  --db-instance-identifier mydb-unencrypted \
  --db-snapshot-identifier mydb-snapshot-001

# Step 2: スナップショットが AVAILABLE になるまで待機
aws rds wait db-snapshot-available \
  --db-snapshot-identifier mydb-snapshot-001

# Step 3: スナップショットを暗号化コピー
aws rds copy-db-snapshot \
  --source-db-snapshot-identifier mydb-snapshot-001 \
  --target-db-snapshot-identifier mydb-snapshot-encrypted \
  --kms-key-id arn:aws:kms:us-east-1:123456789012:key/12345678-... \
  --storage-encrypted

# Step 4: 暗号化スナップショットから新インスタンス復元
aws rds restore-db-instance-from-db-snapshot \
  --db-instance-identifier mydb-encrypted \
  --db-snapshot-identifier mydb-snapshot-encrypted

# Step 5: エンドポイント切り替え（Route 53、セキュリティグループ等）
# → 既存アプリケーションを新インスタンスに指す

クロスリージョンレプリカの暗号化

レプリケーション中も暗号化が保たれます：

# プライマリ（us-east-1）が KMS キー A で暗号化されている場合
aws rds create-db-instance-read-replica \
  --db-instance-identifier mydb-replica \
  --source-db-instance-identifier mydb \
  --region us-west-2 \
  --storage-encrypted \
  --kms-key-id arn:aws:kms:us-west-2:123456789012:key/...  # 別リージョンのキー

重要なポイント：

• プライマリとレプリカは異なるKMSキーを使用可能
• データ転送中も暗号化される
• レプリカ昇格後、プライマリと独立した暗号化ポリシーが適用

---

1-3. 監査ログと監視

監査ログの種類（エンジン別）

MySQL Audit Plugin の有効化

# パラメータグループ作成
aws rds create-db-parameter-group \
  --db-parameter-group-name mysql-audit-params \
  --db-parameter-group-family mysql8.0 \
  --description "MySQL with audit enabled"

# Audit Plugin を有効化
aws rds modify-db-parameter-group \
  --db-parameter-group-name mysql-audit-params \
  --parameters \
    ParameterName=server_audit_logging,ParameterValue=1,ApplyMethod=immediate \
    ParameterName=server_audit_events,ParameterValue="CONNECT,QUERY_DCL,QUERY_DDL",ApplyMethod=immediate

PostgreSQL pgAudit の設定

# RDS パラメータグループで pgAudit を有効化
aws rds modify-db-parameter-group \
  --db-parameter-group-name postgres-audit \
  --parameters \
    ParameterName=pgaudit.log,ParameterValue=DDL,ApplyMethod=pending-reboot

# RDS インスタンスに適用して再起動
aws rds modify-db-instance \
  --db-instance-identifier mydb \
  --db-parameter-group-name postgres-audit \
  --apply-immediately

pgaudit.log の値：

pgaudit.log = 'ALL'      # すべてのログ
pgaudit.log = 'DDL'      # CREATE, ALTER, DROP
pgaudit.log = 'DML'      # SELECT, INSERT, UPDATE, DELETE
pgaudit.log = 'FUNCTION' # 関数実行

---

1-4. Aurora Activity Streams

Aurora 専有の強力な監査機能。すべてのDBアクティビティをリアルタイムで Kinesis ストリームに出力します。

Activity Streams の有効化

aws rds start-activity-stream \
  --resource-arn arn:aws:rds:us-east-1:123456789012:db:myaurora \
  --mode async  # または sync（遅延許容度による）

Activity Stream イベント例

{
  "type": "record",
  "version": "1.0",
  "databaseActivityEvents": [
    {
      "type": "heartbeat"
    },
    {
      "type": "database_activity",
      "timestamp": 1672531200000000,
      "statementStartTime": 1672531200000000,
      "statementStopTime": 1672531200000100,
      "databaseUser": "app_user",
      "databaseHost": "10.0.1.100",
      "sid": "123456",
      "sqlText": "SELECT * FROM customers WHERE id = ?",
      "statementType": "SELECT",
      "objectType": "TABLE",
      "objectName": "customers"
    }
  ]
}

Kinesis → Lambda → CloudWatch Logs へのパイプライン

# Lambda handler
import json
import boto3
import base64

logs_client = boto3.client('logs')

def lambda_handler(event, context):
    for record in event['Records']:
        # Kinesis レコードをデコード
        payload = base64.b64decode(record['kinesis']['data'])
        activity_events = json.loads(payload)
        
        # CloudWatch Logs に出力
        for event_data in activity_events.get('databaseActivityEvents', []):
            if event_data['type'] == 'database_activity':
                log_message = f"User: {event_data['databaseUser']}, SQL: {event_data['sqlText']}"
                
                logs_client.put_log_events(
                    logGroupName='/aws/rds/activity-stream',
                    logStreamName='aurora-activity',
                    logEvents=[{
                        'timestamp': int(event_data['timestamp'] / 1000),
                        'message': log_message
                    }]
                )
    
    return {'statusCode': 200}

Activity Streams のモード

---

1-5. AWS Secrets Manager 自動ローテーション

DB認証情報をアプリケーションに直接保存せず、Secrets Manager で管理・自動ローテーション。

ローテーションの仕組み

┌─────────────────────────────────────────┐
│ Secrets Manager                         │
│  Secret: RDS/mysql/prod                 │
│  ├─ Current Version (Version A)          │
│  └─ Previous Version (Version B)         │
└─────────────┬───────────────────────────┘
              │
           毎日 1:00 AM
         Lambda Rotation関数実行
              │
     ┌────────┴─────────┐
     ↓                  ↓
Step 1: New           Step 2: Set
Secret生成            RDS に新パス登録
                     
     ↓                  ↓
Step 3: Test         Step 4: Finish
接続確認              Secrets更新

Secrets Manager でシークレット作成

aws secretsmanager create-secret \
  --name prod/rds/mysql \
  --secret-string '{"username":"admin","password":"MySecurePass123!","engine":"mysql","host":"mydb.xxxxx.us-east-1.rds.amazonaws.com","port":3306,"dbname":"production"}'

Lambda ローテーション関数

import boto3
import json
import pymysql

sm_client = boto3.client('secretsmanager')
rds_client = boto3.client('rds')

def lambda_handler(event, context):
    secret_id = event['SecretId']
    operation = event['ClientRequestToken']
    
    # ステップ1: 新しいシークレットを生成
    if 'secret' not in event:
        # 現在のシークレット取得
        response = sm_client.get_secret_value(SecretId=secret_id)
        secret = json.loads(response['SecretString'])
        
        # 新しいパスワード生成
        new_password = sm_client.get_random_password(PasswordLength=32)['RandomPassword']
        
        # 新しいシークレットをSecrets Managerに保存（未確定）
        new_secret = secret.copy()
        new_secret['password'] = new_password
        
        sm_client.put_secret_value(
            SecretId=secret_id,
            ClientRequestToken=operation,
            SecretString=json.dumps(new_secret),
            VersionStages=['AWSPENDING']
        )
    
    # ステップ2: RDS で新しいパスワードを設定
    secret = json.loads(sm_client.get_secret_value(
        SecretId=secret_id,
        VersionId=operation,
        VersionStage='AWSPENDING'
    )['SecretString'])
    
    connection = pymysql.connect(
        host=secret['host'],
        user=secret['username'],
        password=secret['password'],
        database=secret['dbname']
    )
    
    cursor = connection.cursor()
    # 新しいパスワードでRDSユーザーを更新
    cursor.execute(
        f"ALTER USER '{secret['username']}'@'%' IDENTIFIED BY %s",
        (secret['password'],)
    )
    connection.commit()
    connection.close()
    
    # ステップ3: 接続テスト
    try:
        test_conn = pymysql.connect(
            host=secret['host'],
            user=secret['username'],
            password=secret['password'],
            database=secret['dbname']
        )
        test_conn.close()
    except Exception as e:
        # テスト失敗時はローテーションを中止
        sm_client.cancel_rotate_secret(SecretId=secret_id)
        raise e
    
    # ステップ4: シークレット確定
    sm_client.update_secret_version_stage(
        SecretId=secret_id,
        VersionStage='AWSCURRENT',
        MoveToVersionId=operation,
        RemoveFromVersionId=event['ClientRequestToken']
    )
    
    return {'statusCode': 200, 'message': 'Rotation successful'}

ローテーション設定

# Lambda実行ロール作成（Secrets Manager と RDS へのアクセス許可）
aws iam create-role \
  --role-name RDSSecretsRotationRole \
  --assume-role-policy-document '{
    "Version":"2012-10-17",
    "Statement":[{
      "Effect":"Allow",
      "Principal":{"Service":"lambda.amazonaws.com"},
      "Action":"sts:AssumeRole"
    }]
  }'

# インラインポリシーアタッチ
aws iam put-role-policy \
  --role-name RDSSecretsRotationRole \
  --policy-name SecretsRotationPolicy \
  --policy-document '{...Secrets Manager, RDS へのアクセス...}'

# ローテーション設定
aws secretsmanager rotate-secret \
  --secret-id prod/rds/mysql \
  --rotation-lambda-arn arn:aws:lambda:us-east-1:123456789012:function:rotate-rds-secret \
  --rotation-rules AutomaticallyAfterDays=30

試験で狙われるポイント

• 4段階：Secret生成 → パスワード設定 → テスト → 確定
• Lambda関数が VPC エンドポイント経由で Secrets Manager と RDS にアクセス
• ローテーション中も既存接続は有効（AWSCURRENT と AWSPREVIOUS が併存）

---

1-6. RDS Proxy

データベース接続の「プール」を管理し、IAM認証、Secrets Manager との連携、自動フェイルオーバーをサポート。

RDS Proxy のメリット

┌──────────────────────────────────┐
│  アプリケーション多数            │
│  ├─ App1 (connection pool: 10)   │
│  ├─ App2 (connection pool: 10)   │
│  └─ App3 (connection pool: 10)   │
│  Total: 30 connections           │
└──────────────────────────────────┘
            ↓ (TCP)
┌──────────────────────────────────┐
│  RDS Proxy                       │
│  ├─ Connection pooling (max 100) │
│  └─ IAM authentication           │
└──────────────────────────────────┘
            ↓ (TCP)
┌──────────────────────────────────┐
│  RDS Instance                    │
│  └─ Max connections: 20 ✓        │
└──────────────────────────────────┘

❌ なし: 30接続 > 20接続制限（エラー多発）
✅ あり: 30接続を効率的に20接続に圧縮

RDS Proxy 作成

aws rds create-db-proxy \
  --db-proxy-name prod-mysql-proxy \
  --engine MYSQL \
  --role-arn arn:aws:iam::123456789012:role/RDSProxyRole \
  --max-connections 100 \
  --max-idle-connections 50 \
  --connection-borrow-timeout 120 \
  --init-query "SET SESSION max_connections = 1000" \
  --debug-logging false

IAM認証との連携

# プロキシターゲットグループ作成（IAM認証有効）
aws rds register-db-proxy-targets \
  --db-proxy-name prod-mysql-proxy \
  --target-group-name default \
  --db-instance-identifiers mydb \
  --auth-scheme SECRETS  # または BASIC

# プロキシエンドポイント経由で接続
# mydb.123456789012.us-east-1.rds.amazonaws.com → prod-mysql-proxy.proxy-123456789012.us-east-1.rds.amazonaws.com

Secrets Manager 連携

# Secrets Manager に認証情報保存
aws secretsmanager create-secret \
  --name prod/rds/mysql-proxy \
  --secret-string '{"username":"app_user","password":"SecurePass123"}'

# プロキシが Secrets Manager から認証情報を自動取得・使用

試験で狙われるポイント

• Connection multiplexing で物理接続数を削減
• IAM認証 + Secrets Manager で認証を一元化
• アプリケーション側は接続文字列をプロキシエンドポイントに変更するだけ

---

第2部：DynamoDB セキュリティ

2-1. 細粒度アクセス制御（Fine-Grained Access Control）

DynamoDB は IAM ポリシーでテーブルレベル・アイテムレベルの制御が可能。

テーブルレベルの制御

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:GetItem",
        "dynamodb:Query"
      ],
      "Resource": "arn:aws:dynamodb:us-east-1:123456789012:table/Orders"
    },
    {
      "Effect": "Deny",
      "Action": "dynamodb:DeleteItem",
      "Resource": "arn:aws:dynamodb:us-east-1:123456789012:table/Orders"
    }
  ]
}

リーディングキー（Partition Key）に基づく制御

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:Query",
        "dynamodb:GetItem"
      ],
      "Resource": "arn:aws:dynamodb:us-east-1:123456789012:table/Orders",
      "Condition": {
        "ForAllValues:StringEquals": {
          "dynamodb:LeadingKeys": ["${aws:username}"]
        }
      }
    }
  ]
}

説明：

• ユーザー alice は Partition Key が alice のアイテムのみ読み取り可能
• ユーザー bob は Partition Key が bob のアイテムのみ読み取り可能
• ユーザーが他人のデータにアクセスしようとするとアクセス拒否

属性レベルの制御

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "dynamodb:GetItem",
      "Resource": "arn:aws:dynamodb:us-east-1:123456789012:table/Employees",
      "Condition": {
        "ForAllValues:StringEquals": {
          "dynamodb:Attributes": ["Name", "Title", "Department"]
        }
      }
    }
  ]
}

説明：

• Name, Title, Department の属性のみ読み取り可能
• Salary, SSN などの機密属性は読み取り不可

複合制御の例

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "UserOwnDataOnly",
      "Effect": "Allow",
      "Action": [
        "dynamodb:GetItem",
        "dynamodb:UpdateItem"
      ],
      "Resource": "arn:aws:dynamodb:us-east-1:123456789012:table/UserProfiles",
      "Condition": {
        "ForAllValues:StringEquals": {
          "dynamodb:LeadingKeys": ["${aws:username}"],
          "dynamodb:Attributes": ["Email", "Phone", "Address"]
        }
      }
    }
  ]
}

試験で狙われるポイント：

• LeadingKeys（Partition Key）での行レベル制御
• Attributes での列レベル制御
• テーブルレベルではなくアイテムレベルのキー選択は不可（テーブル全体が制御単位）

---

2-2. DynamoDB 暗号化

キータイプの選択

AWS管理キーを使用

# テーブル作成時に暗号化有効化（デフォルト）
aws dynamodb create-table \
  --table-name Orders \
  --attribute-definitions AttributeName=OrderId,AttributeType=S \
  --key-schema AttributeName=OrderId,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST \
  --sse-specification Enabled=true,SSEType=KMS

カスタマー管理キーを使用

# Step 1: KMS キー作成
KEY_ID=$(aws kms create-key \
  --description "DynamoDB encryption key" \
  --region us-east-1 \
  --query 'Key.KeyId' \
  --output text)

# Step 2: DynamoDB テーブル作成
aws dynamodb create-table \
  --table-name Orders \
  --attribute-definitions AttributeName=OrderId,AttributeType=S \
  --key-schema AttributeName=OrderId,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST \
  --sse-specification \
    Enabled=true,SSEType=KMS,KeyArn=arn:aws:kms:us-east-1:123456789012:key/$KEY_ID

# Step 3: キーポリシーで DynamoDB サービスプリンシパルを許可
aws kms put-key-policy \
  --key-id $KEY_ID \
  --policy-name default \
  --policy '{
    "Sid": "AllowDynamoDB",
    "Principal": {
      "Service": "dynamodb.amazonaws.com"
    },
    "Action": ["kms:Decrypt", "kms:GenerateDataKey"],
    "Resource": "*"
  }'

Global Tables での暗号化

複数リージョン間でレプリケートされるGlobal Tablesでも、各リージョンで個別にキーを設定可能：

# プライマリ（us-east-1）: キーA で暗号化
aws dynamodb create-global-table \
  --global-table-name OrdersGlobal \
  --replication-group RegionName=us-east-1,RegionName=eu-west-1

# レプリカ（eu-west-1）: キーB で暗号化
aws dynamodb update-table \
  --table-name OrdersGlobal \
  --sse-specification Enabled=true,SSEType=KMS,KeyArn=arn:aws:kms:eu-west-1:123456789012:key/...

試験で狙われるポイント：

• キー選択はテーブルレベル（アイテムレベルの異なるキーは不可）
• Global Tables は各リージョンで独立したキー管理

---

2-3. VPC Endpoints（Gateway型）

DynamoDB へのアクセスをVPC内に限定し、インターネット経由を避ける。

VPC Endpoint 作成

# Gateway型 VPC Endpoint 作成
aws ec2 create-vpc-endpoint \
  --vpc-id vpc-12345678 \
  --service-name com.amazonaws.us-east-1.dynamodb \
  --route-table-ids rtb-12345678 \b  --region us-east-1

ルートテーブルの自動更新

VPC Endpointを作成すると、ルートテーブルが自動更新：

宛先              ターゲット
0.0.0.0/0         igw-xxxxx（インターネットゲートウェイ）
<DynamoDBプレフィックスリスト>  vpce-xxxxx（VPC Endpoint）

VPC Endpoint ポリシー（アクセス制限）

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": "*",
      "Action": "dynamodb:*",
      "Resource": "*",
      "Condition": {
        "StringEquals": {
          "aws:SourceVpc": "vpc-12345678"
        }
      }
    },
    {
      "Effect": "Deny",
      "Principal": "*",
      "Action": "dynamodb:DeleteTable",
      "Resource": "*"
    }
  ]
}

説明：

• VPC 内からのすべての DynamoDB アクセスを許可
• DeleteTable は明示的に拒否

EC2 / ECS / Lambda からのアクセス

import boto3

# VPC内のEC2/Lambda から実行
# → 自動的に VPC Endpoint 経由でルーティング
# → インターネット経由を経由しない

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Orders')

response = table.get_item(
    Key={
        'OrderId': 'order-001'
    }
)

試験で狙われるポイント：

• DynamoDB VPC Endpoint は Gateway型（Interface型ではない）
• ポリシーで特定テーブルへのアクセスのみ許可する制御可能

---

2-4. バックアップと復旧

PITR（Point-In-Time Recovery）

最大35日前の任意の時点にテーブルを復旧。

# PITR 有効化
aws dynamodb update-continuous-backups \
  --table-name Orders \
  --point-in-time-recovery-specification PointInTimeRecoveryEnabled=true

# 復旧（7時間前の状態に戻す）
aws dynamodb restore-table-to-point-in-time \
  --source-table-name Orders \
  --target-table-name Orders-Restored \
  --restore-date-time $(date -u -d '7 hours ago' +%s)

オンデマンドバックアップ

手動でスナップショットを作成。保持期間に制限なし。

# バックアップ作成
aws dynamodb create-backup \
  --table-name Orders \
  --backup-name Orders-Backup-$(date +%Y%m%d)

# バックアップから復旧
aws dynamodb restore-table-from-backup \
  --target-table-name Orders-Restored \
  --backup-arn arn:aws:dynamodb:us-east-1:123456789012:table/Orders/backup/01234567890123-abc12345

暗号化されたテーブルのバックアップ

PITR・オンデマンドバックアップとも、元のテーブルの暗号化設定が引き継がれます。復旧後も同じキーで暗号化されたテーブルが作成されます。

---

2-5. DynamoDB Streams + Lambda（監査パターン）

すべてのアイテム変更をログに記録するシステムを構築。

┌─────────────────────────┐
│ DynamoDB テーブル       │
│  └─ Item Insert/Update  │
└────────────┬────────────┘
             │
        ┌────┴─────────────┐
        ↓                  ↓
  [DynamoDB Streams]  [STREAM_SPEC]
  ├─ NEW_IMAGE        ├─ KEYS_ONLY
  ├─ OLD_IMAGE        ├─ NEW_IMAGE
  ├─ NEW_AND_OLD_IMAGE ├─ NEW_AND_OLD_IMAGE
  └─ KEYS_ONLY        └─ STREAM_ENABLED
                           ↓
                      [Lambda トリガー]
                        ↓
                  [CloudWatch Logs]
                  [Elasticsearch]
                  [S3 (監査ログ)]

DynamoDB Streams の有効化

aws dynamodb update-table \
  --table-name Orders \
  --stream-specification \
    StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGE \
  --region us-east-1

StreamViewType の違い：

Lambda ハンドラ（監査ロギング）

import json
import boto3
from datetime import datetime

logs_client = boto3.client('logs')
s3_client = boto3.client('s3')

def lambda_handler(event, context):
    audit_logs = []
    
    for record in event['Records']:
        # DynamoDB Stream イベントをパース
        stream_event = {
            'eventID': record['eventID'],
            'eventName': record['eventName'],  # INSERT, MODIFY, REMOVE
            'eventSource': record['eventSource'],
            'dynamodb': record['dynamodb']
        }
        
        # 変更内容を抽出
        if record['eventName'] == 'INSERT':
            item = record['dynamodb']['NewImage']
            action = 'CREATE'
        elif record['eventName'] == 'MODIFY':
            old_item = record['dynamodb']['OldImage']
            new_item = record['dynamodb']['NewImage']
            action = 'UPDATE'
        elif record['eventName'] == 'REMOVE':
            item = record['dynamodb']['OldImage']
            action = 'DELETE'
        
        # 監査ログ作成
        audit_log = {
            'timestamp': datetime.utcnow().isoformat(),
            'action': action,
            'itemId': record['dynamodb']['Keys'],
            'oldImage': record['dynamodb'].get('OldImage'),
            'newImage': record['dynamodb'].get('NewImage')
        }
        
        audit_logs.append(audit_log)
        
        # CloudWatch Logs に出力
        logs_client.put_log_events(
            logGroupName='/aws/dynamodb/audit',
            logStreamName='orders-audit',
            logEvents=[{
                'timestamp': int(datetime.utcnow().timestamp() * 1000),
                'message': json.dumps(audit_log)
            }]
        )
    
    # S3 にも長期保存
    s3_client.put_object(
        Bucket='audit-logs-bucket',
        Key=f"dynamodb-audit/{datetime.utcnow().strftime('%Y-%m-%d/%H:%M:%S')}.json",
        Body=json.dumps(audit_logs),
        ServerSideEncryption='AES256'
    )
    
    return {'statusCode': 200, 'processedRecords': len(event['Records'])}

Lambda トリガー設定

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:dynamodb:us-east-1:123456789012:table/Orders/stream/2024-01-01T00:00:00.000 \
  --function-name AuditLoggingFunction \
  --enabled true \
  --starting-position LATEST \
  --batch-size 100

試験で狙われるポイント：

• NEW_AND_OLD_IMAGE で変更前後を両方記録
• Lambda は自動的に Stream イベントを消費
• CloudWatch Logs + S3 で複数層の監査ログを確保

---

2-6. Global Tables のセキュリティ

複数リージョン間でレプリケート。各リージョンで独立したキー・IAMポリシーを適用可能。

# Global Table 作成
aws dynamodb create-global-table \
  --global-table-name OrdersGlobal \
  --replication-group RegionName=us-east-1 RegionName=eu-west-1 RegionName=ap-northeast-1

# 各リージョンで異なるKMSキーを適用
# us-east-1 側: キー A で暗号化
# eu-west-1 側: キー B で暗号化
# ap-northeast-1 側: キー C で暗号化

# 各リージョンで異なるIAMポリシーを適用可能
# us-east-1: 特定のロールのみアクセス
# eu-west-1: 別のロールのみアクセス

試験で狙われるポイント：

• 各レプリカリージョンで独立した暗号化キーを指定可能
• 各リージョンで個別のIAMポリシーを適用

---

第3部：ElastiCache セキュリティ

3-1. 暗号化（保存時・転送時）

保存時暗号化（At-Rest Encryption）

Redis クラスタで保存データを KMS で暗号化。

aws elasticache create-cache-cluster \
  --cache-cluster-id prod-redis \
  --cache-node-type cache.t3.micro \
  --engine redis \
  --engine-version 7.0 \
  --num-cache-nodes 1 \
  --at-rest-encryption-enabled

転送時暗号化（In-Transit Encryption）

クライアント ↔ ElastiCache 間を TLS で暗号化。

aws elasticache create-cache-cluster \
  --cache-cluster-id prod-redis-tls \
  --cache-node-type cache.t3.micro \
  --engine redis \
  --transit-encryption-enabled \
  --transit-encryption-mode preferred  # preferred または required

TLS モード：

両方を有効化（推奨）

aws elasticache create-replication-group \
  --replication-group-id prod-redis-secure \
  --replication-group-description "Production Redis with encryption" \
  --engine redis \
  --cache-node-type cache.r6g.large \
  --num-cache-clusters 2 \
  --automatic-failover-enabled \
  --at-rest-encryption-enabled \
  --transit-encryption-enabled \
  --transit-encryption-mode required \
  --auth-token xxxxxxxxxxxxxxxx  # Redis AUTH トークン

---

3-2. Redis AUTH と ACL

Redis AUTH（シンプル認証）

# AUTH トークン指定（推奨：32文字以上）
aws elasticache create-cache-cluster \
  --cache-cluster-id prod-redis \
  --auth-token "$(openssl rand -base64 32)" \
  --engine redis

クライアント側で接続：

import redis

r = redis.Redis(
    host='prod-redis.xxxxx.xxxxxx.cache.amazonaws.com',
    port=6379,
    password='<AUTH_TOKEN>',  # AUTH トークン
    ssl=True,
    ssl_certfile='/path/to/ca.pem'
)

r.set('key', 'value')

Redis ACL（アクセス制御リスト）Redis 6.0+

複数ユーザー、権限ベースのアクセス制御。

# default ユーザーのパスワード変更
aws elasticache modify-user \
  --user-id default \
  --auth-string "NewPassword123!" \
  --no-password-required false

# 新しいユーザー作成（読み取りのみ）
aws elasticache create-user \
  --user-id readonly-user \
  --user-name readonly \
  --access-string "on >Password123! ~* +@read" \
  --engine redis

# ユーザーグループ作成
aws elasticache create-user-group \
  --user-group-id readonly-group \
  --user-ids readonly-user

ACL 権限フォーマット：

+@read      : 読み取り命令を許可
+@write     : 書き込み命令を許可
+@all       : すべての命令を許可
-@write     : 書き込み命令を拒否
~*          : すべてのキーにアクセス可能
~user:*     : user: で始まるキーのみ

---

3-3. サブネットグループ

ElastiCache をVPC内の特定サブネットに限定。

# サブネットグループ作成
aws elasticache create-cache-subnet-group \
  --cache-subnet-group-name prod-redis-subnet \
  --cache-subnet-group-description "Private subnets for Redis" \
  --subnet-ids subnet-12345678 subnet-87654321

# キャッシュクラスタ作成（サブネットグループを指定）
aws elasticache create-cache-cluster \
  --cache-cluster-id prod-redis \
  --cache-subnet-group-name prod-redis-subnet \
  --engine redis

セキュリティグループ設定：

# Redis 用セキュリティグループ作成
SG_ID=$(aws ec2 create-security-group \
  --group-name redis-sg \
  --description "ElastiCache Redis" \
  --vpc-id vpc-12345678 \
  --query 'GroupId' --output text)

# アプリケーション層からの 6379 へのアクセスを許可
aws ec2 authorize-security-group-ingress \
  --group-id $SG_ID \
  --protocol tcp \
  --port 6379 \
  --source-security-group-id sg-app-layer

# キャッシュクラスタに適用
aws elasticache modify-cache-cluster \
  --cache-cluster-id prod-redis \
  --security-group-ids $SG_ID

---

第4部：Redshift セキュリティ

4-1. 暗号化（KMS / CloudHSM）

KMS での暗号化

# キー作成
KEY_ARN=$(aws kms create-key \
  --description "Redshift encryption key" \
  --region us-east-1 \
  --query 'Key.Arn' --output text)

# Redshift クラスタ作成（KMS 暗号化）
aws redshift create-cluster \
  --cluster-identifier prod-redshift \
  --node-type dc2.large \
  --master-username admin \
  --master-user-password MySecurePass123! \
  --number-of-nodes 2 \
  --encrypted \
  --kms-key-id $KEY_ARN

CloudHSM での暗号化（FIPS 140-2 Level 3）

厳格なコンプライアンス要件がある場合。

# CloudHSM クラスタ作成
aws cloudhsm create-cluster \
  --hsm-type hsm1.medium \
  --subnet-ids subnet-12345678 subnet-87654321 \
  --availability-zone us-east-1a

# CloudHSM クラスタIDを取得
CLUSTER_ID=$(aws cloudhsm describe-clusters \
  --query 'Clusters[0].ClusterId' --output text)

# Redshift で CloudHSM を使用
aws redshift create-cluster \
  --cluster-identifier prod-redshift-hsm \
  --node-type dc2.large \
  --master-username admin \
  --master-user-password MySecurePass123! \
  --number-of-nodes 2 \
  --hsm-client-certificate-identifier <CERT_ID> \
  --hsm-configuration-identifier <CONFIG_ID>

KMS vs CloudHSM：

---

4-2. 監査ログ

クエリ実行ログ

# ロギング有効化
aws redshift modify-cluster \
  --cluster-identifier prod-redshift \
  --enable-logging \
  --logging-properties BucketName=audit-logs-bucket,S3KeyPrefix=redshift-logs/

ログに記録される情報

timestamp | xid | userid | database | query | rows_returned | conn_source | app_name | client_address
2024-01-15 10:30:45 | 12345 | awsuser | analytics | SELECT * FROM sales | 10000 | INTERNAL | psql | 10.0.1.100

CloudTrail でのAPI監査

# CloudTrail で Redshift API 呼び出しを記録
aws cloudtrail create-trail \
  --name redshift-api-audit \
  --s3-bucket-name api-audit-bucket \
  --include-global-service-events

aws cloudtrail start-logging --trail-name redshift-api-audit

記録されるイベント：

• CreateCluster, DeleteCluster, ModifyCluster
• CreateUser, DeleteUser, ModifyUser
• AuthorizeClusterSecurityGroupIngress
• クエリなど

---

4-3. Redshift Spectrum のアクセス制御

Redshift が S3 データにアクセスするための IAM ロール設定。

# IAM ロール作成（Redshift 用）
ROLE_ARN=$(aws iam create-role \
  --role-name RedshiftSpectrumRole \
  --assume-role-policy-document '{
    "Version":"2012-10-17",
    "Statement":[{
      "Effect":"Allow",
      "Principal":{"Service":"redshift.amazonaws.com"},
      "Action":"sts:AssumeRole"
    }]
  }' \
  --query 'Role.Arn' --output text)

# S3 アクセス権限追加
aws iam put-role-policy \
  --role-name RedshiftSpectrumRole \
  --policy-name S3Access \
  --policy-document '{
    "Version":"2012-10-17",
    "Statement":[{
      "Effect":"Allow",
      "Action":"s3:GetObject",
      "Resource":"arn:aws:s3:::data-lake-bucket/*"
    }]
  }'

# Redshift クラスタに IAM ロール関連付け
aws redshift associate-iam-role \
  --cluster-identifier prod-redshift \
  --iam-role-arn $ROLE_ARN

Spectrum 外部テーブルから S3 データを読み取り：

CREATE EXTERNAL SCHEMA spectrum_schema
  FROM DATA CATALOG DATABASE 'default'
  IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftSpectrumRole';

CREATE EXTERNAL TABLE spectrum_sales(
  sales_id INT,
  amount DECIMAL(10,2),
  sale_date DATE
)
STORED AS PARQUET
LOCATION 's3://data-lake-bucket/sales/2024/'
TABLE PROPERTIES ('classification' = 'parquet');

SELECT * FROM spectrum_schema.spectrum_sales WHERE amount > 1000;

---

第5部：連携パターン・ベストプラクティス

5-1. IAM認証 + Secrets Manager + RDS Proxy

┌──────────────────────────┐
│ アプリケーション          │
└────────────┬─────────────┘
             │
        (IAMロール)
             │
        ┌────┴─────────────────┐
        ↓                      ↓
    Secrets Manager      IAM STS
      (DB credentials)   (トークン)
             │                │
        ┌────┴────────────────┴───┐
        ↓                         ↓
   ┌─────────────────────────────────┐
   │ RDS Proxy                       │
   │ ├─ Connection Pooling           │
   │ └─ Credential Rotation          │
   └─────────────────────────────────┘
             │
        ┌────┴────────┐
        ↓             ↓
    RDS Instance  RDS Read Replica
    (Encrypted)   (Encrypted)

実装例（Python）：

import boto3
import sqlalchemy as sa
from sqlalchemy import create_engine

# Secrets Manager から認証情報を取得
sm_client = boto3.client('secretsmanager')
secret = sm_client.get_secret_value(SecretId='prod/rds/mysql')

credentials = json.loads(secret['SecretString'])

# RDS Proxy エンドポイントに接続（認証情報は Secrets Manager から）
engine = create_engine(
    f"mysql+pymysql://{credentials['username']}:{credentials['password']}@"
    f"prod-mysql-proxy.proxy-123456789012.us-east-1.rds.amazonaws.com:3306"
    f"/{credentials['dbname']}",
    pool_size=10,
    max_overflow=20,
    pool_timeout=30
)

with engine.connect() as conn:
    result = conn.execute(sa.text("SELECT * FROM users"))
    for row in result:
        print(row)

5-2. DynamoDB + Secrets Manager + IAM

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ReadOwnData",
      "Effect": "Allow",
      "Action": "dynamodb:GetItem",
      "Resource": "arn:aws:dynamodb:us-east-1:123456789012:table/UserData",
      "Condition": {
        "ForAllValues:StringEquals": {
          "dynamodb:LeadingKeys": ["${aws:username}"]
        }
      }
    },
    {
      "Sid": "SecretsAccess",
      "Effect": "Allow",
      "Action": "secretsmanager:GetSecretValue",
      "Resource": "arn:aws:secretsmanager:us-east-1:123456789012:secret:dynamodb/*"
    }
  ]
}

5-3. Aurora Activity Streams → Elasticsearch

リアルタイム監査ログの可視化。

import json
import boto3
from opensearchpy import OpenSearch, RequestsHttpConnection

kinesis_client = boto3.client('kinesis')
opensearch_client = OpenSearch(
    hosts=['my-domain.us-east-1.es.amazonaws.com'],
    http_auth=('username', 'password'),
    use_ssl=True,
    verify_certs=True,
    connection_class=RequestsHttpConnection
)

def process_activity_stream(event, context):
    for record in event['Records']:
        payload = base64.b64decode(record['kinesis']['data'])
        activities = json.loads(payload)
        
        for activity in activities['databaseActivityEvents']:
            if activity['type'] == 'database_activity':
                # Elasticsearch に索引化
                opensearch_client.index(
                    index='aurora-activity-logs',
                    body={
                        'timestamp': activity['timestamp'],
                        'user': activity['databaseUser'],
                        'sql': activity['sqlText'],
                        'statementType': activity['statementType']
                    }
                )

5-4. よくある失敗パターン

---

まとめ

試験では「できないこと」「後付けできないこと」「デフォルト値」が頻出。本講義の内容を何度も反復し、ハンズオンで体験することが合格の鍵です。