Cloud Compliance and Governance: Meeting Regulatory Requi...

requirements.txt:

boto3>=1.34.0
python-dateutil>=2.8.2

Complete Compliance Framework Implementation:

#!/usr/bin/env python3
"""
Cloud Compliance & Governance - Compliance Framework Manager
Production-ready compliance framework management with comprehensive validation
"""

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

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


class ComplianceError(Exception):
    """Base exception for compliance errors."""
    pass


class FrameworkNotFoundError(ComplianceError):
    """Raised when compliance framework is not found."""
    pass


class ComplianceLevel(Enum):
    """Compliance requirement levels."""
    REQUIRED = "required"
    RECOMMENDED = "recommended"
    OPTIONAL = "optional"
    NOT_APPLICABLE = "not_applicable"
    
    def __str__(self) -> str:
        return self.value


class ControlStatus(Enum):
    """Status of a compliance control."""
    COMPLIANT = "compliant"
    NON_COMPLIANT = "non_compliant"
    NOT_ASSESSED = "not_assessed"
    IN_PROGRESS = "in_progress"
    EXEMPT = "exempt"
    
    def __str__(self) -> str:
        return self.value


@dataclass
class ComplianceRequirement:
    """Represents a compliance requirement with comprehensive metadata."""
    name: str
    level: ComplianceLevel
    description: str
    category: str
    validation_function: Optional[str] = None
    remediation_guidance: Optional[str] = None
    evidence_required: bool = False
    frequency: str = "continuous"  # continuous, daily, weekly, monthly, annually
    
    def to_dict(self) -> Dict:
        """Convert to dictionary for serialization."""
        result = asdict(self)
        result['level'] = self.level.value
        return result


@dataclass
class ControlAssessment:
    """Represents an assessment of a compliance control."""
    control_name: str
    status: ControlStatus
    assessed_date: datetime
    assessed_by: str
    evidence: Optional[str] = None
    findings: List[str] = field(default_factory=list)
    remediation_plan: Optional[str] = None
    next_assessment_date: Optional[datetime] = None
    
    def to_dict(self) -> Dict:
        """Convert to dictionary for serialization."""
        result = asdict(self)
        result['status'] = self.status.value
        result['assessed_date'] = self.assessed_date.isoformat() if self.assessed_date else None
        result['next_assessment_date'] = self.next_assessment_date.isoformat() if self.next_assessment_date else None
        return result


@dataclass
class ComplianceFramework:
    """Represents a compliance framework with comprehensive controls."""
    name: str
    version: str
    description: str
    requirements: List[ComplianceRequirement]
    controls: Dict[str, ComplianceLevel]
    applicable_regions: List[str] = field(default_factory=lambda: ['global'])
    reference_url: Optional[str] = None
    last_updated: Optional[datetime] = None
    
    def __post_init__(self):
        """Set default last_updated if not provided."""
        if self.last_updated is None:
            self.last_updated = datetime.utcnow()
    
    def to_dict(self) -> Dict:
        """Convert to dictionary for serialization."""
        result = asdict(self)
        result['last_updated'] = self.last_updated.isoformat() if self.last_updated else None
        result['requirements'] = [req.to_dict() for req in self.requirements]
        result['controls'] = {k: v.value for k, v in self.controls.items()}
        return result


@dataclass
class ComplianceReport:
    """Comprehensive compliance assessment report."""
    framework_name: str
    assessment_date: datetime
    overall_compliance_percentage: float
    total_controls: int
    compliant_controls: int
    non_compliant_controls: int
    assessments: Dict[str, ControlAssessment]
    summary: str
    recommendations: List[str] = field(default_factory=list)
    
    def to_dict(self) -> Dict:
        """Convert to dictionary for serialization."""
        result = asdict(self)
        result['assessment_date'] = self.assessment_date.isoformat()
        result['assessments'] = {k: v.to_dict() for k, v in self.assessments.items()}
        return result
    
    def to_json(self) -> str:
        """Convert to JSON string."""
        return json.dumps(self.to_dict(), indent=2)


