OpenAI SDK Track Part 17: Enterprise & Compliance

                flowchart TD
                    A[Organization Owner] --> B[Admin API]
                    B --> C[Project: Production]
                    B --> D[Project: Staging]
                    B --> E[Project: Development]

                    C --> F[Service Account - Prod]
                    D --> G[Service Account - Staging]
                    E --> H[Developer Keys]

                    F --> I{Data Classification}
                    I -->|PII / PHI| J[ZDR Endpoint + Azure Private Link]
                    I -->|Internal| K[Standard Endpoint + Audit Log]
                    I -->|Public| L[Cached Endpoint]

                    J --> M[Immutable Audit Trail]
                    K --> M
                    L --> M

                    M --> N[SIEM / Compliance Dashboard]
                    N --> O[SOC 2 Evidence Collection]
                    N --> P[GDPR Data Subject Requests]
                    N --> Q[Incident Response]
            

1. Organization & Admin API

OpenAI’s Admin API provides programmatic control over your organization’s membership, roles, projects, and rate limits. For enterprises managing dozens of teams and hundreds of developers, manual dashboard management is untenable. The Admin API enables Infrastructure-as-Code (IaC) patterns where organization structure is defined declaratively and enforced automatically — new hires get provisioned, leavers get revoked, and project access follows your identity provider’s group memberships.

Roles & Permissions

Programmatic Organization Management

import os
import json
import httpx
from datetime import datetime, timezone


class OpenAIAdminClient:
    """Client for OpenAI's Admin API — manage members, roles, and invitations.

    Requires an organization-level admin key (not a project key).
    API docs: https://platform.openai.com/docs/api-reference/organization
    """

    BASE_URL = "https://api.openai.com/v1/organization"

    def __init__(self, admin_key: str = None):
        self.admin_key = admin_key or os.environ.get("OPENAI_ADMIN_KEY", "sk-admin-demo-key")
        self.headers = {
            "Authorization": f"Bearer {self.admin_key}",
            "Content-Type": "application/json",
            "OpenAI-Organization": os.environ.get("OPENAI_ORG_ID", "org-demo123"),
        }

    def list_members(self, limit: int = 100) -> dict:
        """List all organization members with their roles."""
        # In production, this calls the real API
        # response = httpx.get(f"{self.BASE_URL}/users", headers=self.headers, params={"limit": limit})
        # return response.json()

        # Demo: simulated response
        return {
            "data": [
                {"id": "user-abc123", "email": "alice@company.com", "role": "owner", "added_at": "2024-01-15"},
                {"id": "user-def456", "email": "bob@company.com", "role": "admin", "added_at": "2024-03-20"},
                {"id": "user-ghi789", "email": "carol@company.com", "role": "member", "added_at": "2025-06-10"},
                {"id": "user-jkl012", "email": "dave@company.com", "role": "reader", "added_at": "2025-11-01"},
            ],
            "has_more": False,
        }

    def invite_member(self, email: str, role: str = "member", projects: list = None) -> dict:
        """Invite a new member to the organization with a specific role."""
        payload = {"email": email, "role": role}
        if projects:
            payload["projects"] = [{"id": p, "role": "member"} for p in projects]

        # In production: httpx.post(f"{self.BASE_URL}/invites", headers=self.headers, json=payload)
        return {
            "id": f"invite-{email.split('@')[0]}",
            "email": email,
            "role": role,
            "status": "pending",
            "created_at": datetime.now(timezone.utc).isoformat(),
            "expires_at": "2026-06-01T00:00:00Z",
        }

    def update_member_role(self, user_id: str, new_role: str) -> dict:
        """Update a member's organization role."""
        # In production: httpx.post(f"{self.BASE_URL}/users/{user_id}", headers=self.headers, json={"role": new_role})
        return {"id": user_id, "role": new_role, "updated": True}

    def remove_member(self, user_id: str) -> dict:
        """Remove a member from the organization (revokes all keys)."""
        # In production: httpx.delete(f"{self.BASE_URL}/users/{user_id}", headers=self.headers)
        return {"id": user_id, "deleted": True, "keys_revoked": 3}

    def list_projects(self) -> dict:
        """List all projects in the organization."""
        return {
            "data": [
                {"id": "proj-prod-001", "name": "production", "status": "active", "created_at": "2024-02-01"},
                {"id": "proj-stg-002", "name": "staging", "status": "active", "created_at": "2024-02-01"},
                {"id": "proj-dev-003", "name": "development", "status": "active", "created_at": "2024-02-01"},
                {"id": "proj-ml-004", "name": "ml-research", "status": "active", "created_at": "2025-01-15"},
            ],
        }


# Usage demonstration
admin = OpenAIAdminClient()

print("=== OpenAI Admin API: Organization Management ===\n")

