Gestion des données GDPR

Guide pratique de mise en œuvre pour le traitement des données conforme au RGPD, la gestion du consentement et les contrôles de confidentialité.

Quand utiliser cette compétence

• Construire des systèmes qui traitent les données personnelles de l'UE
• Mettre en œuvre la gestion du consentement
• Gérer les demandes des personnes concernées (DSR)
• Conduire des audits de conformité RGPD
• Concevoir des architectures respectueuses de la vie privée
• Créer des accords de traitement des données

Concepts fondamentaux

1. Catégories de données personnelles

2. Bases légales du traitement

Article 6 - Bases légales :
├── Consentement : Donné librement, spécifique, éclairé
├── Contrat : Nécessaire pour l'exécution du contrat
├── Obligation légale : Exigée par la loi
├── Intérêts vitaux : Protection de la vie de quelqu'un
├── Intérêt public : Fonctions officielles
└── Intérêt légitime : Équilibré par rapport aux droits

3. Droits des personnes concernées

Droit d'accès (Art. 15)          ─┐
Droit de rectification (Art. 16)   │
Droit à l'effacement (Art. 17)     │ Répondre
Droit à la limitation (Art. 18)    │ dans 1 mois
Droit à la portabilité (Art. 20)   │
Droit d'opposition (Art. 21)      ─┘

Modèles d'implémentation

Modèle 1 : Gestion du consentement

// Modèle de données de consentement
const consentSchema = {
  userId: String,
  consents: [
    {
      purpose: String, // 'marketing', 'analytics', etc.
      granted: Boolean,
      timestamp: Date,
      source: String, // 'web_form', 'api', etc.
      version: String, // Version de la politique de confidentialité
      ipAddress: String, // Pour la preuve
      userAgent: String, // Pour la preuve
    },
  ],
  auditLog: [
    {
      action: String, // 'granted', 'withdrawn', 'updated'
      purpose: String,
      timestamp: Date,
      source: String,
    },
  ],
};

// Service de consentement
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,
    };

    // Stocker le consentement
    await this.db.consents.updateOne(
      { userId },
      {
        $push: {
          consents: consent,
          auditLog: {
            action: granted ? "granted" : "withdrawn",
            purpose,
            timestamp: consent.timestamp,
            source: metadata.source,
          },
        },
      },
      { upsert: true },
    );

    // Émettre un événement pour les systèmes en aval
    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 || [];
  }
}

<!-- Interface de consentement conforme au RGPD -->
<div class="consent-banner" role="dialog" aria-labelledby="consent-title">
  <h2 id="consent-title">Préférences de cookies</h2>

  <p>
    Nous utilisons des cookies pour améliorer votre expérience. Sélectionnez vos préférences ci-dessous.
  </p>

  <form id="consent-form">
    <!-- Nécessaires - toujours activés, pas de consentement requis -->
    <div class="consent-category">
      <input type="checkbox" id="necessary" checked disabled />
      <label for="necessary">
        <strong>Nécessaires</strong>
        <span>Requis pour le fonctionnement du site web. Impossible à désactiver.</span>
      </label>
    </div>

    <!-- Analyse - nécessite le consentement -->
    <div class="consent-category">
      <input type="checkbox" id="analytics" name="analytics" />
      <label for="analytics">
        <strong>Analyse</strong>
        <span>Nous aider à comprendre comment vous utilisez notre site.</span>
      </label>
    </div>

    <!-- Marketing - nécessite le consentement -->
    <div class="consent-category">
      <input type="checkbox" id="marketing" name="marketing" />
      <label for="marketing">
        <strong>Marketing</strong>
        <span>Annonces personnalisées basées sur vos intérêts.</span>
      </label>
    </div>

    <div class="consent-actions">
      <button type="button" id="accept-all">Tout accepter</button>
      <button type="button" id="reject-all">Tout refuser</button>
      <button type="submit">Enregistrer les préférences</button>
    </div>

    <p class="consent-links">
      <a href="/privacy-policy">Politique de confidentialité</a> |
      <a href="/cookie-policy">Politique relative aux cookies</a>
    </p>
  </form>
</div>

Modèle 2 : Demande d'accès aux données (DSAR)

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

class DSARHandler:
    """Gérer les demandes d'accès aux données des personnes concernées."""

    RESPONSE_DEADLINE_DAYS = 30
    EXTENSION_ALLOWED_DAYS = 60  # Pour les demandes complexes

    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:
        """Soumettre une nouvelle 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:
        """Traiter une demande d'accès aux données."""
        request = await self.get_request(request_id)

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

        # Collecter les données de toutes les sources
        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)}

        # Formater la réponse
        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()
        }

        # Mettre à jour le statut de la demande
        await self.update_request(request_id, 'completed', response)

        return response

    async def process_erasure_request(self, request_id: str) -> Dict:
        """Traiter une demande de droit à l'effacement."""
        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:
                # Vérifier les exceptions légales
                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  # ex. '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:
        """Générer un export de données portable."""
        request = await self.get_request(request_id)
        user_data = await self.process_access_request(request_id)

        # Convertir en format lisible par machine (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()

