Cloud Data Loss Prevention: Protecting Sensitive Data

requirements.txt:

boto3>=1.34.0
python-dateutil>=2.8.2

Complete DLP Data Classification Implementation:

#!/usr/bin/env python3
"""
Cloud Data Loss Prevention - Data Classification Module
Production-ready data classification with comprehensive pattern matching
"""

import re
from typing import Optional, List, Dict, Set
from enum import Enum
from dataclasses import dataclass, asdict
from datetime import datetime
import logging
import hashlib
import json

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class DLPError(Exception):
    """Base exception for DLP errors."""
    pass


class ClassificationError(DLPError):
    """Raised when classification fails."""
    pass


class DataClassification(Enum):
    """Data classification levels."""
    PUBLIC = "public"
    INTERNAL = "internal"
    CONFIDENTIAL = "confidential"
    RESTRICTED = "restricted"
    
    def __str__(self) -> str:
        return self.value


@dataclass
class ClassificationResult:
    """Result of data classification with comprehensive metadata."""
    classification: DataClassification
    confidence: float
    matched_patterns: List[str]
    recommendations: List[str]
    timestamp: datetime
    content_hash: str
    pattern_details: Dict[str, int]  # Pattern name -> count of matches
    
    def __post_init__(self):
        """Set default timestamp if not provided."""
        if self.timestamp is None:
            self.timestamp = datetime.utcnow()
    
    def to_dict(self) -> Dict:
        """Convert to dictionary for serialization."""
        result = asdict(self)
        result['classification'] = self.classification.value
        result['timestamp'] = self.timestamp.isoformat()
        return result
    
    def to_json(self) -> str:
        """Convert to JSON string."""
        return json.dumps(self.to_dict(), indent=2)

class DataClassification(Enum):
    """Data classification levels."""
    PUBLIC = "public"
    INTERNAL = "internal"
    CONFIDENTIAL = "confidential"
    RESTRICTED = "restricted"

@dataclass
class ClassificationResult:
    """Result of data classification."""
    classification: DataClassification
    confidence: float
    matched_patterns: List[str]
    recommendations: List[str]

