class DataPoisoningThreatModel:
    """Understanding training data poisoning attacks."""
 
    attack_types = {
        'label_flipping': {
            'description': 'Adversary changes labels on training examples',
            'goal': 'Cause misclassification on specific inputs',
            'required_access': 'Write access to labels',
            'detection_difficulty': 'Medium - statistical analysis can detect',
            'example': 'Flip 5% of "spam" emails to "not spam" to evade detection'
        },
 
        'backdoor_injection': {
            'description': 'Insert samples with trigger pattern and target label',
            'goal': 'Model behaves normally except when trigger present',
            'required_access': 'Ability to add training samples',
            'detection_difficulty': 'High - trigger pattern may be subtle',
            'example': 'Images with small pixel pattern always classified as target class'
        },
 
        'clean_label_poisoning': {
            'description': 'Add correctly-labeled but adversarial samples',
            'goal': 'Degrade model performance on specific classes',
            'required_access': 'Ability to add training samples',
            'detection_difficulty': 'Very High - samples appear legitimate',
            'example': 'Add hard-to-classify edge cases for target class'
        },
 
        'gradient_based_poisoning': {
            'description': 'Craft samples that maximally shift model parameters',
            'goal': 'Efficient degradation with fewer poisoned samples',
            'required_access': 'Knowledge of model architecture',
            'detection_difficulty': 'High - requires specialized detection',
            'example': 'Optimized perturbations that amplify gradient updates'
        },
 
        'model_replication_via_data': {
            'description': 'Extract training data to replicate proprietary models',
            'goal': 'Steal intellectual property',
            'required_access': 'Query access to model',
            'detection_difficulty': 'Medium - unusual query patterns',
            'example': 'Systematic queries to reconstruct training distribution'
        }
    }

class DataLeakageVectors:
    """Vectors through which training data can leak."""
 
    vectors = {
        'model_memorization': {
            'description': 'Model memorizes and can reproduce training examples',
            'risk_level': 'High for LLMs and generative models',
            'detection': 'Membership inference attacks, extraction attacks',
            'mitigation': 'Differential privacy, deduplication, training guardrails'
        },
 
        'gradient_leakage': {
            'description': 'Training gradients reveal information about training data',
            'risk_level': 'High in federated learning settings',
            'detection': 'Gradient inversion attacks',
            'mitigation': 'Gradient compression, differential privacy, secure aggregation'
        },
 
        'model_inversion': {
            'description': 'Reconstruct training data from model parameters',
            'risk_level': 'Medium - depends on model type',
            'detection': 'Inversion attack testing',
            'mitigation': 'Model architecture choices, output perturbation'
        },
 
        'side_channel_leakage': {
            'description': 'Training infrastructure leaks data through timing, etc.',
            'risk_level': 'Medium in shared computing environments',
            'detection': 'Side-channel analysis',
            'mitigation': 'Isolated training environments, constant-time operations'
        },
 
        'unauthorized_access': {
            'description': 'Direct access to training data storage',
            'risk_level': 'High if access controls inadequate',
            'detection': 'Access logging and monitoring',
            'mitigation': 'Encryption, access controls, audit logging'
        }
    }

from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
import hashlib
 
class DataSourceTrust(Enum):
    INTERNAL = "internal"         # Organization's own data
    VERIFIED_PARTNER = "partner"   # Vetted third-party
    PUBLIC_CURATED = "curated"     # Public but quality-checked
    PUBLIC_RAW = "raw"             # Public without validation
    UNKNOWN = "unknown"            # Unverified source
 
@dataclass
class DataSourceMetadata:
    source_id: str
    source_name: str
    trust_level: DataSourceTrust
    collection_date: str
    collection_method: str
    legal_basis: str
    data_subjects: Optional[str]
    retention_policy: str
    contact_info: str
 
