Sécurité et Confidentialité des : Analyse Technique

# Checklist DPIA simplifiée
☑ Description du traitement :
  - Finalité : Chatbot RH interne pour répondre aux questions des employés
  - Données traitées : CV, contrats, évaluations annuelles, emails RH
  - Base légale : Intérêt légitime (gestion RH)

☑ Nécessité et proportionnalité :
  - Embeddings nécessaires pour la recherche sémantique (RAG)
  - Alternative non vectorielle : recherche mot-clé (moins efficace)
  - Mesure de minimisation : exclusion des données médicales

☑ Risques identifiés :
  - Risque élevé : Inversion d'embedding révélant des évaluations confidentielles
  - Risque moyen : Membership inference pour détecter les licenciements
  - Risque faible : Fuite via métadonnées (mitigé par pseudonymisation)

☑ Mesures de sécurité :
  - Chiffrement AES-256 au repos (AWS KMS)
  - TLS 1.3 en transit
  - RBAC avec authentification SSO (Okta)
  - Audit trail sur CloudWatch (rétention 1 an)
  - Differential privacy (DP-SGD, ε=0.5) lors du fine-tuning

☑ Droits des personnes :
  - Procédure d'effacement : script automatique de suppression d'embeddings par user_id
  - Délai de réponse : 30 jours
  - Information préalable : mention dans la politique de confidentialité interne

import weaviate

class EmbeddingTracker:
    def __init__(self, weaviate_client):
        self.client = weaviate_client

    def create_embedding_with_tracking(self, text, user_id, source_doc_id):
        """
        Génère un embedding avec métadonnées de traçabilité
        """
        embedding = model.encode(text)

        # Stockage avec métadonnées de traçabilité
        self.client.data_object.create(
            data_object={
                "vector": embedding.tolist(),
                "user_id": user_id,  # Identifiant de la personne concernée
                "source_doc_id": source_doc_id,
                "created_at": datetime.now().isoformat(),
            },
            class_name="Document"
        )

    def delete_user_embeddings(self, user_id):
        """
        Supprime tous les embeddings associés à un utilisateur (droit à l'oubli)
        """
        # Recherche de tous les embeddings de l'utilisateur
        result = self.client.query.get(
            "Document",
            ["user_id", "source_doc_id"]
        ).with_where({
            "path": ["user_id"],
            "operator": "Equal",
            "valueString": user_id
        }).do()

        # Suppression
        deleted_count = 0
        for obj in result['data']['Get']['Document']:
            self.client.data_object.delete(
                uuid=obj['_additional']['id'],
                class_name="Document"
            )
            deleted_count += 1

        # Logging pour audit trail
        log_deletion(user_id, deleted_count, reason="RGPD Article 17")

        return deleted_count

# Exemple d'usage
tracker = EmbeddingTracker(weaviate_client)
deleted = tracker.delete_user_embeddings("user_12345")
print(f"{deleted} embeddings supprimés")

import re
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

class DataMinimizer:
    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

    def minimize_before_embedding(self, text):
        """
        Supprime les PII (Personally Identifiable Information) avant embedding
        """
        # 1. Détection des PII
        results = self.analyzer.analyze(
            text=text,
            language='fr',
            entities=["PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER", "IBAN_CODE", "CREDIT_CARD"]
        )

        # 2. Anonymisation
        anonymized_text = self.anonymizer.anonymize(
            text=text,
            analyzer_results=results,
            operators={
                "PERSON": {"type": "replace", "new_value": "[PERSONNE]"},
                "EMAIL_ADDRESS": {"type": "replace", "new_value": "[EMAIL]"},
                "PHONE_NUMBER": {"type": "replace", "new_value": "[TEL]"},
                "IBAN_CODE": {"type": "redact"},
                "CREDIT_CARD": {"type": "redact"},
            }
        )

        return anonymized_text.text

# Exemple
minimizer = DataMinimizer()
original = "Jean Dupont (jean.dupont@acme.fr) a appelé au 06 12 34 56 78"
minimized = minimizer.minimize_before_embedding(original)
print(minimized)  # "[PERSONNE] ([EMAIL]) a appelé au [TEL]"

# Génération de l'embedding sur le texte minimisé
embedding = model.encode(minimized)

from sklearn.decomposition import PCA
import numpy as np

def reduce_embedding_dimension(embedding, target_dim=384):
    """
    Réduit la dimensionnalité d'un embedding pour minimiser le risque d'inversion
    """
    pca = PCA(n_components=target_dim)
    reduced_emb = pca.fit_transform(embedding.reshape(1, -1))
    return reduced_emb[0]

# Embedding original (1536 dimensions)
original_emb = model.encode("Texte sensible")

# Réduction à 384 dimensions (perte d'information contrôlée)
reduced_emb = reduce_embedding_dimension(original_emb, target_dim=384)

print(f"Dimension originale: {original_emb.shape[0]}")
print(f"Dimension réduite: {reduced_emb.shape[0]}")
print(f"Réduction: {(1 - 384/1536)*100:.1f}% (risque d'inversion diminué)")

import boto3
import numpy as np
from botocore.exceptions import ClientError

class EncryptedVectorStore:
    def __init__(self, kms_key_id, region='eu-west-1'):
        self.kms_client = boto3.client('kms', region_name=region)
        self.kms_key_id = kms_key_id

    def encrypt_embedding(self, embedding):
        """
        Chiffre un embedding avec AWS KMS (AES-256-GCM)
        """
        # Conversion du vecteur en bytes
        embedding_bytes = embedding.astype(np.float32).tobytes()

        try:
            # Chiffrement via KMS
            response = self.kms_client.encrypt(
                KeyId=self.kms_key_id,
                Plaintext=embedding_bytes,
                EncryptionAlgorithm='RSAES_OAEP_SHA_256'
            )
            return response['CiphertextBlob']
        except ClientError as e:
            print(f"Erreur de chiffrement: {e}")
            raise

    def decrypt_embedding(self, ciphertext, shape):
        """
        Déchiffre un embedding
        """
        try:
            # Déchiffrement via KMS
            response = self.kms_client.decrypt(
                CiphertextBlob=ciphertext,
                KeyId=self.kms_key_id,
                EncryptionAlgorithm='RSAES_OAEP_SHA_256'
            )

            # Reconstruction du vecteur
            embedding = np.frombuffer(response['Plaintext'], dtype=np.float32)
            return embedding.reshape(shape)
        except ClientError as e:
            print(f"Erreur de déchiffrement: {e}")
            raise

# Exemple d'usage
store = EncryptedVectorStore(kms_key_id='arn:aws:kms:eu-west-1:123456789:key/abc-def')

# Génération d'un embedding
embedding = model.encode("Donnée confidentielle")

# Chiffrement avant stockage
encrypted = store.encrypt_embedding(embedding)
print(f"Taille originale: {embedding.nbytes} bytes")
print(f"Taille chiffrée: {len(encrypted)} bytes")

# Stockage dans la base vectorielle (chiffré)
# vector_db.insert(encrypted)

# Récupération et déchiffrement
decrypted = store.decrypt_embedding(encrypted, embedding.shape)
print(f"Vérification: {np.allclose(embedding, decrypted)}")

Nginx - Configuration sécurisée pour API vectorielle

server {
    listen 443 ssl http2;
    server_name api.vectordb.company.com;

    # Certificats SSL (Let's Encrypt ou certificat d'entreprise)
    ssl_certificate /etc/ssl/certs/vectordb.crt;
    ssl_certificate_key /etc/ssl/private/vectordb.key;

    # TLS 1.3 uniquement (désactiver TLS 1.2 et versions antérieures)
    ssl_protocols TLSv1.3;

    # Ciphersuites sécurisées
    ssl_ciphers 'TLS_AES_256_GCM_SHA384:TLS_CHACHA20_POLY1305_SHA256';
    ssl_prefer_server_ciphers on;

    # HSTS (HTTP Strict Transport Security)
    add_header Strict-Transport-Security "max-age=31536000; includeSubDomains; preload" always;

    # Perfect Forward Secrecy
    ssl_session_cache shared:SSL:10m;
    ssl_session_timeout 10m;
    ssl_session_tickets off;

    # OCSP Stapling (vérification de révocation de certificat)
    ssl_stapling on;
    ssl_stapling_verify on;
    resolver 8.8.8.8 8.8.4.4 valid=300s;

    location /api/v1/search {
        # Proxy vers service backend
        proxy_pass http://localhost:8000;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # Limite de taille pour requêtes (embeddings)
        client_max_body_size 10M;
    }
}

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.ssl_ import create_urllib3_context