# List current members
members = admin.list_members()
print("Current Members:")
for m in members["data"]:
    print(f"  {m['email']:<25} role={m['role']:<8} added={m['added_at']}")

# Invite a new team member
print("\n--- Inviting new member ---")
invite = admin.invite_member("newdev@company.com", role="member", projects=["proj-dev-003"])
print(f"  Invited: {invite['email']} as {invite['role']} (status={invite['status']})")

# List projects
print("\n--- Organization Projects ---")
projects = admin.list_projects()
for p in projects["data"]:
    print(f"  [{p['id']}] {p['name']:<15} status={p['status']}")

# Offboarding: remove a departing employee
print("\n--- Offboarding user-jkl012 ---")
result = admin.remove_member("user-jkl012")
print(f"  Removed: {result['id']} | Keys revoked: {result['keys_revoked']}")

2. Project Isolation

OpenAI projects provide environment isolation similar to AWS accounts or GCP projects. Each project has its own API keys, rate limits, usage tracking, and access controls. This separation is critical for enterprises that need to prevent development workloads from consuming production quotas, enforce different data handling policies per environment, and track costs by business unit.

Multi-Project Architecture

3. Access Control & RBAC

Enterprise access control for OpenAI goes beyond the platform’s built-in roles. You need application-level RBAC that maps your organization’s identity provider groups to OpenAI projects and capabilities, service accounts for CI/CD pipelines, and automated key rotation to limit the blast radius of credential compromise.

Key Rotation Automation

API keys should be rotated regularly (every 30-90 days) and immediately upon any suspected compromise. The rotation process must be zero-downtime: create the new key, update all consumers, verify the new key works, then revoke the old key. This pattern uses a dual-key overlap window to prevent any request failures during rotation.

import os
import json
import hashlib
from datetime import datetime, timezone, timedelta
from dataclasses import dataclass, field


@dataclass
class APIKeyRecord:
    """Metadata record for a managed API key."""
    key_id: str
    key_prefix: str           # First 8 chars for identification
    project: str
    environment: str
    created_at: str
    expires_at: str
    rotated_from: str = None  # Previous key ID this replaced
    status: str = "active"    # active, rotating, revoked


class KeyRotationManager:
    """Automated API key rotation with zero-downtime overlap window.

    Implements the pattern:
    1. Create new key (both keys active during overlap)
    2. Update consumers (config/secrets manager)
    3. Verify new key works (health check)
    4. Revoke old key after grace period
    """

    def __init__(self, rotation_days: int = 60, overlap_hours: int = 24):
        self.rotation_days = rotation_days
        self.overlap_hours = overlap_hours
        self.key_registry: dict[str, APIKeyRecord] = {}
        self._seed_demo_keys()

    def _seed_demo_keys(self):
        """Initialize with sample key records for demonstration."""
        now = datetime.now(timezone.utc)
        keys = [
            APIKeyRecord(
                key_id="key-prod-001",
                key_prefix="sk-proj-a",
                project="proj-prod",
                environment="production",
                created_at=(now - timedelta(days=55)).isoformat(),
                expires_at=(now + timedelta(days=5)).isoformat(),
                status="active",
            ),
            APIKeyRecord(
                key_id="key-stg-001",
                key_prefix="sk-proj-b",
                project="proj-staging",
                environment="staging",
                created_at=(now - timedelta(days=20)).isoformat(),
                expires_at=(now + timedelta(days=40)).isoformat(),
                status="active",
            ),
            APIKeyRecord(
                key_id="key-dev-001",
                key_prefix="sk-proj-c",
                project="proj-dev",
                environment="development",
                created_at=(now - timedelta(days=80)).isoformat(),
                expires_at=(now - timedelta(days=20)).isoformat(),
                status="active",
            ),
        ]
        for k in keys:
            self.key_registry[k.key_id] = k

    def check_rotation_needed(self) -> list[dict]:
        """Identify keys that need rotation (expired or expiring within 7 days)."""
        now = datetime.now(timezone.utc)
        threshold = now + timedelta(days=7)
        needs_rotation = []

        for key in self.key_registry.values():
            if key.status == "revoked":
                continue
            expires = datetime.fromisoformat(key.expires_at)
            if expires <= threshold:
                days_remaining = (expires - now).days
                needs_rotation.append({
                    "key_id": key.key_id,
                    "project": key.project,
                    "environment": key.environment,
                    "days_remaining": days_remaining,
                    "urgency": "CRITICAL" if days_remaining <= 0 else "WARNING",
                })
        return needs_rotation

    def rotate_key(self, old_key_id: str) -> dict:
        """Execute key rotation with overlap window."""
        old_key = self.key_registry.get(old_key_id)
        if not old_key:
            return {"error": f"Key {old_key_id} not found"}

        now = datetime.now(timezone.utc)
        new_key_id = f"key-{old_key.environment[:4]}-{hashlib.sha256(now.isoformat().encode()).hexdigest()[:6]}"

        # Create new key record
        new_key = APIKeyRecord(
            key_id=new_key_id,
            key_prefix=f"sk-proj-{hashlib.sha256(new_key_id.encode()).hexdigest()[:4]}",
            project=old_key.project,
            environment=old_key.environment,
            created_at=now.isoformat(),
            expires_at=(now + timedelta(days=self.rotation_days)).isoformat(),
            rotated_from=old_key_id,
            status="active",
        )
        self.key_registry[new_key_id] = new_key

        # Mark old key for revocation after overlap window
        old_key.status = "rotating"

        return {
            "action": "rotation_initiated",
            "old_key": old_key_id,
            "new_key": new_key_id,
            "overlap_until": (now + timedelta(hours=self.overlap_hours)).isoformat(),
            "old_key_revocation": (now + timedelta(hours=self.overlap_hours)).isoformat(),
            "steps": [
                "1. New key created and active",
                "2. Update secrets manager / environment variables",
                "3. Verify new key with health check",
                f"4. Old key will be revoked in {self.overlap_hours}h",
            ],
        }

    def get_rotation_report(self) -> dict:
        """Generate a compliance report of all key statuses."""
        report = {"generated_at": datetime.now(timezone.utc).isoformat(), "keys": []}
        for key in self.key_registry.values():
            report["keys"].append({
                "key_id": key.key_id,
                "prefix": key.key_prefix,
                "project": key.project,
                "env": key.environment,
                "status": key.status,
                "age_days": (datetime.now(timezone.utc) - datetime.fromisoformat(key.created_at)).days,
            })
        return report


