GDPR データハンドリング

GDPR に準拠したデータ処理、同意管理、プライバシーコントロールの実装ガイドです。

このスキルを使用する場合

• EU の個人データを処理するシステムの構築
• 同意管理の実装
• データ主体リクエスト (DSR) の処理
• GDPR コンプライアンスレビューの実施
• プライバシーファーストアーキテクチャの設計
• データ処理契約の作成

コア概念

1. 個人データの分類

カテゴリ	例	保護レベル
基本データ	名前、メール、電話	標準
機密データ (第9条)	健康、宗教、民族性	明示的な同意
犯罪データ (第10条)	有罪判決、犯罪	公的機関のみ
子どものデータ	16歳未満のデータ	親の同意

2. 処理の法的根拠

第6条 - 適法な処理根拠:
├── 同意: 自由意思で与えられた、具体的で、情報に基づいた同意
├── 契約: 契約の履行に必要
├── 法的義務: 法律で要求される
├── 重大な利益: 生命を守る必要性
├── 公共の利益: 公式機能
└── 正当な利益: 権利と利益のバランス考慮

3. データ主体の権利

アクセス権 (第15条)      ─┐
訂正権 (第16条)          │
削除権 (第17条)          │ 1ヶ月以内に
制限権 (第18条)          │ 対応が必要
可搬性権 (第20条)        │
異議権 (第21条)         ─┘

実装パターン

パターン1: 同意管理

// 同意データモデル
const consentSchema = {
  userId: String,
  consents: [
    {
      purpose: String, // 'marketing', 'analytics' など
      granted: Boolean,
      timestamp: Date,
      source: String, // 'web_form', 'api' など
      version: String, // プライバシーポリシーのバージョン
      ipAddress: String, // 証拠用
      userAgent: String, // 証拠用
    },
  ],
  auditLog: [
    {
      action: String, // 'granted', 'withdrawn', 'updated'
      purpose: String,
      timestamp: Date,
      source: String,
    },
  ],
};

// 同意サービス
class ConsentManager {
  async recordConsent(userId, purpose, granted, metadata) {
    const consent = {
      purpose,
      granted,
      timestamp: new Date(),
      source: metadata.source,
      version: await this.getCurrentPolicyVersion(),
      ipAddress: metadata.ipAddress,
      userAgent: metadata.userAgent,
    };

    // 同意を保存
    await this.db.consents.updateOne(
      { userId },
      {
        $push: {
          consents: consent,
          auditLog: {
            action: granted ? "granted" : "withdrawn",
            purpose,
            timestamp: consent.timestamp,
            source: metadata.source,
          },
        },
      },
      { upsert: true },
    );

    // 下流システムのためにイベント発行
    await this.eventBus.emit("consent.changed", {
      userId,
      purpose,
      granted,
      timestamp: consent.timestamp,
    });
  }

  async hasConsent(userId, purpose) {
    const record = await this.db.consents.findOne({ userId });
    if (!record) return false;

    const latestConsent = record.consents
      .filter((c) => c.purpose === purpose)
      .sort((a, b) => b.timestamp - a.timestamp)[0];

    return latestConsent?.granted === true;
  }

  async getConsentHistory(userId) {
    const record = await this.db.consents.findOne({ userId });
    return record?.auditLog || [];
  }
}

<!-- GDPR準拠の同意UI -->
<div class="consent-banner" role="dialog" aria-labelledby="consent-title">
  <h2 id="consent-title">Cookie設定</h2>

  <p>
    ご体験を向上させるためにクッキーを使用しています。以下で設定を選択してください。
  </p>

  <form id="consent-form">
    <!-- 必須 - 常にオン、同意不要 -->
    <div class="consent-category">
      <input type="checkbox" id="necessary" checked disabled />
      <label for="necessary">
        <strong>必須</strong>
        <span>ウェブサイトの機能に必要です。無効にできません。</span>
      </label>
    </div>

    <!-- 分析 - 同意が必要 -->
    <div class="consent-category">
      <input type="checkbox" id="analytics" name="analytics" />
      <label for="analytics">
        <strong>分析</strong>
        <span>サイトの使用方法を理解するのに役立ちます。</span>
      </label>
    </div>

    <!-- マーケティング - 同意が必要 -->
    <div class="consent-category">
      <input type="checkbox" id="marketing" name="marketing" />
      <label for="marketing">
        <strong>マーケティング</strong>
        <span>ご興味に基づいてパーソナライズされた広告。</span>
      </label>
    </div>

    <div class="consent-actions">
      <button type="button" id="accept-all">すべて同意</button>
      <button type="button" id="reject-all">すべて拒否</button>
      <button type="submit">設定を保存</button>
    </div>

    <p class="consent-links">
      <a href="/privacy-policy">プライバシーポリシー</a> |
      <a href="/cookie-policy">クッキーポリシー</a>
    </p>
  </form>