class SecureVectorDBClient:
    def __init__(self, api_url, api_key):
        self.api_url = api_url
        self.api_key = api_key
        self.session = self._create_secure_session()

    def _create_secure_session(self):
        """
        Crée une session HTTPS sécurisée (TLS 1.3 uniquement)
        """
        session = requests.Session()

        # Adapter SSL personnalisé
        class TLS13Adapter(HTTPAdapter):
            def init_poolmanager(self, *args, **kwargs):
                context = create_urllib3_context()
                context.minimum_version = ssl.TLSVersion.TLSv1_3  # TLS 1.3 minimum
                context.maximum_version = ssl.TLSVersion.TLSv1_3
                kwargs['ssl_context'] = context
                return super().init_poolmanager(*args, **kwargs)

        session.mount('https://', TLS13Adapter())
        return session

    def search_vectors(self, query_embedding, top_k=5):
        """
        Recherche vectorielle sécurisée via TLS 1.3
        """
        try:
            response = self.session.post(
                f"{self.api_url}/api/v1/search",
                json={
                    "vector": query_embedding.tolist(),
                    "top_k": top_k
                },
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                timeout=10,
                verify=True  # Vérification du certificat SSL
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.SSLError as e:
            print(f"Erreur SSL: {e}")
            raise
        except requests.exceptions.RequestException as e:
            print(f"Erreur requête: {e}")
            raise

# Usage
client = SecureVectorDBClient(
    api_url="https://api.vectordb.company.com",
    api_key="sk_prod_abc123"
)
results = client.search_vectors(query_embedding)

from seal import *
import numpy as np

class HomomorphicVectorSearch:
    def __init__(self):
        # Configuration SEAL (BFV scheme)
        self.parms = EncryptionParameters(scheme_type.bfv)
        poly_modulus_degree = 8192
        self.parms.set_poly_modulus_degree(poly_modulus_degree)
        self.parms.set_coeff_modulus(CoeffModulus.BFVDefault(poly_modulus_degree))
        self.parms.set_plain_modulus(PlainModulus.Batching(poly_modulus_degree, 20))

        # Génération des clés
        self.context = SEALContext(self.parms)
        self.keygen = KeyGenerator(self.context)
        self.public_key = self.keygen.create_public_key()
        self.secret_key = self.keygen.secret_key()
        self.encryptor = Encryptor(self.context, self.public_key)
        self.decryptor = Decryptor(self.context, self.secret_key)
        self.evaluator = Evaluator(self.context)

    def encrypt_vector(self, vector):
        """
        Chiffre un vecteur d'embedding
        """
        # Conversion en entiers (SEAL BFV nécessite des entiers)
        scaled_vector = (vector * 10000).astype(np.int64)

        # Encodage et chiffrement
        plain = Plaintext()
        self.encryptor.encrypt(plain, encrypted)
        return encrypted

    def compute_similarity_encrypted(self, enc_vector1, enc_vector2):
        """
        Calcule la similarité entre deux vecteurs chiffrés (produit scalaire)
        """
        # Multiplication homomorphe (composante par composante)
        encrypted_product = Ciphertext()
        self.evaluator.multiply(enc_vector1, enc_vector2, encrypted_product)

        # Somme des composantes (approximation du produit scalaire)
        # Note : implémentation simplifiée, nécessite rotation keys pour somme complète
        return encrypted_product

    def decrypt_result(self, encrypted_result):
        """
        Déchiffre le résultat de similarité
        """
        plain_result = Plaintext()
        self.decryptor.decrypt(encrypted_result, plain_result)
        return plain_result

# Exemple d'usage
he_search = HomomorphicVectorSearch()

# Chiffrement de deux embeddings
vec1 = np.array([0.5, 0.3, 0.8])  # Embedding requête
vec2 = np.array([0.6, 0.2, 0.7])  # Embedding document

enc_vec1 = he_search.encrypt_vector(vec1)
enc_vec2 = he_search.encrypt_vector(vec2)

# Calcul de similarité sur données chiffrées
enc_similarity = he_search.compute_similarity_encrypted(enc_vec1, enc_vec2)

# Déchiffrement du résultat uniquement
similarity_score = he_search.decrypt_result(enc_similarity)
print(f"Similarité (chiffrée): {similarity_score}")

import boto3
import json
from datetime import datetime, timedelta

class KeyRotationManager:
    def __init__(self, kms_client, key_alias='alias/embeddings-master-key'):
        self.kms = kms_client
        self.key_alias = key_alias

    def create_master_key(self):
        """
        Crée une clé maître avec rotation automatique
        """
        response = self.kms.create_key(
            Description='Master key for embeddings encryption',
            KeyUsage='ENCRYPT_DECRYPT',
            Origin='AWS_KMS',
            MultiRegion=False,
            KeySpec='SYMMETRIC_DEFAULT',
            Tags=[
                {'TagKey': 'Environment', 'TagValue': 'Production'},
                {'TagKey': 'Purpose', 'TagValue': 'Embeddings'},
                {'TagKey': 'Compliance', 'TagValue': 'RGPD'},
            ]
        )

        key_id = response['KeyMetadata']['KeyId']

        # Activation de la rotation automatique (annuelle)
        self.kms.enable_key_rotation(KeyId=key_id)

        # Création d'un alias
        self.kms.create_alias(
            AliasName=self.key_alias,
            TargetKeyId=key_id
        )

        return key_id

    def generate_data_key(self):
        """
        Génère une clé de données (DEK) chiffrée par la clé maître
        """
        response = self.kms.generate_data_key(
            KeyId=self.key_alias,
            KeySpec='AES_256'
        )

        return {
            'plaintext_key': response['Plaintext'],  # À utiliser pour chiffrement, puis supprimer de la mémoire
            'encrypted_key': response['CiphertextBlob']  # À stocker avec les données
        }

    def check_rotation_needed(self, key_id):
        """
        Vérifie si une rotation manuelle est nécessaire
        """
        response = self.kms.describe_key(KeyId=key_id)
        creation_date = response['KeyMetadata']['CreationDate']

        # Rotation si la clé a plus de 90 jours
        if datetime.now() - creation_date > timedelta(days=90):
            return True
        return False

    def rotate_key_manual(self, old_key_id):
        """
        Rotation manuelle : crée une nouvelle clé et met à jour l'alias
        """
        # Création d'une nouvelle clé
        new_key_id = self.create_master_key()

        # Mise à jour de l'alias (pointe vers la nouvelle clé)
        self.kms.update_alias(
            AliasName=self.key_alias,
            TargetKeyId=new_key_id
        )

        # Programmation de la suppression de l'ancienne clé (30 jours)
        self.kms.schedule_key_deletion(
            KeyId=old_key_id,
            PendingWindowInDays=30
        )

        return new_key_id

# Exemple d'usage
kms = boto3.client('kms', region_name='eu-west-1')
rotation_mgr = KeyRotationManager(kms)

# Création de la clé maître
master_key_id = rotation_mgr.create_master_key()
print(f"Master key créée: {master_key_id}")

# Génération d'une DEK pour chiffrer les embeddings
dek = rotation_mgr.generate_data_key()
print(f"DEK générée (encrypted key stockée avec les données)")

import boto3
import subprocess
import tempfile
from datetime import datetime

class EncryptedBackupManager:
    def __init__(self, kms_key_id, s3_bucket, db_connection_string):
        self.kms_key_id = kms_key_id
        self.s3_bucket = s3_bucket
        self.db_connection = db_connection_string
        self.s3_client = boto3.client('s3')

    def backup_vector_database(self):
        """
        Sauvegarde chiffrée de la base vectorielle (PostgreSQL + pgvector)
        """
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        backup_file = f"embeddings_backup_{timestamp}.sql.enc"

        with tempfile.NamedTemporaryFile(mode='w', delete=False) as tmp:
            # 1. Dump de la base (pg_dump)
            dump_cmd = f"pg_dump {self.db_connection} -t embeddings > {tmp.name}"
            subprocess.run(dump_cmd, shell=True, check=True)

            # 2. Chiffrement du dump avec KMS
            with open(tmp.name, 'rb') as f:
                plaintext_data = f.read()

            kms_client = boto3.client('kms')
            encrypted_data = kms_client.encrypt(
                KeyId=self.kms_key_id,
                Plaintext=plaintext_data
            )['CiphertextBlob']

            # 3. Upload vers S3 avec server-side encryption
            self.s3_client.put_object(
                Bucket=self.s3_bucket,
                Key=f"backups/{backup_file}",
                Body=encrypted_data,
                ServerSideEncryption='aws:kms',
                SSEKMSKeyId=self.kms_key_id,
                StorageClass='STANDARD_IA',  # Infrequent Access (coût optimisé)
                Metadata={
                    'backup-date': timestamp,
                    'database': 'embeddings',
                    'encrypted': 'true'
                }
            )

            # 4. Vérification de l'intégrité (checksum)
            checksum = hashlib.sha256(encrypted_data).hexdigest()
            self.s3_client.put_object(
                Bucket=self.s3_bucket,
                Key=f"backups/{backup_file}.sha256",
                Body=checksum
            )

        return backup_file

    def restore_from_backup(self, backup_file):
        """
        Restauration depuis une sauvegarde chiffrée
        """
        # 1. Téléchargement depuis S3
        response = self.s3_client.get_object(
            Bucket=self.s3_bucket,
            Key=f"backups/{backup_file}"
        )
        encrypted_data = response['Body'].read()

        # 2. Déchiffrement avec KMS
        kms_client = boto3.client('kms')
        decrypted_data = kms_client.decrypt(
            CiphertextBlob=encrypted_data,
            KeyId=self.kms_key_id
        )['Plaintext']

        # 3. Restauration dans la base
        with tempfile.NamedTemporaryFile(mode='wb', delete=False) as tmp:
            tmp.write(decrypted_data)
            tmp.flush()

            restore_cmd = f"psql {self.db_connection} < {tmp.name}"
            subprocess.run(restore_cmd, shell=True, check=True)

        print(f"Restauration réussie depuis {backup_file}")

    def setup_lifecycle_policy(self):
        """
        Politique de rétention des sauvegardes
        """
        lifecycle_config = {
            'Rules': [
                {
                    'Id': 'Transition to Glacier after 90 days',
                    'Status': 'Enabled',
                    'Prefix': 'backups/',
                    'Transitions': [
                        {
                            'Days': 90,
                            'StorageClass': 'GLACIER'
                        },
                    ],
                    'Expiration': {
                        'Days': 2555  # 7 ans (conformité RGPD pour données financières)
                    }
                },
            ]
        }

        self.s3_client.put_bucket_lifecycle_configuration(
            Bucket=self.s3_bucket,
            LifecycleConfiguration=lifecycle_config
        )

# Exemple d'usage
backup_mgr = EncryptedBackupManager(
    kms_key_id='arn:aws:kms:eu-west-1:123456789:key/abc-def',
    s3_bucket='company-embeddings-backups',
    db_connection_string='postgresql://user:pass@localhost/vectordb'
)

# Sauvegarde quotidienne (via cron)
backup_file = backup_mgr.backup_vector_database()
print(f"Sauvegarde créée: {backup_file}")

# Configuration de la politique de rétention
backup_mgr.setup_lifecycle_policy()

from flask import Flask, request, jsonify
from functools import wraps
import jwt
import pyotp
import boto3

app = Flask(__name__)

class SecureVectorAPI:
    def __init__(self, jwt_secret, totp_secret):
        self.jwt_secret = jwt_secret
        self.totp_secret = totp_secret
        self.cognito_client = boto3.client('cognito-idp')

    def require_auth(self, f):
        """
        Décorateur : authentification JWT + MFA TOTP
        """
        @wraps(f)
        def decorated_function(*args, **kwargs):
            # 1. Vérification du token JWT
            token = request.headers.get('Authorization', '').replace('Bearer ', '')
            if not token:
                return jsonify({'error': 'Token manquant'}), 401

            try:
                payload = jwt.decode(token, self.jwt_secret, algorithms=['HS256'])
                user_id = payload['user_id']
            except jwt.ExpiredSignatureError:
                return jsonify({'error': 'Token expiré'}), 401
            except jwt.InvalidTokenError:
                return jsonify({'error': 'Token invalide'}), 401

            # 2. Vérification MFA (TOTP)
            mfa_code = request.headers.get('X-MFA-Code')
            if not mfa_code:
                return jsonify({'error': 'Code MFA manquant'}), 401

            totp = pyotp.TOTP(self.totp_secret)
            if not totp.verify(mfa_code, valid_window=1):
                return jsonify({'error': 'Code MFA invalide'}), 401

            # 3. Vérification dans Cognito (optionnel : check user status)
            try:
                response = self.cognito_client.get_user(
                    AccessToken=token
                )
                if response['UserStatus'] != 'CONFIRMED':
                    return jsonify({'error': 'Utilisateur non confirmé'}), 403
            except Exception as e:
                return jsonify({'error': 'Erreur Cognito'}), 500

            # Authentification réussie : exécution de la fonction
            return f(user_id=user_id, *args, **kwargs)

        return decorated_function

# Initialisation
auth = SecureVectorAPI(
    jwt_secret='your-jwt-secret-256-bits',
    totp_secret='BASE32ENCODEDSECRET'
)

@app.route('/api/v1/search', methods=['POST'])
@auth.require_auth
def search_vectors(user_id):
    """
    Endpoint de recherche vectorielle sécurisé
    """
    query_vector = request.json.get('vector')
    top_k = request.json.get('top_k', 5)

    # Recherche dans la base vectorielle (avec filtrage par user_id pour RBAC)
    results = vector_db.search(
        vector=query_vector,
        top_k=top_k,
        filter={'user_id': user_id}  # Isolation des données par utilisateur
    )

    return jsonify(results)

import weaviate
from enum import Enum

class Role(Enum):
    ADMIN = "admin"
    DATA_MANAGER = "data_manager"
    ANALYST = "analyst"
    END_USER = "end_user"
    AUDITOR = "auditor"

class RBACVectorStore:
    def __init__(self, weaviate_client):
        self.client = weaviate_client

        # Matrice de permissions
        self.permissions = {
            Role.ADMIN: ['read', 'write', 'delete', 'manage_roles'],
            Role.DATA_MANAGER: ['read', 'write'],
            Role.ANALYST: ['read'],
            Role.END_USER: ['read_filtered'],
            Role.AUDITOR: ['read_logs'],
        }

    def check_permission(self, user_role, action):
        """
        Vérifie si un rôle a la permission pour une action
        """
        return action in self.permissions.get(user_role, [])

    def search_vectors(self, query_vector, user_id, user_role, department=None):
        """
        Recherche vectorielle avec filtrage RBAC
        """
        if not self.check_permission(user_role, 'read') and \
           not self.check_permission(user_role, 'read_filtered'):
            raise PermissionError("Accès refusé : permission insuffisante")

        # Construction du filtre selon le rôle
        filters = {}

        if user_role == Role.END_USER:
            # End users : accès uniquement à leur département
            filters = {
                "path": ["department"],
                "operator": "Equal",
                "valueString": department
            }
        elif user_role == Role.ANALYST:
            # Analysts : accès à tous les documents non-confidentiels
            filters = {
                "path": ["confidentiality_level"],
                "operator": "NotEqual",
                "valueString": "confidential"
            }
        # Admin et Data Manager : pas de filtre (accès complet)

        # Recherche avec filtres RBAC
        query = self.client.query.get("Document", ["title", "content"]) \
            .with_near_vector({"vector": query_vector})

        if filters:
            query = query.with_where(filters)

        results = query.with_limit(10).do()

        # Audit trail
        self.log_access(user_id, user_role, "search", len(results))

        return results

    def insert_vector(self, vector, metadata, user_id, user_role):
        """
        Insertion avec contrôle RBAC
        """
        if not self.check_permission(user_role, 'write'):
            raise PermissionError("Accès refusé : permission write requise")

        # Insertion dans Weaviate
        self.client.data_object.create(
            data_object={
                "vector": vector,
                **metadata,
                "created_by": user_id,
                "created_at": datetime.now().isoformat()
            },
            class_name="Document"
        )

        self.log_access(user_id, user_role, "insert", 1)

    def log_access(self, user_id, role, action, record_count):
        """
        Logging des accès pour audit
        """
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "user_id": user_id,
            "role": role.value,
            "action": action,
            "record_count": record_count
        }
        # Stockage dans CloudWatch, Datadog, ou base de logs
        print(f"[AUDIT] {log_entry}")