# Usage demonstration
manager = KeyRotationManager(rotation_days=60, overlap_hours=24)

print("=== API Key Rotation Manager ===\n")

# Check which keys need rotation
needs_rotation = manager.check_rotation_needed()
print("Keys Needing Rotation:")
for item in needs_rotation:
    print(f"  [{item['urgency']:>8}] {item['key_id']} ({item['environment']}) — "
          f"{item['days_remaining']} days remaining")

# Rotate the most urgent key
if needs_rotation:
    urgent = needs_rotation[0]
    print(f"\n--- Rotating {urgent['key_id']} ---")
    result = manager.rotate_key(urgent["key_id"])
    print(f"  New key: {result['new_key']}")
    print(f"  Overlap until: {result['overlap_until']}")
    for step in result["steps"]:
        print(f"    {step}")

# Generate compliance report
print("\n--- Key Inventory Report ---")
report = manager.get_rotation_report()
for k in report["keys"]:
    print(f"  {k['key_id']:<20} env={k['env']:<12} status={k['status']:<10} age={k['age_days']}d")

4. Data Privacy & Compliance

Data privacy is the top concern for enterprises adopting OpenAI. The key questions: Is my data used to train models? Where is my data stored? How long is it retained? Can I satisfy GDPR data subject requests? OpenAI offers Zero Data Retention (ZDR) for eligible API customers, which means prompts and completions are not stored by OpenAI after the response is delivered — they pass through memory only.

Zero Data Retention Configuration

import os
import re
import hashlib
from dataclasses import dataclass, field
from datetime import datetime, timezone


@dataclass
class ComplianceConfig:
    """Enterprise compliance configuration for OpenAI API usage."""

    zero_data_retention: bool = True
    data_processing_agreement: bool = True
    pii_redaction_enabled: bool = True
    audit_all_requests: bool = True
    allowed_models: list = field(default_factory=lambda: ["gpt-4.1", "gpt-4.1-mini"])
    blocked_data_patterns: list = field(default_factory=lambda: [
        r"\b\d{3}-\d{2}-\d{4}\b",          # SSN
        r"\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b",  # Credit card
        r"\b[A-Z]{2}\d{2}[A-Z0-9]{4}\d{7}([A-Z0-9]?){0,16}\b",  # IBAN
    ])


