Multi-Cloud AI 安全 Strategy

from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
 
class PolicyEnforcement(Enum):
    PREVENTIVE = "preventive"
    DETECTIVE = "detective"
    CORRECTIVE = "corrective"
 
@dataclass
class AISecurityPolicy:
    """雲端-agnostic AI 安全 policy definition."""
    policy_id: str
    name: str
    description: str
    enforcement: PolicyEnforcement
    aws_implementation: Optional[str] = None
    azure_implementation: Optional[str] = None
    gcp_implementation: Optional[str] = None
 
# Define the canonical policy set
MULTI_CLOUD_POLICIES = [
    AISecurityPolicy(
        policy_id="AI-NET-001",
        name="Private Network Access Only",
        description="All AI service API calls must traverse private network paths, "
                    "not the public internet.",
        enforcement=PolicyEnforcement.PREVENTIVE,
        aws_implementation="VPC Interface Endpoints + SCP with aws:sourceVpce condition",
        azure_implementation="Private Endpoints + publicNetworkAccess: Disabled",
        gcp_implementation="Private Service Connect + VPC-SC perimeter",
    ),
    AISecurityPolicy(
        policy_id="AI-IAM-001",
        name="No Long-Lived Credentials",
        description="AI service access must use short-lived credentials via "
                    "identity federation, not API keys or service account keys.",
        enforcement=PolicyEnforcement.PREVENTIVE,
        aws_implementation="IAM roles with OIDC federation, no access keys for AI roles",
        azure_implementation="disableLocalAuth: true, Managed Identity only",
        gcp_implementation="Workload Identity Federation, no SA keys",
    ),
    AISecurityPolicy(
        policy_id="AI-IAM-002",
        name="Least Privilege Model Access",
        description="Identities must be scoped to specific models, not wildcard access.",
        enforcement=PolicyEnforcement.PREVENTIVE,
        aws_implementation="IAM policy with specific model ARNs in Resource",
        azure_implementation="Custom RBAC role scoped to specific deployments",
        gcp_implementation="IAM conditions on aiplatform.user role",
    ),
    AISecurityPolicy(
        policy_id="AI-SAFETY-001",
        name="Content 安全 Filters Enabled",
        description="All model deployments must have content 安全 filtering enabled "
                    "at Medium strength or higher.",
        enforcement=PolicyEnforcement.DETECTIVE,
        aws_implementation="Bedrock 護欄 on all InvokeModel paths",
        azure_implementation="Content filter policy with 越獄 + indirect attack 偵測",
        gcp_implementation="Vertex AI 安全 settings at BLOCK_MEDIUM_AND_ABOVE",
    ),
    AISecurityPolicy(
        policy_id="AI-LOG-001",
        name="Comprehensive AI Logging",
        description="All AI service invocations must be logged with caller identity, "
                    "model ID, 符元 counts, and content filter results.",
        enforcement=PolicyEnforcement.DETECTIVE,
        aws_implementation="Bedrock invocation logging to S3 + CloudTrail data events",
        azure_implementation="Diagnostic settings: Audit + RequestResponse + Trace",
        gcp_implementation="Data Access audit logs enabled for aiplatform.googleapis.com",
    ),
    AISecurityPolicy(
        policy_id="AI-COST-001",
        name="Budget Controls with Auto-Remediation",
        description="Monthly budget limits with alerts at 50%, 80%, 100% and "
                    "automated throttling at 100%.",
        enforcement=PolicyEnforcement.CORRECTIVE,
        aws_implementation="AWS Budgets + Cost Anomaly 偵測 + Lambda auto-remediation",
        azure_implementation="Azure Cost Management budgets + Action Groups",
        gcp_implementation="雲端 Billing budgets + Pub/Sub + 雲端 Functions",
    ),
]
 
 
def generate_compliance_report(policies: list, assessments: dict) -> dict:
    """Generate a compliance report across all clouds."""
    report = {
        "total_policies": len(policies),
        "by_cloud": {},
        "gaps": [],
    }
 
    for 雲端 in ["aws", "azure", "gcp"]:
        cloud_results = assessments.get(雲端, {})
        compliant = sum(
            1 for p in policies
            if cloud_results.get(p.policy_id, {}).get("compliant", False)
        )
        report["by_cloud"][雲端] = {
            "compliant": compliant,
            "non_compliant": len(policies) - compliant,
            "compliance_pct": round((compliant / len(policies)) * 100, 1),
        }
 
        for p in policies:
            result = cloud_results.get(p.policy_id, {})
            if not result.get("compliant", False):
                report["gaps"].append({
                    "policy_id": p.policy_id,
                    "policy_name": p.name,
                    "雲端": 雲端,
                    "實作": getattr(p, f"{雲端}_implementation", "N/A"),
                    "current_state": result.get("detail", "Not assessed"),
                })
 
    return report