class SecureDataCollector:
    """Collect training data with security controls."""
 
    def __init__(self, config: dict):
        self.allowed_sources = config.get('allowed_sources', [])
        self.required_trust_level = DataSourceTrust(
            config.get('min_trust_level', 'partner')
        )
        self.validators = self._load_validators(config)
 
    def collect_from_source(self, source: DataSourceMetadata,
                           data: bytes) -> dict:
        """Securely collect data from a source."""
 
        collection_result = {
            'source': source.source_id,
            'timestamp': datetime.utcnow().isoformat(),
            'status': 'pending',
            'validations': []
        }
 
        # Validate source trust level
        trust_order = [DataSourceTrust.UNKNOWN, DataSourceTrust.PUBLIC_RAW,
                      DataSourceTrust.PUBLIC_CURATED, DataSourceTrust.VERIFIED_PARTNER,
                      DataSourceTrust.INTERNAL]
 
        if trust_order.index(source.trust_level) < trust_order.index(self.required_trust_level):
            collection_result['status'] = 'rejected'
            collection_result['reason'] = f'Source trust level {source.trust_level.value} below required {self.required_trust_level.value}'
            return collection_result
 
        # Validate source is in allowlist
        if source.source_id not in self.allowed_sources and self.allowed_sources:
            collection_result['status'] = 'rejected'
            collection_result['reason'] = 'Source not in allowlist'
            return collection_result
 
        # Run content validators
        for validator in self.validators:
            result = validator.validate(data, source)
            collection_result['validations'].append({
                'validator': validator.name,
                'passed': result['passed'],
                'details': result.get('details')
            })
 
            if not result['passed'] and validator.is_blocking:
                collection_result['status'] = 'rejected'
                collection_result['reason'] = f'Validation failed: {validator.name}'
                return collection_result
 
        # Calculate integrity hash
        collection_result['data_hash'] = hashlib.sha256(data).hexdigest()
 
        # Store provenance
        collection_result['provenance'] = {
            'source_metadata': source.__dict__,
            'collector_version': self.version,
            'collection_timestamp': collection_result['timestamp']
        }
 
        collection_result['status'] = 'accepted'
        return collection_result
 
 
class ContentValidator:
    """Base class for data content validators."""
 
    def __init__(self, name: str, is_blocking: bool = True):
        self.name = name
        self.is_blocking = is_blocking
 
    def validate(self, data: bytes, source: DataSourceMetadata) -> dict:
        raise NotImplementedError
 
 
class MalwareValidator(ContentValidator):
    """Scan data for malware before ingestion."""
 
    def validate(self, data: bytes, source: DataSourceMetadata) -> dict:
        # Use antivirus scanning
        scan_result = self._scan_with_av(data)
 
        return {
            'passed': not scan_result['threats_found'],
            'details': scan_result
        }
 
 
class PIIValidator(ContentValidator):
    """Detect PII in training data."""
 
    def __init__(self):
        super().__init__('pii_detection', is_blocking=False)
        self.pii_patterns = self._load_pii_patterns()
 
    def validate(self, data: bytes, source: DataSourceMetadata) -> dict:
        text = data.decode('utf-8', errors='ignore')
        detected_pii = []
 
        for pii_type, pattern in self.pii_patterns.items():
            matches = re.findall(pattern, text)
            if matches:
                detected_pii.append({
                    'type': pii_type,
                    'count': len(matches),
                    'sample_redacted': True  # Don't log actual PII
                })
 
        return {
            'passed': len(detected_pii) == 0,
            'details': {
                'pii_detected': detected_pii,
                'recommendation': 'Review and redact PII before training'
            }
        }

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64
 