class ZDRCompliantClient:
    """OpenAI client wrapper enforcing Zero Data Retention compliance.

    Implements defense-in-depth:
    1. PII detection and redaction before API calls
    2. Request validation against compliance policy
    3. Audit logging of all interactions (without storing prompts in logs)
    4. Model allowlisting to prevent use of non-compliant endpoints
    """

    def __init__(self, config: ComplianceConfig = None):
        self.config = config or ComplianceConfig()
        self.audit_log: list[dict] = []

    def scan_for_pii(self, text: str) -> dict:
        """Scan text for PII patterns that should never reach the API."""
        findings = []
        for pattern in self.config.blocked_data_patterns:
            matches = re.findall(pattern, text)
            if matches:
                findings.append({
                    "pattern": pattern,
                    "count": len(matches),
                    "severity": "CRITICAL",
                })
        return {"has_pii": len(findings) > 0, "findings": findings}

    def redact_pii(self, text: str) -> str:
        """Replace detected PII with placeholder tokens."""
        redacted = text
        replacements = {
            r"\b\d{3}-\d{2}-\d{4}\b": "[SSN_REDACTED]",
            r"\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b": "[CARD_REDACTED]",
            r"\b[A-Z]{2}\d{2}[A-Z0-9]{4}\d{7}([A-Z0-9]?){0,16}\b": "[IBAN_REDACTED]",
        }
        for pattern, replacement in replacements.items():
            redacted = re.sub(pattern, replacement, redacted)
        return redacted

    def validate_request(self, model: str, messages: list) -> dict:
        """Validate a request against compliance policy before sending."""
        issues = []

        # Check model allowlist
        if model not in self.config.allowed_models:
            issues.append(f"Model '{model}' not in allowlist: {self.config.allowed_models}")

        # Scan all message content for PII
        for msg in messages:
            content = msg.get("content", "")
            scan = self.scan_for_pii(content)
            if scan["has_pii"]:
                issues.append(f"PII detected in {msg['role']} message: {scan['findings']}")

        return {
            "compliant": len(issues) == 0,
            "issues": issues,
            "zdr_active": self.config.zero_data_retention,
            "dpa_signed": self.config.data_processing_agreement,
        }

    def prepare_compliant_request(self, model: str, messages: list) -> dict:
        """Prepare a request with all compliance controls applied."""
        # Step 1: Validate
        validation = self.validate_request(model, messages)

        # Step 2: Redact PII if enabled (defense in depth)
        cleaned_messages = []
        for msg in messages:
            cleaned = {**msg}
            if self.config.pii_redaction_enabled:
                cleaned["content"] = self.redact_pii(msg.get("content", ""))
            cleaned_messages.append(cleaned)

        # Step 3: Create audit record (hash content, don't store it)
        content_hash = hashlib.sha256(
            json.dumps(cleaned_messages).encode()
        ).hexdigest()[:16]

        audit_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "model": model,
            "content_hash": content_hash,
            "message_count": len(messages),
            "compliant": validation["compliant"],
            "pii_redacted": self.config.pii_redaction_enabled,
            "zdr_active": self.config.zero_data_retention,
        }
        self.audit_log.append(audit_entry)

        return {
            "model": model,
            "messages": cleaned_messages,
            "validation": validation,
            "audit_id": content_hash,
            "headers": {
                "X-ZDR-Enabled": "true",
                "X-Audit-ID": content_hash,
                "X-Compliance-Version": "2.1",
            },
        }


# Import json for the hash
import json

# Usage demonstration
config = ComplianceConfig()
client = ZDRCompliantClient(config)

print("=== Zero Data Retention Compliance Client ===\n")

# Test with clean message
clean_messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "Summarize the quarterly revenue trends for Q1 2026."},
]

result = client.prepare_compliant_request("gpt-4.1", clean_messages)
print(f"Clean request — compliant: {result['validation']['compliant']}")
print(f"  ZDR active: {result['headers']['X-ZDR-Enabled']}")
print(f"  Audit ID: {result['audit_id']}")

# Test with PII-containing message
pii_messages = [
    {"role": "user", "content": "Process refund for customer SSN 123-45-6789 card 4111 1111 1111 1111"},
]

print("\n--- PII Detection Test ---")
result = client.prepare_compliant_request("gpt-4.1", pii_messages)
print(f"PII request — compliant: {result['validation']['compliant']}")
for issue in result["validation"]["issues"]:
    print(f"  ISSUE: {issue}")
print(f"  Redacted content: {result['messages'][0]['content']}")

# Test with disallowed model
print("\n--- Model Allowlist Test ---")
result = client.prepare_compliant_request("gpt-3.5-turbo", clean_messages)
print(f"Disallowed model — compliant: {result['validation']['compliant']}")
for issue in result["validation"]["issues"]:
    print(f"  ISSUE: {issue}")

5. Audit & Governance

Enterprise governance requires a complete audit trail of every AI interaction: who called what model, when, with what parameters, and what the outcome was. This data feeds compliance reporting, cost allocation, incident investigation, and model inventory management. The audit system must be tamper-evident (append-only), queryable for compliance reports, and efficient enough to handle thousands of requests per second without impacting latency.

Immutable Audit Logger

import json
import hashlib
import time
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from enum import Enum


class AuditAction(Enum):
    """Auditable actions in the OpenAI integration layer."""
    API_CALL = "api_call"
    KEY_CREATED = "key_created"
    KEY_ROTATED = "key_rotated"
    KEY_REVOKED = "key_revoked"
    MODEL_CHANGED = "model_changed"
    POLICY_VIOLATION = "policy_violation"
    PII_DETECTED = "pii_detected"
    RATE_LIMITED = "rate_limited"
    FALLBACK_TRIGGERED = "fallback_triggered"