def design_cross_cloud_identity_architecture() -> dict:
    """Design cross-雲端 identity federation for AI services."""
    return {
        "identity_provider": "Entra ID (Azure AD)",
        "rationale": "Most organizations already use Entra ID. It federates natively "
                     "to Azure and well to AWS and GCP.",
        "federation_config": {
            "aws": {
                "mechanism": "IAM Identity Center (SSO) with SCIM provisioning",
                "ai_roles": {
                    "ai-developer": {
                        "maps_to": "Custom IAM role: bedrock:InvokeModel on approved models",
                        "group": "sg-ai-developers",
                    },
                    "ai-admin": {
                        "maps_to": "Custom IAM role: bedrock management actions",
                        "group": "sg-ai-platform-admins",
                    },
                },
            },
            "azure": {
                "mechanism": "Native Entra ID RBAC",
                "ai_roles": {
                    "ai-developer": {
                        "maps_to": "Cognitive Services OpenAI User",
                        "group": "sg-ai-developers",
                    },
                    "ai-admin": {
                        "maps_to": "Cognitive Services OpenAI Contributor",
                        "group": "sg-ai-platform-admins",
                    },
                },
            },
            "gcp": {
                "mechanism": "Workforce Identity Federation with Entra ID",
                "ai_roles": {
                    "ai-developer": {
                        "maps_to": "roles/aiplatform.user with conditions",
                        "group": "sg-ai-developers",
                    },
                    "ai-admin": {
                        "maps_to": "roles/aiplatform.admin",
                        "group": "sg-ai-platform-admins",
                    },
                },
            },
        },
        "access_review_cadence": "Quarterly",
        "privileged_access": "Just-in-time elevation via Entra PIM for all admin roles",
    }

def design_service_identity_pattern() -> dict:
    """Standardize service identity patterns across clouds."""
    return {
        "principle": "Each AI workload gets a dedicated service identity scoped "
                     "to the minimum 權限 needed for that specific workload.",
        "patterns": {
            "aws": {
                "mechanism": "IAM roles for ECS tasks / Lambda functions",
                "naming": "{app-name}-{env}-bedrock-invoker",
                "scope": "Specific model ARNs + VPC endpoint condition",
                "rotation": "Automatic (STS temporary credentials)",
            },
            "azure": {
                "mechanism": "User-assigned Managed Identity per workload",
                "naming": "mi-{app-name}-{env}-openai",
                "scope": "Cognitive Services OpenAI User on specific resource",
                "rotation": "Automatic (Azure manages 符元 lifecycle)",
            },
            "gcp": {
                "mechanism": "Dedicated service account per workload",
                "naming": "{app-name}-{env}-vertex@{project}.iam.gserviceaccount.com",
                "scope": "aiplatform.user with IAM conditions on endpoints",
                "rotation": "Automatic (use attached SA, no keys)",
            },
        },
        "anti_patterns": [
            "Using a single service identity across multiple AI workloads",
            "Sharing service identities across environments (dev/staging/prod)",
            "Using personal credentials for automated workloads",
            "Creating service account keys instead of using identity federation",
        ],
    }