# Exemple d'usage
rbac_store = RBACVectorStore(weaviate_client)

# Recherche en tant qu'end user (accès limité)
results = rbac_store.search_vectors(
    query_vector=[0.5, 0.3, ...],
    user_id="user_12345",
    user_role=Role.END_USER,
    department="RH"
)

import os
import boto3
import hashlib
import secrets
from datetime import datetime, timedelta

class SecureAPIKeyManager:
    def __init__(self, secrets_manager_client):
        self.sm_client = secrets_manager_client

    def generate_api_key(self, user_id, expiration_days=90):
        """
        Génère une API key sécurisée avec expiration
        """
        # Génération d'une clé aléatoire (32 bytes = 256 bits)
        raw_key = secrets.token_urlsafe(32)

        # Hash de la clé pour stockage (ne jamais stocker la clé en clair)
        key_hash = hashlib.sha256(raw_key.encode()).hexdigest()

        # Métadonnées
        metadata = {
            "user_id": user_id,
            "created_at": datetime.now().isoformat(),
            "expires_at": (datetime.now() + timedelta(days=expiration_days)).isoformat(),
            "key_hash": key_hash,
            "revoked": False
        }

        # Stockage dans AWS Secrets Manager
        self.sm_client.create_secret(
            Name=f"vectordb/api-key/{user_id}",
            SecretString=json.dumps(metadata),
            Tags=[
                {'Key': 'User', 'Value': user_id},
                {'Key': 'ExpiresAt', 'Value': metadata['expires_at']}
            ]
        )

        # Retourner la clé en clair (une seule fois)
        return raw_key, metadata['expires_at']

    def validate_api_key(self, api_key):
        """
        Valide une API key (vérifie hash + expiration + révocation)
        """
        key_hash = hashlib.sha256(api_key.encode()).hexdigest()

        # Recherche dans Secrets Manager (ou base de données)
        # Simplified : assume we store in DynamoDB
        table = boto3.resource('dynamodb').Table('api-keys')
        response = table.get_item(Key={'key_hash': key_hash})

        if 'Item' not in response:
            raise ValueError("API key invalide")

        metadata = response['Item']

        # Vérifications
        if metadata.get('revoked'):
            raise ValueError("API key révoquée")

        expires_at = datetime.fromisoformat(metadata['expires_at'])
        if datetime.now() > expires_at:
            raise ValueError("API key expirée")

        return metadata['user_id']

    def revoke_api_key(self, user_id):
        """
        Révoque une API key (sans suppression pour audit trail)
        """
        self.sm_client.update_secret(
            SecretId=f"vectordb/api-key/{user_id}",
            SecretString=json.dumps({"revoked": True})
        )

# Exemple d'usage
sm_client = boto3.client('secretsmanager', region_name='eu-west-1')
key_manager = SecureAPIKeyManager(sm_client)

# Génération d'une clé pour un utilisateur
api_key, expires_at = key_manager.generate_api_key("user_12345", expiration_days=90)
print(f"API Key: {api_key}")
print(f"Expire le: {expires_at}")

# Validation lors d'une requête
try:
    user_id = key_manager.validate_api_key(api_key)
    print(f"Accès autorisé pour {user_id}")
except ValueError as e:
    print(f"Accès refusé: {e}")

# Security Group pour la base vectorielle (Data Tier)
resource "aws_security_group" "vector_db_sg" {
  name        = "vector-db-security-group"
  description = "Security group for vector database (private subnet)"
  vpc_id      = aws_vpc.main.id

  # Autoriser accès depuis Application Tier uniquement
  ingress {
    description     = "Weaviate from Application Tier"
    from_port       = 8080
    to_port         = 8080
    protocol        = "tcp"
    security_groups = [aws_security_group.app_tier_sg.id]
  }

  ingress {
    description     = "PostgreSQL from Application Tier"
    from_port       = 5432
    to_port         = 5432
    protocol        = "tcp"
    security_groups = [aws_security_group.app_tier_sg.id]
  }

  # BLOQUER tout trafic sortant vers Internet
  egress {
    description = "No outbound internet access"
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = []
  }

  tags = {
    Name        = "VectorDB-SecurityGroup"
    Environment = "Production"
    Compliance  = "RGPD"
  }
}

# Network ACL (couche supplémentaire)
resource "aws_network_acl" "data_tier_acl" {
  vpc_id     = aws_vpc.main.id
  subnet_ids = [aws_subnet.data_tier.id]

  # Autoriser uniquement trafic depuis Application Tier
  ingress {
    protocol   = "tcp"
    rule_no    = 100
    action     = "allow"
    cidr_block = aws_subnet.app_tier.cidr_block
    from_port  = 8080
    to_port    = 8080
  }

  # Bloquer tout le reste
  ingress {
    protocol   = "-1"
    rule_no    = 200
    action     = "deny"
    cidr_block = "0.0.0.0/0"
    from_port  = 0
    to_port    = 0
  }
}

import re
import spacy
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

class TextAnonymizer:
    def __init__(self):
        # Chargement du modèle NLP français
        self.nlp = spacy.load("fr_core_news_sm")
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

    def anonymize_text(self, text, method="replace"):
        """
        Anonymise un texte avant génération d'embedding
        """
        # 1. Détection automatique des entités nommées
        analyzer_results = self.analyzer.analyze(
            text=text,
            language='fr',
            entities=[
                "PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER",
                "IBAN_CODE", "CREDIT_CARD", "IP_ADDRESS",
                "FR_NIR", "FR_PASSPORT"  # Entités françaises spécifiques
            ]
        )

        # 2. Anonymisation selon la méthode choisie
        if method == "replace":
            # Remplacement par des placeholders génériques
            anonymized_result = self.anonymizer.anonymize(
                text=text,
                analyzer_results=analyzer_results,
                operators={
                    "PERSON": {"type": "replace", "new_value": "[PERSONNE]"},
                    "EMAIL_ADDRESS": {"type": "replace", "new_value": "[EMAIL]"},
                    "PHONE_NUMBER": {"type": "replace", "new_value": "[TEL]"},
                    "IBAN_CODE": {"type": "redact"},
                    "CREDIT_CARD": {"type": "redact"},
                    "IP_ADDRESS": {"type": "replace", "new_value": "[IP]"},
                    "FR_NIR": {"type": "redact"},
                    "FR_PASSPORT": {"type": "redact"},
                }
            )
            return anonymized_result.text

        elif method == "synthetic":
            # Remplacement par des données synthétiques réalistes
            anonymized_result = self.anonymizer.anonymize(
                text=text,
                analyzer_results=analyzer_results,
                operators={
                    "PERSON": {"type": "replace", "new_value": "Jean Martin"},
                    "EMAIL_ADDRESS": {"type": "replace", "new_value": "contact@exemple.fr"},
                    "PHONE_NUMBER": {"type": "replace", "new_value": "01 23 45 67 89"},
                }
            )
            return anonymized_result.text

        elif method == "k_anonymity":
            # K-anonymité : regroupement par catégories
            return self._apply_k_anonymity(text, analyzer_results)

    def _apply_k_anonymity(self, text, analyzer_results, k=5):
        """
        Application du principe de k-anonymité
        """
        anonymized_text = text

        for result in analyzer_results:
            entity_type = result.entity_type
            start = result.start
            end = result.end

            # Remplacement par catégorie générale
            if entity_type == "PERSON":
                anonymized_text = anonymized_text[:start] + "[Employé_Catégorie_A]" + anonymized_text[end:]
            elif entity_type == "EMAIL_ADDRESS":
                anonymized_text = anonymized_text[:start] + "[Email_Professionnel]" + anonymized_text[end:]

        return anonymized_text