class ComplianceFrameworkManager:
    """Manages compliance frameworks with comprehensive error handling and assessment capabilities."""
    
    def __init__(self):
        """Initialize compliance framework manager with comprehensive frameworks."""
        self.frameworks: Dict[str, ComplianceFramework] = {}
        self.assessments: Dict[str, Dict[str, ControlAssessment]] = {}  # framework -> control -> assessment
        
        # Initialize standard frameworks
        self._initialize_standard_frameworks()
        
        logger.info(f"Initialized ComplianceFrameworkManager with {len(self.frameworks)} frameworks")
    
    def _initialize_standard_frameworks(self) -> None:
        """Initialize standard compliance frameworks."""
        # GDPR Framework
        self.frameworks['GDPR'] = ComplianceFramework(
                name="GDPR",
            version="1.0",
            description="General Data Protection Regulation (EU) 2016/679",
            reference_url="https://gdpr.eu/",
                requirements=[
                    ComplianceRequirement(
                        name="data_encryption",
                        level=ComplianceLevel.REQUIRED,
                    description="Encrypt personal data at rest and in transit",
                    category="Technical Safeguards",
                    remediation_guidance="Use AES-256 encryption for data at rest and TLS 1.3 for data in transit",
                    evidence_required=True,
                    frequency="continuous"
                    ),
                    ComplianceRequirement(
                        name="access_controls",
                        level=ComplianceLevel.REQUIRED,
                    description="Implement access controls and authentication for personal data",
                    category="Access Control",
                    remediation_guidance="Implement role-based access control (RBAC) with least privilege principle",
                    evidence_required=True,
                    frequency="continuous"
                    ),
                    ComplianceRequirement(
                        name="audit_logging",
                        level=ComplianceLevel.REQUIRED,
                    description="Log all access, modification, and deletion of personal data",
                    category="Audit & Monitoring",
                    remediation_guidance="Enable comprehensive audit logging with tamper-proof storage",
                    evidence_required=True,
                    frequency="continuous"
                    ),
                    ComplianceRequirement(
                        name="data_retention",
                        level=ComplianceLevel.REQUIRED,
                    description="Implement data retention and deletion policies",
                    category="Data Lifecycle",
                    remediation_guidance="Define and enforce retention periods. Automate data deletion when retention expires",
                    evidence_required=True,
                    frequency="monthly"
                ),
                ComplianceRequirement(
                    name="right_to_erasure",
                    level=ComplianceLevel.REQUIRED,
                    description="Support data subject's right to erasure (Article 17)",
                    category="Data Subject Rights",
                    remediation_guidance="Implement automated data deletion workflows for subject access requests",
                    evidence_required=True,
                    frequency="on_demand"
                ),
                ComplianceRequirement(
                    name="data_breach_notification",
                    level=ComplianceLevel.REQUIRED,
                    description="Notify supervisory authority within 72 hours of data breach",
                    category="Incident Response",
                    remediation_guidance="Establish incident response procedures and breach notification workflows",
                    evidence_required=True,
                    frequency="on_demand"
                    )
                ],
                controls={
                    "data_encryption": ComplianceLevel.REQUIRED,
                    "access_controls": ComplianceLevel.REQUIRED,
                    "audit_logging": ComplianceLevel.REQUIRED,
                "data_retention": ComplianceLevel.REQUIRED,
                "right_to_erasure": ComplianceLevel.REQUIRED,
                "data_breach_notification": ComplianceLevel.REQUIRED
            },
            applicable_regions=['EU', 'EEA']
        )
        
        # HIPAA Framework
        self.frameworks['HIPAA'] = ComplianceFramework(
                name="HIPAA",
            version="1.0",
            description="Health Insurance Portability and Accountability Act",
            reference_url="https://www.hhs.gov/hipaa/index.html",
                requirements=[
                    ComplianceRequirement(
                        name="encryption_at_rest",
                        level=ComplianceLevel.REQUIRED,
                    description="Encrypt Protected Health Information (PHI) at rest",
                    category="Technical Safeguards",
                    remediation_guidance="Use NIST-approved encryption (AES-256) for PHI storage",
                    evidence_required=True,
                    frequency="continuous"
                    ),
                    ComplianceRequirement(
                        name="encryption_in_transit",
                        level=ComplianceLevel.REQUIRED,
                    description="Encrypt PHI in transit",
                    category="Technical Safeguards",
                    remediation_guidance="Use TLS 1.2 or higher for all PHI transmission",
                    evidence_required=True,
                    frequency="continuous"
                    ),
                    ComplianceRequirement(
                        name="access_controls",
                        level=ComplianceLevel.REQUIRED,
                    description="Implement access controls for PHI",
                    category="Administrative Safeguards",
                    remediation_guidance="Enforce least privilege, role-based access, and regular access reviews",
                    evidence_required=True,
                    frequency="quarterly"
                    ),
                    ComplianceRequirement(
                        name="audit_trails",
                        level=ComplianceLevel.REQUIRED,
                    description="Maintain audit trails for PHI access and modifications",
                    category="Technical Safeguards",
                    remediation_guidance="Log all PHI access with user, timestamp, and action details",
                    evidence_required=True,
                    frequency="continuous"
                ),
                ComplianceRequirement(
                    name="baa_requirements",
                    level=ComplianceLevel.REQUIRED,
                    description="Execute Business Associate Agreements (BAAs) with third-party vendors",
                    category="Administrative Safeguards",
                    remediation_guidance="Ensure all vendors handling PHI have signed BAAs",
                    evidence_required=True,
                    frequency="annually"
                ),
                ComplianceRequirement(
                    name="workforce_training",
                    level=ComplianceLevel.REQUIRED,
                    description="Train workforce on HIPAA requirements and PHI handling",
                    category="Administrative Safeguards",
                    remediation_guidance="Provide annual HIPAA training to all staff handling PHI",
                    evidence_required=True,
                    frequency="annually"
                    )
                ],
                controls={
                    "encryption_at_rest": ComplianceLevel.REQUIRED,
                    "encryption_in_transit": ComplianceLevel.REQUIRED,
                    "access_controls": ComplianceLevel.REQUIRED,
                "audit_trails": ComplianceLevel.REQUIRED,
                "baa_requirements": ComplianceLevel.REQUIRED,
                "workforce_training": ComplianceLevel.REQUIRED
            },
            applicable_regions=['US']
        )
        
        # PCI-DSS Framework
        self.frameworks['PCI-DSS'] = ComplianceFramework(
            name="PCI-DSS",
            version="3.2.1",
            description="Payment Card Industry Data Security Standard",
            reference_url="https://www.pcisecuritystandards.org/",
            requirements=[
                ComplianceRequirement(
                    name="firewall_configuration",
                    level=ComplianceLevel.REQUIRED,
                    description="Install and maintain firewall configuration",
                    category="Network Security",
                    remediation_guidance="Implement network segmentation and firewall rules",
                    evidence_required=True,
                    frequency="quarterly"
                ),
                ComplianceRequirement(
                    name="cardholder_data_encryption",
                    level=ComplianceLevel.REQUIRED,
                    description="Protect stored cardholder data with encryption",
                    category="Data Protection",
                    remediation_guidance="Encrypt cardholder data using strong cryptography (AES-256)",
                    evidence_required=True,
                    frequency="continuous"
                ),
                ComplianceRequirement(
                    name="transmission_encryption",
                    level=ComplianceLevel.REQUIRED,
                    description="Encrypt transmission of cardholder data across public networks",
                    category="Data Protection",
                    remediation_guidance="Use TLS 1.2+ for all cardholder data transmission",
                    evidence_required=True,
                    frequency="continuous"
                ),
                ComplianceRequirement(
                    name="anti_virus",
                    level=ComplianceLevel.REQUIRED,
                    description="Use and regularly update anti-virus software",
                    category="Malware Protection",
                    remediation_guidance="Deploy and maintain anti-virus on all systems",
                    evidence_required=True,
                    frequency="daily"
                ),
                ComplianceRequirement(
                    name="vulnerability_management",
                    level=ComplianceLevel.REQUIRED,
                    description="Develop and maintain secure systems and applications",
                    category="Vulnerability Management",
                    remediation_guidance="Regular vulnerability scanning and patch management",
                    evidence_required=True,
                    frequency="quarterly"
                )
            ],
            controls={
                "firewall_configuration": ComplianceLevel.REQUIRED,
                "cardholder_data_encryption": ComplianceLevel.REQUIRED,
                "transmission_encryption": ComplianceLevel.REQUIRED,
                "anti_virus": ComplianceLevel.REQUIRED,
                "vulnerability_management": ComplianceLevel.REQUIRED
            },
            applicable_regions=['global']
        )
        
        # SOC 2 Framework
        self.frameworks['SOC2'] = ComplianceFramework(
            name="SOC2",
            version="2017",
            description="Service Organization Control 2 - Trust Service Criteria",
            reference_url="https://www.aicpa.org/interestareas/frc/assuranceadvisoryservices/aicpasoc2report.html",
            requirements=[
                ComplianceRequirement(
                    name="security",
                    level=ComplianceLevel.REQUIRED,
                    description="Protect against unauthorized access",
                    category="Security",
                    remediation_guidance="Implement access controls, encryption, and security monitoring",
                    evidence_required=True,
                    frequency="continuous"
                ),
                ComplianceRequirement(
                    name="availability",
                    level=ComplianceLevel.RECOMMENDED,
                    description="Ensure system availability",
                    category="Availability",
                    remediation_guidance="Implement redundancy, monitoring, and disaster recovery",
                    evidence_required=True,
                    frequency="continuous"
                ),
                ComplianceRequirement(
                    name="processing_integrity",
                    level=ComplianceLevel.RECOMMENDED,
                    description="Ensure processing integrity",
                    category="Processing Integrity",
                    remediation_guidance="Implement data validation and error handling",
                    evidence_required=True,
                    frequency="continuous"
                ),
                ComplianceRequirement(
                    name="confidentiality",
                    level=ComplianceLevel.RECOMMENDED,
                    description="Maintain confidentiality of information",
                    category="Confidentiality",
                    remediation_guidance="Encrypt sensitive data and control access",
                    evidence_required=True,
                    frequency="continuous"
                ),
                ComplianceRequirement(
                    name="privacy",
                    level=ComplianceLevel.OPTIONAL,
                    description="Protect privacy of personal information",
                    category="Privacy",
                    remediation_guidance="Implement privacy controls and data handling procedures",
                    evidence_required=True,
                    frequency="continuous"
                )
            ],
            controls={
                "security": ComplianceLevel.REQUIRED,
                "availability": ComplianceLevel.RECOMMENDED,
                "processing_integrity": ComplianceLevel.RECOMMENDED,
                "confidentiality": ComplianceLevel.RECOMMENDED,
                "privacy": ComplianceLevel.OPTIONAL
            },
            applicable_regions=['global']
        )
    
    def get_framework(self, framework_name: str) -> ComplianceFramework:
        """Get compliance framework by name.
        
        Args:
            framework_name: Name of framework (GDPR, HIPAA, PCI-DSS, SOC2, etc.)
            
        Returns:
            ComplianceFramework object
            
        Raises:
            FrameworkNotFoundError: If framework not found
        """
        if framework_name not in self.frameworks:
            available = ", ".join(self.frameworks.keys())
            raise FrameworkNotFoundError(
                f"Framework '{framework_name}' not found. Available frameworks: {available}"
            )
        
        return self.frameworks[framework_name]
    
    def list_frameworks(self) -> List[str]:
        """List all available compliance frameworks.
        
        Returns:
            List of framework names
        """
        return list(self.frameworks.keys())
    
    def assess_control(
        self,
        framework_name: str,
        control_name: str,
        status: ControlStatus,
        assessed_by: str,
        evidence: Optional[str] = None,
        findings: Optional[List[str]] = None,
        remediation_plan: Optional[str] = None
    ) -> ControlAssessment:
        """Assess a compliance control.
        
        Args:
            framework_name: Name of compliance framework
            control_name: Name of control to assess
            status: Assessment status
            assessed_by: Name/ID of person performing assessment
            evidence: Evidence of compliance (documentation, screenshots, etc.)
            findings: List of findings/issues
            remediation_plan: Plan to remediate non-compliance
            
        Returns:
            ControlAssessment object
            
        Raises:
            FrameworkNotFoundError: If framework not found
        """
        framework = self.get_framework(framework_name)
        
        if control_name not in framework.controls:
            raise ComplianceError(
                f"Control '{control_name}' not found in framework '{framework_name}'"
            )
        
        # Determine next assessment date based on requirement frequency
        requirement = next(
            (r for r in framework.requirements if r.name == control_name),
            None
        )
        
        next_assessment_date = None
        if requirement:
            if requirement.frequency == "continuous":
                next_assessment_date = datetime.utcnow() + timedelta(days=1)
            elif requirement.frequency == "daily":
                next_assessment_date = datetime.utcnow() + timedelta(days=1)
            elif requirement.frequency == "weekly":
                next_assessment_date = datetime.utcnow() + timedelta(weeks=1)
            elif requirement.frequency == "monthly":
                next_assessment_date = datetime.utcnow() + timedelta(days=30)
            elif requirement.frequency == "quarterly":
                next_assessment_date = datetime.utcnow() + timedelta(days=90)
            elif requirement.frequency == "annually":
                next_assessment_date = datetime.utcnow() + timedelta(days=365)
        
        assessment = ControlAssessment(
            control_name=control_name,
            status=status,
            assessed_date=datetime.utcnow(),
            assessed_by=assessed_by,
            evidence=evidence,
            findings=findings or [],
            remediation_plan=remediation_plan,
            next_assessment_date=next_assessment_date
        )
        
        # Store assessment
        if framework_name not in self.assessments:
            self.assessments[framework_name] = {}
        self.assessments[framework_name][control_name] = assessment
        
        logger.info(
            f"Assessed control '{control_name}' in framework '{framework_name}': {status.value}"
        )
        
        return assessment
    
    def generate_compliance_report(self, framework_name: str) -> ComplianceReport:
        """Generate comprehensive compliance assessment report.
        
        Args:
            framework_name: Name of compliance framework
            
        Returns:
            ComplianceReport with assessment results
            
        Raises:
            FrameworkNotFoundError: If framework not found
        """
        framework = self.get_framework(framework_name)
        
        # Get assessments for this framework
        framework_assessments = self.assessments.get(framework_name, {})
        
        total_controls = len(framework.controls)
        compliant_controls = sum(
            1 for a in framework_assessments.values()
            if a.status == ControlStatus.COMPLIANT
        )
        non_compliant_controls = sum(
            1 for a in framework_assessments.values()
            if a.status == ControlStatus.NON_COMPLIANT
        )
        
        compliance_percentage = (
            (compliant_controls / total_controls * 100)
            if total_controls > 0 else 0.0
        )
        
        # Generate summary
        summary = (
            f"Compliance assessment for {framework_name}:\n"
            f"- Total Controls: {total_controls}\n"
            f"- Compliant: {compliant_controls}\n"
            f"- Non-Compliant: {non_compliant_controls}\n"
            f"- Not Assessed: {total_controls - len(framework_assessments)}\n"
            f"- Overall Compliance: {compliance_percentage:.1f}%"
        )
        
        # Generate recommendations
        recommendations = []
        for control_name, level in framework.controls.items():
            if control_name not in framework_assessments:
                recommendations.append(
                    f"Assess control '{control_name}' ({level.value})"
                )
            elif framework_assessments[control_name].status == ControlStatus.NON_COMPLIANT:
                assessment = framework_assessments[control_name]
                recommendations.append(
                    f"Remediate non-compliant control '{control_name}': "
                    f"{assessment.remediation_plan or 'No remediation plan provided'}"
                )
        
        # Create assessments dict (include all controls)
        all_assessments = {}
        for control_name in framework.controls:
            if control_name in framework_assessments:
                all_assessments[control_name] = framework_assessments[control_name]
            else:
                # Create default "not assessed" assessment
                all_assessments[control_name] = ControlAssessment(
                    control_name=control_name,
                    status=ControlStatus.NOT_ASSESSED,
                    assessed_date=datetime.utcnow(),
                    assessed_by="system"
                )
        
        report = ComplianceReport(
            framework_name=framework_name,
            assessment_date=datetime.utcnow(),
            overall_compliance_percentage=compliance_percentage,
            total_controls=total_controls,
            compliant_controls=compliant_controls,
            non_compliant_controls=non_compliant_controls,
            assessments=all_assessments,
            summary=summary,
            recommendations=recommendations
        )
        
        logger.info(f"Generated compliance report for {framework_name}: {compliance_percentage:.1f}% compliant")
        
        return report


