How to Reduce Cloud Security Risks by 50% with AWS Config 5 and Azure Policy 8 for Compliance

import boto3
import json
import logging
from botocore.exceptions import ClientError, NoCredentialsError
from typing import Dict, List, Optional

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("config_deployment.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

class AWSConfig5Deployer:
    """Deploys AWS Config 5 managed rules with error handling and idempotency."""

    def __init__(self, region: str = "us-east-1"):
        try:
            self.config_client = boto3.client("config", region_name=region)
            self.s3_client = boto3.client("s3", region_name=region)
            logger.info(f"Initialized Config client for region {region}")
        except NoCredentialsError:
            logger.error("No AWS credentials found. Configure via AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY or IAM role")
            raise
        except ClientError as e:
            logger.error(f"Failed to initialize AWS clients: {e.response['Error']['Message']}")
            raise

    def deploy_managed_rule(self, rule_name: str, rule_identifier: str, resource_types: List[str], 
                           evaluation_mode: str = "DETECTIVE", maximum_execution_frequency: Optional[str] = None) -> bool:
        """
        Deploys a single AWS Config 5 managed rule.

        Args:
            rule_name: Human-readable name for the rule (e.g., "s3-bucket-public-read-prohibited")
            rule_identifier: Config 5 managed rule ID (e.g., "S3_BUCKET_PUBLIC_READ_PROHIBITED")
            resource_types: List of resource types to scope the rule to (e.g., ["AWS::S3::Bucket"])
            evaluation_mode: "DETECTIVE" (passive check) or "PROACTIVE" (blocks non-compliant deploys)
            maximum_execution_frequency: For periodic rules, e.g., "One_Hour", "Six_Hours"

        Returns:
            bool: True if deployment succeeded, False otherwise
        """
        try:
            # Check if rule already exists to avoid duplicates
            existing_rules = self.config_client.describe_config_rules(ConfigRuleNames=[rule_name])
            if existing_rules.get("ConfigRules"):
                logger.warning(f"Rule {rule_name} already exists. Updating instead of creating.")
                return self.update_managed_rule(rule_name, rule_identifier, resource_types, evaluation_mode, maximum_execution_frequency)

            # Construct rule parameters
            rule_params = {
                "ConfigRuleName": rule_name,
                "Description": f"AWS Config 5 managed rule: {rule_identifier}",
                "Source": {
                    "Owner": "AWS",
                    "SourceIdentifier": rule_identifier,
                    "SourceDetails": [
                        {
                            "EventSource": "aws.config",
                            "MessageType": "ConfigurationItemChangeNotification",
                            "MaximumExecutionFrequency": maximum_execution_frequency or "Six_Hours"
                        }
                    ]
                },
                "Scope": {
                    "ComplianceResourceTypes": resource_types
                },
                "EvaluationModes": [evaluation_mode]
            }

            # Add execution frequency only for periodic rules
            if maximum_execution_frequency and evaluation_mode == "DETECTIVE":
                rule_params["MaximumExecutionFrequency"] = maximum_execution_frequency

            response = self.config_client.put_config_rule(ConfigRule=rule_params)
            logger.info(f"Successfully deployed rule {rule_name} (Rule ID: {rule_identifier})")
            return True

        except ClientError as e:
            error_code = e.response["Error"]["Code"]
            error_msg = e.response["Error"]["Message"]
            if error_code == "ResourceInUseException":
                logger.warning(f"Rule {rule_name} is in use. Retry after 10 seconds.")
                import time
                time.sleep(10)
                return self.deploy_managed_rule(rule_name, rule_identifier, resource_types, evaluation_mode, maximum_execution_frequency)
            logger.error(f"Failed to deploy rule {rule_name}: {error_msg}")
            return False
        except Exception as e:
            logger.error(f"Unexpected error deploying rule {rule_name}: {str(e)}")
            return False

    def update_managed_rule(self, rule_name: str, rule_identifier: str, resource_types: List[str],
                           evaluation_mode: str, maximum_execution_frequency: Optional[str]) -> bool:
        """Updates an existing Config rule with new parameters."""
        try:
            update_params = {
                "ConfigRuleName": rule_name,
                "Source": {
                    "Owner": "AWS",
                    "SourceIdentifier": rule_identifier
                },
                "Scope": {
                    "ComplianceResourceTypes": resource_types
                },
                "EvaluationModes": [evaluation_mode]
            }
            if maximum_execution_frequency:
                update_params["MaximumExecutionFrequency"] = maximum_execution_frequency

            self.config_client.put_config_rule(ConfigRule=update_params)
            logger.info(f"Updated existing rule {rule_name}")
            return True
        except ClientError as e:
            logger.error(f"Failed to update rule {rule_name}: {e.response['Error']['Message']}")
            return False

if __name__ == "__main__":
    # Deploy 3 high-priority Config 5 rules for S3, IAM, and Lambda
    deployer = AWSConfig5Deployer(region="us-east-1")

    rules_to_deploy = [
        {
            "rule_name": "s3-no-public-read",
            "rule_identifier": "S3_BUCKET_PUBLIC_READ_PROHIBITED",
            "resource_types": ["AWS::S3::Bucket"],
            "evaluation_mode": "DETECTIVE",
            "max_freq": "One_Hour"
        },
        {
            "rule_name": "iam-user-mfa-enabled",
            "rule_identifier": "IAM_USER_MFA_ENABLED",
            "resource_types": ["AWS::IAM::User"],
            "evaluation_mode": "DETECTIVE",
            "max_freq": "Six_Hours"
        },
        {
            "rule_name": "lambda-public-access-prohibited",
            "rule_identifier": "LAMBDA_FUNCTION_PUBLIC_ACCESS_PROHIBITED",
            "resource_types": ["AWS::Lambda::Function"],
            "evaluation_mode": "PROACTIVE",
            "max_freq": None
        }
    ]

    success_count = 0
    for rule in rules_to_deploy:
        if deployer.deploy_managed_rule(
            rule_name=rule["rule_name"],
            rule_identifier=rule["rule_identifier"],
            resource_types=rule["resource_types"],
            evaluation_mode=rule["evaluation_mode"],
            maximum_execution_frequency=rule["max_freq"]
        ):
            success_count += 1

    logger.info(f"Deployed {success_count}/{len(rules_to_deploy)} Config 5 rules successfully")

import os
import json
import logging
from typing import Dict, List, Optional
from azure.identity import DefaultAzureCredential
from azure.mgmt.policy import PolicyClient
from azure.mgmt.policy.models import (
    PolicyDefinition, PolicyAssignment, PolicySku,
    EnforcementMode, NonComplianceMessage
)
from azure.core.exceptions import HttpResponseError, ClientAuthenticationError

# Configure logging for Azure policy operations
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("azure_policy_deployment.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

class AzurePolicy8Deployer:
    """Deploys Azure Policy 8 custom policies with cross-subscription support."""

    def __init__(self, subscription_id: str):
        try:
            self.credential = DefaultAzureCredential()
            self.policy_client = PolicyClient(self.credential, subscription_id)
            self.subscription_id = subscription_id
            logger.info(f"Initialized Policy client for subscription {subscription_id}")
        except ClientAuthenticationError:
            logger.error("Azure authentication failed. Ensure you're logged in via az login or have managed identity configured.")
            raise
        except HttpResponseError as e:
            logger.error(f"Failed to initialize Azure Policy client: {e.message}")
            raise

    def create_custom_policy_definition(self, policy_name: str, display_name: str, description: str,
                                       policy_rule: Dict, parameters: Optional[Dict] = None) -> Optional[str]:
        """
        Creates an Azure Policy 8 custom policy definition.

        Args:
            policy_name: Unique name for the policy (e.g., "require-s3-private-buckets")
            display_name: Human-readable name shown in Azure Portal
            description: Detailed policy description
            policy_rule: Policy rule JSON (if/then logic)
            parameters: Optional policy parameters

        Returns:
            str: Policy definition ID if successful, None otherwise
        """
        try:
            # Check if policy already exists
            existing_policies = self.policy_client.policy_definitions.list()
            for policy in existing_policies:
                if policy.name == policy_name:
                    logger.warning(f"Policy {policy_name} already exists. Returning existing ID.")
                    return policy.id

            policy_def = PolicyDefinition(
                display_name=display_name,
                description=description,
                policy_rule=policy_rule,
                parameters=parameters or {},
                policy_type="Custom",
                sku=PolicySku(name="A", tier="Standard")  # Policy 8 uses Sku A for custom policies
            )

            result = self.policy_client.policy_definitions.create_or_update(
                policy_definition_name=policy_name,
                parameters=policy_def
            )
            logger.info(f"Created custom policy definition {policy_name} (ID: {result.id})")
            return result.id

        except HttpResponseError as e:
            logger.error(f"Failed to create policy {policy_name}: {e.message}")
            return None
        except Exception as e:
            logger.error(f"Unexpected error creating policy {policy_name}: {str(e)}")
            return None

    def assign_policy_to_subscription(self, policy_definition_id: str, assignment_name: str,
                                     display_name: str, scope: str, enforcement_mode: EnforcementMode = EnforcementMode.DEFAULT,
                                     non_compliance_messages: Optional[List[NonComplianceMessage]] = None) -> bool:
        """
        Assigns a policy to a subscription scope (Policy 8 supports cross-subscription groups).

        Args:
            policy_definition_id: ID of the policy definition to assign
            assignment_name: Unique name for the assignment
            display_name: Human-readable assignment name
            scope: Scope to assign (e.g., /subscriptions/{sub-id})
            enforcement_mode: DEFAULT (audit) or DO_NOT_ENFORCE (disabled)
            non_compliance_messages: Optional messages shown when resource is non-compliant

        Returns:
            bool: True if assignment succeeded, False otherwise
        """
        try:
            # Check for existing assignment
            existing_assignments = self.policy_client.policy_assignments.list_for_scope(scope)
            for assignment in existing_assignments:
                if assignment.name == assignment_name:
                    logger.warning(f"Assignment {assignment_name} already exists. Updating.")
                    return self.update_policy_assignment(assignment_name, scope, policy_definition_id, enforcement_mode)

            assignment_params = PolicyAssignment(
                display_name=display_name,
                policy_definition_id=policy_definition_id,
                scope=scope,
                enforcement_mode=enforcement_mode,
                non_compliance_messages=non_compliance_messages or [],
                sku=PolicySku(name="A", tier="Standard")
            )

            self.policy_client.policy_assignments.create(
                scope=scope,
                policy_assignment_name=assignment_name,
                parameters=assignment_params
            )
            logger.info(f"Assigned policy {policy_definition_id} to scope {scope}")
            return True

        except HttpResponseError as e:
            if e.status_code == 409:
                logger.warning(f"Assignment {assignment_name} conflict. Retrying after 5 seconds.")
                import time
                time.sleep(5)
                return self.assign_policy_to_subscription(policy_definition_id, assignment_name, display_name, scope, enforcement_mode)
            logger.error(f"Failed to assign policy: {e.message}")
            return False

    def update_policy_assignment(self, assignment_name: str, scope: str, policy_definition_id: str,
                                enforcement_mode: EnforcementMode) -> bool:
        """Updates an existing policy assignment."""
        try:
            assignment = self.policy_client.policy_assignments.get(scope, assignment_name)
            assignment.policy_definition_id = policy_definition_id
            assignment.enforcement_mode = enforcement_mode
            self.policy_client.policy_assignments.create(
                scope=scope,
                policy_assignment_name=assignment_name,
                parameters=assignment
            )
            logger.info(f"Updated policy assignment {assignment_name}")
            return True
        except HttpResponseError as e:
            logger.error(f"Failed to update assignment {assignment_name}: {e.message}")
            return False

if __name__ == "__main__":
    # Configure Azure subscription (set via AZURE_SUBSCRIPTION_ID env var)
    subscription_id = os.getenv("AZURE_SUBSCRIPTION_ID")
    if not subscription_id:
        logger.error("AZURE_SUBSCRIPTION_ID environment variable not set.")
        exit(1)

    deployer = AzurePolicy8Deployer(subscription_id)

    # Define a Policy 8 custom rule to require NSGs on all VMs
    nsg_policy_rule = {
        "if": {
            "allOf": [
                {"field": "type", "equals": "Microsoft.Compute/virtualMachines"},
                {"field": "Microsoft.Compute/virtualMachines/networkInterfaceConfigurations/networkSecurityGroup.id", "exists": False}
            ]
        },
        "then": {
            "effect": "audit"
        }
    }

    # Create policy definition
    policy_id = deployer.create_custom_policy_definition(
        policy_name="require-nsg-on-vms",
        display_name="Require NSG on all Virtual Machines",
        description="Azure Policy 8 rule to audit VMs without Network Security Groups attached",
        policy_rule=nsg_policy_rule
    )

    if policy_id:
        # Assign to current subscription
        assignment_success = deployer.assign_policy_to_subscription(
            policy_definition_id=policy_id,
            assignment_name="require-nsg-assignment",
            display_name="Require NSG on VMs - Subscription Wide",
            scope=f"/subscriptions/{subscription_id}",
            enforcement_mode=EnforcementMode.DEFAULT
        )
        if assignment_success:
            logger.info("Successfully assigned NSG policy to subscription")
        else:
            logger.error("Failed to assign NSG policy")
    else:
        logger.error("Failed to create NSG policy definition")

import boto3
import os
import json
import logging
from datetime import datetime
from typing import Dict, List
from azure.identity import DefaultAzureCredential
from azure.mgmt.policy import PolicyClient
from botocore.exceptions import ClientError

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("cross_cloud_compliance.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

class CrossCloudComplianceReporter:
    """Generates unified compliance reports from AWS Config 5 and Azure Policy 8."""

    def __init__(self, aws_region: str = "us-east-1", azure_subscription_id: str = None):
        # Initialize AWS Config client
        try:
            self.aws_config = boto3.client("config", region_name=aws_region)
            logger.info(f"Initialized AWS Config client for {aws_region}")
        except ClientError as e:
            logger.error(f"AWS Config init failed: {e.response['Error']['Message']}")
            self.aws_config = None

        # Initialize Azure Policy client
        self.azure_policy = None
        if azure_subscription_id:
            try:
                credential = DefaultAzureCredential()
                self.azure_policy = PolicyClient(credential, azure_subscription_id)
                logger.info(f"Initialized Azure Policy client for sub {azure_subscription_id}")
            except Exception as e:
                logger.error(f"Azure Policy init failed: {str(e)}")

    def get_aws_compliance_status(self) -> List[Dict]:
        """Fetches compliance status for all AWS Config 5 rules."""
        if not self.aws_config:
            return []

        compliance_results = []
        try:
            # List all Config rules
            rules_paginator = self.aws_config.get_paginator("describe_config_rules")
            for rules_page in rules_paginator.paginate():
                for rule in rules_page.get("ConfigRules", []):
                    rule_name = rule["ConfigRuleName"]
                    rule_id = rule["ConfigRuleId"]

                    # Get compliance details for the rule
                    compliance = self.aws_config.get_compliance_details_by_config_rule(
                        ConfigRuleName=rule_name,
                        ComplianceTypes=["COMPLIANT", "NON_COMPLIANT"]
                    )

                    non_compliant_count = 0
                    for result in compliance.get("EvaluationResults", []):
                        if result["ComplianceType"] == "NON_COMPLIANT":
                            non_compliant_count += 1

                    compliance_results.append({
                        "cloud": "AWS",
                        "rule_name": rule_name,
                        "rule_id": rule_id,
                        "compliance_status": "NON_COMPLIANT" if non_compliant_count > 0 else "COMPLIANT",
                        "non_compliant_resources": non_compliant_count,
                        "rule_version": "Config 5" if "5" in rule.get("Source", {}).get("SourceIdentifier", "") else "Older"
                    })

            logger.info(f"Fetched {len(compliance_results)} AWS Config rule compliance results")
            return compliance_results

        except ClientError as e:
            logger.error(f"Failed to fetch AWS compliance: {e.response['Error']['Message']}")
            return []

    def get_azure_compliance_status(self) -> List[Dict]:
        """Fetches compliance status for all Azure Policy 8 assignments."""
        if not self.azure_policy:
            return []

        compliance_results = []
        try:
            # List all policy assignments
            assignments = self.azure_policy.policy_assignments.list()
            for assignment in assignments:
                assignment_name = assignment.name
                policy_def_id = assignment.policy_definition_id

                # Get compliance state for the assignment
                compliance_states = self.azure_policy.policy_states.list_query_results_for_policy_assignment(
                    policy_assignment_id=assignment.id
                )

                non_compliant = 0
                compliant = 0
                for state in compliance_states:
                    if state.compliance_state == "NonCompliant":
                        non_compliant += 1
                    elif state.compliance_state == "Compliant":
                        compliant += 1

                compliance_results.append({
                    "cloud": "Azure",
                    "assignment_name": assignment_name,
                    "policy_definition_id": policy_def_id,
                    "compliance_status": "NON_COMPLIANT" if non_compliant > 0 else "COMPLIANT",
                    "non_compliant_resources": non_compliant,
                    "compliant_resources": compliant,
                    "policy_version": "Policy 8" if "2024" in policy_def_id else "Older"
                })

            logger.info(f"Fetched {len(compliance_results)} Azure Policy compliance results")
            return compliance_results

        except Exception as e:
            logger.error(f"Failed to fetch Azure compliance: {str(e)}")
            return []

    def generate_report(self, output_path: str = "compliance_report.json") -> Dict:
        """Generates a unified compliance report and saves to JSON."""
        aws_results = self.get_aws_compliance_status()
        azure_results = self.get_azure_compliance_status()

        total_resources = sum(r.get("non_compliant_resources", 0) for r in aws_results + azure_results)
        compliant_resources = sum(r.get("compliant_resources", 0) for r in azure_results) + len([r for r in aws_results if r["compliance_status"] == "COMPLIANT"])

        report = {
            "generated_at": datetime.utcnow().isoformat() + "Z",
            "aws_config_version": "5",
            "azure_policy_version": "8",
            "total_rules": len(aws_results) + len(azure_results),
            "aws_results": aws_results,
            "azure_results": azure_results,
            "summary": {
                "total_non_compliant_resources": total_resources,
                "total_compliant_resources": compliant_resources,
                "compliance_percentage": (compliant_resources / (compliant_resources + total_resources)) * 100 if (compliant_resources + total_resources) > 0 else 0
            }
        }

        with open(output_path, "w") as f:
            json.dump(report, f, indent=2)

        logger.info(f"Saved compliance report to {output_path}")
        return report

if __name__ == "__main__":
    # Configure from environment variables
    aws_region = os.getenv("AWS_REGION", "us-east-1")
    azure_sub = os.getenv("AZURE_SUBSCRIPTION_ID")

    reporter = CrossCloudComplianceReporter(
        aws_region=aws_region,
        azure_subscription_id=azure_sub
    )

    report = reporter.generate_report()
    print(f"Compliance Summary: {json.dumps(report['summary'], indent=2)}")

# Idempotent check for AWS Config rule existence
import boto3
config = boto3.client("config")
def rule_exists(rule_name):
    try:
        config.describe_config_rules(ConfigRuleNames=[rule_name])
        return True
    except config.exceptions.NoSuchConfigRuleException:
        return False

# Terraform snippet for Azure Policy 8 initiative group
resource "azurerm_policy_set_definition" "soc2_initiative" {
  name         = "soc2-compliance-initiative"
  policy_type  = "Custom"
  display_name = "SOC2 Compliance Initiative (Policy 8)"
  management_group_id = data.azurerm_management_group.example.id
  policy_definition_reference {
    policy_definition_id = azurerm_policy_definition.nsg_requirement.id
  }
}

# GitHub Actions step for AWS Config 5 compliance check
- name: Run Config 5 Compliance Check
  run: |
    python -c "import boto3; config = boto3.client('config'); config.start_config_rules_evaluation()"
    sleep 60  # Wait for evaluation to complete
    python compliance_check.py --cloud aws --output json