HIPAA準拠AWSデータレイクの設計図：取り込みから分析まで

# バケット名の一意の接尾辞
UNIQUE_ID="your-unique-id-12345"

# 1. RAWデータバケットを作成
aws s3api create-bucket \
    --bucket phi-raw-data-${UNIQUE_ID} \
    --region us-east-1 \
    --object-lock-enabled-for-bucket

# 2. すべての公開アクセスをブロック
aws s3api put-public-access-block \
    --bucket phi-raw-data-${UNIQUE_ID} \
    --public-access-block-configuration "BlockPublicAcls=true,IgnorePublicAcls=true,BlockPublicPolicy=true,RestrictPublicBuckets=true"

# 3. デフォルトでサーバーサイド暗号化を有効化（SSE-S3）
aws s3api put-bucket-encryption \
    --bucket phi-raw-data-${UNIQUE_ID} \
    --server-side-encryption-configuration '{"Rules": [{"ApplyServerSideEncryptionByDefault": {"SSEAlgorithm": "AES256"}}]}'

# 4. 偶発的な削除/変更から保護するためにバージョニングを有効化
aws s3api put-bucket-versioning \
    --bucket phi-raw-data-${UNIQUE_ID} \
    --versioning-configuration Status=Enabled

# 5. サーバーアクセスログを有効化（独自のログバケットに置き換えてください）
# まず、ログバケットを作成
LOG_BUCKET="phi-access-logs-${UNIQUE_ID}"
aws s3api create-bucket --bucket ${LOG_BUCKET} --region us-east-1
aws s3api put-bucket-acl --bucket ${LOG_BUCKET} --grant-write URI=http://acs.amazonaws.com/groups/s3/LogDelivery --grant-read-acp URI=http://acs.amazonaws.com/groups/s3/LogDelivery

# ログ設定を適用
aws s3api put-bucket-logging \
    --bucket phi-raw-data-${UNIQUE_ID} \
    --bucket-logging-status "{\"LoggingEnabled\":{\"TargetBucket\":\"${LOG_BUCKET}\",\"TargetPrefix\":\"raw-data-logs/\"}}"

# 6. 厳格なデータ保持ライフサイクルポリシーを適用
# このポリシーは、30日後に非現行バージョンをGlacierに移動し、365日後に削除します。
cat > lifecycle.json <<EOL
{
  "Rules": [
    {
      "ID": "HIPAA-Retention-Policy",
      "Status": "Enabled",
      "Filter": {},
      "NoncurrentVersionTransitions": [
        {
          "NoncurrentDays": 30,
          "StorageClass": "GLACIER"
        }
      ],
      "NoncurrentVersionExpiration": {
        "NoncurrentDays": 365
      }
    }
  ]
}
EOL

aws s3api put-bucket-lifecycle-configuration \
    --bucket phi-raw-data-${UNIQUE_ID} \
    --lifecycle-configuration file://lifecycle.json

# 'phi-processed-data'バケットにも同じプロセスを繰り返します
aws s3api create-bucket --bucket phi-processed-data-${UNIQUE_ID} --region us-east-1
aws s3api put-public-access-block --bucket phi-processed-data-${UNIQUE_ID} --public-access-block-configuration "BlockPublicAcls=true,IgnorePublicAcls=true,BlockPublicPolicy=true,RestrictPublicBuckets=true"
aws s3api put-bucket-encryption --bucket phi-processed-data-${UNIQUE_ID} --server-side-encryption-configuration '{"Rules": [{"ApplyServerSideEncryptionByDefault": {"SSEAlgorithm": "AES256"}}]}'
aws s3api put-bucket-versioning --bucket phi-processed-data-${UNIQUE_ID} --versioning-configuration Status=Enabled
aws s3api put-bucket-logging --bucket phi-processed-data-${UNIQUE_ID} --bucket-logging-status "{\"LoggingEnabled\":{\"TargetBucket\":\"${LOG_BUCKET}\",\"TargetPrefix\":\"processed-data-logs/\"}}"

# iam-role-policy.json
cat > iam-role-policy.json <<EOL
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:*:*:*"
        },
        {
            "Effect": "Allow",
            "Action": "s3:PutObject",
            "Resource": "arn:aws:s3:::phi-raw-data-${UNIQUE_ID}/*"
        }
    ]
}
EOL