# Exemple d'usage
anonymizer = TextAnonymizer()

original_text = """
Jean Dupont (jean.dupont@acme.fr) a signé le contrat le 15/03/2024.
Son numéro de sécurité sociale est 1 85 03 75 116 027 86.
Contact: +33 6 12 34 56 78
"""

# Méthode 1: Remplacement par placeholders
anonymized_replace = anonymizer.anonymize_text(original_text, method="replace")
print("Anonymisé (replace):", anonymized_replace)

# Méthode 2: Données synthétiques
anonymized_synthetic = anonymizer.anonymize_text(original_text, method="synthetic")
print("Anonymisé (synthetic):", anonymized_synthetic)

# Génération de l'embedding sur le texte anonymisé
embedding = model.encode(anonymized_replace)
print(f"Embedding généré (dimension: {embedding.shape[0]})")

import numpy as np
from opacus import PrivacyEngine
import torch
import torch.nn as nn
from sentence_transformers import SentenceTransformer

class DifferentiallyPrivateEmbedder:
    def __init__(self, model_name="sentence-transformers/all-MiniLM-L6-v2", epsilon=1.0, delta=1e-6):
        self.model = SentenceTransformer(model_name)
        self.epsilon = epsilon
        self.delta = delta
        self.privacy_engine = PrivacyEngine()

    def add_laplace_noise(self, embeddings, sensitivity=1.0):
        """
        Ajoute du bruit de Laplace aux embeddings (mécanisme de Laplace)
        """
        # Calcul de l'échelle du bruit selon le budget ε
        scale = sensitivity / self.epsilon

        # Génération du bruit de Laplace
        noise = np.random.laplace(0, scale, embeddings.shape)

        # Addition du bruit
        noisy_embeddings = embeddings + noise

        return noisy_embeddings

    def add_gaussian_noise(self, embeddings, sensitivity=1.0):
        """
        Ajoute du bruit gaussien (mécanisme gaussien)
        """
        # Calcul de la variance selon le budget (ε, δ)
        variance = 2 * np.log(1.25 / self.delta) * (sensitivity ** 2) / (self.epsilon ** 2)
        std_dev = np.sqrt(variance)

        # Génération du bruit gaussien
        noise = np.random.normal(0, std_dev, embeddings.shape)

        return embeddings + noise

    def train_dp_model(self, texts, target_epsilon=1.0):
        """
        Entraînement d'un modèle d'embedding avec DP-SGD
        """
        # Préparation des données
        train_dataset = [(text, text) for text in texts]  # Auto-encodage
        train_loader = torch.utils.data.DataLoader(train_dataset, batch_size=32)

        # Modèle simple pour l'exemple
        model = nn.Linear(768, 768)
        optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

        # Attachement du privacy engine
        model, optimizer, train_loader = self.privacy_engine.make_private(
            module=model,
            optimizer=optimizer,
            data_loader=train_loader,
            noise_multiplier=1.0,  # Intensité du bruit
            max_grad_norm=1.0,     # Clipping des gradients
        )

        # Entraînement avec DP
        for epoch in range(10):
            for batch in train_loader:
                optimizer.zero_grad()
                # ... logique d'entraînement ...
                loss.backward()
                optimizer.step()

            # Calcul du budget de confidentialité consommé
            epsilon_spent = self.privacy_engine.accountant.get_epsilon(self.delta)
            print(f"Epoch {epoch}, ε spent: {epsilon_spent:.3f}")

    def embed_with_privacy(self, texts):
        """
        Génère des embeddings avec differential privacy
        """
        # Génération d'embeddings standards
        embeddings = self.model.encode(texts)

        # Application de la differential privacy
        private_embeddings = self.add_gaussian_noise(embeddings)

        # Normalisation (optionnelle)
        private_embeddings = private_embeddings / np.linalg.norm(private_embeddings, axis=1, keepdims=True)

        return private_embeddings

# Exemple d'usage
dp_embedder = DifferentiallyPrivateEmbedder(epsilon=0.5)  # Budget strict

texts = [
    "Contrat de travail de Jean Dupont",
    "Facture n° F2024-001 client Acme Corp",
    "Rapport médical confidentiel"
]

# Génération d'embeddings avec DP
private_embeddings = dp_embedder.embed_with_privacy(texts)
print(f"Embeddings privés générés: {private_embeddings.shape}")

# Comparaison de l'utilité (similarité avant/après bruit)
original_embeddings = dp_embedder.model.encode(texts)
cosine_sim_before = np.dot(original_embeddings[0], original_embeddings[1])
cosine_sim_after = np.dot(private_embeddings[0], private_embeddings[1])

print(f"Similarité originale: {cosine_sim_before:.3f}")
print(f"Similarité avec DP: {cosine_sim_after:.3f}")
print(f"Perte d'utilité: {abs(cosine_sim_before - cosine_sim_after):.3f}")

import flwr as fl
import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from transformers import AutoModel, AutoTokenizer

class FederatedEmbeddingClient(fl.client.NumPyClient):
    def __init__(self, client_id, local_data):
        self.client_id = client_id
        self.local_data = local_data

        # Modèle d'embedding local
        self.model = AutoModel.from_pretrained('sentence-transformers/all-MiniLM-L6-v2')
        self.tokenizer = AutoTokenizer.from_pretrained('sentence-transformers/all-MiniLM-L6-v2')

    def get_parameters(self, config):
        """
        Retourne les paramètres du modèle local
        """
        return [val.cpu().numpy() for _, val in self.model.state_dict().items()]

    def set_parameters(self, parameters):
        """
        Met à jour le modèle local avec les paramètres reçus du serveur
        """
        params_dict = zip(self.model.state_dict().keys(), parameters)
        state_dict = {k: torch.tensor(v) for k, v in params_dict}
        self.model.load_state_dict(state_dict, strict=True)

    def fit(self, parameters, config):
        """
        Entraînement local sur les données du participant
        """
        # Mise à jour du modèle avec les paramètres globaux
        self.set_parameters(parameters)

        # Entraînement local
        optimizer = torch.optim.Adam(self.model.parameters(), lr=1e-5)
        self.model.train()

        for epoch in range(config.get("local_epochs", 1)):
            for texts in self.local_data:
                # Tokenisation
                inputs = self.tokenizer(texts, return_tensors="pt", padding=True, truncation=True)

                # Forward pass
                outputs = self.model(**inputs)
                embeddings = outputs.last_hidden_state[:, 0, :]  # [CLS] token

                # Loss (exemple: contrastive learning)
                loss = self.compute_contrastive_loss(embeddings)

                # Backward pass
                optimizer.zero_grad()
                loss.backward()
                optimizer.step()

        # Retour des paramètres mis à jour + métrique
        return self.get_parameters(config={}), len(self.local_data), {"loss": loss.item()}

    def evaluate(self, parameters, config):
        """
        Évaluation locale du modèle
        """
        self.set_parameters(parameters)

        # Évaluation sur un dataset de test local
        test_loss = self.compute_test_loss()

        return test_loss, len(self.local_data), {"test_loss": test_loss}

    def compute_contrastive_loss(self, embeddings):
        """
        Loss contrastive pour apprendre des embeddings de qualité
        """
        # Simplifié : MSE entre embeddings similaires
        return torch.mean((embeddings[0] - embeddings[1]) ** 2)

class FederatedEmbeddingServer:
    def __init__(self):
        self.strategy = fl.server.strategy.FedAvg(
            min_available_clients=3,
            min_fit_clients=3,
            min_evaluate_clients=3,
            initial_parameters=self.get_initial_parameters(),
        )

    def get_initial_parameters(self):
        """
        Paramètres initiaux du modèle global
        """
        model = AutoModel.from_pretrained('sentence-transformers/all-MiniLM-L6-v2')
        return [val.cpu().numpy() for _, val in model.state_dict().items()]

    def start_server(self, num_rounds=10):
        """
        Lance l'entraînement fédéré
        """
        fl.server.start_server(
            server_address="localhost:8080",
            config=fl.server.ServerConfig(num_rounds=num_rounds),
            strategy=self.strategy,
        )

# Exemple d'usage pour un participant
def start_client(client_id, local_texts):
    """
    Lance un client fédéré
    """
    client = FederatedEmbeddingClient(client_id, local_texts)

    fl.client.start_numpy_client(
        server_address="localhost:8080",
        client=client,
    )

# Données locales de chaque participant (exemples)
hospital_data = ["Dossier patient 1", "Rapport médical 2", ...]
bank_data = ["Contrat client A", "Analyse de risque B", ...]
company_data = ["Email interne 1", "Rapport trimestriel", ...]

# Démarrage des clients (sur machines séparées en réalité)
# start_client("hospital_A", hospital_data)
# start_client("bank_B", bank_data)
# start_client("company_C", company_data)

# Démarrage du serveur
server = FederatedEmbeddingServer()
server.start_server(num_rounds=50)

import openai
from transformers import GPT2LMHeadModel, GPT2Tokenizer
import numpy as np
from sklearn.decomposition import LatentDirichletAllocation
from sklearn.feature_extraction.text import CountVectorizer

