Dataresidentie en Soevereiniteit voor Cloud-AI

import boto3
from botocore.config import Config
 
def create_region_locked_bedrock_client(
    region: str,
    enforce_region: bool = True,
) -> dict:
    """Create a Bedrock client locked to a specific region with verification."""
 
    # Prevent any cross-region fallback
    config = Config(
        region_name=region,
        retries={"max_attempts": 3, "mode": "standard"},
    )
 
    session = boto3.Session(region_name=region)
    bedrock = session.client("bedrock", config=config)
    bedrock_runtime = session.client("bedrock-runtime", config=config)
 
    if enforce_region:
        # Verify we are talking to the intended region
        available_models = bedrock.list_foundation_models()
        model_count = len(available_models.get("modelSummaries", []))
 
        return {
            "region": region,
            "runtime_endpoint": f"https://bedrock-runtime.{region}.amazonaws.com",
            "management_endpoint": f"https://bedrock.{region}.amazonaws.com",
            "models_available": model_count,
            "clients": {
                "bedrock": bedrock,
                "bedrock_runtime": bedrock_runtime,
            },
        }
 
    return {"bedrock": bedrock, "bedrock_runtime": bedrock_runtime}
 
def generate_region_restriction_scp(allowed_regions: list) -> dict:
    """Generate an SCP that restricts Bedrock to specific regions."""
    return {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "RestrictBedrockRegions",
                "Effect": "Deny",
                "Action": [
                    "bedrock:*",
                ],
                "Resource": "*",
                "Condition": {
                    "StringNotEquals": {
                        "aws:RequestedRegion": allowed_regions
                    }
                },
            },
            {
                "Sid": "RestrictSageMakerRegions",
                "Effect": "Deny",
                "Action": [
                    "sagemaker:*",
                ],
                "Resource": "*",
                "Condition": {
                    "StringNotEquals": {
                        "aws:RequestedRegion": allowed_regions
                    }
                },
            },
        ],
    }

def audit_bedrock_data_residency(
    session: boto3.Session,
    target_region: str,
) -> dict:
    """Audit Bedrock deployment for data residency compliance."""
    findings = []
 
    # Check invocation logging destination
    bedrock = session.client("bedrock", region_name=target_region)
    try:
        logging_config = bedrock.get_model_invocation_logging_configuration()
        config = logging_config.get("loggingConfig", {})
 
        # Check S3 logging destination region
        s3_config = config.get("s3Config", {})
        if s3_config:
            bucket_name = s3_config.get("bucketName", "")
            s3 = session.client("s3")
            try:
                location = s3.get_bucket_location(Bucket=bucket_name)
                bucket_region = location.get("LocationConstraint") or "us-east-1"
                if bucket_region != target_region:
                    findings.append({
                        "severity": "HIGH",
                        "finding": "Invocation logs stored in different region",
                        "detail": f"Bedrock is in {target_region} but logs go to "
                                  f"S3 bucket in {bucket_region}. This creates "
                                  f"cross-region data transfer of prompts/responses.",
                    })
            except Exception:
                pass
 
        # Check CloudWatch logging destination
        cw_config = config.get("cloudWatchConfig", {})
        if cw_config:
            log_group = cw_config.get("logGroupName", "")
            findings.append({
                "severity": "INFO",
                "finding": f"CloudWatch logging to {log_group}",
                "detail": "CloudWatch log groups are regional. Verify this log group "
                          "is in the same region as the Bedrock endpoint.",
            })
 
    except Exception:
        findings.append({
            "severity": "MEDIUM",
            "finding": "Cannot verify invocation logging configuration",
        })
 
    # Check CloudTrail configuration
    cloudtrail = session.client("cloudtrail", region_name=target_region)
    try:
        trails = cloudtrail.describe_trails()
        for trail in trails.get("trailList", []):
            if trail.get("IsMultiRegionTrail"):
                s3_bucket = trail.get("S3BucketName", "")
                findings.append({
                    "severity": "MEDIUM",
                    "finding": "Multi-region CloudTrail trail detected",
                    "detail": f"Trail {trail['Name']} aggregates logs from all regions "
                              f"to bucket {s3_bucket}. Bedrock management events from "
                              f"{target_region} may be stored in a different region.",
                })
    except Exception:
        pass
 
    return {"target_region": target_region, "findings": findings}

from azure.identity import DefaultAzureCredential
from azure.mgmt.cognitiveservices import CognitiveServicesManagementClient
 