Modèle 3 : Rétention des données

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:
    """Définir et appliquer les politiques de rétention des données."""

    POLICIES = {
        'user_account': {
            'retention_period_days': 365 * 3,  # 3 ans après dernière activité
            'basis': RetentionBasis.CONTRACT,
            'trigger': 'last_activity_date',
            'archive_before_delete': True
        },
        'transaction_records': {
            'retention_period_days': 365 * 7,  # 7 ans pour la fiscalité
            '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 ans
            '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 an
            'basis': RetentionBasis.CONSENT,
            'trigger': 'collection_date',
            'archive_before_delete': False,
            'anonymize_instead': True
        }
    }

    async def apply_retention_policies(self):
        """Exécuter l'application des politiques de rétention."""
        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):
        """Anonymiser les données au lieu de les supprimer."""
        # Exemple : Remplacer les champs d'identification par des hachis
        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()
                }}
            )

Modèle 4 : Respect de la vie privée par conception

class PrivacyFirstDataModel:
    """Exemple de modèle de données conçu dans une optique de respect de la vie privée."""

    # Séparer les données à caractère personnel du comportement
    user_profile_schema = {
        'user_id': str,  # UUID, non séquentiel
        'email_hash': str,  # Haché pour les recherches
        'created_at': datetime,
        # Collecte minimale de données
        'preferences': {
            'language': str,
            'timezone': str
        }
    }

    # Chiffré au repos
    user_pii_schema = {
        'user_id': str,
        'email': str,  # Chiffré
        'name': str,   # Chiffré
        'phone': str,  # Chiffré (optionnel)
        'address': dict,  # Chiffré (optionnel)
        'encryption_key_id': str
    }

    # Données comportementales pseudonymisées
    analytics_schema = {
        'session_id': str,  # Non lié à user_id
        'pseudonym_id': str,  # Pseudonyme en rotation
        'events': list,
        'device_category': str,  # Généralisé, non spécifique
        'country': str,  # Pas au niveau ville
    }

class DataMinimization:
    """Implémenter les principes de minimisation des données."""

    @staticmethod
    def collect_only_needed(form_data: dict, purpose: str) -> dict:
        """Filtrer les données de formulaire pour ne garder que les champs nécessaires à la finalité."""
        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:
        """Généraliser l'IP au niveau du pays uniquement."""
        import geoip2.database
        reader = geoip2.database.Reader('GeoLite2-Country.mmdb')
        try:
            response = reader.country(ip_address)
            return response.country.iso_code
        except:
            return 'UNKNOWN'

Modèle 5 : Notification de violation

from datetime import datetime
from enum import Enum

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

class BreachNotificationHandler:
    """Gérer les exigences de notification de violation du RGPD."""

    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:
        """Signaler et gérer une violation de données."""
        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)

        # Notifications immédiates
        await self.notify_dpo(breach)
        await self.notify_security_team(breach)

        # Notification à l'autorité requise dans 72 heures
        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)

        # Notification des personnes concernées
        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:
        """Déterminer si l'autorité de contrôle doit être notifiée."""
        # Toujours notifier pour les données sensibles
        sensitive_types = ['health', 'financial', 'credentials', 'biometric']
        if any(t in sensitive_types for t in data_types):
            return True

        # Notifier pour la sévérité moyenne ou supérieure
        return severity in [BreachSeverity.MEDIUM, BreachSeverity.HIGH, BreachSeverity.CRITICAL]

    async def generate_authority_report(self, breach_id: str) -> dict:
        """Générer un rapport pour l'autorité de contrôle."""
        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()
        }

Liste de contrôle de conformité

## Liste de contrôle de mise en œuvre du RGPD

### Base légale

- [ ] Base légale documentée pour chaque activité de traitement
- [ ] Les mécanismes de consentement respectent les exigences du RGPD
- [ ] Les évaluations d'intérêt légitime sont complétées

### Transparence

- [ ] La politique de confidentialité est claire et accessible
- [ ] Les finalités du traitement sont clairement indiquées
- [ ] Les périodes de rétention des données sont documentées

### Droits des personnes concernées

- [ ] Processus de demande d'accès mis en œuvre
- [ ] Processus de demande d'effacement mis en œuvre
- [ ] Export de portabilité disponible
- [ ] Processus de rectification disponible
- [ ] Réponse dans le délai de 30 jours

### Sécurité

- [ ] Chiffrement au repos mis en œuvre
- [ ] Chiffrement en transit (TLS)
- [ ] Contrôles d'accès en place
- [ ] Journalisation d'audit activée

### Réponse aux violations

- [ ] Mécanismes de détection de violation
- [ ] Processus de notification dans 72 heures
- [ ] Système de documentation des violations

### Documentation

- [ ] Registre des activités de traitement (Art. 30)
- [ ] Analyses d'impact relatives à la protection des données
- [ ] Accords de traitement des données avec les prestataires

Bonnes pratiques

À faire

• Minimiser la collecte de données - Ne collecter que ce qui est nécessaire
• Documenter tout - Activités de traitement, bases légales
• Chiffrer les données à caractère personnel - Au repos et en transit
• Mettre en œuvre des contrôles d'accès - Sur la base du besoin de savoir
• Audits réguliers - Vérifier la conformité en continu

À ne pas faire

• Ne pas pré-cocher les cases de consentement - Doit être un opt-in
• Ne pas regrouper les consentements - Finalités séparées séparément
• Ne pas conserver indéfiniment - Définir et appliquer la rétention
• Ne pas ignorer les DSAR - Réponse requise dans 30 jours
• Ne pas transférer sans garanties - Clauses contractuelles types ou décisions d'adéquation