class SyntheticTextGenerator:
    def __init__(self, method="gpt"):
        self.method = method

        if method == "gpt":
            self.tokenizer = GPT2Tokenizer.from_pretrained('gpt2')
            self.model = GPT2LMHeadModel.from_pretrained('gpt2')
            self.tokenizer.pad_token = self.tokenizer.eos_token

    def generate_synthetic_corpus(self, original_texts, num_synthetic=100):
        """
        Génère un corpus synthétique à partir d'un corpus original
        """
        if self.method == "template_based":
            return self._generate_template_based(original_texts, num_synthetic)
        elif self.method == "lda_guided":
            return self._generate_lda_guided(original_texts, num_synthetic)
        elif self.method == "gpt":
            return self._generate_gpt_based(original_texts, num_synthetic)

    def _generate_template_based(self, original_texts, num_synthetic):
        """
        Génération basée sur des templates extraits du corpus
        """
        # Extraction de patterns structurels
        templates = self._extract_templates(original_texts)

        synthetic_texts = []
        for i in range(num_synthetic):
            template = np.random.choice(templates)
            synthetic_text = self._fill_template(template)
            synthetic_texts.append(synthetic_text)

        return synthetic_texts

    def _extract_templates(self, texts):
        """
        Extrait des templates en remplaçant les entités nommées par des placeholders
        """
        import spacy
        nlp = spacy.load("fr_core_news_sm")

        templates = []
        for text in texts:
            doc = nlp(text)
            template = text

            # Remplacement des entités nommées
            for ent in doc.ents:
                if ent.label_ == "PERSON":
                    template = template.replace(ent.text, "[PERSONNE]")
                elif ent.label_ == "ORG":
                    template = template.replace(ent.text, "[ORGANISATION]")
                elif ent.label_ == "DATE":
                    template = template.replace(ent.text, "[DATE]")

            templates.append(template)

        return list(set(templates))  # Dédoublonnage

    def _fill_template(self, template):
        """
        Remplit un template avec des entités synthétiques
        """
        synthetic_entities = {
            "[PERSONNE]": ["Jean Martin", "Marie Dubois", "Pierre Leroy", "Sophie Bernard"],
            "[ORGANISATION]": ["Acme Corp", "TechStart", "GlobalInc", "DataSoft"],
            "[DATE]": ["15/03/2024", "22/07/2024", "08/11/2024", "30/12/2024"]
        }

        filled_template = template
        for placeholder, options in synthetic_entities.items():
            if placeholder in filled_template:
                replacement = np.random.choice(options)
                filled_template = filled_template.replace(placeholder, replacement, 1)

        return filled_template

    def _generate_lda_guided(self, original_texts, num_synthetic):
        """
        Génération guidée par analyse topique (LDA)
        """
        # Vectorisation et LDA
        vectorizer = CountVectorizer(max_features=1000, stop_words='english')
        doc_term_matrix = vectorizer.fit_transform(original_texts)

        lda = LatentDirichletAllocation(n_components=5, random_state=42)
        lda.fit(doc_term_matrix)

        # Génération de textes synthétiques basés sur les topics
        feature_names = vectorizer.get_feature_names_out()
        synthetic_texts = []

        for i in range(num_synthetic):
            # Sélection d'un topic aléatoire
            topic_idx = np.random.randint(0, lda.n_components)
            topic_words = lda.components_[topic_idx]

            # Sélection des mots les plus probables
            top_words_idx = topic_words.argsort()[-10:][::-1]
            selected_words = [feature_names[idx] for idx in top_words_idx[:5]]

            # Construction d'un texte synthétique
            synthetic_text = f"Document concernant {' et '.join(selected_words[:3])}. "
            synthetic_text += f"Analyse des {selected_words[3]} dans le contexte de {selected_words[4]}."

            synthetic_texts.append(synthetic_text)

        return synthetic_texts

    def _generate_gpt_based(self, original_texts, num_synthetic):
        """
        Génération avec un modèle GPT fine-tuné
        """
        # Création d'un prompt représentatif
        sample_texts = np.random.choice(original_texts, min(5, len(original_texts)), replace=False)
        prompt = "Voici des exemples de documents:\n\n"
        for i, text in enumerate(sample_texts):
            prompt += f"{i+1}. {text[:100]}...\n"
        prompt += "\nGénérez un document similaire:"

        synthetic_texts = []
        for i in range(num_synthetic):
            try:
                # Génération via OpenAI (ou modèle local)
                response = openai.ChatCompletion.create(
                    model="gpt-3.5-turbo",
                    messages=[
                        {"role": "system", "content": "Vous êtes un générateur de documents synthétiques pour préserver la confidentialité. Générez des textes similaires sans reproduire d'informations sensibles réelles."},
                        {"role": "user", "content": prompt}
                    ],
                    max_tokens=200,
                    temperature=0.8  # Variabilité
                )

                synthetic_text = response.choices[0].message.content.strip()
                synthetic_texts.append(synthetic_text)

            except Exception as e:
                print(f"Erreur génération GPT: {e}")
                # Fallback vers template
                synthetic_texts.append(self._generate_template_based([original_texts[0]], 1)[0])

        return synthetic_texts

# Exemple d'usage
generator = SyntheticTextGenerator(method="template_based")

original_corpus = [
    "Jean Dupont a signé un contrat avec Acme Corp le 15/03/2024 pour un montant de 50,000€.",
    "Marie Martin travaille chez TechStart depuis 2022 en tant que développeuse senior.",
    "Le rapport trimestriel de GlobalInc montre une croissance de 15% ce trimestre."
]

# Génération de 20 textes synthétiques
synthetic_corpus = generator.generate_synthetic_corpus(original_corpus, num_synthetic=20)

print("Corpus synthétique:")
for i, text in enumerate(synthetic_corpus[:5]):
    print(f"{i+1}. {text}")

# Génération d'embeddings sur le corpus synthétique
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('all-MiniLM-L6-v2')
synthetic_embeddings = model.encode(synthetic_corpus)

print(f"\nEmbeddings synthétiques générés: {synthetic_embeddings.shape}")
print("Avantage: Préservation de la sémantique sans exposition des données réelles")

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from sklearn.cluster import KMeans
from sklearn.metrics import adjusted_rand_score

class PrivacyUtilityEvaluator:
    def __init__(self):
        pass

    def evaluate_utility_preservation(self, original_embeddings, private_embeddings):
        """
        Évalue la préservation de l'utilité après application d'une technique privacy-preserving
        """
        results = {}

        # 1. Préservation de la similarité cosine
        original_similarities = cosine_similarity(original_embeddings)
        private_similarities = cosine_similarity(private_embeddings)

        similarity_correlation = np.corrcoef(
            original_similarities.flatten(),
            private_similarities.flatten()
        )[0, 1]
        results['similarity_preservation'] = similarity_correlation

        # 2. Préservation du clustering
        original_clusters = KMeans(n_clusters=5, random_state=42).fit_predict(original_embeddings)
        private_clusters = KMeans(n_clusters=5, random_state=42).fit_predict(private_embeddings)

        clustering_score = adjusted_rand_score(original_clusters, private_clusters)
        results['clustering_preservation'] = clustering_score

        # 3. Préservation des distances relatives
        def relative_distance_preservation(orig, priv):
            orig_dists = np.linalg.norm(orig[:, None] - orig[None, :], axis=2)
            priv_dists = np.linalg.norm(priv[:, None] - priv[None, :], axis=2)
            return np.corrcoef(orig_dists.flatten(), priv_dists.flatten())[0, 1]

        results['distance_preservation'] = relative_distance_preservation(
            original_embeddings, private_embeddings
        )

        # 4. Score global d'utilité (moyenne pondérée)
        results['overall_utility'] = (
            0.4 * similarity_correlation +
            0.3 * clustering_score +
            0.3 * results['distance_preservation']
        )

        return results

    def benchmark_privacy_techniques(self, original_embeddings, techniques_results):
        """
        Compare plusieurs techniques privacy-preserving
        """
        print("Benchmarking des techniques de privacy-preserving:")
        print("=" * 60)

        for technique_name, private_embeddings in techniques_results.items():
            scores = self.evaluate_utility_preservation(original_embeddings, private_embeddings)

            print(f"\n{technique_name}:")
            print(f"  Préservation similarité: {scores['similarity_preservation']:.3f}")
            print(f"  Préservation clustering: {scores['clustering_preservation']:.3f}")
            print(f"  Préservation distances: {scores['distance_preservation']:.3f}")
            print(f"  Score global d'utilité: {scores['overall_utility']:.3f}")

            # Interprétation
            if scores['overall_utility'] >= 0.8:
                print(f"  ➡️ Évaluation: EXCELLENTE utilité préservée")
            elif scores['overall_utility'] >= 0.6:
                print(f"  ➡️ Évaluation: BONNE utilité préservée")
            elif scores['overall_utility'] >= 0.4:
                print(f"  ➡️ Évaluation: UTILITÉ MODÉRÉE")
            else:
                print(f"  ➡️ Évaluation: UTILITÉ FORTEMENT DÉGRADÉE")

# Exemple d'usage
evaluator = PrivacyUtilityEvaluator()

# Embeddings originaux
original_embs = model.encode(["Document 1", "Document 2", "Document 3"])

# Résultats de différentes techniques
techniques = {
    "Anonymisation simple": anonymized_embeddings,
    "Differential Privacy (ε=1.0)": dp_embeddings_eps1,
    "Differential Privacy (ε=0.1)": dp_embeddings_eps01,
    "Données synthétiques": synthetic_embeddings,
}

# Benchmark comparatif
evaluator.benchmark_privacy_techniques(original_embs, techniques)

import json
import logging
from datetime import datetime
import hashlib
import boto3
from dataclasses import dataclass, asdict
from typing import Optional, Dict, Any

@dataclass
class AuditLogEntry:
    timestamp: str
    user_id: str
    action: str
    resource_type: str
    resource_id: Optional[str] = None
    ip_address: Optional[str] = None
    user_agent: Optional[str] = None
    metadata: Optional[Dict[str, Any]] = None
    risk_level: str = "LOW"

