企业级 AI Agent 私有化部署跨网络安全域（DMZ / 核心生产区 / GPU算力集群）架构实战：应对严格防火墙阻断、mTLS 双向认证与流式代理中继

# 禁用响应缓冲，防止 SSE 切片被聚集积压
proxy_buffering off;
proxy_cache off;
# 强制关闭 TCP 延迟发送算法，确保小包实时推入网络套接字
tcp_nodelay on;
# 将读超时时间延长至大模型最长容忍单步推理时间
proxy_read_timeout 600s;
proxy_send_timeout 600s;
# 开启长连接保活
proxy_http_version 1.1;
proxy_set_header Connection "";

/**
 * 文件名：CrossZoneStreamingRelay.ts
 * 描述：企业级 AI Agent 跨网络安全域流式代理中继器核心实现
 * 职责：负责在严格网络隔离环境下，安全中继客户端与 GPU 算力区/核心生产区之间的 SSE 长连接通信
 * 特性：mTLS 双向认证、动态心跳注入保活、背压流控、请求头安全清洗与审计归档
 */

import http from 'node:http';
import https from 'node:https';
import fs from 'node:fs';
import { Readable, Transform } from 'node:stream';

/**
 * 跨安全域通信配置接口
 */
export interface CrossZoneProxyConfig {
  /** 监听端口 */
  listenPort: number;
  /** 上游目标服务地址（如专有 GPU 算力集群推理服务） */
  upstreamUrl: string;
  /** 内部专有 CA 根证书路径 */
  caCertPath: string;
  /** 本代理服务端的证书与私钥（用于面对下游 mTLS） */
  serverCertPath: string;
  serverKeyPath: string;
  /** 本代理访问上游时所使用的客户端证书与私钥 */
  clientCertPath: string;
  clientKeyPath: string;
  /** 心跳注入间隔时间（毫秒，默认 10000ms） */
  heartbeatIntervalMs?: number;
  /** 单次推理最大允许超时时间（毫秒，默认 300000ms / 5分钟） */
  requestTimeoutMs?: number;
  /** 允许通行的请求头白名单 */
  allowedHeaders?: string[];
}

/**
 * 结构化跨区审计日志记录接口
 */
export interface CrossZoneAuditEntry {
  traceId: string;
  timestamp: string;
  sourceIp: string;
  clientCertSubject: string;
  targetEndpoint: string;
  durationMs: number;
  bytesRelayed: number;
  statusCode: number;
  heartbeatsInjected: number;
}

export class CrossZoneStreamingRelay {
  private config: Required<CrossZoneProxyConfig>;
  private server: https.Server | null = null;
  private httpsAgent: https.Agent;

  constructor(config: CrossZoneProxyConfig) {
    this.config = {
      listenPort: config.listenPort,
      upstreamUrl: config.upstreamUrl,
      caCertPath: config.caCertPath,
      serverCertPath: config.serverCertPath,
      serverKeyPath: config.serverKeyPath,
      clientCertPath: config.clientCertPath,
      clientKeyPath: config.clientKeyPath,
      heartbeatIntervalMs: config.heartbeatIntervalMs ?? 10000,
      requestTimeoutMs: config.requestTimeoutMs ?? 300000,
      allowedHeaders: config.allowedHeaders ?? [
        'content-type',
        'accept',
        'authorization',
        'x-request-id',
        'x-tenant-id',
        'x-bxagent-trace-id'
      ]
    };

    // 初始化访问上游 GPU 专区的专属高可靠 mTLS HTTPS Agent（启用连接池复用与长连接）
    this.httpsAgent = new https.Agent({
      keepAlive: true,
      maxSockets: 256,
      maxFreeSockets: 64,
      keepAliveMsecs: 60000,
      ca: fs.readFileSync(this.config.caCertPath),
      cert: fs.readFileSync(this.config.clientCertPath),
      key: fs.readFileSync(this.config.clientKeyPath),
      rejectUnauthorized: true // 严格校验上游服务端证书完整性
    });
  }

