FastAPI、PostgreSQL、VaultでHIPAA対応の健康データパイプラインを構築する

# docker-compose.yml
version: '3.8'

services:
  db:
    image: postgres:14-alpine
    restart: always
    environment:
      - POSTGRES_USER=admin
      - POSTGRES_PASSWORD=supersecretpassword
      - POSTGRES_DB=health_db
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data/

  vault:
    image: vault:latest
    ports:
      - "8200:8200"
    environment:
      VAULT_DEV_ROOT_TOKEN_ID: "dev-root-token"
      VAULT_DEV_LISTEN_ADDRESS: "0.0.0.0:8200"
    cap_add:
      - IPC_LOCK

volumes:
  postgres_data:

mkdir hipaa_pipeline && cd hipaa_pipeline
python -m venv venv
source venv/bin/activate
pip install fastapi uvicorn "sqlalchemy[asyncio]" asyncpg pydantic python-jose[cryptography] passlib[bcrypt] hvac

# src/vault_service.py
import hvac
import os
import base64

VAULT_ADDR = os.getenv("VAULT_ADDR", "http://127.0.0.1:8200")
VAULT_TOKEN = os.getenv("VAULT_TOKEN", "dev-root-token")

client = hvac.Client(url=VAULT_ADDR, token=VAULT_TOKEN)

ENCRYPTION_KEY = "phi-key"

def encrypt_data(plain_text: str) -> str:
    """VaultのTransitエンジンを使ってデータを暗号化します。"""
    if not plain_text:
        return plain_text
        
    encoded_text = base64.b64encode(plain_text.encode()).decode('ascii')
    
    response = client.secrets.transit.encrypt_data(
        name=ENCRYPTION_KEY,
        plaintext=encoded_text,
    )
    return response['data']['ciphertext']

def decrypt_data(cipher_text: str) -> str:
    """VaultのTransitエンジンを使ってデータを復号化します。"""
    if not cipher_text:
        return cipher_text
        
    response = client.secrets.transit.decrypt_data(
        name=ENCRYPTION_KEY,
        ciphertext=cipher_text,
    )
    decoded_text = base64.b64decode(response['data']['plaintext']).decode('ascii')
    return decoded_text

# src/schemas.py
from pydantic import BaseModel
from enum import Enum

class UserRole(str, Enum):
    DOCTOR = "doctor"
    NURSE = "nurse"
    ADMIN = "admin"
    RESEARCHER = "researcher"

class User(BaseModel):
    username: str
    role: UserRole
    
# デモ用のダミーユーザーDB
DUMMY_USERS = {
    "doctor_davis": User(username="doctor_davis", role=UserRole.DOCTOR),
    "nurse_nancy": User(username="nurse_nancy", role=UserRole.NURSE),
    "admin_arnold": User(username="admin_arnold", role=UserRole.ADMIN),
}

# トークンから現在のユーザーを取得する単純な関数
def get_current_user_from_token(token: str) -> User:
    # 実際のアプリでは、ここでJWTをデコードします
    username = token.split("_") + "_" + token.split("_")
    return DUMMY_USERS.get(username)

# src/security.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from typing import List
from .schemas import User, UserRole, get_current_user_from_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class RoleChecker:
    def __init__(self, allowed_roles: List[UserRole]):
        self.allowed_roles = allowed_roles

    def __call__(self, token: str = Depends(oauth2_scheme)):
        user = get_current_user_from_token(token)
        if not user:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="ユーザーが見つかりません",
            )
        if user.role not in self.allowed_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="この役割では操作が許可されていません",
            )
        return user

# ロールベースの依存関係を定義
allow_doctors_and_admins = RoleChecker([UserRole.DOCTOR, UserRole.ADMIN])
allow_all_clinical_staff = RoleChecker([UserRole.DOCTOR, UserRole.NURSE, UserRole.ADMIN])
allow_admins_only = RoleChecker([UserRole.ADMIN])

# src/audit_log.py
import logging
from functools import wraps
from fastapi import Request, Response
from .schemas import User

