# Security-first: Non-root user with minimal permissions
FROM python:3.11-slim
# Create non-root user
RUN useradd -m -u 1000 appuser && \
    mkdir -p /app && \
    chown appuser:appuser /app
# Install dependencies as root, run as non-root
COPY requirements.txt /tmp/
RUN pip install --no-cache-dir -r /tmp/requirements.txt && \
    rm /tmp/requirements.txt
# Switch to non-root user
USER appuser
WORKDIR /app
# Copy application code
COPY --chown=appuser:appuser . .
# Run application as non-root
CMD ["python", "app.py"]

# Phala Cloud deployment with defense-in-depth
app_name: secure-ai-api
tee_type: intel-tdx
# Layer 1: Infrastructure Security
infrastructure:  
  network_isolation: true  
  firewall_rules:    
    - port: 443      
      protocol: tcp      
      source: "0.0.0.0/0"  # HTTPS only    
    - port: 9944      
      protocol: tcp      
      source: "10.0.0.0/8"  # Internal attestation only
# Layer 2: Application Container Security
container:  
  image: "ghcr.io/myorg/secure-ai-api:v1.2.3"  
  image_verification:    
    signature_check: true    
    sbom_scan: true  
  readonly_rootfs: true  
  no_new_privileges: true
# Layer 3: Runtime Security
runtime:  
  seccomp_profile: "runtime/default"  
  apparmor_profile: "docker-default"  
  capabilities_drop:    
    - ALL  
  capabilities_add:    
    - NET_BIND_SERVICE  # Only what's needed
# Layer 4: Application Security
application:  
  environment_variables:    
    - name: LOG_LEVEL      
      value: "INFO"    
    - name: ENABLE_DEBUG      
      value: "false"  
  secrets:    
    - name: API_KEY      
      source: "phala-kms"      
      encrypted: true    
    - name: MODEL_ENCRYPTION_KEY      
      source: "phala-kms"      
      encrypted: true
# Layer 5: Monitoring and Attestation
security:  
  attestation:    
    continuous_verification: true    
    interval_seconds: 300    
    alert_on_failure: true  
  audit_logging:    
    enabled: true    
    destination: "syslog://secure-log-collector:514"    
    pii_redaction: true

# secure_ai_service.py
import os
import ssl
import httpx
from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer
import hashlib
import hmac

app = FastAPI()
security = HTTPBearer()

# Configuration from environment (set securely via Phala Cloud)
API_SECRET = os.environ["API_SECRET"]  # Retrieved from KMS
TRUSTED_CERT_HASH = os.environ["TRUSTED_CERT_HASH"]

# Create SSL context for mTLS
def create_ssl_context():
    """Create SSL context with certificate pinning"""
    context = ssl.create_default_context()
    # Load client certificate (for mutual TLS)
    context.load_cert_chain(
        certfile="/app/certs/client.crt",
        keyfile="/app/certs/client.key"
    )
    # Enable certificate verification
    context.verify_mode = ssl.CERT_REQUIRED
    context.check_hostname = True
    return context

# Verify request authenticity
def verify_request_signature(authorization: str = Depends(security)):
    """Verify HMAC signature on requests"""
    try:
        token = authorization.credentials
        # Verify HMAC signature (implementation depends on your protocol)
        # This prevents tampering even within TEE network
        expected = hmac.new(
            API_SECRET.encode(),
            msg=token.encode(),
            digestmod=hashlib.sha256
        ).hexdigest()
        if not hmac.compare_digest(token[:64], expected):
            raise HTTPException(status_code=401, detail="Invalid signature")
    except Exception as e:
        raise HTTPException(status_code=401, detail="Authentication failed")

# Example: Secure model inference endpoint
@app.post("/v1/inference")
async def secure_inference(
    request: dict,
    auth: None = Depends(verify_request_signature)
):
    """Inference endpoint with security controls"""
    # Input validation (see Input Validation section)
    if not validate_input(request):
        raise HTTPE
...