class DataClassifier:
    """Classifies data based on content patterns with comprehensive detection."""
    
    def __init__(self, custom_patterns: Optional[Dict] = None):
        """Initialize data classifier with comprehensive pattern definitions.
        
        Args:
            custom_patterns: Optional custom patterns to add or override defaults
        """
        # Comprehensive pattern definitions
        self.patterns = {
            # PII Patterns
            'ssn': {
                'pattern': r'\b\d{3}-\d{2}-\d{4}\b',
                'classification': DataClassification.CONFIDENTIAL,
                'description': 'Social Security Number',
                'confidence_boost': 0.3
            },
            'ssn_no_dash': {
                'pattern': r'\b\d{9}\b',
                'classification': DataClassification.CONFIDENTIAL,
                'description': 'Social Security Number (no dashes)',
                'confidence_boost': 0.2
            },
            'credit_card': {
                'pattern': r'\b(?:\d{4}[-\s]?){3}\d{4}\b',
                'classification': DataClassification.RESTRICTED,
                'description': 'Credit Card Number',
                'confidence_boost': 0.4
            },
            'credit_card_luhn': {
                'pattern': r'\b(?:\d{4}[-\s]?){3}\d{4}\b',
                'classification': DataClassification.RESTRICTIAL,
                'description': 'Credit Card Number (Luhn validated)',
                'confidence_boost': 0.5,
                'validate_luhn': True
            },
            'passport_us': {
                'pattern': r'\b\d{9}\b',
                'classification': DataClassification.CONFIDENTIAL,
                'description': 'US Passport Number',
                'confidence_boost': 0.2
            },
            'driver_license': {
                'pattern': r'\b[A-Z]{1,2}\d{6,8}\b',
                'classification': DataClassification.CONFIDENTIAL,
                'description': 'Driver License Number',
                'confidence_boost': 0.2
            },
            
            # Financial Patterns
            'bank_account': {
                'pattern': r'\b\d{8,17}\b',
                'classification': DataClassification.RESTRICTED,
                'description': 'Bank Account Number',
                'confidence_boost': 0.3
            },
            'routing_number': {
                'pattern': r'\b\d{9}\b',
                'classification': DataClassification.CONFIDENTIAL,
                'description': 'Bank Routing Number',
                'confidence_boost': 0.2
            },
            
            # Contact Information
            'email': {
                'pattern': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
                'classification': DataClassification.INTERNAL,
                'description': 'Email Address',
                'confidence_boost': 0.1
            },
            'phone_us': {
                'pattern': r'\b(?:\+1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b',
                'classification': DataClassification.INTERNAL,
                'description': 'US Phone Number',
                'confidence_boost': 0.1
            },
            'ip_address': {
                'pattern': r'\b(?:\d{1,3}\.){3}\d{1,3}\b',
                'classification': DataClassification.INTERNAL,
                'description': 'IP Address',
                'confidence_boost': 0.05
            },
            'mac_address': {
                'pattern': r'\b(?:[0-9A-Fa-f]{2}[:-]){5}(?:[0-9A-Fa-f]{2})\b',
                'classification': DataClassification.INTERNAL,
                'description': 'MAC Address',
                'confidence_boost': 0.05
            },
            
            # API Keys and Secrets
            'aws_access_key': {
                'pattern': r'\bAKIA[0-9A-Z]{16}\b',
                'classification': DataClassification.RESTRICTED,
                'description': 'AWS Access Key ID',
                'confidence_boost': 0.9
            },
            'aws_secret_key': {
                'pattern': r'\b[A-Za-z0-9/+=]{40}\b',
                'classification': DataClassification.RESTRICTED,
                'description': 'AWS Secret Access Key (potential)',
                'confidence_boost': 0.5
            },
            'api_key_generic': {
                'pattern': r'\b(?:api[_-]?key|apikey)[\s:=]+["\']?([A-Za-z0-9\-_]{20,})["\']?',
                'classification': DataClassification.CONFIDENTIAL,
                'description': 'Generic API Key',
                'confidence_boost': 0.6
            },
            'github_token': {
                'pattern': r'\bghp_[A-Za-z0-9]{36}\b',
                'classification': DataClassification.RESTRICTED,
                'description': 'GitHub Personal Access Token',
                'confidence_boost': 0.9
            },
            'private_key': {
                'pattern': r'-----BEGIN (?:RSA |DSA |EC )?PRIVATE KEY-----',
                'classification': DataClassification.RESTRICTED,
                'description': 'Private Key',
                'confidence_boost': 0.95
            },
            
            # Medical Information
            'hipaa_phi': {
                'pattern': r'\b(?:patient|medical record|diagnosis|prescription)\b',
                'classification': DataClassification.RESTRICTED,
                'description': 'Potential HIPAA PHI',
                'confidence_boost': 0.3
            }
        }
        
        # Add custom patterns if provided
        if custom_patterns:
            self.patterns.update(custom_patterns)
        
        # Compile all regex patterns for performance
        self.compiled_patterns = {}
        for pattern_name, pattern_info in self.patterns.items():
            try:
                self.compiled_patterns[pattern_name] = re.compile(
                    pattern_info['pattern'],
                    re.IGNORECASE | re.MULTILINE
                )
            except re.error as e:
                logger.warning(f"Invalid regex pattern for {pattern_name}: {e}")
                continue
    
    def classify_data(
        self, 
        content: str, 
        max_size_mb: int = 10,
        context: Optional[Dict] = None
    ) -> ClassificationResult:
        """Classify data based on patterns with comprehensive error handling.
        
        Args:
            content: Data content to classify
            max_size_mb: Maximum content size in MB (default 10MB)
            context: Optional context information (filename, source, etc.)
            
        Returns:
            ClassificationResult with classification and confidence
            
        Raises:
            ValueError: If content exceeds maximum size
            ClassificationError: If classification fails
        """
        if context is None:
            context = {}
        
        # Handle empty content
        if not content:
            return ClassificationResult(
                classification=DataClassification.PUBLIC,
                confidence=1.0,
                matched_patterns=[],
                recommendations=['Empty content classified as public'],
                timestamp=datetime.utcnow(),
                content_hash=self._hash_content(''),
                pattern_details={}
            )
        
        try:
            # Validate content size
            content_bytes = content.encode('utf-8')
            content_size_mb = len(content_bytes) / (1024 * 1024)
            
            if content_size_mb > max_size_mb:
                raise ValueError(
                    f"Content size {content_size_mb:.2f}MB exceeds maximum {max_size_mb}MB. "
                    f"Consider chunking or increasing max_size_mb."
                )
            
            # Calculate content hash
            content_hash = self._hash_content(content)
            
            # Find all pattern matches
            matched_patterns = []
            pattern_details = {}  # Pattern name -> count of matches
            highest_classification = DataClassification.PUBLIC
            total_confidence_boost = 0.0
            
            # Check each compiled pattern
            for pattern_name, compiled_pattern in self.compiled_patterns.items():
                try:
                    pattern_info = self.patterns[pattern_name]
                    matches = compiled_pattern.findall(content)
                    
                    if matches:
                        # Count matches
                        match_count = len(matches) if isinstance(matches, list) else 1
                        pattern_details[pattern_name] = match_count
                        matched_patterns.append(pattern_name)
                        
                        # Upgrade classification if this pattern requires higher level
                        pattern_classification = pattern_info['classification']
                        if self._is_higher_classification(pattern_classification, highest_classification):
                            highest_classification = pattern_classification
                        
                        # Add confidence boost
                        confidence_boost = pattern_info.get('confidence_boost', 0.1)
                        total_confidence_boost += confidence_boost * min(match_count, 5) / 5  # Cap boost at 5 matches
                        
                        # Validate with additional checks if needed
                        if pattern_info.get('validate_luhn', False):
                            # Validate credit card with Luhn algorithm
                            for match in (matches if isinstance(matches, list) else [matches]):
                                if isinstance(match, tuple):
                                    match = match[0] if match else ''
                                if self._validate_luhn(match.replace('-', '').replace(' ', '')):
                                    total_confidence_boost += 0.2
                        
                        logger.debug(
                            f"Pattern '{pattern_name}' matched {match_count} times "
                            f"({pattern_info['description']})"
                        )
                
                except re.error as e:
                    logger.warning(f"Regex error for pattern {pattern_name}: {e}")
                    continue
                except Exception as e:
                    logger.error(f"Unexpected error checking pattern {pattern_name}: {e}", exc_info=True)
                    continue
            
            # Calculate confidence based on matches and boosts
            base_confidence = 0.5
            if matched_patterns:
                # More patterns matched = higher confidence
                pattern_confidence = min(0.4, len(matched_patterns) * 0.1)
                # Confidence boost from pattern-specific boosts
                boost_confidence = min(0.5, total_confidence_boost)
                confidence = min(1.0, base_confidence + pattern_confidence + boost_confidence)
            else:
                confidence = base_confidence
            
            # Generate recommendations
            recommendations = self._generate_recommendations(
                highest_classification, 
                matched_patterns,
                context
            )
            
            result = ClassificationResult(
                classification=highest_classification,
                confidence=confidence,
                matched_patterns=matched_patterns,
                recommendations=recommendations,
                timestamp=datetime.utcnow(),
                content_hash=content_hash,
                pattern_details=pattern_details
            )
            
            logger.info(
                f"Classified data as {highest_classification.value} "
                f"(confidence: {confidence:.2f}, patterns: {len(matched_patterns)})"
            )
            
            return result
        
        except ValueError:
            raise
        except Exception as e:
            error_msg = f"Failed to classify data: {e}"
            logger.error(error_msg, exc_info=True)
            raise ClassificationError(error_msg) from e
    
    def _hash_content(self, content: str) -> str:
        """Calculate SHA-256 hash of content for tracking.
        
        Args:
            content: Content to hash
            
        Returns:
            Hexadecimal hash string
        """
        return hashlib.sha256(content.encode('utf-8')).hexdigest()
    
    def _validate_luhn(self, card_number: str) -> bool:
        """Validate credit card number using Luhn algorithm.
        
        Args:
            card_number: Credit card number string
            
        Returns:
            True if valid Luhn checksum, False otherwise
        """
        try:
            # Remove non-digit characters
            digits = [int(d) for d in card_number if d.isdigit()]
            
            if len(digits) < 13 or len(digits) > 19:
                return False
            
            # Luhn algorithm
            checksum = 0
            is_even = False
            
            # Process digits from right to left
            for digit in reversed(digits):
                if is_even:
                    digit *= 2
                    if digit > 9:
                        digit -= 9
                checksum += digit
                is_even = not is_even
            
            return checksum % 10 == 0
        
        except (ValueError, IndexError):
            return False
    
    def _is_higher_classification(self, new: DataClassification, current: DataClassification) -> bool:
        """Check if new classification is higher than current."""
        order = {
            DataClassification.PUBLIC: 0,
            DataClassification.INTERNAL: 1,
            DataClassification.CONFIDENTIAL: 2,
            DataClassification.RESTRICTED: 3
        }
        return order[new] > order[current]
    
    def _generate_recommendations(
        self, 
        classification: DataClassification, 
        patterns: List[str],
        context: Optional[Dict] = None
    ) -> List[str]:
        """Generate comprehensive security recommendations based on classification and patterns.
        
        Args:
            classification: Data classification level
            patterns: List of matched pattern names
            context: Optional context information
            
        Returns:
            List of security recommendations
        """
        if context is None:
            context = {}
        
        recommendations = []
        
        # Classification-based recommendations
        if classification == DataClassification.RESTRICTED:
            recommendations.extend([
                'Encrypt data at rest using AES-256 or stronger encryption',
                'Encrypt data in transit using TLS 1.3 or higher',
                'Restrict access to authorized personnel only with MFA required',
                'Enable comprehensive audit logging for all access and modifications',
                'Implement data loss prevention (DLP) policies to prevent exfiltration',
                'Store data in isolated, access-controlled environments',
                'Implement data retention and deletion policies',
                'Require approval workflows for data access and sharing',
                'Monitor for anomalous access patterns',
                'Comply with GDPR, HIPAA, PCI-DSS, and other relevant regulations'
            ])
        elif classification == DataClassification.CONFIDENTIAL:
            recommendations.extend([
                'Encrypt sensitive data at rest',
                'Encrypt data in transit using TLS',
                'Limit access based on need-to-know principle',
                'Enable access logging and monitoring',
                'Implement access controls and authentication',
                'Regular security audits and reviews',
                'Data classification labels and handling procedures'
            ])
        elif classification == DataClassification.INTERNAL:
            recommendations.extend([
                'Limit internal access appropriately',
                'Consider encryption for sensitive internal data',
                'Enable basic access logging',
                'Follow internal data handling policies'
            ])
        else:  # PUBLIC
            recommendations.extend([
                'Ensure no sensitive information is included',
                'Verify public sharing is appropriate',
                'Review data before making public'
            ])
        
        # Pattern-specific recommendations
        if 'aws_access_key' in patterns or 'aws_secret_key' in patterns:
            recommendations.extend([
                'CRITICAL: AWS credentials detected. Rotate immediately.',
                'Review AWS CloudTrail logs for unauthorized access',
                'Check for compromised credentials in public repositories',
                'Use IAM roles instead of access keys when possible',
                'Enable AWS credential rotation policies'
            ])
        
        if 'credit_card' in patterns:
            recommendations.extend([
                'PCI-DSS compliance required for credit card data',
                'Tokenize or encrypt credit card numbers',
                'Do not store CVV codes',
                'Implement PCI-DSS compliant infrastructure',
                'Restrict access to authorized payment processors only'
            ])
        
        if 'ssn' in patterns:
            recommendations.extend([
                'Social Security Numbers require highest protection',
                'Encrypt SSNs with strong encryption',
                'Minimize collection and storage of SSNs',
                'Comply with identity theft protection regulations',
                'Implement strict access controls'
            ])
        
        if 'private_key' in patterns or 'github_token' in patterns:
            recommendations.extend([
                'CRITICAL: Secrets detected. Rotate immediately.',
                'Never commit secrets to version control',
                'Use secret management systems (AWS Secrets Manager, HashiCorp Vault)',
                'Scan repositories for exposed secrets',
                'Implement secret rotation policies'
            ])
        
        if 'hipaa_phi' in patterns:
            recommendations.extend([
                'HIPAA compliance required for protected health information',
                'Encrypt PHI at rest and in transit',
                'Implement access controls and audit logging',
                'Train staff on HIPAA requirements',
                'Execute Business Associate Agreements (BAAs)'
            ])
        
        return recommendations