# Example usage
if __name__ == "__main__":
    manager = ComplianceFrameworkManager()
    
    # List available frameworks
    print("Available frameworks:", manager.list_frameworks())
    
    # Get GDPR framework
    gdpr = manager.get_framework("GDPR")
    print(f"\n{gdpr.name} ({gdpr.version}): {gdpr.description}")
    print(f"Controls: {list(gdpr.controls.keys())}")
    
    # Assess a control
    assessment = manager.assess_control(
        framework_name="GDPR",
        control_name="data_encryption",
        status=ControlStatus.COMPLIANT,
        assessed_by="security-team",
        evidence="Encryption enabled on all databases using AES-256"
    )
    print(f"\nAssessed: {assessment.control_name} - {assessment.status.value}")
    
    # Generate compliance report
    report = manager.generate_compliance_report("GDPR")
    print(f"\n{report.summary}")
    print(f"\nRecommendations:")
    for rec in report.recommendations[:5]:  # Show first 5
        print(f"- {rec}")

Unit Tests:

# test_compliance_framework.py
import pytest
from compliance_framework import (
    ComplianceFrameworkManager,
    ControlStatus,
    FrameworkNotFoundError,
    ComplianceError
)


class TestComplianceFrameworkManager:
    """Unit tests for ComplianceFrameworkManager."""
    
    @pytest.fixture
    def manager(self):
        """Create ComplianceFrameworkManager instance."""
        return ComplianceFrameworkManager()
    
    def test_list_frameworks(self, manager):
        """Test listing available frameworks."""
        frameworks = manager.list_frameworks()
        assert len(frameworks) > 0
        assert "GDPR" in frameworks
        assert "HIPAA" in frameworks
    
    def test_get_framework_success(self, manager):
        """Test getting existing framework."""
        framework = manager.get_framework("GDPR")
        assert framework.name == "GDPR"
        assert len(framework.requirements) > 0
    
    def test_get_framework_not_found(self, manager):
        """Test getting non-existent framework raises error."""
        with pytest.raises(FrameworkNotFoundError):
            manager.get_framework("NONEXISTENT")
    
    def test_assess_control(self, manager):
        """Test assessing a control."""
        assessment = manager.assess_control(
            framework_name="GDPR",
            control_name="data_encryption",
            status=ControlStatus.COMPLIANT,
            assessed_by="test-user"
        )
        
        assert assessment.control_name == "data_encryption"
        assert assessment.status == ControlStatus.COMPLIANT
        assert assessment.assessed_by == "test-user"
    
    def test_generate_compliance_report(self, manager):
        """Test generating compliance report."""
        # Assess some controls
        manager.assess_control(
            "GDPR", "data_encryption", ControlStatus.COMPLIANT, "test-user"
        )
        manager.assess_control(
            "GDPR", "access_controls", ControlStatus.NON_COMPLIANT, "test-user"
        )
        
        report = manager.generate_compliance_report("GDPR")
        
        assert report.framework_name == "GDPR"
        assert report.total_controls > 0
        assert report.compliant_controls >= 1
        assert report.non_compliant_controls >= 1
        assert len(report.recommendations) > 0


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