# trust-policy.json
cat > trust-policy.json <<EOL
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}
EOL

aws iam create-role --role-name PHI-Ingestion-Lambda-Role --assume-role-policy-document file://trust-policy.json
aws iam put-role-policy --role-name PHI-Ingestion-Lambda-Role --policy-name PHI-Ingestion-Lambda-Policy --policy-document file://iam-role-policy.json

# ingest_phi_function.py
import json
import boto3
import os
import uuid
from datetime import datetime

s3_client = boto3.client('s3')

def lambda_handler(event, context):
    """
    API GatewayからPHIデータを受け取り口を処理します。
    - patient_idの存在を検証します。
    - 安全なS3 rawバケットにタイムスタンプ付きのままのイベントを保存します。
    """
    try:
        # 環境変数からバケット名を取得
        raw_bucket = os.environ['RAW_BUCKET_NAME']
        
        # 受信リクエストボディを解析
        body = json.loads(event.get('body', '{}'))

        # --- 基本的な検証 ---
        if 'patient_id' not in body:
            return {
                'statusCode': 400,
                'body': json.dumps({'error': 'patient_idは必須フィールドです'})
            }

        # --- ストレージ用にデータを準備 ---
        timestamp = datetime.utcnow()
        event_id = str(uuid.uuid4())
        
        # レコードにメタデータを追加
        body['_metadata'] = {
            'ingestion_timestamp_utc': timestamp.isoformat(),
            'event_id': event_id,
            'source_ip': event.get('requestContext', {}).get('identity', {}).get('sourceIp')
        }

        # --- S3 Rawバケットに保存 ---
        file_key = f"year={timestamp.year}/month={timestamp.month:02}/day={timestamp.day:02}/{event_id}.json"
        
        s3_client.put_object(
            Bucket=raw_bucket,
            Key=file_key,
            Body=json.dumps(body),
            ServerSideEncryption='AES256'
        )
        
        return {
            'statusCode': 200,
            'body': json.dumps({'status': 'success', 'event_id': event_id})
        }

    except json.JSONDecodeError:
        return {
            'statusCode': 400,
            'body': json.dumps({'error': 'リクエストボディのJSON形式が無効です'})
        }
    except Exception as e:
        print(f"リクエスト処理エラー: {e}")
        # 実装の詳細を漏らさないための汎用エラー
        return {
            'statusCode': 500,
            'body': json.dumps({'error': '内部サーバーエラー'})
        }

# template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: HIPAA対応データ取り込みAPI

Parameters:
  RawDataBucketName:
    Type: String
    Description: 生PHIデータ用のS3バケット名。

Resources:
  PhiIngestionFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: ingest_phi_function.lambda_handler
      Runtime: python3.9
      CodeUri: .
      Description: PHIデータを取り込み、raw S3バケットに保存します。
      MemorySize: 256
      Timeout: 30
      Role: !GetAtt PhiIngestionLambdaRole.Arn
      Environment:
        Variables:
          RAW_BUCKET_NAME: !Ref RawDataBucketName
      Events:
        ApiEvent:
          Type: Api
          Properties:
            Path: /ingest
            Method: post

  PhiIngestionLambdaRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: S3AndCloudWatchPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: 'arn:aws:logs:*:*:*'
              - Effect: Allow
                Action: s3:PutObject
                Resource: !Sub 'arn:aws:s3:::${RawDataBucketName}/*'

# 1. S3ロケーションをLake Formationに登録
# 使用されるロールはバケットへのアクセス許可が必要です
aws lakeformation register-resource --resource-arn arn:aws:s3:::phi-raw-data-${UNIQUE_ID}
aws lakeformation register-resource --resource-arn arn:aws:s3:::phi-processed-data-${UNIQUE_ID}

# 2. AWS Glue Data Catalogにデータベースを作成
aws glue create-database --database-input "{\"Name\":\"phi_database\"}"

