Multi-Cloud Security Nightmare: How to Manage Consistent Security Across AWS, GCP, and Azure

Managing security across multiple cloud providers is one of the most challenging aspects of modern cloud architecture. With 92% of organizations using a multi-cloud strategy, the complexity of maintaining consistent security policies across AWS, GCP, and Azure has become a critical business risk. This comprehensive guide shows you how to overcome multi-cloud security challenges and implement unified security management.

The Multi-Cloud Security Challenge

Why Organizations Go Multi-Cloud

Best-of-breed services: Each cloud provider excels in different areas
Vendor lock-in avoidance: Reduce dependency on single provider
Cost optimization: Leverage competitive pricing across providers
Regulatory compliance: Meet data residency requirements
Disaster recovery: Geographic redundancy across providers
Acquisition integration: Merging different cloud environments

The Security Complexity That Follows

Each cloud provider has:

Different security models: IAM, networking, encryption approaches
Unique terminology: Similar concepts with different names
Varying compliance frameworks: Different audit trails and reporting
Distinct management interfaces: Multiple consoles and APIs
Provider-specific services: Native security tools that don’t cross platforms

Common Multi-Cloud Security Challenges

1. Inconsistent Identity and Access Management

The Problem: Each cloud provider has different IAM models, making it difficult to maintain consistent access controls.

AWS IAM vs GCP IAM vs Azure AD:

# AWS IAM Policy
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::my-bucket/*"
    }
  ]
}

# GCP IAM Policy
{
  "bindings": [
    {
      "role": "roles/storage.objectViewer",
      "members": [
        "user:developer@company.com"
      ]
    }
  ]
}

# Azure RBAC
{
  "properties": {
    "roleDefinitionId": "/subscriptions/{subscription-id}/providers/Microsoft.Authorization/roleDefinitions/2a2b9908-6ea1-4ae2-8e65-a410df84e7d1",
    "principalId": "{principal-id}"
  }
}

The Solution: Unified Identity Management

# unified_iam.py
import boto3
from google.cloud import iam
from azure.identity import DefaultAzureCredential
from azure.mgmt.authorization import AuthorizationManagementClient

class UnifiedIAMManager:
    def __init__(self):
        # AWS
        self.aws_iam = boto3.client('iam')
        
        # GCP
        self.gcp_iam = iam.IAMClient()
        
        # Azure
        self.azure_credential = DefaultAzureCredential()
        self.azure_auth_client = AuthorizationManagementClient(
            self.azure_credential, 
            subscription_id='your-subscription-id'
        )
        
        # Unified role mappings
        self.role_mappings = {
            'read_only': {
                'aws': ['ReadOnlyAccess'],
                'gcp': ['roles/viewer'],
                'azure': ['Reader']
            },
            'storage_admin': {
                'aws': ['AmazonS3FullAccess'],
                'gcp': ['roles/storage.admin'],
                'azure': ['Storage Account Contributor']
            },
            'developer': {
                'aws': ['PowerUserAccess'],
                'gcp': ['roles/editor'],
                'azure': ['Contributor']
            }
        }
    
    def create_user_across_clouds(self, user_email, role_name):
        """Create user with consistent role across all clouds"""
        results = {}
        
        # AWS
        try:
            aws_username = user_email.split('@')[0]
            self.aws_iam.create_user(UserName=aws_username)
            
            # Attach policies
            for policy in self.role_mappings[role_name]['aws']:
                self.aws_iam.attach_user_policy(
                    UserName=aws_username,
                    PolicyArn=f'arn:aws:iam::aws:policy/{policy}'
                )
            
            results['aws'] = 'success'
        except Exception as e:
            results['aws'] = f'error: {e}'
        
        # GCP
        try:
            # Create service account (GCP equivalent)
            project_id = 'your-project-id'
            service_account_id = aws_username
            
            service_account = self.gcp_iam.create_service_account(
                parent=f'projects/{project_id}',
                service_account_id=service_account_id
            )
            
            # Grant roles
            for role in self.role_mappings[role_name]['gcp']:
                self.grant_gcp_role(project_id, service_account.email, role)
            
            results['gcp'] = 'success'
        except Exception as e:
            results['gcp'] = f'error: {e}'
        
        # Azure
        try:
            # Azure AD user creation would be done here
            # For simplicity, showing role assignment
            for role in self.role_mappings[role_name]['azure']:
                self.assign_azure_role(user_email, role)
            
            results['azure'] = 'success'
        except Exception as e:
            results['azure'] = f'error: {e}'
        
        return results
    
    def grant_gcp_role(self, project_id, member_email, role):
        """Grant GCP IAM role"""
        # Implementation depends on your GCP setup
        pass
    
    def assign_azure_role(self, user_email, role_name):
        """Assign Azure role"""
        # Implementation depends on your Azure setup
        pass
    
    def audit_access_across_clouds(self):
        """Audit access permissions across all clouds"""
        audit_results = {
            'aws': self.audit_aws_access(),
            'gcp': self.audit_gcp_access(),
            'azure': self.audit_azure_access()
        }
        
        return audit_results
    
    def audit_aws_access(self):
        """Audit AWS access"""
        users = self.aws_iam.list_users()
        
        audit_data = []
        for user in users['Users']:
            username = user['UserName']
            
            # Get user policies
            attached_policies = self.aws_iam.list_attached_user_policies(UserName=username)
            
            audit_data.append({
                'username': username,
                'policies': [p['PolicyName'] for p in attached_policies['AttachedPolicies']],
                'last_used': user.get('PasswordLastUsed', 'Never')
            })
        
        return audit_data
    
    def audit_gcp_access(self):
        """Audit GCP access"""
        # Implementation for GCP access audit
        return []
    
    def audit_azure_access(self):
        """Audit Azure access"""
        # Implementation for Azure access audit
        return []
    
    def sync_user_access(self, user_email, target_role):
        """Sync user access across all clouds"""
        # Remove existing access
        self.revoke_all_access(user_email)
        
        # Apply new consistent access
        return self.create_user_across_clouds(user_email, target_role)
    
    def revoke_all_access(self, user_email):
        """Revoke access across all clouds"""
        # Implementation to revoke access
        pass

# Usage
iam_manager = UnifiedIAMManager()

# Create user with consistent role across clouds
result = iam_manager.create_user_across_clouds('developer@company.com', 'developer')
print(f"User creation results: {result}")

# Audit access across all clouds
audit_results = iam_manager.audit_access_across_clouds()
print(f"Access audit: {audit_results}")

2. Network Security Inconsistencies

The Problem: Different networking models and security group implementations across providers.

Network Security Comparison:

Feature	AWS	GCP	Azure
Virtual Networks	VPC	VPC	VNet
Subnets	Subnets	Subnets	Subnets
Firewall Rules	Security Groups	Firewall Rules	Network Security Groups
Default Behavior	Deny all inbound	Allow internal	Allow same subnet
Stateful	Yes	Yes	Yes

The Solution: Unified Network Security

# unified_network_security.py
import boto3
from google.cloud import compute_v1
from azure.mgmt.network import NetworkManagementClient

class UnifiedNetworkSecurity:
    def __init__(self):
        # AWS
        self.aws_ec2 = boto3.client('ec2')
        
        # GCP
        self.gcp_compute = compute_v1.FirewallsClient()
        
        # Azure
        self.azure_network = NetworkManagementClient(
            credential=DefaultAzureCredential(),
            subscription_id='your-subscription-id'
        )
        
        # Standard security rules
        self.standard_rules = {
            'web_servers': [
                {'protocol': 'tcp', 'port': 80, 'source': '0.0.0.0/0'},
                {'protocol': 'tcp', 'port': 443, 'source': '0.0.0.0/0'}
            ],
            'database_servers': [
                {'protocol': 'tcp', 'port': 3306, 'source': '10.0.0.0/8'},
                {'protocol': 'tcp', 'port': 5432, 'source': '10.0.0.0/8'}
            ],
            'ssh_access': [
                {'protocol': 'tcp', 'port': 22, 'source': '203.0.113.0/24'}
            ]
        }
    
    def create_consistent_security_groups(self, group_name, rule_set):
        """Create consistent security groups across all clouds"""
        results = {}
        
        # AWS Security Group
        try:
            aws_sg = self.create_aws_security_group(group_name, rule_set)
            results['aws'] = aws_sg
        except Exception as e:
            results['aws'] = f'error: {e}'
        
        # GCP Firewall Rules
        try:
            gcp_rules = self.create_gcp_firewall_rules(group_name, rule_set)
            results['gcp'] = gcp_rules
        except Exception as e:
            results['gcp'] = f'error: {e}'
        
        # Azure Network Security Group
        try:
            azure_nsg = self.create_azure_nsg(group_name, rule_set)
            results['azure'] = azure_nsg
        except Exception as e:
            results['azure'] = f'error: {e}'
        
        return results
    
    def create_aws_security_group(self, group_name, rule_set):
        """Create AWS security group"""
        # Get default VPC
        vpcs = self.aws_ec2.describe_vpcs(Filters=[{'Name': 'isDefault', 'Values': ['true']}])
        vpc_id = vpcs['Vpcs'][0]['VpcId']
        
        # Create security group
        response = self.aws_ec2.create_security_group(
            GroupName=group_name,
            Description=f'Security group for {group_name}',
            VpcId=vpc_id
        )
        
        sg_id = response['GroupId']
        
        # Add ingress rules
        rules = self.standard_rules[rule_set]
        for rule in rules:
            self.aws_ec2.authorize_security_group_ingress(
                GroupId=sg_id,
                IpPermissions=[
                    {
                        'IpProtocol': rule['protocol'],
                        'FromPort': rule['port'],
                        'ToPort': rule['port'],
                        'IpRanges': [{'CidrIp': rule['source']}]
                    }
                ]
            )
        
        return sg_id
    
    def create_gcp_firewall_rules(self, group_name, rule_set):
        """Create GCP firewall rules"""
        project_id = 'your-project-id'
        rules = self.standard_rules[rule_set]
        
        created_rules = []
        
        for i, rule in enumerate(rules):
            rule_name = f"{group_name}-rule-{i}"
            
            firewall_rule = {
                'name': rule_name,
                'direction': 'INGRESS',
                'priority': 1000,
                'target_tags': [group_name],
                'source_ranges': [rule['source']],
                'allowed': [
                    {
                        'I_p_protocol': rule['protocol'],
                        'ports': [str(rule['port'])]
                    }
                ]
            }
            
            operation = self.gcp_compute.insert(
                project=project_id,
                firewall_resource=firewall_rule
            )
            
            created_rules.append(rule_name)
        
        return created_rules
    
    def create_azure_nsg(self, group_name, rule_set):
        """Create Azure Network Security Group"""
        resource_group = 'your-resource-group'
        location = 'East US'
        
        # Create NSG
        nsg_params = {
            'location': location,
            'security_rules': []
        }
        
        rules = self.standard_rules[rule_set]
        for i, rule in enumerate(rules):
            nsg_params['security_rules'].append({
                'name': f'rule-{i}',
                'priority': 1000 + i,
                'direction': 'Inbound',
                'access': 'Allow',
                'protocol': rule['protocol'].upper(),
                'source_address_prefix': rule['source'],
                'source_port_range': '*',
                'destination_address_prefix': '*',
                'destination_port_range': str(rule['port'])
            })
        
        nsg = self.azure_network.network_security_groups.begin_create_or_update(
            resource_group,
            group_name,
            nsg_params
        ).result()
        
        return nsg.id
    
    def audit_network_security(self):
        """Audit network security across all clouds"""
        audit_results = {
            'aws': self.audit_aws_security_groups(),
            'gcp': self.audit_gcp_firewall_rules(),
            'azure': self.audit_azure_nsgs()
        }
        
        return audit_results
    
    def audit_aws_security_groups(self):
        """Audit AWS security groups"""
        security_groups = self.aws_ec2.describe_security_groups()
        
        audit_data = []
        for sg in security_groups['SecurityGroups']:
            # Check for overly permissive rules
            risky_rules = []
            for rule in sg['IpPermissions']:
                for ip_range in rule.get('IpRanges', []):
                    if ip_range.get('CidrIp') == '0.0.0.0/0':
                        risky_rules.append({
                            'port': rule.get('FromPort'),
                            'protocol': rule.get('IpProtocol'),
                            'source': ip_range.get('CidrIp')
                        })
            
            audit_data.append({
                'id': sg['GroupId'],
                'name': sg['GroupName'],
                'risky_rules': risky_rules
            })
        
        return audit_data
    
    def audit_gcp_firewall_rules(self):
        """Audit GCP firewall rules"""
        # Implementation for GCP firewall audit
        return []
    
    def audit_azure_nsgs(self):
        """Audit Azure NSGs"""
        # Implementation for Azure NSG audit
        return []

# Usage
network_security = UnifiedNetworkSecurity()

# Create consistent security groups
result = network_security.create_consistent_security_groups('web-servers', 'web_servers')
print(f"Security group creation results: {result}")

# Audit network security
audit = network_security.audit_network_security()
print(f"Network security audit: {audit}")

3. Inconsistent Encryption and Data Protection

The Problem: Different encryption implementations and key management across providers.

Encryption Comparison:

Feature	AWS	GCP	Azure
Key Management	KMS	Cloud KMS	Key Vault
Storage Encryption	S3 SSE	Cloud Storage	Storage Service Encryption
Database Encryption	RDS Encryption	Cloud SQL	Transparent Data Encryption
Default Behavior	Varies by service	Varies by service	Varies by service

The Solution: Unified Encryption Management

# unified_encryption.py
import boto3
from google.cloud import kms
from azure.keyvault.keys import KeyClient
from azure.identity import DefaultAzureCredential

class UnifiedEncryptionManager:
    def __init__(self):
        # AWS KMS
        self.aws_kms = boto3.client('kms')
        
        # GCP KMS
        self.gcp_kms = kms.KeyManagementServiceClient()
        
        # Azure Key Vault
        self.azure_credential = DefaultAzureCredential()
        self.azure_key_client = KeyClient(
            vault_url="https://your-vault.vault.azure.net/",
            credential=self.azure_credential
        )
        
        # Encryption policies
        self.encryption_policies = {
            'storage': {
                'algorithm': 'AES256',
                'key_rotation': 365,  # days
                'required': True
            },
            'database': {
                'algorithm': 'AES256',
                'key_rotation': 365,
                'required': True
            },
            'transit': {
                'algorithm': 'TLS1.2',
                'required': True
            }
        }
    
    def create_encryption_keys(self, key_name, purpose):
        """Create encryption keys across all clouds"""
        results = {}
        
        # AWS KMS Key
        try:
            aws_key = self.aws_kms.create_key(
                Description=f'Key for {key_name} - {purpose}',
                KeyUsage='ENCRYPT_DECRYPT'
            )
            
            # Create alias
            self.aws_kms.create_alias(
                AliasName=f'alias/{key_name}',
                TargetKeyId=aws_key['KeyMetadata']['KeyId']
            )
            
            results['aws'] = aws_key['KeyMetadata']['KeyId']
        except Exception as e:
            results['aws'] = f'error: {e}'
        
        # GCP KMS Key
        try:
            project_id = 'your-project-id'
            location_id = 'us-central1'
            key_ring_id = 'unified-keyring'
            
            # Create key ring if it doesn't exist
            key_ring_name = self.gcp_kms.key_ring_path(project_id, location_id, key_ring_id)
            
            try:
                self.gcp_kms.create_key_ring(
                    parent=f'projects/{project_id}/locations/{location_id}',
                    key_ring_id=key_ring_id
                )
            except Exception:
                pass  # Key ring might already exist
            
            # Create key
            key = self.gcp_kms.create_crypto_key(
                parent=key_ring_name,
                crypto_key_id=key_name,
                crypto_key={
                    'purpose': kms.CryptoKey.CryptoKeyPurpose.ENCRYPT_DECRYPT,
                    'version_template': {
                        'algorithm': kms.CryptoKeyVersion.CryptoKeyVersionAlgorithm.GOOGLE_SYMMETRIC_ENCRYPTION
                    }
                }
            )
            
            results['gcp'] = key.name
        except Exception as e:
            results['gcp'] = f'error: {e}'
        
        # Azure Key Vault Key
        try:
            key = self.azure_key_client.create_rsa_key(
                name=key_name,
                size=2048
            )
            
            results['azure'] = key.id
        except Exception as e:
            results['azure'] = f'error: {e}'
        
        return results
    
    def enforce_encryption_policies(self):
        """Enforce encryption policies across all clouds"""
        results = {
            'aws': self.enforce_aws_encryption(),
            'gcp': self.enforce_gcp_encryption(),
            'azure': self.enforce_azure_encryption()
        }
        
        return results
    
    def enforce_aws_encryption(self):
        """Enforce AWS encryption policies"""
        issues = []
        
        # Check S3 buckets
        s3 = boto3.client('s3')
        buckets = s3.list_buckets()
        
        for bucket in buckets['Buckets']:
            bucket_name = bucket['Name']
            
            try:
                s3.get_bucket_encryption(Bucket=bucket_name)
            except s3.exceptions.ClientError:
                issues.append(f'S3 bucket {bucket_name} is not encrypted')
        
        # Check RDS instances
        rds = boto3.client('rds')
        instances = rds.describe_db_instances()
        
        for instance in instances['DBInstances']:
            if not instance.get('StorageEncrypted', False):
                issues.append(f'RDS instance {instance["DBInstanceIdentifier"]} is not encrypted')
        
        return issues
    
    def enforce_gcp_encryption(self):
        """Enforce GCP encryption policies"""
        issues = []
        
        # Check Cloud Storage buckets
        # Implementation depends on your GCP setup
        
        return issues
    
    def enforce_azure_encryption(self):
        """Enforce Azure encryption policies"""
        issues = []
        
        # Check Storage Accounts
        # Implementation depends on your Azure setup
        
        return issues
    
    def rotate_keys(self):
        """Rotate encryption keys across all clouds"""
        rotation_results = {}
        
        # AWS key rotation
        try:
            keys = self.aws_kms.list_keys()
            for key in keys['Keys']:
                key_id = key['KeyId']
                
                # Enable automatic rotation
                self.aws_kms.enable_key_rotation(KeyId=key_id)
                
            rotation_results['aws'] = 'enabled'
        except Exception as e:
            rotation_results['aws'] = f'error: {e}'
        
        # GCP key rotation
        try:
            # GCP requires manual key rotation
            rotation_results['gcp'] = 'manual rotation required'
        except Exception as e:
            rotation_results['gcp'] = f'error: {e}'
        
        # Azure key rotation
        try:
            # Azure Key Vault supports automatic rotation
            rotation_results['azure'] = 'automatic rotation configured'
        except Exception as e:
            rotation_results['azure'] = f'error: {e}'
        
        return rotation_results
    
    def audit_encryption_compliance(self):
        """Audit encryption compliance across all clouds"""
        compliance_report = {
            'aws': self.audit_aws_encryption(),
            'gcp': self.audit_gcp_encryption(),
            'azure': self.audit_azure_encryption()
        }
        
        return compliance_report
    
    def audit_aws_encryption(self):
        """Audit AWS encryption compliance"""
        report = {
            'compliant_resources': 0,
            'non_compliant_resources': 0,
            'issues': []
        }
        
        # Check S3 encryption
        s3 = boto3.client('s3')
        buckets = s3.list_buckets()
        
        for bucket in buckets['Buckets']:
            bucket_name = bucket['Name']
            
            try:
                s3.get_bucket_encryption(Bucket=bucket_name)
                report['compliant_resources'] += 1
            except s3.exceptions.ClientError:
                report['non_compliant_resources'] += 1
                report['issues'].append(f'S3 bucket {bucket_name} not encrypted')
        
        return report
    
    def audit_gcp_encryption(self):
        """Audit GCP encryption compliance"""
        # Implementation for GCP encryption audit
        return {'compliant_resources': 0, 'non_compliant_resources': 0, 'issues': []}
    
    def audit_azure_encryption(self):
        """Audit Azure encryption compliance"""
        # Implementation for Azure encryption audit
        return {'compliant_resources': 0, 'non_compliant_resources': 0, 'issues': []}

# Usage
encryption_manager = UnifiedEncryptionManager()

# Create encryption keys across all clouds
key_results = encryption_manager.create_encryption_keys('data-encryption-key', 'data protection')
print(f"Key creation results: {key_results}")

# Enforce encryption policies
policy_results = encryption_manager.enforce_encryption_policies()
print(f"Policy enforcement results: {policy_results}")

# Audit encryption compliance
compliance_report = encryption_manager.audit_encryption_compliance()
print(f"Compliance report: {compliance_report}")

4. Monitoring and Logging Fragmentation

The Problem: Each cloud provider has different logging and monitoring services, making it difficult to get a unified view of security events.

Logging Services Comparison:

Feature	AWS	GCP	Azure
Audit Logging	CloudTrail	Cloud Audit Logs	Activity Log
Application Logs	CloudWatch	Cloud Logging	Azure Monitor
Security Monitoring	GuardDuty	Security Command Center	Azure Security Center
SIEM Integration	Native + 3rd party	Native + 3rd party	Native + 3rd party

The Solution: Unified Logging and Monitoring

# unified_monitoring.py
import boto3
import json
from google.cloud import logging as gcp_logging
from azure.monitor.query import LogsQueryClient
from datetime import datetime, timedelta

class UnifiedMonitoringSystem:
    def __init__(self):
        # AWS
        self.aws_cloudtrail = boto3.client('cloudtrail')
        self.aws_cloudwatch = boto3.client('cloudwatch')
        
        # GCP
        self.gcp_logging = gcp_logging.Client()
        
        # Azure
        self.azure_logs = LogsQueryClient(DefaultAzureCredential())
        
        # Central log aggregation
        self.central_log_store = []
        
        # Standard security events to monitor
        self.security_events = {
            'failed_logins': {
                'aws': 'ConsoleLogin',
                'gcp': 'google.login',
                'azure': 'SigninLogs'
            },
            'resource_creation': {
                'aws': ['RunInstances', 'CreateBucket'],
                'gcp': ['compute.instances.insert', 'storage.buckets.create'],
                'azure': ['Microsoft.Compute/virtualMachines/write']
            },
            'permission_changes': {
                'aws': ['AttachUserPolicy', 'CreateRole'],
                'gcp': ['SetIamPolicy'],
                'azure': ['Microsoft.Authorization/roleAssignments/write']
            }
        }
    
    def collect_security_events(self, hours_back=24):
        """Collect security events from all clouds"""
        end_time = datetime.now()
        start_time = end_time - timedelta(hours=hours_back)
        
        all_events = []
        
        # AWS events
        aws_events = self.collect_aws_events(start_time, end_time)
        all_events.extend(aws_events)
        
        # GCP events
        gcp_events = self.collect_gcp_events(start_time, end_time)
        all_events.extend(gcp_events)
        
        # Azure events
        azure_events = self.collect_azure_events(start_time, end_time)
        all_events.extend(azure_events)
        
        # Normalize and correlate events
        normalized_events = self.normalize_events(all_events)
        
        return normalized_events
    
    def collect_aws_events(self, start_time, end_time):
        """Collect AWS CloudTrail events"""
        events = []
        
        try:
            response = self.aws_cloudtrail.lookup_events(
                StartTime=start_time,
                EndTime=end_time,
                MaxItems=1000
            )
            
            for event in response['Events']:
                normalized_event = {
                    'provider': 'aws',
                    'event_time': event['EventTime'],
                    'event_name': event['EventName'],
                    'user_identity': event.get('UserIdentity', {}),
                    'source_ip': event.get('SourceIPAddress'),
                    'user_agent': event.get('UserAgent'),
                    'aws_region': event.get('AwsRegion'),
                    'error_code': event.get('ErrorCode'),
                    'resources': event.get('Resources', []),
                    'raw_event': event
                }
                
                events.append(normalized_event)
        
        except Exception as e:
            print(f"Error collecting AWS events: {e}")
        
        return events
    
    def collect_gcp_events(self, start_time, end_time):
        """Collect GCP Cloud Audit Logs"""
        events = []
        
        try:
            filter_str = f'timestamp >= "{start_time.isoformat()}Z" AND timestamp <= "{end_time.isoformat()}Z"'
            
            entries = self.gcp_logging.list_entries(
                filter_=filter_str,
                page_size=1000
            )
            
            for entry in entries:
                if hasattr(entry, 'payload') and isinstance(entry.payload, dict):
                    audit_log = entry.payload
                    
                    normalized_event = {
                        'provider': 'gcp',
                        'event_time': entry.timestamp,
                        'event_name': audit_log.get('methodName', ''),
                        'user_identity': audit_log.get('authenticationInfo', {}),
                        'source_ip': audit_log.get('requestMetadata', {}).get('callerIp'),
                        'user_agent': audit_log.get('requestMetadata', {}).get('callerSuppliedUserAgent'),
                        'resource_name': audit_log.get('resourceName'),
                        'service_name': audit_log.get('serviceName'),
                        'raw_event': audit_log
                    }
                    
                    events.append(normalized_event)
        
        except Exception as e:
            print(f"Error collecting GCP events: {e}")
        
        return events
    
    def collect_azure_events(self, start_time, end_time):
        """Collect Azure Activity Logs"""
        events = []
        
        try:
            workspace_id = "your-workspace-id"
            
            query = f"""
            AzureActivity
            | where TimeGenerated between (datetime({start_time.isoformat()}) .. datetime({end_time.isoformat()}))
            | where CategoryValue == "Administrative"
            | project TimeGenerated, OperationNameValue, CallerIpAddress, Caller, ResourceGroup, ResourceId, ActivityStatusValue
            | limit 1000
            """
            
            response = self.azure_logs.query_workspace(
                workspace_id=workspace_id,
                query=query
            )
            
            for row in response.tables[0].rows:
                normalized_event = {
                    'provider': 'azure',
                    'event_time': row[0],
                    'event_name': row[1],
                    'source_ip': row[2],
                    'user_identity': {'principalName': row[3]},
                    'resource_group': row[4],
                    'resource_id': row[5],
                    'status': row[6],
                    'raw_event': dict(zip([col.name for col in response.tables[0].columns], row))
                }
                
                events.append(normalized_event)
        
        except Exception as e:
            print(f"Error collecting Azure events: {e}")
        
        return events
    
    def normalize_events(self, events):
        """Normalize events from different providers"""
        normalized = []
        
        for event in events:
            # Extract common fields
            normalized_event = {
                'timestamp': event['event_time'],
                'provider': event['provider'],
                'event_type': self.categorize_event(event),
                'user': self.extract_user(event),
                'source_ip': event.get('source_ip'),
                'severity': self.calculate_severity(event),
                'description': self.generate_description(event),
                'raw_event': event
            }
            
            normalized.append(normalized_event)
        
        return normalized
    
    def categorize_event(self, event):
        """Categorize event type"""
        event_name = event.get('event_name', '').lower()
        
        if any(keyword in event_name for keyword in ['login', 'signin', 'authenticate']):
            return 'authentication'
        elif any(keyword in event_name for keyword in ['create', 'delete', 'terminate', 'run']):
            return 'resource_change'
        elif any(keyword in event_name for keyword in ['policy', 'role', 'permission']):
            return 'permission_change'
        else:
            return 'other'
    
    def extract_user(self, event):
        """Extract user information from event"""
        user_identity = event.get('user_identity', {})
        
        if event['provider'] == 'aws':
            return user_identity.get('userName') or user_identity.get('type')
        elif event['provider'] == 'gcp':
            return user_identity.get('principalEmail')
        elif event['provider'] == 'azure':
            return user_identity.get('principalName')
        
        return 'unknown'
    
    def calculate_severity(self, event):
        """Calculate event severity"""
        event_name = event.get('event_name', '').lower()
        
        # High severity events
        high_severity_events = [
            'createuser', 'deleteuser', 'attachuserpolicy',
            'putbucketpolicy', 'deletebucket', 'terminateinstances',
            'setbucketacl', 'createaccesskey'
        ]
        
        if any(keyword in event_name for keyword in high_severity_events):
            return 'HIGH'
        
        # Medium severity events
        medium_severity_events = [
            'runinstances', 'createbucket', 'stopsystem',
            'modifydbinstance', 'createsecuritygroup'
        ]
        
        if any(keyword in event_name for keyword in medium_severity_events):
            return 'MEDIUM'
        
        # Check for error codes
        if event.get('error_code'):
            return 'MEDIUM'
        
        return 'LOW'
    
    def generate_description(self, event):
        """Generate human-readable description"""
        user = self.extract_user(event)
        event_name = event.get('event_name', '')
        provider = event['provider'].upper()
        
        return f"User {user} performed {event_name} on {provider}"
    
    def detect_anomalies(self, events):
        """Detect anomalies across multi-cloud events"""
        anomalies = []
        
        # Group events by user
        user_events = {}
        for event in events:
            user = event['user']
            if user not in user_events:
                user_events[user] = []
            user_events[user].append(event)
        
        # Detect anomalies
        for user, user_event_list in user_events.items():
            # Check for cross-cloud activity
            providers = set(event['provider'] for event in user_event_list)
            if len(providers) > 1:
                anomalies.append({
                    'type': 'cross_cloud_activity',
                    'user': user,
                    'providers': list(providers),
                    'severity': 'MEDIUM',
                    'description': f'User {user} active across multiple clouds: {", ".join(providers)}'
                })
            
            # Check for unusual time patterns
            timestamps = [event['timestamp'] for event in user_event_list]
            if self.is_unusual_time_pattern(timestamps):
                anomalies.append({
                    'type': 'unusual_time_pattern',
                    'user': user,
                    'severity': 'LOW',
                    'description': f'User {user} has unusual activity time pattern'
                })
        
        return anomalies
    
    def is_unusual_time_pattern(self, timestamps):
        """Check if timestamps show unusual pattern"""
        # Simple implementation - check for off-hours activity
        for timestamp in timestamps:
            if hasattr(timestamp, 'hour'):
                hour = timestamp.hour
            else:
                hour = datetime.fromisoformat(str(timestamp)).hour
            
            if hour < 6 or hour > 22:  # Outside business hours
                return True
        
        return False
    
    def generate_unified_dashboard(self, events):
        """Generate unified security dashboard data"""
        dashboard_data = {
            'total_events': len(events),
            'events_by_provider': {},
            'events_by_severity': {},
            'top_users': {},
            'recent_high_severity': [],
            'anomalies': self.detect_anomalies(events)
        }
        
        # Events by provider
        for event in events:
            provider = event['provider']
            dashboard_data['events_by_provider'][provider] = dashboard_data['events_by_provider'].get(provider, 0) + 1
        
        # Events by severity
        for event in events:
            severity = event['severity']
            dashboard_data['events_by_severity'][severity] = dashboard_data['events_by_severity'].get(severity, 0) + 1
        
        # Top users
        for event in events:
            user = event['user']
            dashboard_data['top_users'][user] = dashboard_data['top_users'].get(user, 0) + 1
        
        # Recent high severity events
        high_severity_events = [event for event in events if event['severity'] == 'HIGH']
        dashboard_data['recent_high_severity'] = sorted(high_severity_events, key=lambda x: x['timestamp'], reverse=True)[:10]
        
        return dashboard_data

# Usage
monitoring_system = UnifiedMonitoringSystem()

# Collect security events from all clouds
events = monitoring_system.collect_security_events(hours_back=24)
print(f"Collected {len(events)} events")

# Generate unified dashboard
dashboard = monitoring_system.generate_unified_dashboard(events)
print(f"Dashboard data: {json.dumps(dashboard, indent=2, default=str)}")

Multi-Cloud Security Framework

1. Governance and Policy Management

# multi_cloud_governance.py
import yaml
import json
from datetime import datetime

class MultiCloudGovernance:
    def __init__(self):
        self.policies = self.load_policies()
        self.compliance_frameworks = self.load_compliance_frameworks()
        
    def load_policies(self):
        """Load multi-cloud security policies"""
        return {
            'data_classification': {
                'public': {
                    'encryption_required': False,
                    'access_logging': True,
                    'retention_days': 365
                },
                'internal': {
                    'encryption_required': True,
                    'access_logging': True,
                    'retention_days': 2555  # 7 years
                },
                'confidential': {
                    'encryption_required': True,
                    'access_logging': True,
                    'retention_days': 2555,
                    'geographic_restrictions': True
                }
            },
            'access_control': {
                'mfa_required': True,
                'password_policy': {
                    'min_length': 12,
                    'complexity': True,
                    'rotation_days': 90
                },
                'session_timeout': 8  # hours
            },
            'network_security': {
                'default_deny': True,
                'encryption_in_transit': True,
                'vpc_flow_logs': True
            }
        }
    
    def load_compliance_frameworks(self):
        """Load compliance framework mappings"""
        return {
            'SOC2': {
                'controls': [
                    'CC6.1', 'CC6.2', 'CC6.3', 'CC6.6', 'CC6.7',
                    'CC7.1', 'CC7.2', 'CC7.3', 'CC7.4'
                ],
                'requirements': {
                    'access_control': True,
                    'encryption': True,
                    'monitoring': True,
                    'incident_response': True
                }
            },
            'ISO27001': {
                'controls': [
                    'A.9.1.1', 'A.9.1.2', 'A.9.2.1', 'A.9.2.2',
                    'A.10.1.1', 'A.10.1.2', 'A.12.1.1', 'A.12.1.2'
                ],
                'requirements': {
                    'risk_assessment': True,
                    'access_control': True,
                    'encryption': True,
                    'monitoring': True
                }
            }
        }
    
    def validate_policy_compliance(self, resource_config, data_classification):
        """Validate resource configuration against policies"""
        policy = self.policies['data_classification'][data_classification]
        violations = []
        
        # Check encryption requirement
        if policy['encryption_required'] and not resource_config.get('encrypted', False):
            violations.append(f"Encryption required for {data_classification} data")
        
        # Check access logging
        if policy['access_logging'] and not resource_config.get('logging_enabled', False):
            violations.append(f"Access logging required for {data_classification} data")
        
        # Check geographic restrictions
        if policy.get('geographic_restrictions') and resource_config.get('region') not in ['us-east-1', 'us-west-2']:
            violations.append(f"Geographic restrictions violated for {data_classification} data")
        
        return violations
    
    def generate_compliance_report(self, cloud_resources):
        """Generate compliance report across all clouds"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'total_resources': len(cloud_resources),
            'compliant_resources': 0,
            'violations': [],
            'compliance_by_framework': {}
        }
        
        for resource in cloud_resources:
            violations = self.validate_policy_compliance(
                resource['config'], 
                resource['data_classification']
            )
            
            if violations:
                report['violations'].extend([
                    {
                        'resource_id': resource['id'],
                        'provider': resource['provider'],
                        'violation': violation
                    }
                    for violation in violations
                ])
            else:
                report['compliant_resources'] += 1
        
        # Calculate compliance percentages
        for framework in self.compliance_frameworks:
            framework_violations = [v for v in report['violations'] if self.is_framework_violation(v, framework)]
            compliance_rate = (report['total_resources'] - len(framework_violations)) / report['total_resources']
            report['compliance_by_framework'][framework] = {
                'compliance_rate': compliance_rate,
                'violations': len(framework_violations)
            }
        
        return report
    
    def is_framework_violation(self, violation, framework):
        """Check if violation applies to specific framework"""
        # Simplified logic - in reality, this would map violations to specific controls
        return True
    
    def create_remediation_plan(self, violations):
        """Create remediation plan for violations"""
        plan = {
            'high_priority': [],
            'medium_priority': [],
            'low_priority': []
        }
        
        for violation in violations:
            if 'encryption' in violation['violation'].lower():
                plan['high_priority'].append({
                    'resource': violation['resource_id'],
                    'action': 'Enable encryption',
                    'estimated_effort': '2 hours'
                })
            elif 'logging' in violation['violation'].lower():
                plan['medium_priority'].append({
                    'resource': violation['resource_id'],
                    'action': 'Enable logging',
                    'estimated_effort': '1 hour'
                })
            else:
                plan['low_priority'].append({
                    'resource': violation['resource_id'],
                    'action': 'Review configuration',
                    'estimated_effort': '30 minutes'
                })
        
        return plan

# Usage
governance = MultiCloudGovernance()

# Example cloud resources
cloud_resources = [
    {
        'id': 'aws-bucket-1',
        'provider': 'aws',
        'data_classification': 'confidential',
        'config': {
            'encrypted': True,
            'logging_enabled': True,
            'region': 'us-east-1'
        }
    },
    {
        'id': 'gcp-bucket-1',
        'provider': 'gcp',
        'data_classification': 'internal',
        'config': {
            'encrypted': False,
            'logging_enabled': True,
            'region': 'us-central1'
        }
    }
]

# Generate compliance report
compliance_report = governance.generate_compliance_report(cloud_resources)
print(f"Compliance report: {json.dumps(compliance_report, indent=2)}")

# Create remediation plan
remediation_plan = governance.create_remediation_plan(compliance_report['violations'])
print(f"Remediation plan: {json.dumps(remediation_plan, indent=2)}")

2. Multi-Cloud Security Architecture

# multi-cloud-security-architecture.yaml
multi_cloud_security:
  identity_management:
    primary_idp: "Azure AD"
    federation:
      - aws_sso
      - gcp_identity
      - azure_ad
    mfa_required: true
    
  network_security:
    connectivity:
      - aws_transit_gateway
      - gcp_vpc_peering  
      - azure_vnet_peering
    encryption_in_transit: true
    network_segmentation: true
    
  data_protection:
    encryption_standards:
      - algorithm: "AES-256"
        key_rotation: "365 days"
    data_classification:
      - public
      - internal
      - confidential
      - restricted
    
  monitoring:
    centralized_logging: true
    siem_integration: true
    real_time_alerting: true
    
  compliance:
    frameworks:
      - SOC2
      - ISO27001
      - GDPR
    continuous_monitoring: true
    automated_remediation: true

Best Practices for Multi-Cloud Security

1. Standardize Security Policies

Define consistent security policies across all clouds
Use Infrastructure as Code to enforce policies
Implement automated compliance checking
Regular policy reviews and updates

2. Centralize Identity Management

Use a single identity provider (IdP) for all clouds
Implement federated authentication
Enforce multi-factor authentication
Regular access reviews

3. Unified Monitoring and Logging

Centralize security logs from all clouds
Implement correlation across cloud boundaries
Use consistent alerting thresholds
Create unified dashboards

4. Automate Security Operations

Infrastructure as Code for all security controls
Automated compliance checking
Incident response automation
Regular security assessments

5. Skills Development

Cross-cloud security training
Vendor-specific certifications
Security tooling expertise
Incident response procedures

Common Multi-Cloud Security Mistakes

1. Inconsistent Policies

Problem: Different security policies across clouds create gaps.

Solution: Standardize policies and use automation to enforce them.

2. Fragmented Monitoring

Problem: Security events spread across multiple systems.

Solution: Centralize logging and implement cross-cloud correlation.

3. Complex Identity Management

Problem: Multiple identity systems create administrative overhead.

Solution: Implement federated identity with single sign-on.

4. Skill Gaps

Problem: Teams lack expertise across multiple cloud platforms.

Solution: Invest in cross-cloud training and certifications.

5. Tool Proliferation

Problem: Too many security tools create complexity.

Solution: Standardize on multi-cloud security platforms.

Conclusion

Multi-cloud security is complex but manageable with the right approach. By standardizing policies, centralizing identity management, unifying monitoring, and automating operations, organizations can maintain consistent security across all cloud providers.

Key Takeaways:

Standardize security policies across all clouds
Implement unified identity and access management
Centralize monitoring and logging
Automate compliance and security operations
Invest in multi-cloud security skills

Action Items:

Audit current security policies across all clouds
Implement unified identity management
Centralize security logging and monitoring
Automate compliance checking
Develop multi-cloud security expertise

Remember: Multi-cloud security requires discipline, standardization, and automation. The complexity is manageable when you have the right framework and tools in place.

Multi-Cloud Security Nightmare - How to Manage Consistent Security Across AWS, GCP, and Azure