@dataclass
class AuditEntry:
    """Single audit log entry with chain integrity."""
    timestamp: str
    action: str
    actor: str               # user-id or service-account-id
    project: str
    environment: str
    model: str = None
    tokens_used: int = 0
    latency_ms: float = 0.0
    status: str = "success"  # success, error, blocked
    metadata: dict = field(default_factory=dict)
    content_hash: str = None # SHA-256 of request (not content itself)
    previous_hash: str = None # Hash of previous entry (chain integrity)
    entry_hash: str = None    # Hash of this entry


class ImmutableAuditLog:
    """Append-only audit log with hash-chain integrity for tamper detection.

    Each entry includes a hash of the previous entry, creating a blockchain-like
    chain. Any modification to a historical entry breaks the chain and is
    immediately detectable during verification.
    """

    def __init__(self, project: str, environment: str):
        self.project = project
        self.environment = environment
        self.entries: list[AuditEntry] = []
        self._last_hash = "GENESIS"

    def _compute_hash(self, entry: AuditEntry) -> str:
        """Compute SHA-256 hash of an entry for chain integrity."""
        data = json.dumps(asdict(entry), sort_keys=True, default=str)
        return hashlib.sha256(data.encode()).hexdigest()[:32]

    def append(self, action: AuditAction, actor: str, **kwargs) -> AuditEntry:
        """Append a new audit entry (immutable — cannot be modified after creation)."""
        entry = AuditEntry(
            timestamp=datetime.now(timezone.utc).isoformat(),
            action=action.value,
            actor=actor,
            project=self.project,
            environment=self.environment,
            previous_hash=self._last_hash,
            **kwargs,
        )
        entry.entry_hash = self._compute_hash(entry)
        self._last_hash = entry.entry_hash
        self.entries.append(entry)
        return entry

    def verify_integrity(self) -> dict:
        """Verify the hash chain has not been tampered with."""
        if not self.entries:
            return {"valid": True, "entries_checked": 0}

        expected_prev = "GENESIS"
        for i, entry in enumerate(self.entries):
            if entry.previous_hash != expected_prev:
                return {
                    "valid": False,
                    "broken_at": i,
                    "expected": expected_prev,
                    "found": entry.previous_hash,
                }
            expected_prev = entry.entry_hash
        return {"valid": True, "entries_checked": len(self.entries)}

    def query(self, action: str = None, actor: str = None, since: str = None) -> list:
        """Query audit entries with optional filters."""
        results = self.entries
        if action:
            results = [e for e in results if e.action == action]
        if actor:
            results = [e for e in results if e.actor == actor]
        if since:
            results = [e for e in results if e.timestamp >= since]
        return results

    def compliance_summary(self) -> dict:
        """Generate a summary for compliance reporting."""
        total = len(self.entries)
        actions = {}
        total_tokens = 0
        violations = 0

        for entry in self.entries:
            actions[entry.action] = actions.get(entry.action, 0) + 1
            total_tokens += entry.tokens_used
            if entry.action == AuditAction.POLICY_VIOLATION.value:
                violations += 1

        return {
            "period_start": self.entries[0].timestamp if self.entries else None,
            "period_end": self.entries[-1].timestamp if self.entries else None,
            "total_entries": total,
            "total_tokens": total_tokens,
            "policy_violations": violations,
            "action_breakdown": actions,
            "chain_integrity": self.verify_integrity(),
        }


# Usage demonstration
audit = ImmutableAuditLog(project="proj-prod", environment="production")

print("=== Immutable Audit Log ===\n")

# Simulate a series of auditable events
audit.append(AuditAction.API_CALL, actor="svc-chatbot-prod",
             model="gpt-4.1", tokens_used=1250, latency_ms=340.5,
             content_hash="a1b2c3d4e5f6")

audit.append(AuditAction.PII_DETECTED, actor="svc-chatbot-prod",
             model="gpt-4.1", status="blocked",
             metadata={"pattern": "SSN", "action": "redacted_and_continued"})

audit.append(AuditAction.API_CALL, actor="svc-analytics-prod",
             model="gpt-4.1-mini", tokens_used=820, latency_ms=180.2,
             content_hash="f6e5d4c3b2a1")

audit.append(AuditAction.KEY_ROTATED, actor="admin-platform-team",
             metadata={"old_key": "key-prod-001", "new_key": "key-prod-002"})

audit.append(AuditAction.RATE_LIMITED, actor="svc-chatbot-prod",
             model="gpt-4.1", status="error",
             metadata={"retry_after_ms": 5000})

# Verify chain integrity
integrity = audit.verify_integrity()
print(f"Chain Integrity: valid={integrity['valid']}, entries={integrity['entries_checked']}")