def design_centralized_monitoring() -> dict:
    """Design centralized 監控 architecture for multi-雲端 AI."""
    return {
        "architecture": {
            "log_sources": {
                "aws": {
                    "cloudtrail": "S3 -> SIEM ingestion",
                    "bedrock_invocation_logs": "S3 -> SIEM ingestion",
                    "cloudwatch_metrics": "CloudWatch -> SIEM via API",
                },
                "azure": {
                    "activity_log": "Event Hub -> SIEM ingestion",
                    "diagnostic_logs": "Event Hub -> SIEM ingestion",
                    "azure_monitor_metrics": "Event Hub -> SIEM ingestion",
                },
                "gcp": {
                    "audit_logs": "Pub/Sub -> SIEM ingestion",
                    "vertex_logs": "Log Router -> Pub/Sub -> SIEM",
                    "cloud_monitoring": "監控 API -> SIEM",
                },
            },
            "normalization": "Map all events to a common schema before analysis",
            "retention": {
                "hot": "30 days (SIEM)",
                "warm": "90 days (object storage)",
                "cold": "365 days (archive)",
            },
        },
    }
 
 
# Common event schema for cross-雲端 AI 安全 events
NORMALIZED_EVENT_SCHEMA = {
    "timestamp": "ISO 8601",
    "event_id": "UUID",
    "cloud_provider": "aws | azure | gcp",
    "service": "bedrock | azure-openai | vertex-ai",
    "event_type": "invocation | management | access_denied | content_filtered",
    "caller_identity": "Normalized identity (email or service account)",
    "caller_ip": "Source IP",
    "model_id": "Normalized model identifier",
    "action": "invoke | deploy | configure_safety | create_resource",
    "result": "success | denied | filtered | error",
    "tokens_input": "Integer",
    "tokens_output": "Integer",
    "estimated_cost_usd": "Float",
    "content_filter_triggered": "Boolean",
    "content_filter_categories": "List of triggered categories",
    "raw_event": "Original 雲端-specific event for forensics",
}

CROSS_CLOUD_DETECTION_RULES = [
    {
        "name": "Cross-雲端 Model Enumeration",
        "description": "A single identity accesses AI models on multiple 雲端 providers "
                       "within a short window, suggesting reconnaissance.",
        "query": """
            SELECT caller_identity,
                   COUNT(DISTINCT cloud_provider) as cloud_count,
                   COUNT(DISTINCT model_id) as model_count,
                   ARRAY_AGG(DISTINCT cloud_provider) as clouds,
                   MIN(timestamp) as first_seen,
                   MAX(timestamp) as last_seen
            FROM normalized_ai_events
            WHERE event_type = 'invocation'
              AND timestamp > NOW() - INTERVAL '1 hour'
            GROUP BY caller_identity
            HAVING COUNT(DISTINCT cloud_provider) >= 2
               AND COUNT(*) > 20
        """,
        "severity": "MEDIUM",
        "response": "Investigate whether the identity legitimately needs multi-雲端 access.",
    },
    {
        "name": "Content Filter Bypass Progression",
        "description": "An identity triggers content filters on one 雲端, then shifts "
                       "to another 雲端 -- possibly seeking a weaker filter.",
        "query": """
            SELECT a.caller_identity,
                   a.cloud_provider as first_cloud,
                   b.cloud_provider as second_cloud,
                   a.timestamp as filter_trigger_time,
                   b.timestamp as shift_time
            FROM normalized_ai_events a
            JOIN normalized_ai_events b
              ON a.caller_identity = b.caller_identity
            WHERE a.content_filter_triggered = true
              AND b.event_type = 'invocation'
              AND b.cloud_provider != a.cloud_provider
              AND b.timestamp BETWEEN a.timestamp AND a.timestamp + INTERVAL '30 minutes'
        """,
        "severity": "HIGH",
        "response": "Review the prompts. If the content is consistent across clouds, "
                    "這是 likely 攻擊者 測試 filter strength differences.",
    },
    {
        "name": "Multi-雲端 Cost Spike",
        "description": "Aggregate AI spend across all clouds exceeds threshold.",
        "query": """
            SELECT SUM(estimated_cost_usd) as total_cost,
                   SUM(CASE WHEN cloud_provider = 'aws' THEN estimated_cost_usd ELSE 0 END) as aws_cost,
                   SUM(CASE WHEN cloud_provider = 'azure' THEN estimated_cost_usd ELSE 0 END) as azure_cost,
                   SUM(CASE WHEN cloud_provider = 'gcp' THEN estimated_cost_usd ELSE 0 END) as gcp_cost
            FROM normalized_ai_events
            WHERE timestamp > NOW() - INTERVAL '1 hour'
        """,
        "severity": "HIGH",
        "threshold": "total_cost > $500/hour",
        "response": "識別 top consuming identities across all clouds. Apply rate limits.",
    },
]