  /**
   * 启动代理中继服务
   */
  public start(): Promise<void> {
    return new Promise((resolve, reject) => {
      const serverOptions: https.ServerOptions = {
        ca: fs.readFileSync(this.config.caCertPath),
        cert: fs.readFileSync(this.config.serverCertPath),
        key: fs.readFileSync(this.config.serverKeyPath),
        requestCert: true, // 强制要求下游出示客户端证书（mTLS 双向鉴权）
        rejectUnauthorized: true // 拒绝任何未通过内部专用 CA 签名的非授权请求
      };

      this.server = https.createServer(serverOptions, (req, res) => {
        this.handleIncomingRequest(req, res);
      });

      this.server.on('error', (err) => {
        console.error('【跨区代理中继器】服务器发生底层套接字异常：', err);
        reject(err);
      });

      this.server.listen(this.config.listenPort, () => {
        console.log(`【跨区代理中继器】已成功在端口 ${this.config.listenPort} 启动，启用严格 mTLS 隔离防护。`);
        resolve();
      });
    });
  }

  /**
   * 核心请求处理中继逻辑
   */
  private handleIncomingRequest(clientReq: http.IncomingMessage, clientRes: http.ServerResponse): void {
    const startTime = Date.now();
    const traceId = (clientReq.headers['x-bxagent-trace-id'] as string) || `trace-${Date.now()}-${Math.random().toString(36).substring(2, 9)}`;
    const clientCert = (clientReq.socket as any).getPeerCertificate?.();
    const clientSubject = clientCert ? clientCert.subject?.CN || 'UnknownClient' : 'Unauthenticated';

    let totalBytesRelayed = 0;
    let heartbeatsInjected = 0;

    // 1. 请求头清洗与白名单重构：剔除所有未授权的私有调试头与内部主机名
    const sanitizedHeaders: http.OutgoingHttpHeaders = {
      host: new URL(this.config.upstreamUrl).host,
      'x-bxagent-trace-id': traceId,
      'x-bxagent-source-zone': 'DMZ_GATEWAY',
      'x-bxagent-client-cn': clientSubject
    };

    for (const key of Object.keys(clientReq.headers)) {
      const lowerKey = key.toLowerCase();
      if (this.config.allowedHeaders.includes(lowerKey)) {
        sanitizedHeaders[lowerKey] = clientReq.headers[key];
      }
    }

    const upstreamUrlObj = new URL(clientReq.url || '/', this.config.upstreamUrl);

    // 2. 构造面向内部 GPU 算力集群的上游转发请求
    const upstreamReq = https.request(
      upstreamUrlObj,
      {
        method: clientReq.method,
        agent: this.httpsAgent,
        headers: sanitizedHeaders,
        timeout: this.config.requestTimeoutMs
      },
      (upstreamRes) => {
        // 3. 校验上游响应类型是否为 SSE 流式数据
        const contentType = upstreamRes.headers['content-type'] || '';
        const isSseStream = contentType.includes('text/event-stream');

        // 下发给客户端标准响应头，强制禁止任何中间层缓冲
        clientRes.writeHead(upstreamRes.statusCode || 200, {
          'Content-Type': isSseStream ? 'text/event-stream; charset=utf-8' : contentType,
          'Cache-Control': 'no-cache, no-transform',
          'Connection': 'keep-alive',
          'X-Accel-Buffering': 'no', // 强制告知 Nginx/Edge 代理绝不缓冲
          'X-Bxagent-Trace-Id': traceId
        });

        // 4. 定时心跳注入定时器：在智能体深度思考或耗时工具调用时向客户端推送保活注释包
        let heartbeatTimer: NodeJS.Timeout | null = null;

        const resetHeartbeat = () => {
          if (heartbeatTimer) clearInterval(heartbeatTimer);
          if (isSseStream && !clientRes.writableEnded) {
            heartbeatTimer = setInterval(() => {
              if (!clientRes.writableEnded) {
                // 写入合规 SSE 注释，防火墙刷新 TCP 连接，客户端自动忽略
                clientRes.write(': bxagent-keepalive-heartbeat\n\n');
                heartbeatsInjected++;
              }
            }, this.config.heartbeatIntervalMs);
          }
        };

        resetHeartbeat();

        // 5. 背压感知流传输管理（Backpressure Pipeline）
        upstreamRes.on('data', (chunk: Buffer) => {
          totalBytesRelayed += chunk.length;
          resetHeartbeat(); // 收到真实数据块，重置心跳倒计时

          const canContinue = clientRes.write(chunk);
          if (!canContinue) {
            // 下游客户端网络拥塞或读取过慢，暂停拉取上游数据，防止代理内存积压膨胀
            upstreamRes.pause();
          }
        });

        clientRes.on('drain', () => {
          // 下游缓冲区已排空，恢复拉取上游数据
          upstreamRes.resume();
        });

        upstreamRes.on('end', () => {
          if (heartbeatTimer) clearInterval(heartbeatTimer);
          clientRes.end();
          this.logAudit({
            traceId,
            timestamp: new Date().toISOString(),
            sourceIp: clientReq.socket.remoteAddress || '127.0.0.1',
            clientCertSubject: clientSubject,
            targetEndpoint: upstreamUrlObj.pathname,
            durationMs: Date.now() - startTime,
            bytesRelayed: totalBytesRelayed,
            statusCode: upstreamRes.statusCode || 200,
            heartbeatsInjected
          });
        });

        upstreamRes.on('error', (err) => {
          if (heartbeatTimer) clearInterval(heartbeatTimer);
          console.error(`【跨区代理中继器】上游流式传输中断 [TraceId: ${traceId}]：`, err);
          if (!clientRes.headersSent) {
            clientRes.writeHead(502, { 'Content-Type': 'application/json' });
            clientRes.end(JSON.stringify({ error: 'UPSTREAM_STREAM_FAULT', message: '上游算力集群连接异常中断' }));
          } else {
            clientRes.end();
          }
        });
      }
    );

    // 6. 异常与超时治理
    upstreamReq.on('timeout', () => {
      upstreamReq.destroy(new Error('UPSTREAM_REQUEST_TIMEOUT'));
    });

    upstreamReq.on('error', (err) => {
      console.error(`【跨区代理中继器】转发上游请求失败 [TraceId: ${traceId}]：`, err);
      if (!clientRes.headersSent) {
        clientRes.writeHead(504, { 'Content-Type': 'application/json' });
        clientRes.end(JSON.stringify({ error: 'CROSS_ZONE_TIMEOUT', message: '跨安全域网关通信超时或被防火墙拦截' }));
      } else {
        clientRes.end();
      }
    });

    // 将下游客户端的请求体（如用户 Prompt 和工具调用指令）以管道方式送入上游
    clientReq.pipe(upstreamReq);

    // 监听客户端主动中断事件（例如用户在前端点击了“停止生成”按钮）
    clientReq.on('close', () => {
      if (!upstreamReq.destroyed) {
        upstreamReq.destroy(); // 级联中止上游 GPU 的无谓推理计算，节省企业昂贵的 GPU 算力
      }
    });
  }

  /**
   * 记录不可篡改的安全审计日志
   */
  private logAudit(entry: CrossZoneAuditEntry): void {
    // 在生产环境中推入异步高吞吐日志队列或写入本地受保护的审计文件
    const logLine = `[AUDIT-LOG] [${entry.timestamp}] Trace:${entry.traceId} Client:${entry.clientCertSubject} ` +
      `Target:${entry.targetEndpoint} Status:${entry.statusCode} Duration:${entry.durationMs}ms ` +
      `Bytes:${entry.bytesRelayed} Heartbeats:${entry.heartbeatsInjected}`;
    console.log(logLine);
  }

  /**
   * 优雅停止服务
   */
  public stop(): Promise<void> {
    return new Promise((resolve) => {
      this.httpsAgent.destroy();
      if (this.server) {
        this.server.close(() => {
          console.log('【跨区代理中继器】已安全优雅停止。');
          resolve();
        });
      } else {
        resolve();
      }
    });
  }
}