# Generate compliance summary
print("\n--- Compliance Summary ---")
summary = audit.compliance_summary()
print(f"  Period: {summary['period_start'][:19]} to {summary['period_end'][:19]}")
print(f"  Total entries: {summary['total_entries']}")
print(f"  Total tokens: {summary['total_tokens']:,}")
print(f"  Policy violations: {summary['policy_violations']}")
print(f"  Action breakdown:")
for action, count in summary["action_breakdown"].items():
    print(f"    {action:<25} {count}")

# Query specific events
print("\n--- PII Detection Events ---")
pii_events = audit.query(action="pii_detected")
for event in pii_events:
    print(f"  [{event.timestamp[:19]}] actor={event.actor} status={event.status}")

6. Multi-Region Strategy

For enterprises with data residency requirements (EU data must stay in EU, healthcare data must stay in-region), Azure OpenAI provides regional deployments. A multi-region strategy also gives you failover capabilities: if one region is degraded, traffic automatically routes to a healthy region. This is critical for achieving 99.9%+ availability SLAs that enterprise customers demand.

Data Residency & Regional Requirements

import os
import time
import random
from dataclasses import dataclass, field
from datetime import datetime, timezone


@dataclass
class RegionEndpoint:
    """Configuration for a regional OpenAI/Azure OpenAI endpoint."""
    region: str
    provider: str                   # "azure" or "openai"
    endpoint: str
    deployment_name: str = None     # Azure-specific
    priority: int = 1               # Lower = higher priority
    is_healthy: bool = True
    last_check: float = field(default_factory=time.time)
    latency_ms: float = 0.0
    consecutive_failures: int = 0
    data_residency: str = "global"  # eu, us, apac, global


class MultiRegionClient:
    """Multi-region OpenAI client with automatic failover and data residency routing.

    Features:
    - Routes requests based on data residency requirements
    - Automatic failover when a region is unhealthy
    - Latency-based routing for non-restricted data
    - Health checking with circuit breaker per region
    """

    def __init__(self, regions: list[RegionEndpoint] = None):
        self.regions = regions or self._default_regions()
        self.request_count = 0
        self.failover_count = 0

    def _default_regions(self) -> list[RegionEndpoint]:
        """Default multi-region configuration for global enterprise."""
        return [
            RegionEndpoint(
                region="westeurope",
                provider="azure",
                endpoint="https://myorg-eu.openai.azure.com",
                deployment_name="gpt-4.1-eu",
                priority=1,
                data_residency="eu",
                latency_ms=45.0,
            ),
            RegionEndpoint(
                region="eastus",
                provider="azure",
                endpoint="https://myorg-us.openai.azure.com",
                deployment_name="gpt-4.1-us",
                priority=1,
                data_residency="us",
                latency_ms=30.0,
            ),
            RegionEndpoint(
                region="japaneast",
                provider="azure",
                endpoint="https://myorg-apac.openai.azure.com",
                deployment_name="gpt-4.1-apac",
                priority=2,
                data_residency="apac",
                latency_ms=80.0,
            ),
            RegionEndpoint(
                region="global",
                provider="openai",
                endpoint="https://api.openai.com/v1",
                deployment_name=None,
                priority=3,
                data_residency="global",
                latency_ms=50.0,
            ),
        ]

    def select_endpoint(self, data_residency: str = None, prefer_latency: bool = False) -> RegionEndpoint:
        """Select the best endpoint based on data residency and health.

        Args:
            data_residency: Required region for data (eu, us, apac, or None for any)
            prefer_latency: If True, optimize for lowest latency among eligible regions
        """
        # Filter by health
        healthy = [r for r in self.regions if r.is_healthy]
        if not healthy:
            # All regions down — try the one with fewest consecutive failures
            healthy = sorted(self.regions, key=lambda r: r.consecutive_failures)[:1]

        # Filter by data residency requirement
        if data_residency:
            eligible = [r for r in healthy if r.data_residency == data_residency]
            if not eligible:
                # No healthy region matches residency — this is a compliance violation
                return None  # Caller must handle: cannot process this request
        else:
            eligible = healthy

        # Sort by preference
        if prefer_latency:
            eligible.sort(key=lambda r: r.latency_ms)
        else:
            eligible.sort(key=lambda r: (r.priority, r.latency_ms))

        return eligible[0] if eligible else None

    def simulate_request(self, prompt: str, data_residency: str = None) -> dict:
        """Simulate a request with automatic failover."""
        self.request_count += 1
        endpoint = self.select_endpoint(data_residency=data_residency)

        if endpoint is None:
            return {
                "status": "BLOCKED",
                "reason": f"No healthy endpoint satisfies data_residency='{data_residency}'",
                "compliance_violation": True,
            }

        # Simulate potential failure (10% chance for demo)
        if random.random() < 0.1:
            endpoint.consecutive_failures += 1
            if endpoint.consecutive_failures >= 3:
                endpoint.is_healthy = False

            # Attempt failover
            self.failover_count += 1
            fallback = self.select_endpoint(data_residency=data_residency)
            if fallback and fallback.region != endpoint.region:
                return {
                    "status": "SUCCESS_FAILOVER",
                    "primary_region": endpoint.region,
                    "failover_region": fallback.region,
                    "provider": fallback.provider,
                    "latency_ms": fallback.latency_ms + 20,  # Failover adds latency
                    "data_residency": fallback.data_residency,
                }

        # Normal successful request
        endpoint.consecutive_failures = 0
        return {
            "status": "SUCCESS",
            "region": endpoint.region,
            "provider": endpoint.provider,
            "endpoint": endpoint.endpoint,
            "latency_ms": endpoint.latency_ms + random.uniform(-5, 15),
            "data_residency": endpoint.data_residency,
        }

    def health_report(self) -> dict:
        """Generate a health report across all regions."""
        return {
            "total_requests": self.request_count,
            "failovers": self.failover_count,
            "failover_rate": f"{(self.failover_count / max(self.request_count, 1)) * 100:.1f}%",
            "regions": [
                {
                    "region": r.region,
                    "provider": r.provider,
                    "healthy": r.is_healthy,
                    "latency_ms": r.latency_ms,
                    "failures": r.consecutive_failures,
                    "data_residency": r.data_residency,
                }
                for r in self.regions
            ],
        }