</details>

## Step 2) Create DLP policies

<details>
<summary>Click to view complete production-ready code</summary>

**Complete DLP Policy Enforcer Implementation:**

```python
#!/usr/bin/env python3
"""
Cloud Data Loss Prevention - Policy Enforcement Module
Production-ready DLP policy enforcement with comprehensive validation
"""

from typing import Dict, List, Optional, Set, Tuple
from dataclasses import dataclass, asdict
from enum import Enum
from datetime import datetime, timedelta
import logging
import json

logger = logging.getLogger(__name__)


class PolicyViolationError(DLPError):
    """Raised when DLP policy is violated."""
    pass


class ActionType(Enum):
    """Types of actions that can be performed on data."""
    READ = "read"
    WRITE = "write"
    SHARE = "share"
    EXTERNAL_SHARE = "external_share"
    DOWNLOAD = "download"
    UPLOAD = "upload"
    DELETE = "delete"
    EXPORT = "export"
    COPY = "copy"
    PRINT = "print"
    EMAIL = "email"
    
    def __str__(self) -> str:
        return self.value


@dataclass
class DLPPolicy:
    """DLP policy definition with comprehensive controls."""
    encryption_required: bool
    sharing_allowed: bool
    external_sharing_allowed: bool
    access_restricted: bool
    audit_logging_required: bool
    retention_days: Optional[int] = None
    mfa_required: bool = False
    watermark_required: bool = False
    copy_prevention: bool = False
    print_prevention: bool = False
    download_prevention: bool = False
    allowed_domains: Optional[Set[str]] = None
    blocked_domains: Optional[Set[str]] = None
    max_file_size_mb: Optional[int] = None
    allowed_file_types: Optional[Set[str]] = None
    blocked_file_types: Optional[Set[str]] = None
    
    def to_dict(self) -> Dict:
        """Convert to dictionary for serialization."""
        result = asdict(self)
        if self.allowed_domains:
            result['allowed_domains'] = list(self.allowed_domains)
        if self.blocked_domains:
            result['blocked_domains'] = list(self.blocked_domains)
        if self.allowed_file_types:
            result['allowed_file_types'] = list(self.allowed_file_types)
        if self.blocked_file_types:
            result['blocked_file_types'] = list(self.blocked_file_types)
        return result


@dataclass
class PolicyCheckResult:
    """Result of DLP policy check."""
    allowed: bool
    violation_reason: Optional[str] = None
    policy_applied: Optional[DLPPolicy] = None
    warnings: List[str] = None
    required_actions: List[str] = None
    
    def __post_init__(self):
        """Initialize default lists."""
        if self.warnings is None:
            self.warnings = []
        if self.required_actions is None:
            self.required_actions = []


class DLPPolicyEnforcer:
    """Enforces DLP policies based on data classification with comprehensive checks."""
    
    def __init__(self, custom_policies: Optional[Dict] = None):
        """Initialize policy enforcer with default and custom policies.
        
        Args:
            custom_policies: Optional custom policies to add or override defaults
        """
        # Default policies for each classification level
        self.policies = {
            DataClassification.RESTRICTED: DLPPolicy(
                encryption_required=True,
                sharing_allowed=False,
                external_sharing_allowed=False,
                access_restricted=True,
                audit_logging_required=True,
                retention_days=365,
                mfa_required=True,
                watermark_required=True,
                copy_prevention=True,
                print_prevention=True,
                download_prevention=True,
                allowed_domains=None,  # No sharing allowed
                blocked_domains=None,
                max_file_size_mb=100,
                allowed_file_types={'pdf', 'docx', 'xlsx'},  # Only specific types
                blocked_file_types={'exe', 'bat', 'sh', 'ps1'}  # Block executables
            ),
            DataClassification.CONFIDENTIAL: DLPPolicy(
                encryption_required=True,
                sharing_allowed=True,
                external_sharing_allowed=False,
                access_restricted=True,
                audit_logging_required=True,
                retention_days=730,
                mfa_required=True,
                watermark_required=True,
                copy_prevention=False,
                print_prevention=False,
                download_prevention=False,
                allowed_domains={'company.com', 'partner.com'},
                blocked_domains={'competitor.com', 'public-email.com'},
                max_file_size_mb=500,
                allowed_file_types=None,
                blocked_file_types={'exe', 'bat', 'sh', 'ps1'}
            ),
            DataClassification.INTERNAL: DLPPolicy(
                encryption_required=False,
                sharing_allowed=True,
                external_sharing_allowed=False,
                access_restricted=False,
                audit_logging_required=False,
                retention_days=None,
                mfa_required=False,
                watermark_required=False,
                copy_prevention=False,
                print_prevention=False,
                download_prevention=False,
                allowed_domains={'company.com'},
                blocked_domains=None,
                max_file_size_mb=1000,
                allowed_file_types=None,
                blocked_file_types={'exe', 'bat', 'sh', 'ps1'}
            ),
            DataClassification.PUBLIC: DLPPolicy(
                encryption_required=False,
                sharing_allowed=True,
                external_sharing_allowed=True,
                access_restricted=False,
                audit_logging_required=False,
                retention_days=None,
                mfa_required=False,
                watermark_required=False,
                copy_prevention=False,
                print_prevention=False,
                download_prevention=False,
                allowed_domains=None,
                blocked_domains=None,
                max_file_size_mb=None,
                allowed_file_types=None,
                blocked_file_types=None
            )
        }
        
        # Add or override with custom policies
        if custom_policies:
            self.policies.update(custom_policies)
        
        logger.info("Initialized DLPPolicyEnforcer with policies for all classification levels")
    
    def get_policy(self, classification: DataClassification) -> DLPPolicy:
        """Get DLP policy for classification level.
        
        Args:
            classification: Data classification level
            
        Returns:
            DLPPolicy for the classification level
        """
        policy = self.policies.get(classification)
        if policy is None:
            logger.warning(f"No policy found for {classification}, using PUBLIC policy")
            policy = self.policies[DataClassification.PUBLIC]
        
        return policy
    
    def check_policy_violation(
        self, 
        classification: DataClassification,
        action: ActionType,
        context: Dict
    ) -> PolicyCheckResult:
        """Check if action violates DLP policy with comprehensive validation.
        
        Args:
            classification: Data classification level
            action: Action being attempted
            context: Context about the action containing:
                - user: User attempting action
                - destination: Destination for sharing/downloading
                - file_size_mb: File size in MB (for upload/store)
                - file_type: File type/extension
                - encrypted: Whether data is encrypted
                - mfa_authenticated: Whether user has MFA
                - domain: Domain for email/sharing
                - ip_address: Source IP address
                - timestamp: Action timestamp
                
        Returns:
            PolicyCheckResult indicating if action is allowed and any violations
            
        Raises:
            PolicyViolationError: If policy violation is critical
        """
        policy = self.get_policy(classification)
        warnings = []
        required_actions = []
        violation_reason = None
        
        try:
            # Check action-specific policies
            if action == ActionType.SHARE:
                if not policy.sharing_allowed:
                    violation_reason = f"Sharing not allowed for {classification.value} data"
                    return PolicyCheckResult(
                        allowed=False,
                        violation_reason=violation_reason,
                        policy_applied=policy
                    )
                
                # Check domain restrictions
                destination = context.get('destination', '')
                domain = context.get('domain', '')
                
                if domain:
                    if policy.blocked_domains and domain in policy.blocked_domains:
                        violation_reason = f"Sharing to domain {domain} is blocked for {classification.value} data"
                        return PolicyCheckResult(
                            allowed=False,
                            violation_reason=violation_reason,
                            policy_applied=policy
                        )
                    
                    if policy.allowed_domains and domain not in policy.allowed_domains:
                        violation_reason = f"Domain {domain} not in allowed domains for {classification.value} data"
                        return PolicyCheckResult(
                            allowed=False,
                            violation_reason=violation_reason,
                            policy_applied=policy
                        )
            
            elif action == ActionType.EXTERNAL_SHARE:
                if not policy.external_sharing_allowed:
                    violation_reason = f"External sharing not allowed for {classification.value} data"
                    return PolicyCheckResult(
                        allowed=False,
                        violation_reason=violation_reason,
                        policy_applied=policy
                    )
            
            elif action in [ActionType.DOWNLOAD, ActionType.EXPORT]:
                if policy.download_prevention:
                    violation_reason = f"Download/export not allowed for {classification.value} data"
                    return PolicyCheckResult(
                        allowed=False,
                        violation_reason=violation_reason,
                        policy_applied=policy
                    )
            
            elif action == ActionType.COPY:
                if policy.copy_prevention:
                    violation_reason = f"Copying not allowed for {classification.value} data"
                    return PolicyCheckResult(
                        allowed=False,
                        violation_reason=violation_reason,
                        policy_applied=policy
                    )
            
            elif action == ActionType.PRINT:
                if policy.print_prevention:
                    violation_reason = f"Printing not allowed for {classification.value} data"
                    return PolicyCheckResult(
                        allowed=False,
                        violation_reason=violation_reason,
                        policy_applied=policy
                    )
            
            elif action in [ActionType.UPLOAD, ActionType.WRITE, ActionType.STORE]:
                # Check encryption requirements
                if policy.encryption_required and not context.get('encrypted', False):
                    violation_reason = f"Encryption required for storing {classification.value} data"
                    return PolicyCheckResult(
                        allowed=False,
                        violation_reason=violation_reason,
                        policy_applied=policy,
                        required_actions=['encrypt_data']
                    )
                
                # Check file size limits
                file_size_mb = context.get('file_size_mb', 0)
                if policy.max_file_size_mb and file_size_mb > policy.max_file_size_mb:
                    violation_reason = (
                        f"File size {file_size_mb}MB exceeds maximum "
                        f"{policy.max_file_size_mb}MB for {classification.value} data"
                    )
                    return PolicyCheckResult(
                        allowed=False,
                        violation_reason=violation_reason,
                        policy_applied=policy
                    )
                
                # Check file type restrictions
                file_type = context.get('file_type', '').lower().lstrip('.')
                if file_type:
                    if policy.blocked_file_types and file_type in policy.blocked_file_types:
                        violation_reason = f"File type .{file_type} is blocked for {classification.value} data"
                        return PolicyCheckResult(
                            allowed=False,
                            violation_reason=violation_reason,
                            policy_applied=policy
                        )
                    
                    if policy.allowed_file_types and file_type not in policy.allowed_file_types:
                        violation_reason = f"File type .{file_type} not in allowed types for {classification.value} data"
                        return PolicyCheckResult(
                            allowed=False,
                            violation_reason=violation_reason,
                            policy_applied=policy
                        )
            
            # Check MFA requirements
            if policy.mfa_required and not context.get('mfa_authenticated', False):
                warnings.append(f"MFA is recommended for {classification.value} data access")
                required_actions.append('require_mfa')
            
            # Check watermark requirements
            if policy.watermark_required and not context.get('watermarked', False):
                warnings.append(f"Watermark should be applied to {classification.value} data")
                required_actions.append('add_watermark')
            
            # Check audit logging
            if policy.audit_logging_required:
                required_actions.append('log_access')
            
            # All checks passed
            result = PolicyCheckResult(
                allowed=True,
                policy_applied=policy,
                warnings=warnings,
                required_actions=required_actions
            )
            
            logger.info(
                f"Policy check passed: {action.value} on {classification.value} data "
                f"by {context.get('user', 'unknown')}"
            )
            
            return result
        
        except Exception as e:
            error_msg = f"Error checking policy violation: {e}"
            logger.error(error_msg, exc_info=True)
            raise PolicyViolationError(error_msg) from e
    
    def enforce_action(
        self,
        classification: DataClassification,
        action: ActionType,
        context: Dict
    ) -> PolicyCheckResult:
        """Enforce policy and raise exception if violated (strict enforcement).
        
        Args:
            classification: Data classification level
            action: Action being attempted
            context: Context about the action
            
        Returns:
            PolicyCheckResult if action is allowed
            
        Raises:
            PolicyViolationError: If policy violation detected
        """
        result = self.check_policy_violation(classification, action, context)
        
        if not result.allowed:
            raise PolicyViolationError(
                f"DLP policy violation: {result.violation_reason}"
            )
        
        return result


# Example usage
if __name__ == "__main__":
    from data_classifier import DataClassifier, DataClassification
    
    # Initialize classifier and enforcer
    classifier = DataClassifier()
    enforcer = DLPPolicyEnforcer()
    
    # Classify sample data
    sample_data = "Customer SSN: 123-45-6789, Credit Card: 4532-1234-5678-9010"
    classification_result = classifier.classify_data(sample_data)
    
    print(f"Classification: {classification_result.classification.value}")
    print(f"Confidence: {classification_result.confidence:.2f}")
    print(f"Matched patterns: {classification_result.matched_patterns}")
    
    # Check policy for sharing action
    context = {
        'user': 'john.doe@company.com',
        'domain': 'partner.com',
        'mfa_authenticated': True,
        'encrypted': True
    }
    
    try:
        result = enforcer.enforce_action(
            classification_result.classification,
            ActionType.SHARE,
            context
        )
        print(f"\nAction allowed: {result.allowed}")
        if result.warnings:
            print(f"Warnings: {result.warnings}")
        if result.required_actions:
            print(f"Required actions: {result.required_actions}")
    except PolicyViolationError as e:
        print(f"\nPolicy violation: {e}")

Unit Tests:

# test_dlp_policy_enforcer.py
import pytest
from dlp_policy_enforcer import (
    DLPPolicyEnforcer,
    DLPPolicy,
    ActionType,
    PolicyViolationError,
    PolicyCheckResult
)
from data_classifier import DataClassification


class TestDLPPolicyEnforcer:
    """Unit tests for DLPPolicyEnforcer."""
    
    @pytest.fixture
    def enforcer(self):
        """Create DLPPolicyEnforcer instance."""
        return DLPPolicyEnforcer()
    
    def test_get_policy_restricted(self, enforcer):
        """Test getting policy for RESTRICTED classification."""
        policy = enforcer.get_policy(DataClassification.RESTRICTED)
        
        assert policy.encryption_required is True
        assert policy.sharing_allowed is False
        assert policy.mfa_required is True
        assert policy.copy_prevention is True
    
    def test_check_sharing_violation(self, enforcer):
        """Test policy violation for sharing RESTRICTED data."""
        context = {
            'user': 'test@company.com',
            'destination': 'external@partner.com',
            'domain': 'partner.com'
        }
        
        result = enforcer.check_policy_violation(
            DataClassification.RESTRICTED,
            ActionType.SHARE,
            context
        )
        
        assert result.allowed is False
        assert 'not allowed' in result.violation_reason.lower()
    
    def test_check_external_sharing_violation(self, enforcer):
        """Test policy violation for external sharing CONFIDENTIAL data."""
        context = {
            'user': 'test@company.com',
            'destination': 'external@public.com',
            'domain': 'public.com'
        }
        
        result = enforcer.check_policy_violation(
            DataClassification.CONFIDENTIAL,
            ActionType.EXTERNAL_SHARE,
            context
        )
        
        assert result.allowed is False
        assert 'not allowed' in result.violation_reason.lower()
    
    def test_check_encryption_requirement(self, enforcer):
        """Test encryption requirement for CONFIDENTIAL data."""
        context = {
            'user': 'test@company.com',
            'encrypted': False,
            'file_size_mb': 10
        }
        
        result = enforcer.check_policy_violation(
            DataClassification.CONFIDENTIAL,
            ActionType.STORE,
            context
        )
        
        assert result.allowed is False
        assert 'encryption' in result.violation_reason.lower()
        assert 'encrypt_data' in result.required_actions
    
    def test_enforce_action_raises_exception(self, enforcer):
        """Test that enforce_action raises exception on violation."""
        context = {
            'user': 'test@company.com',
            'encrypted': False
        }
        
        with pytest.raises(PolicyViolationError):
            enforcer.enforce_action(
                DataClassification.CONFIDENTIAL,
                ActionType.STORE,
                context
            )
    
    def test_allowed_action_returns_result(self, enforcer):
        """Test that allowed action returns PolicyCheckResult."""
        context = {
            'user': 'test@company.com',
            'encrypted': True,
            'mfa_authenticated': True,
            'file_size_mb': 10
        }
        
        result = enforcer.enforce_action(
            DataClassification.INTERNAL,
            ActionType.READ,
            context
        )
        
        assert isinstance(result, PolicyCheckResult)
        assert result.allowed is True


if __name__ == '__main__':
    pytest.main([__file__, '-v'])

import pytest

class TestDataClassifier:
    """Unit tests for DataClassifier."""
    
    def test_classify_ssn(self):
        """Test SSN classification."""
        classifier = DataClassifier()
        result = classifier.classify_data("SSN: 123-45-6789")
        assert result.classification == DataClassification.CONFIDENTIAL
        assert 'ssn' in result.matched_patterns
    
    def test_classify_credit_card(self):
        """Test credit card classification."""
        classifier = DataClassifier()
        result = classifier.classify_data("Card: 1234-5678-9012-3456")
        assert result.classification == DataClassification.RESTRICTED
        assert 'credit_card' in result.matched_patterns
    
    def test_classify_large_content(self):
        """Test classification with size limit."""
        classifier = DataClassifier()
        large_content = "x" * (11 * 1024 * 1024)  # 11MB
        with pytest.raises(ValueError):
            classifier.classify_data(large_content, max_size_mb=10)

class TestDLPPolicyEnforcer:
    """Unit tests for DLPPolicyEnforcer."""
    
    def test_restricted_sharing_blocked(self):
        """Test that restricted data sharing is blocked."""
        enforcer = DLPPolicyEnforcer()
        is_violation, reason = enforcer.check_policy_violation(
            DataClassification.RESTRICTED,
            'share',
            {}
        )
        assert is_violation
        assert 'not allowed' in reason

# Remove DLP scanning jobs (if created)
aws macie2 delete-findings-filter --id <filter-id>

# Remove classification tags (if added)
aws s3api delete-object-tagging --bucket <bucket> --key <key>

# Remove IAM roles (if created)
aws iam delete-role-policy --role-name DLPClassifierRole --policy-name DLPClassifierPolicy
aws iam delete-role --role-name DLPClassifierRole

# Verify cleanup
aws macie2 list-findings-filters
# Should show no DLP-related filters