class SecureDataStorage:
    """Encrypted storage for training data."""
 
    def __init__(self, config: dict):
        self.encryption_key = self._derive_key(
            config['master_password'],
            config['salt']
        )
        self.cipher = Fernet(self.encryption_key)
        self.access_controller = AccessController(config['access_policy'])
        self.audit_logger = AuditLogger()
 
    def store_dataset(self, dataset_id: str, data: bytes,
                     metadata: dict, user: str) -> dict:
        """Store encrypted training dataset."""
 
        # Check write permission
        if not self.access_controller.can_write(user, dataset_id):
            self.audit_logger.log_access_denied(user, dataset_id, 'write')
            raise PermissionError(f"User {user} cannot write to {dataset_id}")
 
        # Encrypt data
        encrypted_data = self.cipher.encrypt(data)
 
        # Calculate integrity hash of encrypted data
        integrity_hash = hashlib.sha256(encrypted_data).hexdigest()
 
        # Store encrypted data
        storage_path = self._get_storage_path(dataset_id)
        with open(storage_path, 'wb') as f:
            f.write(encrypted_data)
 
        # Store metadata separately
        storage_metadata = {
            'dataset_id': dataset_id,
            'stored_at': datetime.utcnow().isoformat(),
            'stored_by': user,
            'original_size': len(data),
            'encrypted_size': len(encrypted_data),
            'integrity_hash': integrity_hash,
            'user_metadata': metadata
        }
 
        self._store_metadata(dataset_id, storage_metadata)
 
        # Audit log
        self.audit_logger.log_data_stored(user, dataset_id, storage_metadata)
 
        return {
            'dataset_id': dataset_id,
            'integrity_hash': integrity_hash,
            'stored_at': storage_metadata['stored_at']
        }
 
    def retrieve_dataset(self, dataset_id: str, user: str,
                        purpose: str) -> bytes:
        """Retrieve and decrypt training dataset."""
 
        # Check read permission
        if not self.access_controller.can_read(user, dataset_id):
            self.audit_logger.log_access_denied(user, dataset_id, 'read')
            raise PermissionError(f"User {user} cannot read {dataset_id}")
 
        # Load encrypted data
        storage_path = self._get_storage_path(dataset_id)
        with open(storage_path, 'rb') as f:
            encrypted_data = f.read()
 
        # Verify integrity
        metadata = self._load_metadata(dataset_id)
        actual_hash = hashlib.sha256(encrypted_data).hexdigest()
 
        if actual_hash != metadata['integrity_hash']:
            self.audit_logger.log_integrity_violation(dataset_id)
            raise IntegrityError(f"Dataset {dataset_id} integrity check failed")
 
        # Decrypt
        data = self.cipher.decrypt(encrypted_data)
 
        # Audit log
        self.audit_logger.log_data_accessed(user, dataset_id, purpose)
 
        return data
 
    def _derive_key(self, password: str, salt: bytes) -> bytes:
        """Derive encryption key from password."""
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=480000,
        )
        key = base64.urlsafe_b64encode(kdf.derive(password.encode()))
        return key
 
 
class AccessController:
    """Control access to training datasets."""
 
    def __init__(self, policy: dict):
        self.policy = policy
        self.role_permissions = policy.get('role_permissions', {})
        self.dataset_acls = policy.get('dataset_acls', {})
 
    def can_read(self, user: str, dataset_id: str) -> bool:
        """Check if user can read dataset."""
        user_roles = self._get_user_roles(user)
 
        # Check role-based permissions
        for role in user_roles:
            permissions = self.role_permissions.get(role, {})
            if 'read_all' in permissions.get('datasets', []):
                return True
 
        # Check dataset-specific ACL
        dataset_acl = self.dataset_acls.get(dataset_id, {})
        if user in dataset_acl.get('readers', []):
            return True
 
        return False
 
    def can_write(self, user: str, dataset_id: str) -> bool:
        """Check if user can write to dataset."""
        user_roles = self._get_user_roles(user)
 
        for role in user_roles:
            permissions = self.role_permissions.get(role, {})
            if 'write_all' in permissions.get('datasets', []):
                return True
 
        dataset_acl = self.dataset_acls.get(dataset_id, {})
        if user in dataset_acl.get('writers', []):
            return True
 
        return False

import numpy as np
from sklearn.ensemble import IsolationForest
from scipy.stats import zscore
 