# 3. 生データをカタログ化するGlue Crawlerを作成
# <YOUR_ACCOUNT_ID>と<CRAWLER_ROLE_ARN>を実際の値に置き換えてください
aws glue create-crawler --name "phi_raw_data_crawler" \
    --role <CRAWLER_ROLE_ARN> \
    --database-name "phi_database" \
    --targets "{\"S3Targets\":[{\"Path\":\"s3://phi-raw-data-${UNIQUE_ID}/\"}]}" \
    --table-prefix "raw_"

# Crawlerを実行
aws glue start-crawler --name "phi_raw_data_crawler"

# Crawler実行後、'phi_database'に'raw_data'という名前のテーブルが作成されます。

# 4. Glue ETLジョブロールに権限を付与
# <GLUE_JOB_ROLE_ARN>をGlueジョブが使用するロールのARNに置き換えてください
GLUE_JOB_ROLE_ARN="arn:aws:iam::<YOUR_ACCOUNT_ID>:role/Glue-ETL-Job-Role"

# GlueジョブがデータベースをDESCRIBEすることを許可
aws lakeformation grant-permissions --principal "{\"DataLakePrincipalIdentifier\":\"${GLUE_JOB_ROLE_ARN}\"}" \
    --resource "{\"Database\":{\"Name\":\"phi_database\"}}" \
    --permissions "DESCRIBE"

# GlueジョブがrawテーブルからSELECTすることを許可
aws lakeformation grant-permissions --principal "{\"DataLakePrincipalIdentifier\":\"${GLUE_JOB_ROLE_ARN}\"}" \
    --resource "{\"Table\":{\"DatabaseName\":\"phi_database\",\"Name\":\"raw_data\"}}" \
    --permissions "SELECT"

# GlueジョブがprocessedロケーションでCREATEおよびWRITEすることを許可
aws lakeformation grant-permissions --principal "{\"DataLakePrincipalIdentifier\":\"${GLUE_JOB_ROLE_ARN}\"}" \
    --resource "{\"DataLocation\":{\"ResourceArn\":\"arn:aws:s3:::phi-processed-data-${UNIQUE_ID}/\"}}" \
    --permissions "DATA_LOCATION_ACCESS"

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql.functions import col, sha2, concat_ws, lit, udf
from pyspark.sql.types import StringType

# 生年月日を年のみ表示するようにマスクするUDFを定義
def mask_dob(dob_str):
    if dob_str and isinstance(dob_str, str):
        try:
            return dob_str.split('-')[0] + "-01-01"
        except:
            return None
    return None

mask_dob_udf = udf(mask_dob, StringType())

args = getResolvedOptions(sys.argv, ["JOB_NAME"])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args["JOB_NAME"], args)

# Glue Data Catalog（Lake Formationで管理）からデータを読み取り
datasource = glueContext.create_dynamic_frame.from_catalog(
    database="phi_database",
    table_name="raw_data",
    transformation_ctx="datasource",
)

# 操作を容易にするためにSpark DataFrameに変換
df = datasource.toDF()

# --- 匿名化と変換ロジック ---
# 1. SHA-256ハッシュを使用してpatient_idを仮名化します。これは元に戻せません。
# 2. 名前和社会保障番号などの直接識別子を削除します。
# 3. 生年月日は年のみをマスクします。
# 4. メタデータフィールドを削除します。
processed_df = df.withColumn("patient_hash", sha2(col("patient_id"), 256)) \
                 .withColumn("birth_year", mask_dob_udf(col("date_of_birth"))) \
                 .drop("patient_id", "first_name", "last_name", "ssn", "address", "date_of_birth") \
                 .drop("_metadata")

# DynamicFrameに戻す
processed_dynamic_frame = DynamicFrame.fromDF(
    processed_df, glueContext, "processed_dynamic_frame"
)

# --- Parquet形式でProcessed S3バケットに書き込み ---
# Parquetはカラムナー形式で、分析に最適です。
# 一般的なクエリフィールド（例：condition）でパーティション化すると、クエリパフォーマンスが向上します。
datasink = glueContext.write_dynamic_frame.from_options(
    frame=processed_dynamic_frame,
    connection_type="s3",
    connection_options={
        "path": "s3://phi-processed-data-<your-unique-id>/",
        "partitionKeys": ["medical_condition"],
    },
    format="parquet",
    transformation_ctx="datasink",
)

job.commit()