</div>

パターン2: データ主体アクセスリクエスト (DSAR)

from datetime import datetime, timedelta
from typing import Dict, List, Optional
import json

class DSARHandler:
    """データ主体アクセスリクエストを処理します。"""

    RESPONSE_DEADLINE_DAYS = 30
    EXTENSION_ALLOWED_DAYS = 60  # 複雑なリクエスト用

    def __init__(self, data_sources: List['DataSource']):
        self.data_sources = data_sources

    async def submit_request(
        self,
        request_type: str,  # 'access', 'erasure', 'rectification', 'portability'
        user_id: str,
        verified: bool,
        details: Optional[Dict] = None
    ) -> str:
        """新しい DSAR を送信します。"""
        request = {
            'id': self.generate_request_id(),
            'type': request_type,
            'user_id': user_id,
            'status': 'pending_verification' if not verified else 'processing',
            'submitted_at': datetime.utcnow(),
            'deadline': datetime.utcnow() + timedelta(days=self.RESPONSE_DEADLINE_DAYS),
            'details': details or {},
            'audit_log': [{
                'action': 'submitted',
                'timestamp': datetime.utcnow(),
                'details': 'Request received'
            }]
        }

        await self.db.dsar_requests.insert_one(request)
        await self.notify_dpo(request)

        return request['id']

    async def process_access_request(self, request_id: str) -> Dict:
        """データアクセスリクエストを処理します。"""
        request = await self.get_request(request_id)

        if request['type'] != 'access':
            raise ValueError("Not an access request")

        # すべてのソースからデータを収集
        user_data = {}
        for source in self.data_sources:
            try:
                data = await source.get_user_data(request['user_id'])
                user_data[source.name] = data
            except Exception as e:
                user_data[source.name] = {'error': str(e)}

        # レスポンスをフォーマット
        response = {
            'request_id': request_id,
            'generated_at': datetime.utcnow().isoformat(),
            'data_categories': list(user_data.keys()),
            'data': user_data,
            'retention_info': await self.get_retention_info(),
            'processing_purposes': await self.get_processing_purposes(),
            'third_party_recipients': await self.get_recipients()
        }

        # リクエストステータスを更新
        await self.update_request(request_id, 'completed', response)

        return response

    async def process_erasure_request(self, request_id: str) -> Dict:
        """削除権リクエストを処理します。"""
        request = await self.get_request(request_id)

        if request['type'] != 'erasure':
            raise ValueError("Not an erasure request")

        results = {}
        exceptions = []

        for source in self.data_sources:
            try:
                # 法的例外をチェック
                can_delete, reason = await source.can_delete(request['user_id'])

                if can_delete:
                    await source.delete_user_data(request['user_id'])
                    results[source.name] = 'deleted'
                else:
                    exceptions.append({
                        'source': source.name,
                        'reason': reason  # 例: 'legal retention requirement'
                    })
                    results[source.name] = f'retained: {reason}'
            except Exception as e:
                results[source.name] = f'error: {str(e)}'

        response = {
            'request_id': request_id,
            'completed_at': datetime.utcnow().isoformat(),
            'results': results,
            'exceptions': exceptions
        }

        await self.update_request(request_id, 'completed', response)

        return response

    async def process_portability_request(self, request_id: str) -> bytes:
        """可搬性のあるデータエクスポートを生成します。"""
        request = await self.get_request(request_id)
        user_data = await self.process_access_request(request_id)

        # 機械可読形式 (JSON) に変換
        portable_data = {
            'export_date': datetime.utcnow().isoformat(),
            'format_version': '1.0',
            'data': user_data['data']
        }

        return json.dumps(portable_data, indent=2, default=str).encode()

パターン3: データ保持

from datetime import datetime, timedelta
from enum import Enum

class RetentionBasis(Enum):
    CONSENT = "consent"
    CONTRACT = "contract"
    LEGAL_OBLIGATION = "legal_obligation"
    LEGITIMATE_INTEREST = "legitimate_interest"