class PoisoningDetector:
    """Detect potential data poisoning in training datasets."""
 
    def __init__(self, config: dict):
        self.anomaly_threshold = config.get('anomaly_threshold', 0.05)
        self.embedding_model = self._load_embedding_model(config)
 
    def detect_poisoning(self, dataset: List[dict],
                        reference_dataset: Optional[List[dict]] = None) -> dict:
        """Comprehensive poisoning detection."""
 
        results = {
            'total_samples': len(dataset),
            'detection_methods': [],
            'suspicious_samples': [],
            'overall_risk': 'low'
        }
 
        # Method 1: Statistical anomaly detection
        stat_result = self._statistical_detection(dataset)
        results['detection_methods'].append(stat_result)
 
        # Method 2: Embedding-based anomaly detection
        embedding_result = self._embedding_detection(dataset)
        results['detection_methods'].append(embedding_result)
 
        # Method 3: Label consistency check
        label_result = self._label_consistency_check(dataset)
        results['detection_methods'].append(label_result)
 
        # Method 4: Distribution shift detection (if reference provided)
        if reference_dataset:
            dist_result = self._distribution_shift_detection(
                dataset, reference_dataset
            )
            results['detection_methods'].append(dist_result)
 
        # Aggregate suspicious samples
        all_suspicious = set()
        for method in results['detection_methods']:
            all_suspicious.update(method.get('suspicious_indices', []))
 
        results['suspicious_samples'] = list(all_suspicious)
        results['suspicious_rate'] = len(all_suspicious) / len(dataset)
 
        # Determine overall risk
        if results['suspicious_rate'] > 0.1:
            results['overall_risk'] = 'high'
        elif results['suspicious_rate'] > 0.05:
            results['overall_risk'] = 'medium'
 
        return results
 
    def _statistical_detection(self, dataset: List[dict]) -> dict:
        """Detect statistical anomalies in features."""
 
        # Extract numerical features
        features = self._extract_features(dataset)
 
        # Calculate z-scores
        z_scores = np.abs(zscore(features, axis=0))
 
        # Identify outliers
        outlier_mask = np.any(z_scores > 3, axis=1)
        outlier_indices = np.where(outlier_mask)[0].tolist()
 
        return {
            'method': 'statistical',
            'outliers_detected': len(outlier_indices),
            'suspicious_indices': outlier_indices,
            'threshold': '3 sigma'
        }
 
    def _embedding_detection(self, dataset: List[dict]) -> dict:
        """Use embeddings to detect anomalous samples."""
 
        # Generate embeddings
        embeddings = []
        for sample in dataset:
            if 'text' in sample:
                emb = self.embedding_model.encode(sample['text'])
            elif 'image' in sample:
                emb = self.embedding_model.encode_image(sample['image'])
            embeddings.append(emb)
 
        embeddings = np.array(embeddings)
 
        # Isolation Forest for anomaly detection
        iso_forest = IsolationForest(contamination=self.anomaly_threshold)
        predictions = iso_forest.fit_predict(embeddings)
 
        anomaly_indices = np.where(predictions == -1)[0].tolist()
 
        return {
            'method': 'embedding_isolation_forest',
            'anomalies_detected': len(anomaly_indices),
            'suspicious_indices': anomaly_indices,
            'contamination': self.anomaly_threshold
        }
 
    def _label_consistency_check(self, dataset: List[dict]) -> dict:
        """Check for label inconsistencies."""
 
        # Group samples by similar content
        content_groups = self._cluster_by_content(dataset)
 
        inconsistent = []
        for group_id, indices in content_groups.items():
            labels = [dataset[i].get('label') for i in indices]
            unique_labels = set(labels)
 
            if len(unique_labels) > 1:
                # Same content, different labels - suspicious
                inconsistent.extend(indices)
 
        return {
            'method': 'label_consistency',
            'inconsistencies_found': len(inconsistent),
            'suspicious_indices': list(set(inconsistent))
        }
 
    def _distribution_shift_detection(self, new_data: List[dict],
                                     reference: List[dict]) -> dict:
        """Detect if new data distribution differs from reference."""
 
        new_embeddings = self._get_embeddings(new_data)
        ref_embeddings = self._get_embeddings(reference)
 
        # Maximum Mean Discrepancy
        mmd = self._compute_mmd(new_embeddings, ref_embeddings)
 
        # Per-sample distance from reference distribution
        distances = []
        for emb in new_embeddings:
            dist = np.min(np.linalg.norm(ref_embeddings - emb, axis=1))
            distances.append(dist)
 
        # Samples far from reference distribution
        threshold = np.percentile(distances, 95)
        suspicious = [i for i, d in enumerate(distances) if d > threshold]
 
        return {
            'method': 'distribution_shift',
            'mmd_score': float(mmd),
            'suspicious_indices': suspicious,
            'distribution_shift_detected': mmd > 0.1
        }

class DifferentialPrivacyTrainer:
    """Train models with differential privacy guarantees."""
 
    def __init__(self, config: dict):
        self.epsilon = config.get('epsilon', 1.0)
        self.delta = config.get('delta', 1e-5)
        self.max_grad_norm = config.get('max_grad_norm', 1.0)
        self.noise_multiplier = self._calculate_noise_multiplier()
 
    def train_with_dp(self, model, dataset, epochs: int) -> dict:
        """Train model with differential privacy."""
 
        from opacus import PrivacyEngine
 
        # Wrap model with privacy engine
        privacy_engine = PrivacyEngine()
        model, optimizer, dataloader = privacy_engine.make_private_with_epsilon(
            module=model,
            optimizer=optimizer,
            data_loader=dataloader,
            epochs=epochs,
            target_epsilon=self.epsilon,
            target_delta=self.delta,
            max_grad_norm=self.max_grad_norm
        )
 
        # Training loop with privacy accounting
        for epoch in range(epochs):
            for batch in dataloader:
                optimizer.zero_grad()
                loss = self._compute_loss(model, batch)
                loss.backward()
                optimizer.step()
 
            # Check privacy budget
            epsilon_spent = privacy_engine.get_epsilon(self.delta)
            if epsilon_spent > self.epsilon:
                break
 
        return {
            'epsilon_spent': epsilon_engine.get_epsilon(self.delta),
            'delta': self.delta,
            'epochs_completed': epoch + 1
        }
 
 