def audit_azure_openai_data_residency(
    subscription_id: str,
    resource_group: str,
    account_name: str,
    required_geography: str = "europe",
) -> dict:
    """Audit Azure OpenAI resource for data residency compliance."""
    credential = DefaultAzureCredential()
    client = CognitiveServicesManagementClient(credential, subscription_id)
    findings = []
 
    account = client.accounts.get(resource_group, account_name)
    resource_location = account.location
 
    # Map Azure regions to geographies
    eu_regions = [
        "westeurope", "northeurope", "francecentral", "francesouth",
        "germanywestcentral", "germanynorth", "swedencentral",
        "swedensouth", "switzerlandnorth", "switzerlandwest",
        "norwayeast", "norwaywest", "uksouth", "ukwest",
    ]
 
    if required_geography == "europe" and resource_location not in eu_regions:
        findings.append({
            "severity": "CRITICAL",
            "finding": f"Resource deployed outside EU: {resource_location}",
            "detail": "For GDPR compliance, Azure OpenAI resources processing "
                      "EU personal data must be in an EU region.",
        })
 
    # Check data storage and processing commitments
    # Azure OpenAI with abuse monitoring opt-out
    if account.properties.restrict_outbound_network_access:
        findings.append({
            "severity": "INFO",
            "finding": "Outbound network access restricted",
            "detail": "Reduces risk of data leaving the deployment region.",
        })
 
    # Check diagnostic settings for cross-region logging
    # (Diagnostic settings might send data to Log Analytics in another region)
    from azure.mgmt.monitor import MonitorManagementClient
    monitor = MonitorManagementClient(credential, subscription_id)
 
    resource_uri = account.id
    for setting in monitor.diagnostic_settings.list(resource_uri):
        if setting.workspace_id:
            # Check if Log Analytics workspace is in the same region
            workspace_parts = setting.workspace_id.split("/")
            findings.append({
                "severity": "MEDIUM",
                "finding": "Diagnostic logs sent to Log Analytics workspace",
                "detail": f"Verify workspace {setting.workspace_id} is in the same "
                          f"region as the OpenAI resource ({resource_location}). "
                          f"Cross-region diagnostic data transfer can violate residency.",
            })
 
    # List deployments and check model availability
    deployments = client.deployments.list(resource_group, account_name)
    for deployment in deployments:
        findings.append({
            "severity": "INFO",
            "finding": f"Model deployment: {deployment.name}",
            "detail": f"Model: {deployment.properties.model.name}, "
                      f"Version: {deployment.properties.model.version}. "
                      f"Data processed in {resource_location}.",
        })
 
    return {
        "resource_location": resource_location,
        "required_geography": required_geography,
        "findings": findings,
    }

from google.cloud import aiplatform
 
def audit_vertex_data_residency(
    project_id: str,
    required_region: str = "europe-west4",
) -> dict:
    """Audit Vertex AI resources for data residency compliance."""
    findings = []
 
    # Check all locations where Vertex AI resources exist
    locations_to_check = [
        "us-central1", "us-east1", "us-west1",
        "europe-west1", "europe-west4", "europe-west2",
        "asia-southeast1", "asia-northeast1",
    ]
 
    for location in locations_to_check:
        try:
            aiplatform.init(project=project_id, location=location)
            endpoints = aiplatform.Endpoint.list()
            models = aiplatform.Model.list()
 
            if endpoints or models:
                in_required = location == required_region
                severity = "INFO" if in_required else "HIGH"
                findings.append({
                    "severity": severity,
                    "location": location,
                    "finding": f"Vertex AI resources found in {location}",
                    "detail": f"{len(endpoints)} endpoints, {len(models)} models. "
                              + ("In required region." if in_required
                                 else f"Outside required region ({required_region})."),
                })
        except Exception:
            continue
 
    # Check organization policy constraints for location restriction
    try:
        from google.cloud import orgpolicy_v2
        orgpolicy_client = orgpolicy_v2.OrgPolicyClient()
 
        policy = orgpolicy_client.get_policy(
            name=f"projects/{project_id}/policies/gcp.resourceLocations"
        )
        if policy.spec and policy.spec.rules:
            allowed_locations = []
            for rule in policy.spec.rules:
                if rule.values and rule.values.allowed_values:
                    allowed_locations.extend(rule.values.allowed_values)
 
            findings.append({
                "severity": "INFO",
                "finding": "Organization policy restricts resource locations",
                "detail": f"Allowed locations: {', '.join(allowed_locations)}",
            })
        else:
            findings.append({
                "severity": "HIGH",
                "finding": "No location restriction organization policy",
                "detail": "Without gcp.resourceLocations constraint, Vertex AI "
                          "resources can be created in any region.",
            })
    except Exception:
        findings.append({
            "severity": "MEDIUM",
            "finding": "Cannot check organization policy for location restrictions",
        })
 
    return {"required_region": required_region, "findings": findings}

def recommend_residency_org_policies(target_geography: str) -> dict:
    """Recommend organization policies for data residency enforcement."""
    eu_locations = [
        "in:eu-locations",  # GCP's built-in EU location group
    ]
 
    return {
        "resource_location_constraint": {
            "constraint": "constraints/gcp.resourceLocations",
            "description": "Restrict where Vertex AI and other resources can be created",
            "spec": {
                "rules": [{
                    "values": {
                        "allowedValues": eu_locations if target_geography == "eu" else [],
                    }
                }]
            },
        },
        "vertex_ai_specific": {
            "constraint": "constraints/aiplatform.allowedModels",
            "description": "Restrict which models can be deployed (some models "
                           "may have different data processing locations)",
            "note": "Not all models are available in all regions. Restricting "
                    "models to those available in your target region prevents "
                    "users from deploying models that require cross-region processing.",
        },
        "cloud_storage": {
            "constraint": "constraints/storage.locations",
            "description": "Restrict Cloud Storage bucket locations for training data",
            "spec": {
                "rules": [{
                    "values": {
                        "allowedValues": eu_locations if target_geography == "eu" else [],
                    }
                }]
            },
        },
    }