def define_cross_cloud_incident_playbook() -> dict:
    """Define incident response procedures for multi-雲端 AI incidents."""
    return {
        "incident_types": {
            "credential_compromise": {
                "description": "AI service credential (API key, access key, SA key) "
                               "exposed or stolen.",
                "immediate_actions": [
                    "識別 the credential type and scope",
                    "Revoke/rotate the credential on the affected 雲端",
                    "Check all other clouds for the same identity pattern",
                    "Review audit logs for unauthorized usage since exposure",
                    "評估 符元 consumption and cost impact",
                ],
                "cross_cloud_considerations": [
                    "If the compromised identity federates to multiple clouds, "
                    "攻擊者 may have pivoted to other providers",
                    "Check for cross-雲端 service identity chains "
                    "(e.g., AWS role -> assumes GCP SA)",
                    "Review multi-雲端 SIEM for activity from the same source IP",
                ],
            },
            "content_safety_bypass": {
                "description": "攻擊者 successfully bypasses content 安全 "
                               "controls to generate harmful 輸出.",
                "immediate_actions": [
                    "Capture the bypass technique (prompt, context, encoding)",
                    "測試 the same technique against other 雲端 providers' filters",
                    "Update 護欄/content filters to block the technique",
                    "Review logs for prior successful bypasses",
                ],
                "cross_cloud_considerations": [
                    "A technique that bypasses one provider's filter often works "
                    "on others -- 測試 and patch all providers simultaneously",
                    "Content 安全 configurations may differ across clouds; "
                    "normalize policies to consistent strength levels",
                ],
            },
            "model_abuse": {
                "description": "AI models used for unauthorized purposes "
                               "(generating malware, phishing content, etc.).",
                "immediate_actions": [
                    "識別 the abusing identity and disable access",
                    "Preserve all prompts and responses as evidence",
                    "評估 whether the generated content was exfiltrated",
                    "Report to 雲端 provider if required by terms of service",
                ],
                "cross_cloud_considerations": [
                    "攻擊者 may distribute abuse across multiple providers "
                    "to stay below individual provider 偵測 thresholds",
                    "Aggregate abuse indicators across all clouds before assessing scope",
                ],
            },
            "denial_of_wallet": {
                "description": "Attacker generates excessive AI API costs.",
                "immediate_actions": [
                    "Apply emergency rate limits on affected service",
                    "Revoke the abusing credential",
                    "Calculate total financial impact across all clouds",
                    "Engage 雲端 provider support for cost dispute if applicable",
                ],
                "cross_cloud_considerations": [
                    "Sum costs across all providers for total impact 評估",
                    "Check if 攻擊者 spread load across clouds to avoid "
                    "per-provider budget thresholds",
                ],
            },
        },
    }