from typing import Dict, List, Callable

class ComplianceChecker:
    """Checks resources against compliance frameworks."""
    
    def __init__(self):
        """Initialize compliance checker."""
        self.validation_functions = {
            "data_encryption": self._check_encryption,
            "access_controls": self._check_access_controls,
            "audit_logging": self._check_audit_logging,
            "encryption_at_rest": self._check_encryption_at_rest,
            "encryption_in_transit": self._check_encryption_in_transit
        }
    
    def check_compliance(self, resources: List[Dict], 
                        framework: ComplianceFramework) -> Dict:
        """Check resources against compliance framework.
        
        Args:
            resources: List of cloud resources to check
            framework: Compliance framework to check against
            
        Returns:
            Dictionary with compliance results
        """
        violations = []
        compliant_resources = []
        
        for resource in resources:
            resource_violations = []
            for requirement in framework.requirements:
                if requirement.level == ComplianceLevel.REQUIRED:
                    # Check if requirement is met
                    validator = self.validation_functions.get(requirement.name)
                    if validator:
                        is_compliant = validator(resource)
                        if not is_compliant:
                            resource_violations.append({
                                "requirement": requirement.name,
                                "description": requirement.description,
                                "level": requirement.level.value
                            })
            
            if resource_violations:
                violations.append({
                    "resource": resource.get('id', 'unknown'),
                    "violations": resource_violations
                })
            else:
                compliant_resources.append(resource)
        
        compliance_percentage = (len(compliant_resources) / max(len(resources), 1)) * 100
        
        return {
            "framework": framework.name,
            "total_resources": len(resources),
            "compliant_resources": len(compliant_resources),
            "violations": violations,
            "compliance_percentage": compliance_percentage,
            "is_compliant": len(violations) == 0
        }
    
    def _check_encryption(self, resource: Dict) -> bool:
        """Check if resource has encryption enabled."""
        # Implementation depends on resource type
        return resource.get('encryption_enabled', False)
    
    def _check_access_controls(self, resource: Dict) -> bool:
        """Check if resource has access controls."""
        return resource.get('access_controls_enabled', False)
    
    def _check_audit_logging(self, resource: Dict) -> bool:
        """Check if resource has audit logging."""
        return resource.get('audit_logging_enabled', False)
    
    def _check_encryption_at_rest(self, resource: Dict) -> bool:
        """Check if resource has encryption at rest."""
        return resource.get('encryption_at_rest', False)
    
    def _check_encryption_in_transit(self, resource: Dict) -> bool:
        """Check if resource has encryption in transit."""
        return resource.get('encryption_in_transit', False)

