Perimeter Security (Pre-2020)
    │
    ├── Firewalls
    ├── VPNs
    └── "Inside vs. Outside"
    │
    ▼
Zero Trust (2020-2025)
    │
    ├── "Never trust, always verify"
    ├── Identity-based access
    ├── Micro-segmentation
    └── Continuous validation
    │
    ▼
Agent-Aware Security (2026+)
    │
    ├── Intent verification
    ├── Reasoning transparency
    ├── Tool governance
    ├── Action boundaries
    ├── Behavioral baselines
    └── Human-in-the-loop for escalation

# docker-compose.security-hardened.yml
version: '3.8'

services:
  n8n:
    image: n8nio/n8n:latest
    environment:
      # Security Hardening
      - N8N_SECURE_COOKIE=true
      - N8N_SECURITY_AUDIT_DAYS=30
      - N8N_BLOCK_ENV_ACCESS_IN_NODE=true
      - N8N_DEFAULT_BINARY_DATA_MODE=filesystem
      - N8N_DEFAULT_BINARY_DATA_FILESYSTEM_DIRECTORY_PATH=/data
      
      # Sandboxed Code Execution
      - N8N_CODE_SANDBOXED=true
      - N8N_CODE_EXECUTION_TIMEOUT=300000  # 5 minutes
      - N8N_CODE_MEMORY_LIMIT=256  # MB
      
      # Execution
      - EXECUTIONS_MODE=queue
      - EXECUTIONS_TIMEOUT=3600
      - EXECUTIONS_TIMEOUT_MAX=7200
      
      # Redis for secure queue management
      - QUEUE_BULL_REDIS_HOST=redis
      - QUEUE_BULL_REDIS_PORT=6379
      
      # Encryption
      - N8N_ENCRYPTION_KEY=${N8N_ENCRYPTION_KEY}
      
    volumes:
      - n8n_data:/home/node/.n8n
      - /tmp/n8n-binary-data:/data:rw,noexec,nosuid,nodev
      
    security_opt:
      - no-new-privileges:true
      
    cap_drop:
      - ALL
      
    cap_add:
      - CHOWN
      - SETGID
      - SETUID
      
    read_only: true
    tmpfs:
      - /tmp:noexec,nosuid,size=100m
      
    networks:
      - n8n-secure

  redis:
    image: redis:7-alpine
    command: redis-server --requirepass ${REDIS_PASSWORD}
    volumes:
      - redis_data:/data
    networks:
      - n8n-secure

volumes:
  n8n_data:
  redis_data:

networks:
  n8n-secure:
    internal: true

// Input Validation Node
const validateInput = (input) => {
  const rules = {
    email: {
      pattern: /^[^\s@]+@[^\s@]+\.[^\s@]+$/,
      maxLength: 254,
      sanitize: (val) => val.toLowerCase().trim()
    },
    phone: {
      pattern: /^\+?[\d\s-]{10,20}$/,
      maxLength: 20,
      sanitize: (val) => val.replace(/\s/g, '')
    },
    message: {
      maxLength: 5000,
      sanitize: (val) => {
        // Remove potential XSS vectors
        return val
          .replace(/<script[^>]*>.*?<\/script>/gi, '')
          .replace(/<[^>]+>/g, '')  // Remove HTML tags
          .replace(/javascript:/gi, '')
          .replace(/on\w+\s*=/gi, '');  // Remove event handlers
      }
    }
  };

  const validated = {};
  const errors = [];

  for (const [field, rule] of Object.entries(rules)) {
    if (input[field]) {
      // Check length
      if (rule.maxLength && input[field].length > rule.maxLength) {
        errors.push(`${field}: exceeds maximum length`);
        continue;
      }

      // Check pattern
      if (rule.pattern && !rule.pattern.test(input[field])) {
        errors.push(`${field}: invalid format`);
        continue;
      }

      // Sanitize
      validated[field] = rule.sanitize ? rule.sanitize(input[field]) : input[field];
    }
  }

  if (errors.length > 0) {
    return { valid: false, errors };
  }

  return { valid: true, data: validated };
};

const result = validateInput($input.first().json);

if (!result.valid) {
  return [{
    json: {
      error: 'Validation failed',
      details: result.errors,
      statusCode: 400
    }
  }];
}

return [{
  json: result.data
}];

// Rate Limiting Node (requires Redis)
const Redis = require('ioredis');

const redis = new Redis({
  host: process.env.REDIS_HOST,
  password: process.env.REDIS_PASSWORD,
  db: 1  // Separate DB for rate limiting
});