class VectorDBAuditLogger:
    def __init__(self, cloudwatch_client, log_group_name):
        self.cloudwatch = cloudwatch_client
        self.log_group = log_group_name
        self.log_stream = f"vectordb-audit-{datetime.now().strftime('%Y-%m-%d')}"

        # Configuration logging local + CloudWatch
        self.logger = logging.getLogger('vectordb_audit')
        self.logger.setLevel(logging.INFO)

        # Handler pour CloudWatch
        handler = logging.StreamHandler()
        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)

    def log_vector_search(self, user_id: str, query_vector, results_count: int,
                         ip_address: str, execution_time_ms: float):
        """
        Log d'une recherche vectorielle
        """
        # Hash du vecteur de requête (ne pas logger le vecteur complet)
        query_hash = hashlib.sha256(str(query_vector).encode()).hexdigest()[:16]

        audit_entry = AuditLogEntry(
            timestamp=datetime.utcnow().isoformat(),
            user_id=user_id,
            action="VECTOR_SEARCH",
            resource_type="embedding",
            ip_address=ip_address,
            metadata={
                "query_hash": query_hash,
                "results_count": results_count,
                "execution_time_ms": execution_time_ms,
                "query_dimension": len(query_vector)
            },
            risk_level="LOW" if results_count < 100 else "MEDIUM"
        )

        self._send_to_cloudwatch(audit_entry)

    def log_embedding_insertion(self, user_id: str, document_id: str,
                               embedding_size: int, source_type: str):
        """
        Log d'une insertion d'embedding
        """
        audit_entry = AuditLogEntry(
            timestamp=datetime.utcnow().isoformat(),
            user_id=user_id,
            action="EMBEDDING_INSERT",
            resource_type="embedding",
            resource_id=document_id,
            metadata={
                "embedding_size": embedding_size,
                "source_type": source_type,
                "processing_method": "automated"
            },
            risk_level="LOW"
        )

        self._send_to_cloudwatch(audit_entry)

    def log_rgpd_deletion(self, admin_id: str, user_id_deleted: str,
                         deleted_count: int, reason: str):
        """
        Log d'une suppression RGPD (audit critique)
        """
        audit_entry = AuditLogEntry(
            timestamp=datetime.utcnow().isoformat(),
            user_id=admin_id,
            action="RGPD_DELETION",
            resource_type="embedding",
            resource_id=user_id_deleted,
            metadata={
                "deleted_embeddings_count": deleted_count,
                "deletion_reason": reason,
                "compliance_requirement": "RGPD_Article_17",
                "retention_check": True
            },
            risk_level="HIGH"  # Suppression = opération à haut risque
        )

        self._send_to_cloudwatch(audit_entry)

        # Notification additionnelle pour suppressions critiques
        if deleted_count > 1000:
            self._send_security_alert(audit_entry)

    def log_admin_access(self, admin_id: str, action: str, affected_records: int,
                        ip_address: str):
        """
        Log d'accès administratif (niveau élevé de surveillance)
        """
        audit_entry = AuditLogEntry(
            timestamp=datetime.utcnow().isoformat(),
            user_id=admin_id,
            action=f"ADMIN_{action}",
            resource_type="system",
            ip_address=ip_address,
            metadata={
                "affected_records": affected_records,
                "admin_level": "full",
                "requires_approval": affected_records > 10000
            },
            risk_level="HIGH"
        )

        self._send_to_cloudwatch(audit_entry)

    def _send_to_cloudwatch(self, audit_entry: AuditLogEntry):
        """
        Envoi du log vers CloudWatch
        """
        try:
            log_message = json.dumps(asdict(audit_entry), ensure_ascii=False)

            self.cloudwatch.put_log_events(
                logGroupName=self.log_group,
                logStreamName=self.log_stream,
                logEvents=[
                    {
                        'timestamp': int(datetime.utcnow().timestamp() * 1000),
                        'message': log_message
                    }
                ]
            )

            # Log local pour backup
            self.logger.info(log_message)

        except Exception as e:
            self.logger.error(f"Erreur envoi CloudWatch: {e}")
            # Log local obligatoire en cas d'échec CloudWatch
            self.logger.critical(f"AUDIT_LOG_FAILED: {audit_entry}")

    def _send_security_alert(self, audit_entry: AuditLogEntry):
        """
        Alerte sécurité pour opérations critiques
        """
        # Notification SNS pour alertes immédiates
        sns_client = boto3.client('sns')
        message = f"[ALERTE VECTORDB] Opération critique détectée:\n"
        message += f"Action: {audit_entry.action}\n"
        message += f"Utilisateur: {audit_entry.user_id}\n"
        message += f"Timestamp: {audit_entry.timestamp}"

        sns_client.publish(
            TopicArn='arn:aws:sns:eu-west-1:123456789:vectordb-security-alerts',
            Message=message,
            Subject='[VECTORDB] Alerte Sécurité Critique'
        )

# Exemple d'usage
cloudwatch_client = boto3.client('logs', region_name='eu-west-1')
audit_logger = VectorDBAuditLogger(cloudwatch_client, 'vectordb-audit-logs')

# Logging d'une recherche
audit_logger.log_vector_search(
    user_id="user_12345",
    query_vector=[0.1, 0.2, 0.3],
    results_count=25,
    ip_address="192.168.1.100",
    execution_time_ms=150.5
)

# Logging d'une suppression RGPD
audit_logger.log_rgpd_deletion(
    admin_id="admin_67890",
    user_id_deleted="user_12345",
    deleted_count=156,
    reason="Demande utilisateur - Art. 17 RGPD"
)

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
from datetime import datetime, timedelta
import boto3
from typing import List, Dict, Any

class VectorDBAnomalyDetector:
    def __init__(self, cloudwatch_client, sns_topic_arn):
        self.cloudwatch = cloudwatch_client
        self.sns_topic = sns_topic_arn
        self.isolation_forest = IsolationForest(
            contamination=0.1,  # 10% d'anomalies attendues
            random_state=42
        )
        self.scaler = StandardScaler()
        self.baseline_established = False

    def extract_features_from_logs(self, logs: List[Dict]) -> pd.DataFrame:
        """
        Extrait des features d'anomalie depuis les logs d'audit
        """
        features = []

        for log in logs:
            try:
                # Parsing du log JSON
                if isinstance(log, str):
                    log_data = json.loads(log)
                else:
                    log_data = log

                # Extraction des features temporelles
                timestamp = datetime.fromisoformat(log_data['timestamp'])
                hour_of_day = timestamp.hour
                day_of_week = timestamp.weekday()
                is_weekend = day_of_week >= 5

                # Features comportementales
                action = log_data['action']
                user_id = log_data['user_id']
                risk_level = log_data.get('risk_level', 'LOW')

                # Features quantitatives
                metadata = log_data.get('metadata', {})
                results_count = metadata.get('results_count', 0)
                execution_time = metadata.get('execution_time_ms', 0)
                affected_records = metadata.get('affected_records', 0)

                # Construction du vecteur de features
                feature_vector = {
                    'hour_of_day': hour_of_day,
                    'day_of_week': day_of_week,
                    'is_weekend': int(is_weekend),
                    'is_admin_action': int(action.startswith('ADMIN_')),
                    'is_high_risk': int(risk_level == 'HIGH'),
                    'results_count': results_count,
                    'execution_time_ms': execution_time,
                    'affected_records': affected_records,
                    'user_id_hash': hash(user_id) % 10000,  # Anonymisation
                }

                features.append(feature_vector)

            except Exception as e:
                print(f"Erreur parsing log: {e}")
                continue

        return pd.DataFrame(features)

    def train_baseline(self, historical_logs: List[Dict]):
        """
        Entraîne le modèle sur des données historiques "normales"
        """
        features_df = self.extract_features_from_logs(historical_logs)

        if len(features_df) < 100:
            raise ValueError("Pas assez de données historiques pour établir une baseline")

        # Normalisation
        features_scaled = self.scaler.fit_transform(features_df)

        # Entraînement du modèle d'anomalie
        self.isolation_forest.fit(features_scaled)
        self.baseline_established = True

        print(f"Baseline établie sur {len(features_df)} événements")

    def detect_anomalies_realtime(self, recent_logs: List[Dict]) -> List[Dict]:
        """
        Détecte des anomalies dans les logs récents
        """
        if not self.baseline_established:
            raise ValueError("Baseline non établie. Appelez train_baseline() d'abord.")

        features_df = self.extract_features_from_logs(recent_logs)
        if len(features_df) == 0:
            return []

        # Normalisation avec le scaler entraîné
        features_scaled = self.scaler.transform(features_df)

        # Prédiction d'anomalies (-1 = anomalie, 1 = normal)
        anomaly_predictions = self.isolation_forest.predict(features_scaled)
        anomaly_scores = self.isolation_forest.decision_function(features_scaled)

        # Identification des anomalies
        anomalies = []
        for i, (prediction, score) in enumerate(zip(anomaly_predictions, anomaly_scores)):
            if prediction == -1:  # Anomalie détectée
                anomaly_info = {
                    'log_index': i,
                    'anomaly_score': score,
                    'severity': self._calculate_severity(score),
                    'original_log': recent_logs[i],
                    'detected_at': datetime.utcnow().isoformat()
                }
                anomalies.append(anomaly_info)

        return anomalies

    def _calculate_severity(self, anomaly_score: float) -> str:
        """
        Calcule la sévérité d'une anomalie basée sur le score
        """
        if anomaly_score < -0.5:
            return "CRITICAL"
        elif anomaly_score < -0.3:
            return "HIGH"
        elif anomaly_score < -0.1:
            return "MEDIUM"
        else:
            return "LOW"

    def send_anomaly_alert(self, anomalies: List[Dict]):
        """
        Envoie des alertes pour les anomalies détectées
        """
        for anomaly in anomalies:
            severity = anomaly['severity']
            original_log = anomaly['original_log']

            # Création du message d'alerte
            alert_message = f"""[ANOMALIE VECTORDB] {severity}

Action suspecte détectée :
- Action: {original_log.get('action', 'Inconnue')}
- Utilisateur: {original_log.get('user_id', 'Inconnu')}
- Timestamp: {original_log.get('timestamp', 'Inconnu')}
- Score d'anomalie: {anomaly['anomaly_score']:.3f}
- IP: {original_log.get('ip_address', 'Inconnue')}

Métadonnées: {original_log.get('metadata', {})}

Action recommandée: Vérification manuelle requise"""

            # Envoi selon la sévérité
            if severity in ['CRITICAL', 'HIGH']:
                # Alerte immédiate
                sns_client = boto3.client('sns')
                sns_client.publish(
                    TopicArn=self.sns_topic,
                    Message=alert_message,
                    Subject=f'[VECTORDB] Anomalie {severity} détectée'
                )

            # Log structuré pour investigation
            self._log_anomaly_detection(anomaly)

    def _log_anomaly_detection(self, anomaly: Dict):
        """
        Log structuré de la détection d'anomalie
        """
        detection_log = {
            'timestamp': anomaly['detected_at'],
            'event_type': 'ANOMALY_DETECTED',
            'anomaly_score': anomaly['anomaly_score'],
            'severity': anomaly['severity'],
            'original_event': anomaly['original_log'],
            'detector_version': '1.0'
        }

        # Envoi vers CloudWatch pour investigation
        self.cloudwatch.put_log_events(
            logGroupName='vectordb-anomaly-detection',
            logStreamName=f"anomalies-{datetime.now().strftime('%Y-%m-%d')}",
            logEvents=[
                {
                    'timestamp': int(datetime.utcnow().timestamp() * 1000),
                    'message': json.dumps(detection_log)
                }
            ]
        )

# Exemple d'usage
detector = VectorDBAnomalyDetector(
    cloudwatch_client=boto3.client('logs'),
    sns_topic_arn='arn:aws:sns:eu-west-1:123456789:vectordb-anomalies'
)