#!/bin/bash
# dependency-audit.sh - Run before every deployment
set -e
echo "=== Dependency Security Audit ==="
# 1. Check for known vulnerabilities
echo "Scanning for known vulnerabilities..."
pip-audit --desc
# 2. Verify package hashes
echo "Verifying package integrity..."
pip install --require-hashes -r requirements.txt
# 3. Check for outdated packages
echo "Checking for outdated packages..."
pip list --outdated
# 4. Analyze dependency tree
echo "Analyzing dependency tree..."
pipdeptree --warn fail
# 5. License compliance check
echo "Checking license compliance..."
pip-licenses --summary
echo "=== Audit Complete ==="

# requirements.txt - Pin versions and verify hashes
# Generated with: pip-compile --generate-hashes requirements.in
fastapi==0.104.1 \
    --hash=sha256:fc6a8a70c2ad4e0f3f3b7b2b6f8e4a3a3b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7
uvicorn[standard]==0.24.0 \
    --hash=sha256:a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0c1d2e3f4a5b6c7d8e9f0a1b2
pydantic==2.5.0 \
    --hash=sha256:b1c2d3e4f5a6b7c8d9e0f1a2b3c4d5e6f7a8b9a0b1c2d3e4f5a6b7c8d9e0f1a2
# Cryptography with verified hash
cryptography==41.0.7 \
    --hash=sha256:c1d2e3f4a5b6c7d8e9f0a1b2c3d4e5f6a7b8c9a0b1c2d3e4f5a6b7c8d9e0f1a2
# No transitive dependencies beyond what's listed
# Total: 4 direct dependencies only

# secure_validators.py
from pydantic import BaseModel, Field, validator
import re