const rateLimit = async (identifier, rules) => {
  const now = Date.now();
  const key = `ratelimit:${identifier}`;
  
  // Check multiple time windows
  const windows = [
    { name: 'minute', ttl: 60 },
    { name: 'hour', ttl: 3600 },
    { name: 'day', ttl: 86400 }
  ];
  
  const results = {};
  
  for (const window of windows) {
    const windowKey = `${key}:${window.name}`;
    const current = await redis.incr(windowKey);
    
    if (current === 1) {
      await redis.expire(windowKey, window.ttl);
    }
    
    const limit = rules[window.name];
    results[window.name] = {
      current,
      limit,
      remaining: Math.max(0, limit - current),
      exceeded: current > limit
    };
  }
  
  // Check if any limit exceeded
  const exceeded = Object.values(results).some(r => r.exceeded);
  
  return {
    allowed: !exceeded,
    limits: results,
    retryAfter: exceeded ? 
      Math.min(...Object.values(results)
        .filter(r => r.exceeded)
        .map(r => r.reset)) : 0
  };
};

// Apply rate limiting
const identifier = $input.first().json.headers['x-forwarded-for'] || 
                   $input.first().json.headers['x-real-ip'] ||
                   'unknown';

const rules = {
  minute: 60,    // 60 requests per minute
  hour: 1000,    // 1000 requests per hour
  day: 10000     // 10000 requests per day
};

const limitResult = await rateLimit(identifier, rules);

if (!limitResult.allowed) {
  return [{
    json: {
      error: 'Rate limit exceeded',
      limits: limitResult.limits,
      retryAfter: limitResult.retryAfter,
      statusCode: 429
    }
  }];
}

return [{
  json: {
    allowed: true,
    remaining: limitResult.limits.minute.remaining
  }
}];

// Data Classification and Masking
const classifyAndMask = (data) => {
  const patterns = {
    ssn: {
      pattern: /\b\d{3}-\d{2}-\d{4}\b/g,
      mask: (match) => `XXX-XX-${match.slice(-4)}`,
      classification: 'PII_HIGH'
    },
    creditCard: {
      pattern: /\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b/g,
      mask: (match) => `XXXX-XXXX-XXXX-${match.slice(-4)}`,
      classification: 'PCI_DSS'
    },
    email: {
      pattern: /\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b/g,
      mask: (match) => {
        const [local, domain] = match.split('@');
        return `${local[0]}***@${domain}`;
      },
      classification: 'PII_MEDIUM'
    },
    apiKey: {
      pattern: /\b(sk-[a-zA-Z0-9]{48}|Bearer\s+[a-zA-Z0-9_-]+)\b/gi,
      mask: () => '[REDACTED]',
      classification: 'CREDENTIAL'
    }
  };

  let maskedData = JSON.stringify(data);
  const classifications = new Set();

  for (const [type, config] of Object.entries(patterns)) {
    const matches = maskedData.match(config.pattern);
    if (matches) {
      classifications.add(config.classification);
      maskedData = maskedData.replace(config.pattern, config.mask);
    }
  }

  return {
    original: data,
    masked: JSON.parse(maskedData),
    classifications: Array.from(classifications),
    requiresEncryption: classifications.has('PII_HIGH') || 
                       classifications.has('PCI_DSS')
  };
};

const input = $input.first().json;
const result = classifyAndMask(input);

// Log classification (without sensitive data)
console.log('Data classifications:', result.classifications);

// Return masked data for downstream processing
return [{
  json: {
    ...result.masked,
    _securityMeta: {
      classifications: result.classifications,
      requiresEncryption: result.requiresEncryption
    }
  }
}];

