Regulatory Compliance

# HIPAA compliance framework for AI systems
import hashlib
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any
from cryptography.fernet import Fernet
from dataclasses import dataclass
import logging

@dataclass
class PHIAccess:
    """Track PHI access for HIPAA audit"""
    user_id: str
    patient_id: str
    data_accessed: List[str]
    purpose: str
    timestamp: datetime
    ip_address: str

class HIPAACompliantAI:
    """HIPAA-compliant AI system for healthcare"""
    
    def __init__(self):
        self.encryption_key = Fernet.generate_key()
        self.cipher = Fernet(self.encryption_key)
        self.audit_log = []
        self.setup_logging()
    
    def setup_logging(self):
        """Configure HIPAA-compliant audit logging"""
        logging.basicConfig(
            filename='hipaa_audit.log',
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
    
    def encrypt_phi(self, phi_data: Dict) -> bytes:
        """Encrypt PHI data at rest"""
        json_data = json.dumps(phi_data)
        encrypted = self.cipher.encrypt(json_data.encode())
        return encrypted
    
    def decrypt_phi(self, encrypted_data: bytes) -> Dict:
        """Decrypt PHI data with audit trail"""
        decrypted = self.cipher.decrypt(encrypted_data)
        return json.loads(decrypted.decode())
    
    def de_identify_data(self, patient_data: Dict) -> Dict:
        """Remove HIPAA identifiers (Safe Harbor method)"""
        
        # 18 HIPAA identifiers to remove
        identifiers = [
            'name', 'address', 'dates', 'phone', 'fax',
            'email', 'ssn', 'medical_record_number',
            'health_plan_number', 'account_number',
            'license_number', 'vehicle_id', 'device_id',
            'url', 'ip_address', 'biometric',
            'photo', 'unique_identifier'
        ]
        
        de_identified = patient_data.copy()
        
        for identifier in identifiers:
            if identifier in de_identified:
                # Replace with hash for consistency
                de_identified[identifier] = self._hash_identifier(
                    de_identified[identifier]
                )
        
        # Handle dates (keep year only if age < 90)
        if 'birth_date' in de_identified:
            birth_year = de_identified['birth_date'][:4]
            age = datetime.now().year - int(birth_year)
            if age >= 90:
                de_identified['birth_date'] = '90+'
            else:
                de_identified['birth_date'] = birth_year
        
        return de_identified
    
    def _hash_identifier(self, identifier: str) -> str:
        """Create consistent hash for identifier"""
        return hashlib.sha256(str(identifier).encode()).hexdigest()[:10]
    
    def check_minimum_necessary(self,
                                user_role: str,
                                requested_fields: List[str]) -> List[str]:
        """Enforce minimum necessary standard"""
        
        role_permissions = {
            'physician': ['diagnosis', 'treatment', 'medications', 
                          'lab_results', 'medical_history'],
            'nurse': ['vitals', 'medications', 'allergies'],
            'researcher': ['de_identified_data'],
            'billing': ['procedure_codes', 'insurance'],
            'ai_model': ['de_identified_data', 'lab_results']
        }
        
        allowed_fields = role_permissions.get(user_role, [])
        
        # Return only permitted fields
        return [field for field in requested_fields 
                if field in allowed_fields]
    
    def log_phi_access(self, access: PHIAccess):
        """Log PHI access for HIPAA audit trail"""
        
        log_entry = {
            'timestamp': access.timestamp.isoformat(),
            'user_id': access.user_id,
            'patient_id': self._hash_identifier(access.patient_id),
            'data_accessed': access.data_accessed,
            'purpose': access.purpose,
            'ip_address': access.ip_address
        }
        
        self.audit_log.append(log_entry)
        logging.info(f"PHI Access: {json.dumps(log_entry)}")
    
    def detect_potential_breach(self,
                                access_patterns: List[PHIAccess]) -> bool:
        """Detect potential HIPAA breaches"""
        
        # Check for suspicious patterns
        user_access_count = {}
        
        for access in access_patterns:
            user_id = access.user_id
            if user_id not in user_access_count:
                user_access_count[user_id] = 0
            user_access_count[user_id] += 1
        
        # Flag if user accessed too many records
        for user_id, count in user_access_count.items():
            if count > 100:  # Threshold for suspicious activity
                logging.warning(
                    f"Potential breach: User {user_id} accessed {count} records"
                )
                return True
        
        return False
    
    def generate_baa_requirements(self) -> Dict:
        """Generate Business Associate Agreement requirements"""
        
        return {
            'safeguards': [
                'Implement administrative safeguards',
                'Implement physical safeguards',
                'Implement technical safeguards'
            ],
            'permitted_uses': [
                'Perform functions on behalf of covered entity',
                'Provide data aggregation services',
                'Management and administration'
            ],
            'required_provisions': [
                'Report breaches to covered entity',
                'Ensure subcontractors sign BAA',
                'Make PHI available for access',
                'Return or destroy PHI at termination'
            ]
        }

# GDPR compliance framework for AI systems
import uuid
from enum import Enum
from typing import Dict, List, Optional

class LawfulBasis(Enum):
    CONSENT = "consent"
    CONTRACT = "contract"
    LEGAL_OBLIGATION = "legal_obligation"
    VITAL_INTERESTS = "vital_interests"
    PUBLIC_TASK = "public_task"
    LEGITIMATE_INTERESTS = "legitimate_interests"

class GDPRCompliantAI:
    """GDPR-compliant AI system"""
    
    def __init__(self):
        self.consent_records = {}
        self.processing_activities = []
        self.data_subjects = {}
    
    def obtain_consent(self,
                       data_subject_id: str,
                       purposes: List[str],
                       data_categories: List[str]) -> str:
        """Obtain and record GDPR-compliant consent"""
        
        consent_id = str(uuid.uuid4())
        
        self.consent_records[consent_id] = {
            'data_subject_id': data_subject_id,
            'timestamp': datetime.now().isoformat(),
            'purposes': purposes,
            'data_categories': data_categories,
            'freely_given': True,
            'specific': True,
            'informed': True,
            'unambiguous': True,
            'withdrawable': True,
            'version': '1.0'
        }
        
        return consent_id
    
    def withdraw_consent(self, consent_id: str) -> bool:
        """Process consent withdrawal"""
        
        if consent_id in self.consent_records:
            self.consent_records[consent_id]['withdrawn'] = True
            self.consent_records[consent_id]['withdrawn_at'] = \
                datetime.now().isoformat()
            
            # Stop processing based on this consent
            self._halt_processing(consent_id)
            return True
        
        return False
    
    def conduct_dpia(self, ai_system: Dict) -> Dict:
        """Conduct Data Protection Impact Assessment"""
        
        dpia = {
            'system_name': ai_system['name'],
            'assessment_date': datetime.now().isoformat(),
            'high_risk_processing': False,
            'risks': [],
            'mitigations': []
        }
        
        # Check for high-risk processing
        high_risk_indicators = [
            ai_system.get('automated_decision_making', False),
            ai_system.get('large_scale_processing', False),
            ai_system.get('sensitive_data', False),
            ai_system.get('systematic_monitoring', False),
            ai_system.get('profiling', False)
        ]
        
        if sum(high_risk_indicators) >= 2:
            dpia['high_risk_processing'] = True
            dpia['requires_dpo_consultation'] = True
        
        # Identify risks
        if ai_system.get('automated_decision_making'):
            dpia['risks'].append({
                'risk': 'Automated decision without human oversight',
                'severity': 'high',
                'likelihood': 'medium'
            })
            dpia['mitigations'].append(
                'Implement human review process'
            )
        
        if ai_system.get('profiling'):
            dpia['risks'].append({
                'risk': 'Discriminatory profiling',
                'severity': 'high',
                'likelihood': 'medium'
            })
            dpia['mitigations'].append(
                'Regular bias testing and fairness audits'
            )
        
        return dpia
    
    def handle_data_subject_request(self,
                                   request_type: str,
                                   data_subject_id: str) -> Dict:
        """Handle GDPR data subject rights requests"""
        
        response = {
            'request_id': str(uuid.uuid4()),
            'request_type': request_type,
            'data_subject_id': data_subject_id,
            'received_at': datetime.now().isoformat(),
            'status': 'processing'
        }
        
        if request_type == 'access':
            # Right to access (Article 15)
            response['data'] = self._get_subject_data(data_subject_id)
            response['processing_purposes'] = self._get_processing_purposes(
                data_subject_id
            )
            
        elif request_type == 'rectification':
            # Right to rectification (Article 16)
            response['rectification_form'] = self._generate_rectification_form()
            
        elif request_type == 'erasure':
            # Right to erasure/be forgotten (Article 17)
            response['erasure_result'] = self._erase_subject_data(
                data_subject_id
            )
            
        elif request_type == 'portability':
            # Right to data portability (Article 20)
            response['portable_data'] = self._export_portable_data(
                data_subject_id
            )
            
        elif request_type == 'restriction':
            # Right to restriction (Article 18)
            response['restriction_applied'] = self._restrict_processing(
                data_subject_id
            )
            
        elif request_type == 'object':
            # Right to object (Article 21)
            response['objection_result'] = self._process_objection(
                data_subject_id
            )
        
        response['status'] = 'completed'
        response['completed_at'] = datetime.now().isoformat()
        
        return response
    
    def implement_privacy_by_design(self) -> Dict:
        """Implement privacy by design principles"""
        
        return {
            'proactive': 'Preventive not remedial approach',
            'default': 'Privacy as default setting',
            'embedded': 'Privacy embedded into design',
            'positive_sum': 'Full functionality with privacy',
            'lifecycle': 'End-to-end security',
            'visibility': 'Transparency and openness',
            'user_centric': 'Respect for user privacy'
        }

# SOC2 compliance monitoring for AI systems
class SOC2ComplianceMonitor:
    """Monitor and ensure SOC2 compliance"""
    
    def __init__(self):
        self.controls = {}
        self.evidence = []
        self.exceptions = []
    
    def implement_security_controls(self) -> Dict:
        """Implement SOC2 security criteria controls"""
        
        controls = {
            'CC6.1': {
                'description': 'Logical and physical access controls',
                'implementation': [
                    'Multi-factor authentication',
                    'Role-based access control',
                    'Biometric data center access'
                ]
            },
            'CC6.2': {
                'description': 'Encryption at rest and in transit',
                'implementation': [
                    'AES-256 encryption for data at rest',
                    'TLS 1.3 for data in transit',
                    'Key management system'
                ]
            },
            'CC6.3': {
                'description': 'Vulnerability management',
                'implementation': [
                    'Weekly vulnerability scans',
                    'Quarterly penetration testing',
                    'Patch management process'
                ]
            }
        }
        
        self.controls.update(controls)
        return controls
    
    def monitor_availability(self, 
                            uptime_percentage: float,
                            response_time_ms: float) -> Dict:
        """Monitor availability criteria"""
        
        availability_status = {
            'uptime': uptime_percentage,
            'meets_sla': uptime_percentage >= 99.9,
            'response_time': response_time_ms,
            'acceptable_response': response_time_ms < 1000
        }
        
        if not availability_status['meets_sla']:
            self.exceptions.append({
                'criteria': 'Availability',
                'issue': f"Uptime {uptime_percentage}% below 99.9% SLA",
                'timestamp': datetime.now().isoformat()
            })
        
        return availability_status
    
    def validate_processing_integrity(self,
                                     model_outputs: List[Dict]) -> Dict:
        """Validate AI model processing integrity"""
        
        integrity_checks = {
            'completeness': True,
            'validity': True,
            'accuracy': True,
            'timeliness': True,
            'authorization': True
        }
        
        for output in model_outputs:
            # Check completeness
            if not all(k in output for k in ['input', 'output', 'timestamp']):
                integrity_checks['completeness'] = False
            
            # Check validity
            if output.get('confidence', 0) < 0.7:
                integrity_checks['validity'] = False
        
        return integrity_checks
    
    def generate_soc2_report(self) -> Dict:
        """Generate SOC2 Type II report evidence"""
        
        return {
            'report_type': 'SOC2 Type II',
            'period': {
                'start': (datetime.now() - timedelta(days=365)).isoformat(),
                'end': datetime.now().isoformat()
            },
            'controls_tested': len(self.controls),
            'exceptions_noted': len(self.exceptions),
            'opinion': 'Unqualified' if len(self.exceptions) == 0 else 'Qualified',
            'controls': self.controls,
            'exceptions': self.exceptions,
            'management_assertion': True
        }

# EU AI Act compliance assessment
class EUAIActCompliance:
    """Assess and ensure EU AI Act compliance"""
    
    def classify_ai_system(self, system: Dict) -> str:
        """Classify AI system risk level per EU AI Act"""
        
        # Check for prohibited uses
        prohibited_uses = [
            'social_scoring',
            'real_time_biometric_id',
            'subliminal_manipulation',
            'exploiting_vulnerabilities'
        ]
        
        if any(use in system.get('use_cases', []) for use in prohibited_uses):
            return 'UNACCEPTABLE_RISK'
        
        # Check for high-risk applications
        high_risk_areas = [
            'biometric_identification',
            'critical_infrastructure',
            'education_vocational',
            'employment',
            'essential_services',
            'law_enforcement',
            'migration_asylum',
            'justice'
        ]
        
        if any(area in system.get('application_area', []) for area in high_risk_areas):
            return 'HIGH_RISK'
        
        # Check for limited risk
        if system.get('interacts_with_humans') or system.get('generates_content'):
            return 'LIMITED_RISK'
        
        return 'MINIMAL_RISK'
    
    def high_risk_requirements(self) -> Dict:
        """Requirements for high-risk AI systems"""
        
        return {
            'risk_management': 'Establish risk management system',
            'data_governance': 'Training data quality and governance',
            'documentation': 'Technical documentation before market',
            'record_keeping': 'Automatic recording of events',
            'transparency': 'Clear information to users',
            'human_oversight': 'Appropriate human oversight measures',
            'accuracy': 'Appropriate accuracy, robustness, cybersecurity',
            'conformity': 'Conformity assessment before market',
            'registration': 'Register in EU database'
        }