class SecureInferenceRequest(BaseModel):
    """Secure input validation for AI inference requests"""
    prompt: str = Field(
        ...,
        min_length=1,
        max_length=4096,  # Prevent memory exhaustion
        description="User prompt for AI model"
    )
    max_tokens: int = Field(
        default=512,
        ge=1,
        le=2048,  # Prevent resource exhaustion
        description="Maximum tokens to generate"
    )
    temperature: float = Field(
        default=0.7,
        ge=0.0,
        le=2.0,  # Reasonable bounds
        description="Sampling temperature"
    )
    model: str = Field(
        default="deepseek-chat",
        pattern=r"^[a-z0-9\-]+$",  # Alphanumeric and hyphens only
        description="Model identifier"
    )

    @validator('model')
    def validate_model(cls, v):
        """Only allow whitelisted models"""
        allowed_models = {
            "deepseek-chat",
            "llama-3-70b",
            "qwen-2.5-72b",
            "gpt-oss-35"
        }
        if v not in allowed_models:
            raise ValueError(f"Model {v} not allowed. Choose from: {allowed_models}")
        return v

    @validator('prompt')
    def sanitize_prompt(cls, v):
        """Sanitize prompt to prevent injection attacks"""
        v = v.replace('\x00', '')
        sql_patterns = [
            r"(\b(SELECT|INSERT|UPDATE|DELETE|DROP|CREATE|ALTER)\b)",
            r"(--|\#|\/\*|\*\/)",
            r"(\bOR\b.*=.*|1=1|' OR ')"
        ]
        for pattern in sql_patterns:
            if re.search(pattern, v, re.IGNORECASE):
                raise ValueError("Potentially malicious SQL pattern detected")
        cmd_patterns = [r"[;&|`$]", r"\$\(", r"\$\{"]
        for pattern in cmd_patterns:
            if re.search(pattern, v):
                raise ValueError("Potential
...

# secure_secrets.py
import os
from dstack_sdk import DstackClient
from functools import lru_cache
import hashlib

class SecretManager:
    """Secure secret management using Phala Cloud KMS"""
    def __init__(self):
        self.client = DstackClient()
        self._verify_tee_environment()

    def _verify_tee_environment(self) -> bool:
        """Verify we're running in genuine TEE"""
        tee_indicators = [
            os.path.exists("/dev/tdx_guest"),
            os.path.exists("/dev/sev"),
            os.path.exists("/var/run/dstack.sock")
        ]
        if not any(tee_indicators):
            raise RuntimeError("Must run in TEE environment")
        return True

    @lru_cache(maxsize=128)
    def get_key(self, key_path: str) -> bytes:
        """Get deterministic key from Dstack KMS

        Args:
            key_path: Unique identifier like 'my-app/encryption/v1'

        Returns:
            32-byte deterministic key
        """
        try:
            result = self.client.get_key(key_path)
            key_bytes = result.decode_key()  # 32 bytes
            return key_bytes
        except Exception as e:
            print(f"Failed to retrieve key: {type(e).__name__}")
            raise RuntimeError(f"Key retrieval failed for path: {key_path}")

    def derive_secret_key(self, purpose: str, version: str = "v1") -> bytes:
        """Derive application-specific secret key

        Args:
            purpose: Key purpose like 'api-auth', 'db-encryption'
            version: Key version for rotation

        Returns:
            32-byte secret key
        """
        key_path = f"app-secrets/{purpose}/{version}"
        return self.get_key(key_path)

# Global secret manager instance
secrets = SecretManager()

# Usage in application
def get_encryption_key():
    """Get encryption key for sensitive data"""
    # Keys are deterministic - same path always return
...

# phala-secrets.yaml - Secret configuration for Phala Cloud
app_name: secure-ai-api
# Secrets managed by Phala Cloud KMS
secrets:  
  - name: API_KEY    
    description: "External API authentication key"    
    rotation_days: 90  
  - name: API_SECRET    
    description: "HMAC secret for request signing"    
    rotation_days: 90  
  - name: MODEL_ENCRYPTION_KEY    
    description: "AES-256 key for model weight encryption"    
    rotation_days: 180  
  - name: DB_PASSWORD    
    description: "PostgreSQL password"    
    rotation_days: 30  
  - name: OAUTH_CLIENT_SECRET    
    description: "OAuth 2.0 client secret"    
    rotation_days: 60
# Non-secret configuration (can be env vars)
config:  
  LOG_LEVEL: "INFO"  
  MAX_WORKERS: "4"  
  REQUEST_TIMEOUT: "30"  
  MODEL_NAME: "deepseek-chat"

# secure_handlers.py
import re
from typing import Any, Dict
import html

class SecurityValidator:
    """Comprehensive input validation and output sanitization"""
    INJECTION_PATTERNS = {
        'sql': [
            r"(\bSELECT\b.*\bFROM\b)",
            r"(\bINSERT\b.*\bINTO\b)",
            r"(\bUPDATE\b.*\bSET\b)",
            r"(\bDELETE\b.*\bFROM\b)",
            r"(\bDROP\b.*\bTABLE\b)",
            r"(--|\#|\/\*)",
            r"('.*OR.*'.*=.*')"
        ],
        'command': [
            r"[;&|`]",
            r"\$\(",
            r"\$\{",
            r">\s*/",
            r"<\s*script"
        ],
        'path_traversal': [
            r"\.\./",
            r"\.\.\\",
            r"/etc/passwd",
            r"/proc/",
            r"C:\\Windows"
        ]
    }

    @staticmethod
    def validate_input(data: Any, context: str = "general") -> bool:
        """Validate input data for security threats"""
        if isinstance(data, str):
            return SecurityValidator._validate_string(data, context)
        elif isinstance(data, dict):
            return all(
                SecurityValidator.validate_input(v, context)
                for v in data.values()
            )
        elif isinstance(data, list):
            return all(
                SecurityValidator.validate_input(item, context)
                for item in data
            )
        return True

    @staticmethod
    def _validate_string(value: str, context: str) -> bool:
        """Validate string for injection patterns"""
        if len(value) > 100000:  # 100KB max
            raise ValueError("Input exceeds maximum length")
        if '\x00' in value:
            raise ValueError("Null bytes not allowed")
        for category, patterns in SecurityValidator.INJECTION_PATTERNS.items():
            for pattern in patterns:
                if re.search(pattern, value, re.IGNORECASE):
    
...

# continuous_attestation.py
import httpx
import asyncio
from datetime import datetime, timedelta
from typing import Optional

class ContinuousAttestationService:
    """Continuously verify TEE attestation and report to Trust Center"""
    def __init__(
        self,
        app_id: str,
        trust_center_url: str = "https://trust-center.phala.network",
        attestation_interval: int = 300  # 5 minutes
    ):
        self.app_id = app_id
        self.trust_center_url = trust_center_url
        self.attestation_interval = attestation_interval
        self.last_attestation: Optional[datetime] = None
        self.attestation_hash: Optional[str] = None

    async def start(self):
        """Start continuous attestation loop"""
        print(f"Starting continuous attestation (interval: {self.attestation_interval}s)")
        while True:
            try:
                await self.perform_attestation()
                await asyncio.sleep(self.attestation_interval)
            except Exception as e:
                print(f"Attestation failed: {e}")
                await self.alert_attestation_failure(str(e))
                await asyncio.sleep(60)  # Retry after 1 minute

    async def perform_attestation(self):
        """Perform single attestation cycle"""
        report = await self.generate_attestation_report()
        response = await self.submit_to_trust_center(report)
        self.last_attestation = datetime.utcnow()
        self.attestation_hash = response['attestation_hash']
        print(f"Attestation successful: {self.attestation_hash[:16]}...")

    async def generate_attestation_report(self) -> dict:
        """Generate TEE attestation report via Dstack SDK"""
        from dstack_sdk import DstackClient

        try:
            client = DstackClient()
            # Get attestation quote from TEE
            quote_result = client.get_quote(report_data=b"")

...

# Security Checklist for Confidential AI Deployment

## Code Security
- [ ] All dependencies audited (pip-audit, npm audit)
- [ ] Package hashes verified in requirements.txt
- [ ] No secrets in source code or environment files
- [ ] Input validation on all user inputs
- [ ] Output sanitization implemented
- [ ] Error messages don't leak sensitive info
- [ ] Logging redacts PII and secrets

## Configuration Security
- [ ] Running as non-root user
- [ ] Read-only root filesystem
- [ ] Minimal capabilities (no unnecessary privileges)
- [ ] Resource limits configured (CPU, memory, disk)
- [ ] Network policies restrict unnecessary connections
- [ ] Firewall rules follow least-privilege

## Secret Management
- [ ] All secrets stored in Phala Cloud KMS
- [ ] No secrets in environment variables (production)
- [ ] Secret rotation policy configured
- [ ] Secrets never logged or printed
- [ ] Secrets loaded from KMS at runtime only

## TEE Security
- [ ] Continuous attestation enabled (5-minute intervals)
- [ ] Attestation reports published to Trust Center
- [ ] Alert system configured for attestation failures
- [ ] Code hash documented for client verification
- [ ] TEE type and version documented

## Application Security
- [ ] mTLS configured for external connections
- [ ] HTTPS enforced (no HTTP)
- [ ] Request authentication implemented
- [ ] Rate limiting configured
- [ ] CORS policy configured correctly
- [ ] SQL injection defenses (if using database)
- [ ] XSS defenses (if rendering HTML)

## Monitoring and Incident Response
- [ ] Structured logging with security events
- [ ] Log aggregation configured
- [ ] Alert system tested
- [ ] Incident response playbook documented
- [ ] Security contact information current
- [ ] Backup and recovery tested

## Compliance (if applicable)
- [ ] GDPR compliance verified
- [ ] HIPAA compliance verified
- [ ] SOC 2 controls documented
- [ 
...

# security_tests.py
import pytest
import httpx

@pytest.mark.asyncio
class TestSecurityControls:
    """Automated security tests for confidential AI service"""

    @pytest.fixture
    def base_url(self):
        return "https://my-ai-service.phala.cloud"

    async def test_https_enforced(self, base_url):
        """Test that HTTP is redirected to HTTPS"""
        http_url = base_url.replace("https://", "http://")
        async with httpx.AsyncClient(follow_redirects=False) as client:
            response = await client.get(http_url)
            assert response.status_code == 301
            assert response.headers['location'].startswith('https://')

    async def test_authentication_required(self, base_url):
        """Test that endpoints require authentication"""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{base_url}/v1/chat",
                json={"message": "test"}
            )
            assert response.status_code == 401

    async def test_sql_injection_blocked(self, base_url, auth_headers):
        """Test SQL injection prevention"""
        malicious_inputs = [
            "' OR '1'='1",
            "1; DROP TABLE users--",
            "' UNION SELECT * FROM secrets--"
        ]
        for payload in malicious_inputs:
            async with httpx.AsyncClient() as client:
                response = await client.post(
                    f"{base_url}/v1/chat",
                    json={"message": payload},
                    headers=auth_headers
                )
                assert response.status_code == 400

    async def test_xss_prevention(self, base_url, auth_headers):
        """Test XSS attack prevention"""
        xss_payloads = [
            "<script>alert('xss')</script>",
            "<img src=x onerror=alert('xss')>",
            "javascript:alert('xss')"
        ]
        fo
...

# timing_protection.py
import time
import random

def constant_time_compare(a: str, b: str) -> bool:
    """Constant-time string comparison to prevent timing attacks"""
    if len(a) != len(b):
        return False
    result = 0
    for x, y in zip(a, b):
        result |= ord(x) ^ ord(y)
    return result == 0

def add_timing_jitter(min_ms: int = 10, max_ms: int = 50):
    """Add random delay to prevent timing analysis"""
    delay = random.uniform(min_ms / 1000, max_ms / 1000)
    time.sleep(delay)

# Use in authentication
def verify_api_key(provided_key: str, expected_key: str) -> bool:
    """Verify API key with timing attack protection"""
    is_valid = constant_time_compare(provided_key, expected_key)
    add_timing_jitter(10, 50)
    return is_valid

# memory_safety.py
import ctypes

def secure_zero_memory(data: bytearray):
    """Securely zero memory before deallocation"""
    length = len(data)
    ctypes.memset(id(data), 0, length)

def secure_string_handling(secret: str) -> bytearray:
    """Handle sensitive strings with secure memory management"""
    secure_data = bytearray(secret.encode('utf-8'))
    return secure_data

# Example usage with API keys
def process_api_key(api_key: str):
    """Process API key with secure memory handling"""
    secure_key = secure_string_handling(api_key)
    try:
        result = authenticate_with_key(bytes(secure_key))
        return result
    finally:
        secure_zero_memory(secure_key)

#!/bin/bash
# continuous_dependency_scanning.sh
# Run daily as cron job or CI/CD pipeline
set -e
echo "=== Daily Dependency Security Scan ==="
pip install --upgrade pip-audit
echo "Scanning for vulnerabilities..."
pip-audit --desc --format json > audit-report.json
CRITICAL=$(jq '[.vulnerabilities[] | select(.severity=="CRITICAL")] | length' audit-report.json)
HIGH=$(jq '[.vulnerabilities[] | select(.severity=="HIGH")] | length' audit-report.json)
echo "Critical vulnerabilities: $CRITICAL"
echo "High vulnerabilities: $HIGH"
if [ "$CRITICAL" -gt 0 ]; then
    echo "ALERT: Critical vulnerabilities detected!"
    curl -X POST "$ALERT_WEBHOOK" \
        -H "Content-Type: application/json" \
        -d "{\"severity\":\"CRITICAL\",\"message\":\"$CRITICAL critical vulnerabilities detected\",\"report\":\"$(cat audit-report.json)\"}"
    exit 1
fi
if [ "$HIGH" -gt 0 ]; then
    echo "WARNING: High severity vulnerabilities detected"
fi
echo "=== Scan Complete ==="

# secure-deployment-template.yaml
# Drop-in template for secure Phala Cloud deployments
app_name: ${YOUR_APP_NAME}
tee_type: intel-tdx  # or amd-sev, nvidia-h100
# Container security
container:  
  image: ${YOUR_REGISTRY}/${YOUR_IMAGE}:${VERSION}  
  image_verification: true  
  readonly_rootfs: true  
  user: 1000  # Non-root
# Resource limits (prevent DoS)
resources:  
  cpu_limit: "4"  
  memory_limit: "8Gi"  
  storage_limit: "20Gi"
# Network security
network:  
  isolation: true  
  allow_egress:    
    - "api.openai.com:443"  # Whitelist only needed endpoints  
  firewall_rules:    
    - port: 443      
      protocol: tcp
# Secret management
secrets:  
  - name: API_KEY    
    source: phala-kms  
  - name: MODEL_KEY    
    source: phala-kms
# Security monitoring
security:  
  attestation:    
    enabled: true    
    interval: 300  # 5 minutes  
  audit_logging:    
    enabled: true    
    redact_pii: true  
  alerts:    
    webhook: ${ALERT_WEBHOOK_URL}