// Prompt Injection Detection and Prevention
const detectPromptInjection = (input) => {
  // Known injection patterns
  const injectionPatterns = [
    /ignore\s+(?:previous|above|all)\s+instructions?/i,
    /disregard\s+(?:the|your|all)\s+(?:instructions?|system)/i,
    /system\s*:\s*/i,
    /\[system\s*:\s*/i,
    /you\s+are\s+now\s+/i,
    /from\s+now\s+on\s+you\s+are/i,
    /act\s+as\s+/i,
    /pretend\s+to\s+be/i,
    /DAN\s*\(|Do\s+Anything\s+Now/i,
    /jailbreak|prompt\s+leak/i,
    /\{\{[^}]+\}\}/g,  // Template injection
    /<%[^%]+%>/g       // ERB/template injection
  ];

  const suspiciousKeywords = [
    'ignore', 'disregard', 'bypass', 'override',
    'secret', 'password', 'credential', 'token',
    'hidden', 'admin', 'root', 'sudo'
  ];

  const detections = [];

  // Check injection patterns
  for (const pattern of injectionPatterns) {
    if (pattern.test(input)) {
      detections.push({
        type: 'INJECTION_PATTERN',
        pattern: pattern.toString(),
        severity: 'HIGH'
      });
    }
  }

  // Check keyword density
  const keywordCount = suspiciousKeywords.filter(kw => 
    input.toLowerCase().includes(kw)
  ).length;

  if (keywordCount >= 3) {
    detections.push({
      type: 'SUSPICIOUS_KEYWORDS',
      count: keywordCount,
      keywords: suspiciousKeywords.filter(kw => 
        input.toLowerCase().includes(kw)
      ),
      severity: 'MEDIUM'
    });
  }

  // Check for delimiter abuse
  const delimiterCount = (input.match(/["'`<>{}\[\]]/g) || []).length;
  const inputLength = input.length;
  const delimiterRatio = delimiterCount / inputLength;

  if (delimiterRatio > 0.3) {
    detections.push({
      type: 'DELIMITER_ABUSE',
      ratio: delimiterRatio,
      severity: 'MEDIUM'
    });
  }

  return {
    isSafe: detections.length === 0,
    riskScore: detections.reduce((sum, d) => {
      return sum + (d.severity === 'HIGH' ? 10 : 5);
    }, 0),
    detections
  };
};

const userInput = $input.first().json.prompt || $input.first().json.message;
const scanResult = detectPromptInjection(userInput);

if (!scanResult.isSafe) {
  // Log security event
  await $httpRequest({
    method: 'POST',
    url: process.env.SECURITY_WEBHOOK_URL,
    body: {
      event: 'PROMPT_INJECTION_DETECTED',
      timestamp: new Date().toISOString(),
      riskScore: scanResult.riskScore,
      detections: scanResult.detections,
      source: $input.first().json.source
    }
  });

  return [{
    json: {
      error: 'Security violation detected',
      code: 'PROMPT_INJECTION',
      statusCode: 400
    }
  }];
}

return [{
  json: {
    sanitized: userInput,
    securityScan: 'PASSED'
  }
}];

// Tool Permission Governance
class ToolGovernance {
  constructor(agentProfile) {
    this.agentProfile = agentProfile;
    this.auditLog = [];
  }

  async canExecute(toolName, parameters, context) {
    const permission = this.agentProfile.tools[toolName];
    
    if (!permission) {
      return { allowed: false, reason: 'Tool not registered' };
    }

    // Check if tool is enabled
    if (!permission.enabled) {
      return { allowed: false, reason: 'Tool disabled' };
    }

    // Check rate limits
    if (permission.rateLimit) {
      const withinLimit = await this.checkRateLimit(toolName);
      if (!withinLimit) {
        return { allowed: false, reason: 'Rate limit exceeded' };
      }
    }

    // Check parameter constraints
    if (permission.allowedParams) {
      const paramCheck = this.validateParameters(
        parameters, 
        permission.allowedParams
      );
      if (!paramCheck.valid) {
        return { 
          allowed: false, 
          reason: `Invalid parameters: ${paramCheck.errors.join(', ')}` 
        };
      }
    }

    // Check context requirements
    if (permission.requiresContext) {
      for (const req of permission.requiresContext) {
        if (!context[req]) {
          return { 
            allowed: false, 
            reason: `Missing required context: ${req}` 
          };
        }
      }
    }

    // Check approval requirements
    if (permission.requiresApproval) {
      const approvalStatus = await this.requestApproval(
        toolName, 
        parameters,
        permission.requiresApproval
      );
      if (!approvalStatus.granted) {
        return { 
          allowed: false, 
          reason: 'Approval required but not granted',
          approvalStatus 
        };
      }
    }

    // Log approval
    this.auditLog.push({
      timestamp: new Date().toISOString(),
      agent: this.agentProfile.id,
      tool: toolName,
      parameters: this.sanitizeForLog(parameters),
      context: context.requestId,
      decision: 'ALLOWED'
    });

    return { allowed: true };
  }

  validateParameters(params, allowedSchema) {
    const errors = [];
    
    for (const [key, value] of Object.entries(params)) {
      const schema = allowedSchema[key];
      
      if (!schema) {
        errors.push(`Unknown parameter: ${key}`);
        continue;
      }

      if (schema.type && typeof value !== schema.type) {
        errors.push(`${key}: expected ${schema.type}`);
      }

      if (schema.enum && !schema.enum.includes(value)) {
        errors.push(`${key}: must be one of ${schema.enum.join(', ')}`);
      }

      if (schema.maxLength && value.length > schema.maxLength) {
        errors.push(`${key}: exceeds max length`);
      }

      if (schema.pattern && !schema.pattern.test(value)) {
        errors.push(`${key}: invalid format`);
      }
    }

    return { valid: errors.length === 0, errors };
  }
}

// Usage in n8n
const agentProfile = {
  id: 'customer-support-agent',
  role: 'support',
  tools: {
    'database_query': {
      enabled: true,
      rateLimit: { requests: 100, window: '1m' },
      allowedParams: {
        table: { type: 'string', enum: ['tickets', 'customers', 'kb'] },
        action: { type: 'string', enum: ['select', 'count'] },
        where: { type: 'object' },
        limit: { type: 'number', max: 100 }
      },
      requiresContext: ['sessionId', 'userId'],
      requiresApproval: null
    },
    'send_email': {
      enabled: true,
      rateLimit: { requests: 10, window: '1m' },
      allowedParams: {
        to: { type: 'string', pattern: /^[^\s@]+@[^\s@]+\.[^\s@]+$/ },
        template: { type: 'string', enum: ['welcome', 'reset', 'support'] }
      },
      requiresContext: ['userId'],
      requiresApproval: { threshold: 'manager', timeout: 300 }
    },
    'delete_database': {
      enabled: false  // Explicitly disabled
    }
  }
};

const governance = new ToolGovernance(agentProfile);

const toolRequest = $input.first().json;
const context = {
  requestId: $input.first().json.requestId,
  userId: $input.first().json.userId,
  sessionId: $input.first().json.sessionId
};

const authorization = await governance.canExecute(
  toolRequest.toolName,
  toolRequest.parameters,
  context
);

return [{
  json: authorization
}];

┌─────────────────────────────────────────────────────────────────┐
│                    Observability Pipeline                       │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐   │
│  │   Traces     │───▶│   Metrics    │───▶│   Logs       │   │
│  │  (OpenTelemetry)  │ (Prometheus) │    │  (Structured)│   │
│  └──────────────┘    └──────────────┘    └──────────────┘   │
│         │                   │                   │              │
│         └───────────────────┼───────────────────┘              │
│                             │                                  │
│                    ┌────────▼────────┐                        │
│                    │    Langfuse      │                        │
│                    │  (LLM Observability)                     │
│                    └────────┬────────┘                        │
│                             │                                  │
│  ┌──────────────────────────▼──────────────────────────┐      │
│  │                   Analysis & Alerting              │      │
│  │  • Performance Metrics  • Cost Tracking          │      │
│  │  • Quality Scores       • Error Patterns           │      │
│  │  • Latency Analysis     • Behavioral Anomalies    │      │
│  └────────────────────────────────────────────────────┘      │
└─────────────────────────────────────────────────────────────────┘

# docker-compose.langfuse.yml
version: '3.8'

services:
  langfuse:
    image: ghcr.io/langfuse/langfuse:latest
    ports:
      - "3000:3000"
    environment:
      - DATABASE_URL=postgresql://postgres:postgres@postgres:5432/langfuse
      - NEXTAUTH_SECRET=${NEXTAUTH_SECRET}
      - SALT=${SALT}
      - NEXTAUTH_URL=http://localhost:3000
      - LANGFUSE_ENABLE_EXPERIMENTAL_FEATURES=false
    depends_on:
      - postgres
      - redis
    networks:
      - observability

  postgres:
    image: postgres:15-alpine
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=langfuse
    volumes:
      - postgres_data:/var/lib/postgresql/data
    networks:
      - observability

  redis:
    image: redis:7-alpine
    networks:
      - observability

  # Optional: ClickHouse for high-volume event processing
  clickhouse:
    image: clickhouse/clickhouse-server:24
    volumes:
      - clickhouse_data:/var/lib/clickhouse
    networks:
      - observability

volumes:
  postgres_data:
  clickhouse_data:

networks:
  observability:

// Langfuse Integration Node for n8n
const { Langfuse } = require('langfuse');

const langfuse = new Langfuse({
  publicKey: process.env.LANGFUSE_PUBLIC_KEY,
  secretKey: process.env.LANGFUSE_SECRET_KEY,
  baseUrl: process.env.LANGFUSE_BASE_URL || 'http://langfuse:3000'
});

// Trace wrapper for AI agent workflows
const createTracedAgent = async (agentConfig) => {
  const trace = langfuse.trace({
    name: agentConfig.name,
    userId: agentConfig.userId,
    sessionId: agentConfig.sessionId,
    metadata: {
      workflow: agentConfig.workflow,
      version: agentConfig.version
    }
  });

  return {
    trace,
    
    // Wrap LLM calls
    async llmCall(model, messages, options = {}) {
      const generation = trace.generation({
        name: 'llm-completion',
        model,
        input: messages,
        modelParameters: {
          temperature: options.temperature,
          maxTokens: options.maxTokens,
          topP: options.topP
        }
      });

      const startTime = Date.now();
      
      try {
        // Execute actual LLM call
        const response = await $httpRequest({
          method: 'POST',
          url: 'https://api.openai.com/v1/chat/completions',
          body: {
            model,
            messages,
            ...options
          },
          headers: {
            'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`
          }
        });

        const latency = Date.now() - startTime;
        const tokens = response.usage;

        generation.end({
          output: response.choices[0].message,
          usage: {
            input: tokens.prompt_tokens,
            output: tokens.completion_tokens,
            total: tokens.total_tokens
          },
          metadata: {
            latency_ms: latency,
            finish_reason: response.choices[0].finish_reason
          }
        });

        return response;
      } catch (error) {
        generation.end({
          output: { error: error.message },
          level: 'ERROR',
          statusMessage: error.message
        });
        throw error;
      }
    },

    // Wrap tool executions
    async toolExecution(toolName, input) {
      const span = trace.span({
        name: `tool-${toolName}`,
        input
      });

      try {
        const result = await executeTool(toolName, input);
        
        span.end({
          output: result,
          metadata: {
            tool: toolName,
            execution_time_ms: Date.now() - span.startTime
          }
        });

        return result;
      } catch (error) {
        span.end({
          output: { error: error.message },
          level: 'ERROR'
        });
        throw error;
      }
    },

    // Score the trace
    async score(name, value, comment) {
      await langfuse.score({
        traceId: trace.id,
        name,
        value,
        comment
      });
    },

    // Finalize trace
    async finalize(output, level = 'DEFAULT') {
      trace.update({
        output,
        level
      });
      await langfuse.flush();
    }
  };
};

// Example usage in workflow
const agent = await createTracedAgent({
  name: 'customer-support-agent',
  userId: $input.first().json.customer_id,
  sessionId: $input.first().json.session_id,
  workflow: 'support-ticket-resolution',
  version: '2.3.1'
});

try {
  // LLM call with automatic tracing
  const llmResponse = await agent.llmCall('gpt-4o', [
    { role: 'system', content: 'You are a helpful support agent.' },
    { role: 'user', content: $input.first().json.message }
  ]);

  // Tool execution with tracing
  const ticketInfo = await agent.toolExecution('get_ticket', {
    ticket_id: $input.first().json.ticket_id
  });

  // Score the result
  await agent.score('helpfulness', 0.9, 'Resolved on first response');

  // Finalize
  await agent.finalize({
    response: llmResponse.choices[0].message.content,
    ticket_updated: true
  });

  return [{
    json: {
      response: llmResponse.choices[0].message.content,
      trace_id: agent.trace.id
    }
  }];
} catch (error) {
  await agent.finalize({ error: error.message }, 'ERROR');
  throw error;
}

// Comprehensive Observability Wrapper
class AIObservability {
  constructor(config) {
    this.langfuse = new Langfuse(config.langfuse);
    this.metrics = [];
    this.costTracking = new Map();
  }

  async observeWorkflow(workflowFn, context) {
    const trace = this.langfuse.trace({
      id: context.traceId,
      name: context.workflowName,
      userId: context.userId,
      metadata: {
        input: this.sanitizeForLogging(context.input),
        startTime: new Date().toISOString()
      }
    });

    const startTime = Date.now();
    const tokenUsage = { input: 0, output: 0 };
    const toolCalls = [];

    try {
      // Execute workflow with instrumentation
      const result = await workflowFn({
        trace,
        logLLM: async (model, input, output, usage) => {
          tokenUsage.input += usage.input;
          tokenUsage.output += usage.output;
          
          trace.generation({
            name: `llm-${model}`,
            model,
            input,
            output,
            usage
          });
        },
        
        logTool: async (tool, input, output, duration) => {
          toolCalls.push({ tool, duration });
          
          trace.span({
            name: `tool-${tool}`,
            input,
            output,
            metadata: { duration_ms: duration }
          });
        },
        
        logEvent: (name, metadata) => {
          trace.event({ name, metadata });
        }
      });

      // Calculate costs
      const cost = this.calculateCost(tokenUsage);
      const duration = Date.now() - startTime;

      // Update trace with final metrics
      trace.update({
        output: this.sanitizeForLogging(result),
        metadata: {
          duration_ms: duration,
          token_usage: tokenUsage,
          cost_usd: cost,
          tool_calls: toolCalls.length
        }
      });

      // Send metrics
      await this.sendMetrics({
        workflow: context.workflowName,
        duration,
        cost,
        tokenUsage,
        toolCalls: toolCalls.length,
        status: 'success'
      });

      return result;

    } catch (error) {
      trace.update({
        level: 'ERROR',
        output: { error: error.message },
        metadata: {
          error_type: error.name,
          error_stack: error.stack
        }
      });

      await this.sendMetrics({
        workflow: context.workflowName,
        status: 'error',
        error: error.message
      });

      throw error;
    }
  }

  calculateCost(tokenUsage) {
    const pricing = {
      'gpt-4o': { input: 2.50, output: 10.00 },  // per 1M tokens
      'gpt-4o-mini': { input: 0.15, output: 0.60 },
      'claude-3-opus': { input: 15.00, output: 75.00 },
      'claude-3-sonnet': { input: 3.00, output: 15.00 }
    };

    const model = 'gpt-4o'; // Determine from context
    const rates = pricing[model] || pricing['gpt-4o'];
    
    return (
      (tokenUsage.input / 1000000) * rates.input +
      (tokenUsage.output / 1000000) * rates.output
    );
  }

  sanitizeForLogging(data) {
    // Remove PII and sensitive data
    const sensitive = ['password', 'token', 'secret', 'key', 'credit_card'];
    const sanitized = JSON.parse(JSON.stringify(data));
    
    const redact = (obj) => {
      for (const key in obj) {
        if (sensitive.some(s => key.toLowerCase().includes(s))) {
          obj[key] = '[REDACTED]';
        } else if (typeof obj[key] === 'object' && obj[key] !== null) {
          redact(obj[key]);
        }
      }
    };
    
    redact(sanitized);
    return sanitized;
  }

  async sendMetrics(metrics) {
    // Send to Prometheus, Datadog, etc.
    await $httpRequest({
      method: 'POST',
      url: process.env.METRICS_ENDPOINT,
      body: metrics
    });
  }
}

// Usage
const observability = new AIObservability({
  langfuse: {
    publicKey: process.env.LANGFUSE_PUBLIC_KEY,
    secretKey: process.env.LANGFUSE_SECRET_KEY,
    baseUrl: process.env.LANGFUSE_URL
  }
});

const result = await observability.observeWorkflow(
  async ({ trace, logLLM, logTool }) => {
    // Your workflow logic here
    const response = await processWithLLM();
    await logLLM('gpt-4o', input, response.content, response.usage);
    
    const data = await fetchFromAPI();
    await logTool('api_fetch', { url }, data, 150);
    
    return response;
  },
  {
    workflowName: 'customer-support',
    userId: 'user-123',
    traceId: generateUUID(),
    input: $input.first().json
  }
);

// Anomaly Detection for AI Workflows
const analyzeWorkflow = async (traceData) => {
  const alerts = [];

  // Cost anomaly detection
  const avgCost = await getAverageCost(traceData.workflow, '24h');
  if (traceData.cost > avgCost * 3) {
    alerts.push({
      severity: 'HIGH',
      type: 'COST_ANOMALY',
      message: `Cost $${traceData.cost.toFixed(4)} exceeds 3x average ($${avgCost.toFixed(4)})`,
      traceId: traceData.traceId
    });
  }

  // Latency anomaly
  const avgLatency = await getAverageLatency(traceData.workflow, '24h');
  if (traceData.duration > avgLatency * 5) {
    alerts.push({
      severity: 'MEDIUM',
      type: 'LATENCY_ANOMALY',
      message: `Duration ${traceData.duration}ms exceeds 5x average (${avgLatency}ms)`,
      traceId: traceData.traceId
    });
  }

  // Token usage anomaly
  const avgTokens = await getAverageTokens(traceData.workflow, '24h');
  const totalTokens = traceData.tokenUsage.input + traceData.tokenUsage.output;
  if (totalTokens > avgTokens * 4) {
    alerts.push({
      severity: 'MEDIUM',
      type: 'TOKEN_ANOMALY',
      message: `Token usage ${totalTokens} exceeds 4x average (${avgTokens})`,
      traceId: traceData.traceId
    });
  }

  // Error rate check
  const recentErrors = await getRecentErrors(traceData.workflow, '1h');
  if (recentErrors.count > 10) {
    alerts.push({
      severity: 'CRITICAL',
      type: 'ERROR_SPIKE',
      message: `${recentErrors.count} errors in last hour`,
      traceId: traceData.traceId
    });
  }

  // Send alerts
  for (const alert of alerts) {
    await $httpRequest({
      method: 'POST',
      url: process.env.ALERT_WEBHOOK_URL,
      body: alert
    });
  }

  return alerts;
};

// MCP Server Governance
class MCPGovernance {
  constructor(config) {
    this.allowedServers = config.allowedServers || [];
    this.serverMetadata = new Map();
    this.auditLog = [];
  }

  async verifyServer(serverUrl) {
    // Check allowlist
    const allowed = this.allowedServers.find(s => 
      serverUrl.startsWith(s.url)
    );
    
    if (!allowed) {
      throw new Error(`MCP server not allowlisted: ${serverUrl}`);
    }

    // Verify server identity
    const challenge = crypto.randomBytes(32).toString('hex');
    
    const response = await $httpRequest({
      method: 'POST',
      url: `${serverUrl}/verify`,
      body: { challenge }
    });

    if (!this.verifyChallengeResponse(challenge, response)) {
      throw new Error('MCP server verification failed');
    }

    // Fetch and validate tool definitions
    const tools = await $httpRequest({
      method: 'GET',
      url: `${serverUrl}/tools`
    });

    const validatedTools = await this.validateToolDefinitions(tools);

    this.serverMetadata.set(serverUrl, {
      allowed: true,
      verified: true,
      tools: validatedTools,
      permissions: allowed.permissions,
      lastVerified: new Date().toISOString()
    });

    return validatedTools;
  }

  validateToolDefinitions(tools) {
    return tools.map(tool => {
      // Validate schema
      if (!tool.name || !tool.description || !tool.inputSchema) {
        throw new Error(`Invalid tool definition: ${tool.name || 'unnamed'}`);
      }

      // Check for dangerous patterns
      const dangerousPatterns = [
        /exec\s*\(/i,
        /eval\s*\(/i,
        /system\s*\(/i,
        /child_process/i
      ];

      const toolString = JSON.stringify(tool);
      for (const pattern of dangerousPatterns) {
        if (pattern.test(toolString)) {
          throw new Error(`Dangerous pattern detected in tool: ${tool.name}`);
        }
      }

      return {
        ...tool,
        riskLevel: this.assessRiskLevel(tool)
      };
    });
  }

  assessRiskLevel(tool) {
    const highRiskKeywords = [
      'delete', 'remove', 'drop', 'truncate',
      'execute', 'run', 'exec', 'eval',
      'write', 'modify', 'update'
    ];

    const toolString = JSON.stringify(tool).toLowerCase();
    const riskCount = highRiskKeywords.filter(kw => 
      toolString.includes(kw)
    ).length;

    if (riskCount >= 3) return 'HIGH';
    if (riskCount >= 1) return 'MEDIUM';
    return 'LOW';
  }

  async executeTool(serverUrl, toolName, parameters, context) {
    const server = this.serverMetadata.get(serverUrl);
    
    if (!server || !server.verified) {
      throw new Error('Server not verified');
    }

    const tool = server.tools.find(t => t.name === toolName);
    
    if (!tool) {
      throw new Error(`Tool not found: ${toolName}`);
    }

    // Check permissions
    const requiredPerm = `tool:${toolName}`;
    if (!server.permissions.includes(requiredPerm) && 
        !server.permissions.includes('tool:*')) {
      throw new Error(`Permission denied: ${requiredPerm}`);
    }

    // Additional approval for high-risk tools
    if (tool.riskLevel === 'HIGH') {
      const approved = await this.requestHumanApproval({
        server: serverUrl,
        tool: toolName,
        parameters,
        context
      });
      
      if (!approved) {
        throw new Error('High-risk tool execution rejected');
      }
    }

    // Execute with timeout and logging
    const startTime = Date.now();
    
    try {
      const result = await Promise.race([
        $httpRequest({
          method: 'POST',
          url: `${serverUrl}/tools/${toolName}`,
          body: parameters,
          headers: {
            'X-Request-Context': JSON.stringify(context)
          }
        }),
        new Promise((_, reject) => 
          setTimeout(() => reject(new Error('Timeout')), 30000)
        )
      ]);

      this.auditLog.push({
        timestamp: new Date().toISOString(),
        action: 'TOOL_EXECUTE',
        server: serverUrl,
        tool: toolName,
        parameters: this.sanitizeForLog(parameters),
        duration_ms: Date.now() - startTime,
        success: true
      });

      return result;

    } catch (error) {
      this.auditLog.push({
        timestamp: new Date().toISOString(),
        action: 'TOOL_EXECUTE',
        server: serverUrl,
        tool: toolName,
        error: error.message,
        success: false
      });
      throw error;
    }
  }
}

// Usage in n8n
const governance = new MCPGovernance({
  allowedServers: [
    {
      url: 'https://mcp.company.com',
      permissions: ['tool:read_*', 'tool:search_*'],
      requireApproval: ['tool:write_*', 'tool:delete_*']
    },
    {
      url: 'https://mcp.salesforce.com',
      permissions: ['tool:crm_*'],
      requireApproval: []
    }
  ]
});

// Verify and use MCP server
const tools = await governance.verifyServer('https://mcp.company.com');
const result = await governance.executeTool(
  'https://mcp.company.com',
  'search_documents',
  { query: $input.first().json.searchQuery },
  { userId: $input.first().json.user_id, requestId: generateUUID() }
);

// Software Bill of Materials (SBOM) Validation
const validateMCPServerSBOM = async (serverUrl) => {
  // Fetch SBOM
  const sbom = await $httpRequest({
    method: 'GET',
    url: `${serverUrl}/.well-known/sbom.json`
  });

  // Check for known vulnerabilities
  const vulnerabilities = [];
  
  for (const dependency of sbom.dependencies) {
    // Query vulnerability database
    const vulns = await $httpRequest({
      method: 'GET',
      url: `https://api.osv.dev/v1/query`,
      qs: {
        package: {
          name: dependency.name,
          ecosystem: dependency.ecosystem
        },
        version: dependency.version
      }
    });

    if (vulns.vulns && vulns.vulns.length > 0) {
      vulnerabilities.push({
        package: dependency.name,
        version: dependency.version,
        vulnerabilities: vulns.vulns
      });
    }
  }

  // Check signatures
  const signatureValid = await verifySBOMSignature(sbom);

  return {
    valid: vulnerabilities.length === 0 && signatureValid,
    vulnerabilities,
    signatureValid,
    sbom
  };
};

# .github/workflows/secure-n8n-deploy.yml
name: Secure n8n Deployment

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  security-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      # Secret detection
      - name: Detect secrets
        uses: trufflesecurity/trufflehog@main
        with:
          path: ./
          extra_args: --only-verified

      # Dependency vulnerability scan
      - name: Scan dependencies
        uses: aquasecurity/trivy-action@master
        with:
          scan-type: 'fs'
          scan-ref: '.'

      # Workflow security analysis
      - name: Analyze workflow security
        run: |
          npx n8n-workflow-security-scan \
            --input workflows/ \
            --rules security-rules.json

  test:
    runs-on: ubuntu-latest
    needs: security-scan
    steps:
      - uses: actions/checkout@v4

      # Run unit tests
      - name: Run tests
        run: npm test

      # Integration tests with mocked credentials
      - name: Integration tests
        run: npm run test:integration
        env:
          N8N_ENCRYPTION_KEY: test-key

  deploy:
    runs-on: ubuntu-latest
    needs: [security-scan, test]
    if: github.ref == 'refs/heads/main'
    environment: production
    steps:
      - uses: actions/checkout@v4

      # Deploy to staging first
      - name: Deploy to staging
        run: |
          n8n-deploy \
            --target staging \
            --workflows workflows/ \
            --credentials encrypted-credentials.json

      # Smoke tests
      - name: Staging smoke tests
        run: npm run test:smoke:staging

      # Production deployment
      - name: Deploy to production
        run: |
          n8n-deploy \
            --target production \
            --workflows workflows/ \
            --credentials encrypted-credentials.json \
            --canary 10%

      # Production smoke tests
      - name: Production smoke tests
        run: npm run test:smoke:production

      # Full rollout or rollback
      - name: Complete rollout
        if: success()
        run: n8n-deploy --complete --target production

      - name: Rollback on failure
        if: failure()
        run: n8n-deploy --rollback --target production

// Automated Backup System
const backupWorkflows = async () => {
  const timestamp = new Date().toISOString().replace(/[:.]/g, '-');
  const backupPath = `backups/n8n-${timestamp}`;

  // Export all workflows
  const workflows = await $httpRequest({
    method: 'GET',
    url: `${process.env.N8N_BASE_URL}/api/v1/workflows`,
    headers: {
      'X-N8N-API-KEY': process.env.N8N_API_KEY
    }
  });

  // Encrypt and store
  for (const workflow of workflows.data) {
    const encrypted = await encryptData(
      JSON.stringify(workflow),
      process.env.BACKUP_ENCRYPTION_KEY
    );

    await $httpRequest({
      method: 'PUT',
      url: `${process.env.S3_BACKUP_URL}/${backupPath}/workflows/${workflow.id}.enc`,
      body: encrypted,
      headers: {
        'Content-Type': 'application/octet-stream'
      }
    });
  }

  // Backup credentials (encrypted separately)
  const credentials = await $httpRequest({
    method: 'GET',
    url: `${process.env.N8N_BASE_URL}/api/v1/credentials`,
    headers: {
      'X-N8N-API-KEY': process.env.N8N_API_KEY
    }
  });

  for (const credential of credentials.data) {
    const encrypted = await encryptData(
      JSON.stringify(credential),
      process.env.BACKUP_ENCRYPTION_KEY
    );

    await $httpRequest({
      method: 'PUT',
      url: `${process.env.S3_BACKUP_URL}/${backupPath}/credentials/${credential.id}.enc`,
      body: encrypted
    });
  }

  // Cleanup old backups
  await cleanupOldBackups(30); // Keep 30 days

  return {
    backupPath,
    workflowsBackedUp: workflows.data.length,
    credentialsBackedUp: credentials.data.length,
    timestamp: new Date().toISOString()
  };
};

// Scheduled backup trigger
const schedule = require('node-schedule');

// Daily at 2 AM
schedule.scheduleJob('0 2 * * *', async () => {
  const result = await backupWorkflows();
  console.log('Backup completed:', result);
});