import pytest

class TestComplianceFrameworkManager:
    """Unit tests for ComplianceFrameworkManager."""
    
    def test_get_framework_success(self):
        """Test getting existing framework."""
        manager = ComplianceFrameworkManager()
        framework = manager.get_framework("GDPR")
        assert framework.name == "GDPR"
        assert len(framework.requirements) > 0
    
    def test_get_framework_not_found(self):
        """Test getting non-existent framework."""
        manager = ComplianceFrameworkManager()
        with pytest.raises(ValueError):
            manager.get_framework("INVALID")
    
    def test_list_frameworks(self):
        """Test listing all frameworks."""
        manager = ComplianceFrameworkManager()
        frameworks = manager.list_frameworks()
        assert "GDPR" in frameworks
        assert "HIPAA" in frameworks

class TestComplianceChecker:
    """Unit tests for ComplianceChecker."""
    
    def test_check_compliance_compliant(self):
        """Test compliance check for compliant resources."""
        checker = ComplianceChecker()
        manager = ComplianceFrameworkManager()
        framework = manager.get_framework("GDPR")
        
        resources = [{
            'id': 'resource1',
            'encryption_enabled': True,
            'access_controls_enabled': True,
            'audit_logging_enabled': True
        }]
        
        result = checker.check_compliance(resources, framework)
        assert result['is_compliant'] == True
        assert len(result['violations']) == 0

# Remove AWS Config rules (if created)
aws configservice delete-config-rule --config-rule-name compliance-rule

# Remove compliance monitoring resources
aws cloudformation delete-stack --stack-name compliance-monitoring

# Remove IAM roles (if created)
aws iam delete-role-policy --role-name ComplianceMonitorRole --policy-name ComplianceMonitorPolicy
aws iam delete-role --role-name ComplianceMonitorRole

# Verify cleanup
aws configservice describe-config-rules
# Should not show compliance-related rules