class DataRetentionPolicy:
    """データ保持ポリシーを定義して実施します。"""

    POLICIES = {
        'user_account': {
            'retention_period_days': 365 * 3,  # 最後のアクティビティから3年
            'basis': RetentionBasis.CONTRACT,
            'trigger': 'last_activity_date',
            'archive_before_delete': True
        },
        'transaction_records': {
            'retention_period_days': 365 * 7,  # 税務のため7年
            'basis': RetentionBasis.LEGAL_OBLIGATION,
            'trigger': 'transaction_date',
            'archive_before_delete': True,
            'legal_reference': 'Tax regulations require 7 year retention'
        },
        'marketing_consent': {
            'retention_period_days': 365 * 2,  # 2年
            'basis': RetentionBasis.CONSENT,
            'trigger': 'consent_date',
            'archive_before_delete': False
        },
        'support_tickets': {
            'retention_period_days': 365 * 2,
            'basis': RetentionBasis.LEGITIMATE_INTEREST,
            'trigger': 'ticket_closed_date',
            'archive_before_delete': True
        },
        'analytics_data': {
            'retention_period_days': 365,  # 1年
            'basis': RetentionBasis.CONSENT,
            'trigger': 'collection_date',
            'archive_before_delete': False,
            'anonymize_instead': True
        }
    }

    async def apply_retention_policies(self):
        """保持ポリシーの実施を実行します。"""
        for data_type, policy in self.POLICIES.items():
            cutoff_date = datetime.utcnow() - timedelta(
                days=policy['retention_period_days']
            )

            if policy.get('anonymize_instead'):
                await self.anonymize_old_data(data_type, cutoff_date)
            else:
                if policy.get('archive_before_delete'):
                    await self.archive_data(data_type, cutoff_date)
                await self.delete_old_data(data_type, cutoff_date)

            await self.log_retention_action(data_type, cutoff_date)

    async def anonymize_old_data(self, data_type: str, before_date: datetime):
        """削除の代わりにデータを匿名化します。"""
        # 例: 識別フィールドをハッシュに置き換え
        if data_type == 'analytics_data':
            await self.db.analytics.update_many(
                {'collection_date': {'$lt': before_date}},
                {'$set': {
                    'user_id': None,
                    'ip_address': None,
                    'device_id': None,
                    'anonymized': True,
                    'anonymized_date': datetime.utcnow()
                }}
            )

パターン4: プライバシー・バイ・デザイン

class PrivacyFirstDataModel:
    """プライバシー・バイ・デザインデータモデルの例。"""

    # 個人情報と行動データを分離
    user_profile_schema = {
        'user_id': str,  # UUID、連続番号ではない
        'email_hash': str,  # ルックアップ用にハッシュ化
        'created_at': datetime,
        # 最小限のデータ収集
        'preferences': {
            'language': str,
            'timezone': str
        }
    }

    # 保存時に暗号化
    user_pii_schema = {
        'user_id': str,
        'email': str,  # 暗号化
        'name': str,   # 暗号化
        'phone': str,  # 暗号化 (オプション)
        'address': dict,  # 暗号化 (オプション)
        'encryption_key_id': str
    }

    # 擬似化した行動データ
    analytics_schema = {
        'session_id': str,  # user_id とリンクしていない
        'pseudonym_id': str,  # 回転する擬名
        'events': list,
        'device_category': str,  # 具体的ではなく一般化
        'country': str,  # 市レベルではない
    }

class DataMinimization:
    """データ最小化の原則を実装します。"""

    @staticmethod
    def collect_only_needed(form_data: dict, purpose: str) -> dict:
        """フォームデータをフィルタして必要なフィールドのみにします。"""
        REQUIRED_FIELDS = {
            'account_creation': ['email', 'password'],
            'newsletter': ['email'],
            'purchase': ['email', 'name', 'address', 'payment'],
            'support': ['email', 'message']
        }

        allowed = REQUIRED_FIELDS.get(purpose, [])
        return {k: v for k, v in form_data.items() if k in allowed}

    @staticmethod
    def generalize_location(ip_address: str) -> str:
        """IP を国レベルのみに一般化します。"""
        import geoip2.database
        reader = geoip2.database.Reader('GeoLite2-Country.mmdb')
        try:
            response = reader.country(ip_address)
            return response.country.iso_code
        except:
            return 'UNKNOWN'

パターン5: 違反通知

from datetime import datetime
from enum import Enum