# Usage demonstration
random.seed(42)  # Reproducible demo
client = MultiRegionClient()

print("=== Multi-Region OpenAI Client ===\n")

# EU data residency request (GDPR)
print("--- EU Data Residency (GDPR) ---")
result = client.simulate_request("Analyze German customer feedback", data_residency="eu")
print(f"  Status: {result['status']}")
print(f"  Region: {result.get('region', 'N/A')}")
print(f"  Data stays in: {result['data_residency']}")

# US healthcare data
print("\n--- US Data Residency (HIPAA) ---")
result = client.simulate_request("Summarize patient intake notes", data_residency="us")
print(f"  Status: {result['status']}")
print(f"  Region: {result.get('region', 'N/A')}")
print(f"  Data stays in: {result['data_residency']}")

# Global data (no restriction — optimize for latency)
print("\n--- Global Data (Latency Optimized) ---")
for i in range(5):
    result = client.simulate_request("General query", data_residency=None)
    status = result["status"]
    region = result.get("region", result.get("failover_region", "N/A"))
    print(f"  Request {i+1}: {status:<20} region={region:<12} latency={result.get('latency_ms', 0):.0f}ms")

# Health report
print("\n--- Regional Health Report ---")
report = client.health_report()
print(f"  Total requests: {report['total_requests']} | Failovers: {report['failovers']} ({report['failover_rate']})")
for r in report["regions"]:
    health = "HEALTHY" if r["healthy"] else "DEGRADED"
    print(f"  {r['region']:<12} [{r['provider']:<6}] {health:<10} latency={r['latency_ms']:.0f}ms residency={r['data_residency']}")

7. Enterprise Security

Enterprise security for OpenAI deployments encompasses network-level controls, encryption guarantees, and operational security practices. The goal is to ensure that AI API traffic is treated with the same rigor as any other sensitive data flow — encrypted in transit, access-controlled at the network layer, and monitored for anomalies.

Network & Encryption Controls

Service Account Setup for CI/CD

import os
import json
import hashlib
from dataclasses import dataclass, field
from datetime import datetime, timezone, timedelta


@dataclass
class ServiceAccount:
    """Represents a non-human identity for CI/CD and automated systems."""
    account_id: str
    name: str
    project: str
    environment: str
    permissions: list = field(default_factory=list)
    ip_allowlist: list = field(default_factory=list)
    created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    last_used: str = None
    key_expires_at: str = None
    max_rpm: int = 100
    max_tpm: int = 50_000