# 1. Entraînement sur données historiques (30 jours)
historical_logs = fetch_logs_from_cloudwatch(days=30)
detector.train_baseline(historical_logs)

# 2. Détection en temps réel (dernière heure)
recent_logs = fetch_logs_from_cloudwatch(hours=1)
anomalies = detector.detect_anomalies_realtime(recent_logs)

# 3. Envoi d'alertes si anomalies détectées
if anomalies:
    print(f"{len(anomalies)} anomalie(s) détectée(s)")
    detector.send_anomaly_alert(anomalies)
else:
    print("Aucune anomalie détectée")

import boto3
import json
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List, Dict, Callable
from enum import Enum

class AlertSeverity(Enum):
    LOW = "LOW"
    MEDIUM = "MEDIUM"
    HIGH = "HIGH"
    CRITICAL = "CRITICAL"

@dataclass
class Alert:
    id: str
    title: str
    description: str
    severity: AlertSeverity
    timestamp: str
    source: str
    metadata: Dict
    auto_resolve: bool = False

class RealTimeAlertSystem:
    def __init__(self, config: Dict):
        self.config = config
        self.sns_client = boto3.client('sns')
        self.lambda_client = boto3.client('lambda')
        self.slack_webhook = config.get('slack_webhook_url')

        # Canaux de notification par sévérité
        self.notification_channels = {
            AlertSeverity.CRITICAL: ['sns', 'slack', 'pagerduty', 'phone'],
            AlertSeverity.HIGH: ['sns', 'slack', 'email'],
            AlertSeverity.MEDIUM: ['slack', 'email'],
            AlertSeverity.LOW: ['email']
        }

    def trigger_alert(self, alert: Alert):
        """
        Déclenche une alerte avec escalade selon la sévérité
        """
        print(f"[ALERT] {alert.severity.value} - {alert.title}")

        # Métadonnées d'enrichissement
        enriched_alert = self._enrich_alert(alert)

        # Notification selon les canaux configurés
        channels = self.notification_channels[alert.severity]

        for channel in channels:
            try:
                if channel == 'sns':
                    self._send_sns_notification(enriched_alert)
                elif channel == 'slack':
                    self._send_slack_notification(enriched_alert)
                elif channel == 'email':
                    self._send_email_notification(enriched_alert)
                elif channel == 'pagerduty':
                    self._trigger_pagerduty(enriched_alert)
                elif channel == 'phone':
                    self._trigger_phone_call(enriched_alert)

            except Exception as e:
                print(f"Erreur envoi alerte via {channel}: {e}")

        # Actions automatiques pour alertes critiques
        if alert.severity == AlertSeverity.CRITICAL:
            self._trigger_automatic_response(enriched_alert)

    def _enrich_alert(self, alert: Alert) -> Alert:
        """
        Enrichit une alerte avec du contexte additionnel
        """
        # Ajout de contexte AWS
        alert.metadata.update({
            'aws_region': boto3.session.Session().region_name,
            'environment': self.config.get('environment', 'production'),
            'alert_id': alert.id,
            'escalation_policy': self.config.get('escalation_policy', 'default')
        })

        return alert

    def _send_sns_notification(self, alert: Alert):
        """
        Notification SNS pour intégrations enterprise (PagerDuty, OpsGenie)
        """
        message = f"""
[VECTORDB ALERT] {alert.severity.value}

Title: {alert.title}
Description: {alert.description}
Source: {alert.source}
Timestamp: {alert.timestamp}

Metadata:
{json.dumps(alert.metadata, indent=2)}

Dashboard: https://dashboard.vectordb.company.com/alerts/{alert.id}
        """.strip()

        self.sns_client.publish(
            TopicArn=self.config['sns_topic_arn'],
            Subject=f'[VECTORDB] {alert.severity.value} Alert: {alert.title}',
            Message=message,
            MessageAttributes={
                'severity': {
                    'DataType': 'String',
                    'StringValue': alert.severity.value
                },
                'source': {
                    'DataType': 'String',
                    'StringValue': alert.source
                }
            }
        )

    def _send_slack_notification(self, alert: Alert):
        """
        Notification Slack avec formatting rich
        """
        import requests

        # Couleur selon sévérité
        colors = {
            AlertSeverity.CRITICAL: '#FF0000',
            AlertSeverity.HIGH: '#FF8C00',
            AlertSeverity.MEDIUM: '#FFD700',
            AlertSeverity.LOW: '#90EE90'
        }

        slack_payload = {
            "attachments": [
                {
                    "color": colors[alert.severity],
                    "title": f"{alert.severity.value} Alert: {alert.title}",
                    "text": alert.description,
                    "fields": [
                        {
                            "title": "Source",
                            "value": alert.source,
                            "short": True
                        },
                        {
                            "title": "Timestamp",
                            "value": alert.timestamp,
                            "short": True
                        }
                    ],
                    "actions": [
                        {
                            "type": "button",
                            "text": "View Dashboard",
                            "url": f"https://dashboard.vectordb.company.com/alerts/{alert.id}"
                        },
                        {
                            "type": "button",
                            "text": "Acknowledge",
                            "url": f"https://api.vectordb.company.com/alerts/{alert.id}/ack"
                        }
                    ]
                }
            ]
        }

        requests.post(self.slack_webhook, json=slack_payload)

    def _trigger_automatic_response(self, alert: Alert):
        """
        Réponses automatiques pour alertes critiques
        """
        if "exfiltration" in alert.title.lower():
            # Blocage automatique de l'IP suspecte
            suspect_ip = alert.metadata.get('ip_address')
            if suspect_ip:
                self._block_ip_address(suspect_ip, duration_minutes=60)

        elif "authentication" in alert.title.lower():
            # Blocage temporaire du compte
            user_id = alert.metadata.get('user_id')
            if user_id:
                self._temporary_account_lock(user_id, duration_minutes=30)

        elif "rgpd_violation" in alert.source:
            # Notification automatique DPO
            self._notify_dpo(alert)

    def _block_ip_address(self, ip_address: str, duration_minutes: int):
        """
        Blocage automatique d'une IP via AWS WAF
        """
        # Implémentation AWS WAF pour blocage IP
        print(f"[AUTO-RESPONSE] Blocage IP {ip_address} pour {duration_minutes} minutes")

    def _temporary_account_lock(self, user_id: str, duration_minutes: int):
        """
        Verrouillage temporaire d'un compte utilisateur
        """
        print(f"[AUTO-RESPONSE] Verrouillage compte {user_id} pour {duration_minutes} minutes")

    def _notify_dpo(self, alert: Alert):
        """
        Notification spéciale DPO pour violations RGPD
        """
        print(f"[AUTO-RESPONSE] Notification DPO pour violation RGPD: {alert.title}")

# Configuration du système d'alertes
alert_config = {
    'sns_topic_arn': 'arn:aws:sns:eu-west-1:123456789:vectordb-alerts',
    'slack_webhook_url': 'https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX',
    'environment': 'production',
    'escalation_policy': 'vectordb_security'
}

alert_system = RealTimeAlertSystem(alert_config)

# Exemple d'utilisation : détection d'exfiltration
exfiltration_alert = Alert(
    id="alert_001",
    title="Exfiltration massive détectée",
    description="Un utilisateur a téléchargé 15,000 embeddings en 30 minutes",
    severity=AlertSeverity.CRITICAL,
    timestamp=datetime.utcnow().isoformat(),
    source="anomaly_detector",
    metadata={
        'user_id': 'user_suspicious_123',
        'ip_address': '203.0.113.42',
        'embeddings_downloaded': 15000,
        'time_window_minutes': 30
    }
)

alert_system.trigger_alert(exfiltration_alert)

import boto3
import json
from datetime import datetime, timedelta
from typing import Dict, List, Any
from dataclasses import dataclass

@dataclass
class AuditFinding:
    severity: str  # CRITICAL, HIGH, MEDIUM, LOW
    category: str
    title: str
    description: str
    remediation: str
    evidence: Dict[str, Any]
    compliance_impact: List[str]  # ["RGPD", "ISO27001", "SOC2"]