class BreachSeverity(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class BreachNotificationHandler:
    """GDPR 違反通知要件を処理します。"""

    AUTHORITY_NOTIFICATION_HOURS = 72
    AFFECTED_NOTIFICATION_REQUIRED_SEVERITY = BreachSeverity.HIGH

    async def report_breach(
        self,
        description: str,
        data_types: List[str],
        affected_count: int,
        severity: BreachSeverity
    ) -> dict:
        """データ違反を報告して処理します。"""
        breach = {
            'id': self.generate_breach_id(),
            'reported_at': datetime.utcnow(),
            'description': description,
            'data_types_affected': data_types,
            'affected_individuals_count': affected_count,
            'severity': severity.value,
            'status': 'investigating',
            'timeline': [{
                'event': 'breach_reported',
                'timestamp': datetime.utcnow(),
                'details': description
            }]
        }

        await self.db.breaches.insert_one(breach)

        # 即座の通知
        await self.notify_dpo(breach)
        await self.notify_security_team(breach)

        # 監督機関への通知が72時間以内に必要
        if self.requires_authority_notification(severity, data_types):
            breach['authority_notification_deadline'] = (
                datetime.utcnow() + timedelta(hours=self.AUTHORITY_NOTIFICATION_HOURS)
            )
            await self.schedule_authority_notification(breach)

        # 影響を受けた個人への通知
        if severity.value in [BreachSeverity.HIGH.value, BreachSeverity.CRITICAL.value]:
            await self.schedule_individual_notifications(breach)

        return breach

    def requires_authority_notification(
        self,
        severity: BreachSeverity,
        data_types: List[str]
    ) -> bool:
        """監督機関への通知が必要か判断します。"""
        # 機密データについては常に通知
        sensitive_types = ['health', 'financial', 'credentials', 'biometric']
        if any(t in sensitive_types for t in data_types):
            return True

        # 中程度以上の深刻度については通知
        return severity in [BreachSeverity.MEDIUM, BreachSeverity.HIGH, BreachSeverity.CRITICAL]

    async def generate_authority_report(self, breach_id: str) -> dict:
        """監督機関向けのレポートを生成します。"""
        breach = await self.get_breach(breach_id)

        return {
            'organization': {
                'name': self.config.org_name,
                'contact': self.config.dpo_contact,
                'registration': self.config.registration_number
            },
            'breach': {
                'nature': breach['description'],
                'categories_affected': breach['data_types_affected'],
                'approximate_number_affected': breach['affected_individuals_count'],
                'likely_consequences': self.assess_consequences(breach),
                'measures_taken': await self.get_remediation_measures(breach_id),
                'measures_proposed': await self.get_proposed_measures(breach_id)
            },
            'timeline': breach['timeline'],
            'submitted_at': datetime.utcnow().isoformat()
        }

コンプライアンスチェックリスト

## GDPR 実装チェックリスト

### 法的根拠

- [ ] 各処理活動の法的根拠が文書化されている
- [ ] 同意メカニズムが GDPR 要件を満たしている
- [ ] 正当な利益評価が完了している

### 透明性

- [ ] プライバシーポリシーが明確でアクセス可能
- [ ] 処理目的が明確に記載されている
- [ ] データ保持期間が文書化されている

### データ主体の権利

- [ ] アクセスリクエストプロセスが実装されている
- [ ] 削除リクエストプロセスが実装されている
- [ ] ポータビリティエクスポートが利用可能
- [ ] 訂正プロセスが利用可能
- [ ] 30日の期限内に対応できる

### セキュリティ

- [ ] 保存時の暗号化が実装されている
- [ ] 送信時の暗号化 (TLS)
- [ ] アクセスコントロールが実装されている
- [ ] 監査ログが有効になっている

### 違反対応

- [ ] 違反検出メカニズムがある
- [ ] 72時間の通知プロセスがある
- [ ] 違反ドキュメンテーションシステムがある

### ドキュメンテーション

- [ ] 処理活動の記録 (第30条)
- [ ] データ保護影響評価
- [ ] ベンダーとのデータ処理契約

ベストプラクティス

すべきこと

• データ収集を最小化する - 必要な情報のみを収集
• すべてを文書化する - 処理活動、法的根拠
• 個人情報を暗号化する - 保存時と送信時
• アクセス制御を実装する - 必要な範囲でのアクセス
• 定期的な監査を実行する - コンプライアンスを継続的に検証

すべきでないこと

• 同意ボックスを事前にチェックしない - オプトインが必須
• 同意をバンドルしない - 目的ごとに分離する
• 無期限に保持しない - 保持期間を定義して実施する
• DSAR を無視しない - 30日以内の対応が必須
• 保護なしで転送しない - SCC または適切性決定が必要