class ServiceAccountManager:
    """Manage non-human identities for OpenAI API access in CI/CD pipelines.

    Principles:
    - Least privilege: service accounts get minimum permissions needed
    - Scoped keys: each account is bound to a single project + environment
    - IP restriction: only allow calls from known CI/CD runner IPs
    - Short-lived keys: auto-expire, forcing regular rotation
    - Audit trail: all service account actions are logged
    """

    def __init__(self):
        self.accounts: dict[str, ServiceAccount] = {}
        self._seed_accounts()

    def _seed_accounts(self):
        """Create sample service accounts for demonstration."""
        accounts = [
            ServiceAccount(
                account_id="svc-chatbot-prod",
                name="Production Chatbot",
                project="proj-prod",
                environment="production",
                permissions=["chat.completions", "embeddings"],
                ip_allowlist=["10.0.1.0/24", "10.0.2.0/24"],
                max_rpm=5000,
                max_tpm=1_000_000,
                key_expires_at=(datetime.now(timezone.utc) + timedelta(days=30)).isoformat(),
            ),
            ServiceAccount(
                account_id="svc-cicd-staging",
                name="CI/CD Pipeline (Staging)",
                project="proj-staging",
                environment="staging",
                permissions=["chat.completions", "embeddings", "fine-tuning.read"],
                ip_allowlist=["192.168.1.100/32"],  # GitHub Actions runner
                max_rpm=200,
                max_tpm=100_000,
                key_expires_at=(datetime.now(timezone.utc) + timedelta(days=14)).isoformat(),
            ),
            ServiceAccount(
                account_id="svc-eval-runner",
                name="Evaluation Runner",
                project="proj-staging",
                environment="staging",
                permissions=["chat.completions"],
                ip_allowlist=["10.0.5.0/24"],
                max_rpm=1000,
                max_tpm=500_000,
                key_expires_at=(datetime.now(timezone.utc) + timedelta(days=7)).isoformat(),
            ),
        ]
        for acc in accounts:
            self.accounts[acc.account_id] = acc

    def validate_request(self, account_id: str, source_ip: str, permission: str) -> dict:
        """Validate whether a service account request should be allowed."""
        account = self.accounts.get(account_id)
        if not account:
            return {"allowed": False, "reason": "Unknown service account"}

        # Check IP allowlist
        ip_allowed = any(
            self._ip_in_cidr(source_ip, cidr) for cidr in account.ip_allowlist
        )
        if not ip_allowed:
            return {"allowed": False, "reason": f"IP {source_ip} not in allowlist"}

        # Check permission
        if permission not in account.permissions:
            return {"allowed": False, "reason": f"Permission '{permission}' not granted"}

        # Check key expiration
        if account.key_expires_at:
            expires = datetime.fromisoformat(account.key_expires_at)
            if datetime.now(timezone.utc) > expires:
                return {"allowed": False, "reason": "API key expired — rotation required"}

        return {
            "allowed": True,
            "account": account.name,
            "project": account.project,
            "rate_limits": {"rpm": account.max_rpm, "tpm": account.max_tpm},
        }

    def _ip_in_cidr(self, ip: str, cidr: str) -> bool:
        """Simplified CIDR check (production would use ipaddress module)."""
        # For demo: check if IP starts with the CIDR network prefix
        network = cidr.split("/")[0]
        prefix_parts = network.split(".")
        ip_parts = ip.split(".")
        # Compare based on CIDR prefix length (simplified)
        mask_bits = int(cidr.split("/")[1])
        octets_to_check = mask_bits // 8
        return ip_parts[:octets_to_check] == prefix_parts[:octets_to_check]

    def generate_policy_document(self) -> dict:
        """Generate a policy document for compliance documentation."""
        return {
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "total_service_accounts": len(self.accounts),
            "accounts": [
                {
                    "id": acc.account_id,
                    "name": acc.name,
                    "project": acc.project,
                    "environment": acc.environment,
                    "permissions": acc.permissions,
                    "ip_restricted": len(acc.ip_allowlist) > 0,
                    "key_expiry": acc.key_expires_at,
                    "rate_limits": f"{acc.max_rpm} RPM / {acc.max_tpm:,} TPM",
                }
                for acc in self.accounts.values()
            ],
        }


# Usage demonstration
manager = ServiceAccountManager()

print("=== Service Account Manager ===\n")

# Validate requests from different sources
test_cases = [
    ("svc-chatbot-prod", "10.0.1.50", "chat.completions"),
    ("svc-chatbot-prod", "203.0.113.1", "chat.completions"),  # Wrong IP
    ("svc-cicd-staging", "192.168.1.100", "fine-tuning.write"),  # Wrong permission
    ("svc-eval-runner", "10.0.5.10", "chat.completions"),
    ("svc-unknown", "10.0.1.1", "chat.completions"),  # Unknown account
]

print("Request Validation:")
for account_id, ip, permission in test_cases:
    result = manager.validate_request(account_id, ip, permission)
    status = "ALLOWED" if result["allowed"] else "DENIED"
    reason = result.get("reason", result.get("account", ""))
    print(f"  [{status:>7}] {account_id:<20} IP={ip:<16} perm={permission}")
    if not result["allowed"]:
        print(f"           Reason: {reason}")

# Generate policy document
print("\n--- Service Account Policy Document ---")
policy = manager.generate_policy_document()
for acc in policy["accounts"]:
    print(f"  {acc['id']:<20} project={acc['project']:<12} "
          f"perms={len(acc['permissions'])} ip_restricted={acc['ip_restricted']}")
    print(f"  {'':<20} limits={acc['rate_limits']} expires={acc['key_expiry'][:10]}")

SOC 2 Alignment Checklist

Governance Best Practices