class VectorDBAuditChecker:
    def __init__(self):
        self.findings: List[AuditFinding] = []
        self.iam_client = boto3.client('iam')
        self.rds_client = boto3.client('rds')
        self.logs_client = boto3.client('logs')
        self.kms_client = boto3.client('kms')

    def run_full_audit(self) -> Dict[str, Any]:
        """
        Exécute un audit complet de sécurité
        """
        print("[AUDIT] Démarrage audit de sécurité VectorDB...")

        # Catégories d'audit
        self._audit_access_control()
        self._audit_encryption()
        self._audit_logging()
        self._audit_network_security()
        self._audit_rgpd_compliance()
        self._audit_backup_recovery()

        # Génération du rapport
        report = self._generate_audit_report()
        return report

    def _audit_access_control(self):
        """
        Audit des contrôles d'accès
        """
        print("[AUDIT] Contrôles d'accès...")

        # 1. Vérification des politiques IAM
        try:
            response = self.iam_client.list_policies(
                Scope='Local',
                OnlyAttached=True
            )

            for policy in response['Policies']:
                policy_doc = self.iam_client.get_policy_version(
                    PolicyArn=policy['Arn'],
                    VersionId=policy['DefaultVersionId']
                )['PolicyVersion']['Document']

                # Vérification : pas de wildcards dangereux
                if self._has_dangerous_wildcards(policy_doc):
                    self.findings.append(AuditFinding(
                        severity="HIGH",
                        category="Access Control",
                        title="Politique IAM avec wildcards dangereux",
                        description=f"La politique {policy['PolicyName']} contient des wildcards non sécurisés",
                        remediation="Remplacer les wildcards par des ressources spécifiques",
                        evidence={"policy_arn": policy['Arn'], "policy_doc": policy_doc},
                        compliance_impact=["ISO27001", "SOC2"]
                    ))

        except Exception as e:
            self.findings.append(AuditFinding(
                severity="MEDIUM",
                category="Access Control",
                title="Erreur audit IAM",
                description=f"Impossible d'auditer les politiques IAM: {e}",
                remediation="Vérifier les permissions d'audit",
                evidence={"error": str(e)},
                compliance_impact=["SOC2"]
            ))

        # 2. Vérification MFA
        self._check_mfa_compliance()

    def _audit_encryption(self):
        """
        Audit du chiffrement
        """
        print("[AUDIT] Chiffrement...")

        # 1. Vérification chiffrement RDS
        try:
            db_instances = self.rds_client.describe_db_instances()['DBInstances']

            for db in db_instances:
                if 'vector' in db['DBInstanceIdentifier'].lower():
                    if not db.get('StorageEncrypted', False):
                        self.findings.append(AuditFinding(
                            severity="CRITICAL",
                            category="Encryption",
                            title="Base vectorielle non chiffrée",
                            description=f"L'instance RDS {db['DBInstanceIdentifier']} n'est pas chiffrée",
                            remediation="Activer le chiffrement at-rest avec AWS KMS",
                            evidence={"db_instance": db['DBInstanceIdentifier']},
                            compliance_impact=["RGPD", "ISO27001", "SOC2"]
                        ))

        except Exception as e:
            print(f"Erreur audit RDS: {e}")

        # 2. Vérification clés KMS
        self._audit_kms_keys()

    def _audit_logging(self):
        """
        Audit de la journalisation
        """
        print("[AUDIT] Logging...")

        # Vérification existence logs groupe
        required_log_groups = [
            'vectordb-audit-logs',
            'vectordb-anomaly-detection',
            'vectordb-access-logs'
        ]

        try:
            existing_groups = self.logs_client.describe_log_groups()['logGroups']
            existing_names = [group['logGroupName'] for group in existing_groups]

            for required_group in required_log_groups:
                if required_group not in existing_names:
                    self.findings.append(AuditFinding(
                        severity="HIGH",
                        category="Logging",
                        title="Groupe de logs manquant",
                        description=f"Le groupe de logs {required_group} n'existe pas",
                        remediation="Créer le groupe de logs et configurer l'envoi",
                        evidence={"missing_log_group": required_group},
                        compliance_impact=["RGPD", "SOC2"]
                    ))

        except Exception as e:
            print(f"Erreur audit logs: {e}")

    def _audit_rgpd_compliance(self):
        """
        Audit spécifique RGPD
        """
        print("[AUDIT] Conformité RGPD...")

        # Vérifications RGPD critiques
        rgpd_checks = [
            self._check_data_retention_policy(),
            self._check_right_to_erasure_capability(),
            self._check_dpia_documentation(),
            self._check_breach_notification_procedure()
        ]

        for check_result in rgpd_checks:
            if not check_result['compliant']:
                self.findings.append(AuditFinding(
                    severity="HIGH",
                    category="RGPD Compliance",
                    title=check_result['title'],
                    description=check_result['description'],
                    remediation=check_result['remediation'],
                    evidence=check_result['evidence'],
                    compliance_impact=["RGPD"]
                ))

    def _check_right_to_erasure_capability(self) -> Dict:
        """
        Vérifie la capacité à effacer des données (droit à l'oubli)
        """
        # Simulation de vérification
        # En réalité, tester la procédure d'effacement
        return {
            'compliant': True,  # ou False selon test
            'title': 'Capacité d\'effacement RGPD',
            'description': 'Procédure d\'effacement d\'embeddings fonctionnelle',
            'remediation': 'Tester régulièrement la procédure d\'effacement',
            'evidence': {'test_date': datetime.now().isoformat()}
        }

    def _generate_audit_report(self) -> Dict[str, Any]:
        """
        Génère un rapport d'audit structuré
        """
        # Classification par sévérité
        severity_counts = {
            'CRITICAL': len([f for f in self.findings if f.severity == 'CRITICAL']),
            'HIGH': len([f for f in self.findings if f.severity == 'HIGH']),
            'MEDIUM': len([f for f in self.findings if f.severity == 'MEDIUM']),
            'LOW': len([f for f in self.findings if f.severity == 'LOW'])
        }

        # Score global de sécurité
        security_score = self._calculate_security_score(severity_counts)

        report = {
            'audit_date': datetime.utcnow().isoformat(),
            'audit_version': '1.0',
            'total_findings': len(self.findings),
            'severity_breakdown': severity_counts,
            'security_score': security_score,
            'compliance_status': self._assess_compliance_status(),
            'findings': [{
                'severity': f.severity,
                'category': f.category,
                'title': f.title,
                'description': f.description,
                'remediation': f.remediation,
                'compliance_impact': f.compliance_impact
            } for f in self.findings],
            'recommendations': self._generate_recommendations()
        }

        return report

    def _calculate_security_score(self, severity_counts: Dict) -> int:
        """
        Calcule un score de sécurité global (0-100)
        """
        # Pondération par sévérité
        penalty_points = (
            severity_counts['CRITICAL'] * 25 +
            severity_counts['HIGH'] * 10 +
            severity_counts['MEDIUM'] * 5 +
            severity_counts['LOW'] * 1
        )

        # Score sur 100 (maximum 100 points de pénalité)
        score = max(0, 100 - penalty_points)
        return score

# Exécution de l'audit
auditor = VectorDBAuditChecker()
audit_results = auditor.run_full_audit()

print(f"\n=== RAPPORT D'AUDIT VECTORDB ===")
print(f"Score de sécurité: {audit_results['security_score']}/100")
print(f"Findings total: {audit_results['total_findings']}")
print(f"  - Critiques: {audit_results['severity_breakdown']['CRITICAL']}")
print(f"  - Élevés: {audit_results['severity_breakdown']['HIGH']}")
print(f"  - Moyens: {audit_results['severity_breakdown']['MEDIUM']}")
print(f"  - Faibles: {audit_results['severity_breakdown']['LOW']}")

# RUNBOOK: Incident exfiltration massive d'embeddings
# Classification: P0 - CRITIQUE
# Temps de réponse: 15 minutes

# === PHASE 1: DÉTECTION (0-5 min) ===
# Indicateurs:
# - >10,000 embeddings téléchargés en <1h par un utilisateur
# - Patterns de requêtes séquentielles (scraping)
# - Accès depuis nouvelle géolocalisation

# Actions immédiates:
echo "[$(date)] INCIDENT P0: Exfiltration détectée" >> /var/log/incidents.log

# Identifier l'utilisateur et l'IP suspect
SUSPECT_USER_ID="user_12345"  # Depuis alerte
SUSPECT_IP="203.0.113.42"     # Depuis alerte

# === PHASE 2: CONTAINMENT (5-20 min) ===

# 1. Blocage immédiat de l'IP
aws wafv2 create-ip-set \
  --scope CLOUDFRONT \
  --ip-address-version IPV4 \
  --addresses $SUSPECT_IP \
  --name "blocked-ips-$(date +%s)"

# 2. Désactivation du compte utilisateur
aws cognito-idp admin-disable-user \
  --user-pool-id us-east-1_XXXXXXXXX \
  --username $SUSPECT_USER_ID

# 3. Révocation des tokens actifs
aws cognito-idp admin-user-global-sign-out \
  --user-pool-id us-east-1_XXXXXXXXX \
  --username $SUSPECT_USER_ID

# 4. Snapshot de la base vectorielle (préservation preuves)
aws rds create-db-snapshot \
  --db-instance-identifier vectordb-prod \
  --db-snapshot-identifier "incident-$(date +%Y%m%d-%H%M%S)"

# 5. Isolation réseau (Security Group)
aws ec2 revoke-security-group-ingress \
  --group-id sg-xxxxxxxxx \
  --protocol tcp \
  --port 5432 \
  --source-group sg-yyyyyyyyyy

# === PHASE 3: INVESTIGATION (20 min - 2h) ===

# 1. Extraction logs de l'utilisateur suspect
aws logs filter-log-events \
  --log-group-name vectordb-audit-logs \
  --start-time $(date -d "1 hour ago" +%s)000 \
  --filter-pattern "{ $.user_id = \"$SUSPECT_USER_ID\" }" \
  --output json > incident_logs_$(date +%s).json

# 2. Analyse des embeddings accédés
psql -h vectordb-prod.cluster-xyz.eu-west-1.rds.amazonaws.com -U forensic_user -d vectordb << EOF
\copy (
  SELECT document_id, created_at, metadata
  FROM embeddings
  WHERE accessed_by = '$SUSPECT_USER_ID'
    AND accessed_at > NOW() - INTERVAL '2 hours'
) TO 'accessed_embeddings_$(date +%s).csv' CSV HEADER;
EOF

# 3. Vérification intégrité base (détection modification/suppression)
psql -h vectordb-prod.cluster-xyz.eu-west-1.rds.amazonaws.com -U forensic_user -d vectordb << EOF
SELECT
  COUNT(*) as total_embeddings,
  MIN(created_at) as oldest_embedding,
  MAX(created_at) as newest_embedding
FROM embeddings;
EOF

# === PHASE 4: NOTIFICATION (Immédiat) ===

# 1. Notification équipe sécurité
aws sns publish \
  --topic-arn arn:aws:sns:eu-west-1:123456789:security-incidents \
  --subject "[P0] Exfiltration VectorDB détectée" \
  --message "Incident critique: Utilisateur $SUSPECT_USER_ID depuis IP $SUSPECT_IP. Containment activé."

# 2. Notification DPO (si données personnelles)
if [[ "$DATA_TYPE" == "personal" ]]; then
  aws sns publish \
    --topic-arn arn:aws:sns:eu-west-1:123456789:dpo-alerts \
    --subject "[RGPD] Violation données personnelles" \
    --message "Exfiltration potentielle de données personnelles. DPIA et notification CNIL requises."
fi

# 3. Notification management
aws sns publish \
  --topic-arn arn:aws:sns:eu-west-1:123456789:management-alerts \
  --subject "[CRITICAL] Incident sécurité VectorDB" \
  --message "Incident P0 en cours. Équipe sécurité mobilisée. Update dans 30 min."

# === PHASE 5: DOCUMENTATION ===

# Création rapport incident initial
cat << EOF > incident_report_$(date +%Y%m%d-%H%M%S).md
# Rapport d'incident - Exfiltration VectorDB

**Date:** $(date)
**Classification:** P0 - Critique
**Détecté par:** Système d'anomalie

## Résumé
- Utilisateur: $SUSPECT_USER_ID
- IP source: $SUSPECT_IP
- Données accédées: Estimation X,XXX embeddings
- Période: $(date -d "1 hour ago") - $(date)

## Actions prises
- [x] Blocage IP source
- [x] Désactivation compte utilisateur
- [x] Révocation tokens
- [x] Isolation réseau
- [x] Préservation preuves (snapshot)
- [x] Notifications équipes

## Prochaines étapes
- [ ] Investigation forensique approfondie
- [ ] Évaluation impact exact
- [ ] Notification autorités si requis (CNIL)
- [ ] Communication clients si impact
- [ ] Post-mortem et améliorations

**Responsable incident:** [NOM]
**Statut:** CONTAINMENT
EOF

echo "[$(date)] Incident P0 - Containment terminé. Investigation en cours." >> /var/log/incidents.log