# 監査証跡用の専用ロガーを設定
audit_logger = logging.getLogger("audit")
audit_logger.setLevel(logging.INFO)
# 本番環境では、このハンドラーは安全で改ざんできないファイルまたはログサービスに出力します
handler = logging.StreamHandler() 
formatter = logging.Formatter('%(asctime)s - AUDIT - %(message)s')
handler.setFormatter(formatter)
audit_logger.addHandler(handler)

def log_event(user: User, action: str, resource_id: str, status: str):
    """監査イベントをフォーマットしてログに記録します。"""
    audit_logger.info(
        f"User='{user.username}' Role='{user.role.value}' Action='{action}' "
        f"ResourceID='{resource_id}' Status='{status}'"
    )

def audit(action: str):
    """
    エンドポイント関数をラップして監査証跡イベントをログに記録するデコレータ。
    """
    def wrapper(func):
        @wraps(func)
        async def decorator(patient_id: str, request: Request, *args, **kwargs):
            # ユーザーはすでに解決された依存関係から取得
            user = kwargs.get('current_user')
            if not user:
                # エンドポイントがセキュリティで保護されている場合、これは本来発生するべきではありません
                log_event(User(username="anonymous", role="unknown"), action, patient_id, "failed_auth")
                return Response(status_code=401)
            
            try:
                response = await func(patient_id, request, *args, **kwargs)
                status = "success" if response.status_code < 400 else "failure"
                log_event(user, action, patient_id, status)
                return response
            except Exception as e:
                log_event(user, action, patient_id, f"error: {e}")
                raise e
        return decorator
    return wrapper

# src/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "postgresql+asyncpg://admin:supersecretpassword@localhost/health_db"

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()

# src/models.py
from sqlalchemy import Column, Integer, String
from .database import Base

class Patient(Base):
    __tablename__ = 'patients'
    id = Column(Integer, primary_key=True, index=True)
    patient_ref_id = Column(String, unique=True, index=True)
    # これらのフィールドには暗号化されたデータが保存されます
    name = Column(String) 
    ssn = Column(String)
    address = Column(String)

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.future import select
from sqlalchemy.exc import IntegrityError
import uvicorn
import uuid

from src.database import async_session, engine, Base
from src.models import Patient
from src.schemas import User
from src.vault_service import encrypt_data, decrypt_data
from src.security import allow_all_clinical_staff
from src.audit_log import audit

app = FastAPI()

@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

@app.post("/patients/")
@audit(action="CREATE_PATIENT_RECORD")
async def create_patient(
    name: str, 
    ssn: str, 
    address: str, 
    current_user: User = Depends(allow_all_clinical_staff)
):
    patient_ref_id = str(uuid.uuid4())
    
    # 保存前にPHIを暗号化
    encrypted_name = encrypt_data(name)
    encrypted_ssn = encrypt_data(ssn)
    encrypted_address = encrypt_data(address)
    
    new_patient = Patient(
        patient_ref_id=patient_ref_id,
        name=encrypted_name,
        ssn=encrypted_ssn,
        address=encrypted_address,
    )
    
    async with async_session() as session:
        session.add(new_patient)
        try:
            await session.commit()
            await session.refresh(new_patient)
        except IntegrityError:
            raise HTTPException(status_code=400, detail="患者は既に存在します")

    return {"patient_ref_id": new_patient.patient_ref_id}


@app.get("/patients/{patient_id}")
@audit(action="VIEW_PATIENT_RECORD")
async def get_patient(
    patient_id: str, 
    current_user: User = Depends(allow_all_clinical_staff)
):
    async with async_session() as session:
        result = await session.execute(
            select(Patient).where(Patient.patient_ref_id == patient_id)
        )
        patient = result.scalars().first()
        
        if not patient:
            raise HTTPException(status_code=404, detail="患者が見つかりません")
            
    # 返す前にデータを復号化
    decrypted_name = decrypt_data(patient.name)
    decrypted_ssn = decrypt_data(patient.ssn)
    decrypted_address = decrypt_data(patient.address)
    
    return {
        "patient_ref_id": patient.patient_ref_id,
        "name": decrypted_name,
        "ssn": "********" + decrypted_ssn[-4:], # 機密データをマスク
        "address": decrypted_address,
    }

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)