class FederatedLearningSecure:
    """Secure federated learning for distributed training data."""
 
    def __init__(self, config: dict):
        self.aggregation_method = config.get('aggregation', 'secure_aggregation')
        self.min_participants = config.get('min_participants', 3)
 
    def aggregate_updates(self, client_updates: List[dict]) -> dict:
        """Securely aggregate model updates from clients."""
 
        if len(client_updates) < self.min_participants:
            raise ValueError(f"Need at least {self.min_participants} participants")
 
        if self.aggregation_method == 'secure_aggregation':
            return self._secure_aggregate(client_updates)
        elif self.aggregation_method == 'differential_privacy':
            return self._dp_aggregate(client_updates)
        else:
            return self._simple_average(client_updates)
 
    def _secure_aggregate(self, updates: List[dict]) -> dict:
        """Use secure aggregation protocol."""
        # Implement secure multi-party computation
        # Each client's update is masked with random values
        # Only the sum is revealed
        pass
 
    def _dp_aggregate(self, updates: List[dict]) -> dict:
        """Add differential privacy noise to aggregation."""
        # Clip updates
        clipped = [self._clip_update(u) for u in updates]
 
        # Average
        averaged = self._average_updates(clipped)
 
        # Add noise
        noise_scale = self._calculate_noise_scale(len(updates))
        noisy = self._add_gaussian_noise(averaged, noise_scale)
 
        return noisy

class TrainingDataAudit:
    """Audit trail for training data usage."""
 
    def __init__(self, config: dict):
        self.retention_days = config.get('retention_days', 365)
        self.storage = AuditStorage(config['storage'])
 
    def log_data_usage(self, event: dict):
        """Log training data usage event."""
 
        audit_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'event_type': event['type'],
            'user': event['user'],
            'dataset_id': event['dataset_id'],
            'purpose': event.get('purpose'),
            'model_id': event.get('model_id'),
            'legal_basis': event.get('legal_basis'),
            'data_subjects_count': event.get('data_subjects_count'),
        }
 
        # Add integrity hash
        audit_entry['integrity_hash'] = self._compute_hash(audit_entry)
 
        self.storage.store(audit_entry)
 
    def generate_compliance_report(self, dataset_id: str,
                                  time_range: dict) -> dict:
        """Generate compliance report for dataset usage."""
 
        events = self.storage.query(
            dataset_id=dataset_id,
            start_time=time_range['start'],
            end_time=time_range['end']
        )
 
        report = {
            'dataset_id': dataset_id,
            'report_period': time_range,
            'generated_at': datetime.utcnow().isoformat(),
            'summary': {
                'total_accesses': len(events),
                'unique_users': len(set(e['user'] for e in events)),
                'purposes': list(set(e.get('purpose') for e in events if e.get('purpose'))),
                'models_trained': list(set(e.get('model_id') for e in events if e.get('model_id')))
            },
            'events': events,
            'compliance_checks': self._run_compliance_checks(events)
        }
 
        return report
 
    def _run_compliance_checks(self, events: List[dict]) -> List[dict]:
        """Run compliance checks on usage events."""
 
        checks = []
 
        # Check: All accesses have purpose
        missing_purpose = [e for e in events if not e.get('purpose')]
        checks.append({
            'check': 'purpose_documented',
            'passed': len(missing_purpose) == 0,
            'violations': len(missing_purpose)
        })
 
        # Check: All accesses have legal basis
        missing_legal = [e for e in events if not e.get('legal_basis')]
        checks.append({
            'check': 'legal_basis_documented',
            'passed': len(missing_legal) == 0,
            'violations': len(missing_legal)
        })
 
        # Check: Retention policy compliance
        old_events = [
            e for e in events
            if self._days_old(e['timestamp']) > self.retention_days
        ]
        checks.append({
            'check': 'retention_policy',
            'passed': len(old_events) == 0,
            'violations': len(old_events)
        })
 
        return checks