构建基于动态令牌与 DynamoDB 的零信任 WebSocket 安全网关

长连接的 WebSocket 协议，在本质上与现代云原生架构推崇的无状态、短生命周期原则存在冲突。一个典型的 WebSocket 认证流程——在连接建立时验证一次凭证，然后信任该 TCP 连接直至其断开——在零信任安全模型下是极度脆弱的。一旦初始令牌泄露或用户权限需要被实时撤销，已建立的连接将成为一个难以管理的攻击向量。问题核心在于：如何在持久化的连接上实现动态、可即时撤销的访问控制？

传统的 JWT (JSON Web Tokens) 方案在这里显得力不从心。JWT 的无状态特性使其一旦签发，在有效期内便无法作废。即使我们通过非常短的有效期来缓解，也无法解决在有效期内实时撤销权限的需求，例如管理员强制用户下线或实时变更其数据访问范围。这种场景要求我们必须引入一个中心化的状态存储来追踪每一个活跃的会话。

本文将探讨一种架构决策：放弃纯粹的无状态模型，转而构建一个有状态、支持动态授权的 WebSocket 安全网关。此方案的核心权衡在于，牺牲一定的系统简洁性，以换取对长连接生命周期内每一刻的绝对安全控制。我们将使用 Nginx 作为边缘接入点，执行连接初期的认证卸载；使用 DynamoDB 作为高性能、高可用的会话存储，实现会话的实时校验与权限变更。

方案对比与架构选型

在真实项目中，我们面临两种主流的设计路径。

方案 A: 增强型无状态 JWT 方案

这个方案尝试在不引入重度状态存储的情况下增强安全性。其核心思路是使用生命周期极短的 JWT（例如60秒），并配合一个 Refresh Token。客户端在 WebSocket 连接建立后，需要通过带外（out-of-band）的 HTTP 请求，使用 Refresh Token 定期刷新 JWT，并通过 WebSocket PING/PONG 或自定义消息将新的 JWT 发送给服务端进行“续命”。

• 优势:

  • WebSocket 服务本身保持无状态，易于水平扩展。
  • 架构相对简单，对现有系统的侵入性较低。

• 劣势:

  • 无法即时撤销: 即使 JWT 只有60秒有效期，这仍然意味着存在长达60秒的攻击窗口。对于金融交易或敏感数据协作等场景，这是不可接受的。
  • 复杂的客户端逻辑: 客户端需要维护一个定时器来刷新令牌，并处理刷新失败的逻辑，增加了客户端的复杂度和不确定性。
  • 权限变更延迟: 如果用户权限发生变更，需要等待当前 JWT 过期后，新签发的 JWT 才能体现，存在明显的延迟。
  • 撤销风暴: 如果需要撤销 Refresh Token，通常依赖一个黑名单。在高并发场景下，查询黑名单本身就会成为新的性能瓶颈，使系统向“有状态”靠拢。

方案 B: 基于 DynamoDB 的中心化会话存储方案

该方案明确接受“有状态”是实现动态授权的必要代价。它将认证与会话管理完全分离，通过一个中心化的、高性能的 K-V 存储来维护每个连接的实时状态。

• 优势:

  • 即时撤销与权限变更: 通过直接修改或删除 DynamoDB 中的会话记录，可以瞬间终止任何 WebSocket 连接的合法性或更新其权限，真正做到零延迟。
  • 服务端逻辑简化: WebSocket 服务本身只需查询会话存储，逻辑清晰。连接的生命周期管理由外部机制（如 DynamoDB TTL）和认证服务负责。
  • 符合零信任原则: “永不信任，始终验证”。每一次关键操作都可以（并且应该）重新查询 DynamoDB 以获取最新的权限状态。

• 劣势:

  • 引入新的依赖: 整个系统强依赖于 DynamoDB 的可用性和性能。
  • 增加操作延迟: 每次消息处理都可能需要一次对 DynamoDB 的网络请求，这会给消息处理链路增加毫秒级的延迟。
  • 成本考量: DynamoDB 的读写操作会产生费用，在高频消息场景下需要进行成本评估。

最终决策

考虑到项目的核心需求是金融级别的安全性和权限控制的实时性，我们最终选择了方案 B。虽然它引入了额外的复杂度和延迟，但其提供的即时撤销能力是方案 A 无法比拟的。我们认为，对于高安全要求的应用，这种可预测的、可控的延迟是换取安全性的合理代价。DynamoDB 的高可用性、可预测的低延迟以及 TTL 自动清理过期会话的功能，使其成为这个架构中理想的会话存储组件。

架构实现概览

整个系统的交互流程被设计为多个独立的阶段，由 Nginx 作为流量编排的核心。

sequenceDiagram
    participant Client
    participant Nginx
    participant Auth_Service
    participant WebSocket_Service
    participant DynamoDB

    Client->>+Nginx: 1. 发起 HTTPS 请求 /ws/ticket (携带认证凭证)
    Nginx->>+Auth_Service: 2. auth_request 转发请求
    Auth_Service->>+DynamoDB: 3. 生成 Session, 写入会话数据 (含权限, TTL)
    DynamoDB-->>-Auth_Service: 4. 写入成功
    Auth_Service-->>-Nginx: 5. 返回 200 OK, 并在 Header 中携带一次性 Ticket
    Nginx-->>-Client: 6. 返回一次性 Ticket
    Client->>+Nginx: 7. 发起 WebSocket 升级请求 /ws/connect?ticket=...
    Nginx->>+WebSocket_Service: 8. 代理 WebSocket 请求 (携带 Ticket)
    WebSocket_Service->>+DynamoDB: 9. 验证 Ticket, 消耗 Ticket, 激活会话
    DynamoDB-->>-WebSocket_Service: 10. 验证通过
    WebSocket_Service-->>-Nginx: 11. 101 Switching Protocols
    Nginx-->>-Client: 12. 101 Switching Protocols (连接建立)
    
    loop 消息处理循环
        Client->>Nginx: 13. 发送 WebSocket 消息 (携带操作指令)
        Nginx->>WebSocket_Service: 14. 转发消息
        WebSocket_Service->>DynamoDB: 15. [关键] 检查 Session 有效性与操作权限
        DynamoDB-->>WebSocket_Service: 16. 返回当前权限
        alt 权限足够
            WebSocket_Service->>WebSocket_Service: 17. 处理业务逻辑
            WebSocket_Service->>Nginx: 18. 发送响应消息
            Nginx->>Client: 19. 转发响应
        else 权限不足或会话失效
            WebSocket_Service->>Nginx: 20. 发送错误信息并关闭连接
            Nginx->>Client: 21. 关闭连接
        end
    end

这个流程将认证、票据获取和实际连接建立的过程完全分离，确保了 WebSocket 服务本身不处理任何长期的用户凭证。

核心组件实现

1. Nginx: 边缘安全与流量调度

Nginx 在此架构中扮演两个关键角色：TLS 终端和认证前置代理。所有外部流量必须通过 Nginx，确保了内部服务的网络隔离。

# /etc/nginx/nginx.conf

# 定义上游服务
upstream auth_service {
    server 127.0.0.1:8081; # 认证服务
}

upstream websocket_service {
    # 使用 IP Hash 确保来自同一客户端的连接尽可能路由到同一实例
    # 虽然我们的设计是无状态的，但在某些场景下这有助于缓存局部性
    ip_hash;
    server ws_backend_1:8080;
    server ws_backend_2:8080;
}

server {
    listen 443 ssl http2;
    server_name your.domain.com;

    # TLS 配置 (生产环境请使用更强的配置)
    ssl_certificate /etc/nginx/certs/fullchain.pem;
    ssl_certificate_key /etc/nginx/certs/privkey.pem;
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers 'ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256';
    ssl_prefer_server_ciphers on;

    # 日志记录，包含认证服务的响应，便于排错
    log_format main_ext '$remote_addr - $remote_user [$time_local] "$request" '
                      '$status $body_bytes_sent "$http_referer" '
                      '"$http_user_agent" "$http_x_forwarded_for" '
                      'auth_status=$auth_status auth_response_time=$auth_response_time';
    access_log /var/log/nginx/access.log main_ext;

    # 位置 1: 获取一次性 Ticket 的 HTTP 端点
    location = /ws/ticket {
        # 核心：使用 auth_request 模块将认证决策委托给 auth_service
        auth_request /_validate_auth;
        
        # 将认证服务返回的 Header (包含 Ticket) 暴露给客户端
        auth_request_set $auth_ticket $upstream_http_x_ws_ticket;
        add_header 'X-WS-Ticket' $auth_ticket;

        # 认证通过后，直接返回 204 No Content，Ticket 在 Header 中
        return 204;
    }

    # 内部 location，用于执行 auth_request
    location = /_validate_auth {
        internal; # 只能被内部请求调用

        proxy_pass http://auth_service;
        proxy_pass_request_body off; # 我们只关心 Header 中的凭证
        proxy_set_header Content-Length "";
        proxy_set_header X-Original-URI $request_uri;
        # 将原始请求的 Authorization Header 传递给认证服务
        proxy_set_header Authorization $http_authorization; 
    }

    # 位置 2: WebSocket 连接端点
    location = /ws/connect {
        # 将请求代理到 WebSocket 服务集群
        proxy_pass http://websocket_service;

        # WebSocket 代理的必要头信息
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "Upgrade";
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # 设置超时，防止连接无限期挂起
        proxy_connect_timeout 7d;
        proxy_send_timeout 7d;
        proxy_read_timeout 7d;
    }
}

这段配置的关键在于 auth_request 模块。它将 /ws/ticket 的请求拦截，并向 /_validate_auth 发起一个子请求。如果认证服务返回 2xx 状态码，Nginx 认为认证成功并继续处理；否则，它将直接返回认证服务的错误码（例如 401 或 403）。这种方式将认证逻辑完全从 Nginx 和业务服务中解耦。

2. Auth Service & DynamoDB Session Schema

认证服务是一个简单的 HTTP 服务，它的职责是验证用户凭证（例如，通过 Authorization 头传递的长期 token），如果合法，则在 DynamoDB 中创建一个会话记录，并返回一个一次性的、短时效的 Ticket。

我们使用 Go 语言和 AWS SDK 来实现这个服务。

package main

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue"
	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
	"github.comcom/aws/aws-sdk-go-v2/service/dynamodb/types"
	"github.com/google/uuid"
)

const (
	tableName     = "WebSocketSessions"
	ticketTTL     = 60  // Ticket 有效期 60 秒
	sessionTTL    = 24 * 60 * 60 // 会话最长有效期 24 小时
)

// Session represents the data stored in DynamoDB for a WebSocket session.
type Session struct {
	SessionID   string              `dynamodbav:"SessionID"` // Partition Key
	UserID      string              `dynamodbav:"UserID"`
	Permissions map[string]bool     `dynamodbav:"Permissions"`
	Ticket      string              `dynamodbav:"Ticket"`
	Status      string              `dynamodbav:"Status"` // PENDING, ACTIVE, EXPIRED
	CreatedAt   int64               `dynamodbav:"CreatedAt"`
	ExpiresAt   int64               `dynamodbav:"ExpiresAt"` // DynamoDB TTL attribute
	TicketExpiresAt int64           `dynamodbav:"TicketExpiresAt"`
}

var dynamoClient *dynamodb.Client

func main() {
	// 初始化 AWS SDK 配置
	cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithRegion("us-east-1"))
	if err != nil {
		log.Fatalf("unable to load SDK config, %v", err)
	}
	dynamoClient = dynamodb.NewFromConfig(cfg)

	http.HandleFunc("/", handleAuthRequest)
	log.Println("Auth service listening on :8081")
	log.Fatal(http.ListenAndServe(":8081", nil))
}

func handleAuthRequest(w http.ResponseWriter, r *http.Request) {
	// 生产级代码需要一个健壮的认证函数
	// 这里我们简化为直接从 Header 获取 UserID
	userID := validateLongLivedToken(r.Header.Get("Authorization"))
	if userID == "" {
		http.Error(w, "Unauthorized", http.StatusUnauthorized)
		return
	}

	// 生成会话和一次性 Ticket
	sessionID := uuid.New().String()
	ticket := uuid.New().String()
	now := time.Now().Unix()

	session := Session{
		SessionID:   sessionID,
		UserID:      userID,
		Permissions: loadInitialPermissions(userID), // 从数据库或其他服务加载初始权限
		Ticket:      ticket,
		Status:      "PENDING",
		CreatedAt:   now,
		ExpiresAt:   now + sessionTTL,
		TicketExpiresAt: now + ticketTTL,
	}

	// 将会话数据写入 DynamoDB
	av, err := attributevalue.MarshalMap(session)
	if err != nil {
		log.Printf("Failed to marshal session: %v", err)
		http.Error(w, "Internal Server Error", http.StatusInternalServerError)
		return
	}

	_, err = dynamoClient.PutItem(context.TODO(), &dynamodb.PutItemInput{
		TableName: aws.String(tableName),
		Item:      av,
	})
	if err != nil {
		log.Printf("Failed to put item to DynamoDB: %v", err)
		http.Error(w, "Internal Server Error", http.StatusInternalServerError)
		return
	}

	// 在响应头中设置 Ticket
	w.Header().Set("X-WS-Ticket", ticket)
	w.WriteHeader(http.StatusOK)
}

// 伪代码: 验证长期凭证并返回用户ID
func validateLongLivedToken(token string) string {
	// 在真实项目中, 这里会解析 JWT 或查询数据库来验证 token
	if token == "Bearer valid-user-token" {
		return "user-123"
	}
	return ""
}

// 伪代码: 加载用户的初始权限
func loadInitialPermissions(userID string) map[string]bool {
	// 这里可以从权限中心、用户数据库等加载
	return map[string]bool{
		"read:data_stream_A": true,
		"write:control_B":    false,
	}
}

DynamoDB 表设计:

• 表名: WebSocketSessions
• 分区键 (Partition Key): SessionID (String)
• 属性: UserID, Permissions (Map), Ticket (String), Status (String), CreatedAt (Number), ExpiresAt (Number), TicketExpiresAt (Number)
• TTL 设置: 启用 TTL (Time to Live)，并指定 ExpiresAt 属性。这让 DynamoDB 能够自动清理过期的会话记录，避免了手动清理的麻烦。

3. WebSocket Service: 连接处理与实时授权

WebSocket 服务是整个架构的核心业务逻辑承载者。它的主要职责是在连接建立时消耗 Ticket，并在处理关键消息时实时查询 DynamoDB 以验证会话和权限。

package main

import (
	"context"
	"log"
	"net/http"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.comcom/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue"
	"github.comcom/aws/aws-sdk-go-v2/service/dynamodb"
	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
	"github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{
	CheckOrigin: func(r *http.Request) bool {
		// 生产环境应配置严格的来源检查
		return true
	},
}

var dynamoClient *dynamodb.Client

func main() {
	cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithRegion("us-east-1"))
	if err != nil {
		log.Fatalf("unable to load SDK config, %v", err)
	}
	dynamoClient = dynamodb.NewFromConfig(cfg)

	http.HandleFunc("/", handleWebSocket)
	log.Println("WebSocket service listening on :8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

func handleWebSocket(w http.ResponseWriter, r *http.Request) {
	ticket := r.URL.Query().Get("ticket")
	if ticket == "" {
		http.Error(w, "Ticket is required", http.StatusBadRequest)
		return
	}

	// 1. 验证并消耗 Ticket
	session, err := activateSession(ticket)
	if err != nil {
		log.Printf("Failed to activate session with ticket %s: %v", ticket, err)
		http.Error(w, "Invalid or expired ticket", http.StatusForbidden)
		return
	}

	// 2. 升级为 WebSocket 连接
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Printf("Failed to upgrade connection: %v", err)
		return
	}
	defer conn.Close()
	log.Printf("Client connected with SessionID: %s", session.SessionID)

	// 3. 消息处理循环
	for {
		_, message, err := conn.ReadMessage()
		if err != nil {
			log.Printf("Read error for SessionID %s: %v", session.SessionID, err)
			break
		}

		// 解析消息 (假设为 JSON 格式: {"action": "read:data_stream_A", "payload": ...})
		// production code would use a proper struct and json.Unmarshal
		action := parseActionFromMessage(message)

		// 4. [核心] 每次操作都检查权限
		hasPermission, checkErr := checkPermission(session.SessionID, action)
		if checkErr != nil {
			log.Printf("Permission check failed for SessionID %s: %v", session.SessionID, checkErr)
			// 发送错误信息并关闭连接
			conn.WriteMessage(websocket.TextMessage, []byte("error: internal server error"))
			break
		}

		if !hasPermission {
			log.Printf("Permission denied for SessionID %s, action %s", session.SessionID, action)
			conn.WriteMessage(websocket.TextMessage, []byte("error: permission denied"))
			continue // 继续监听，但不处理该消息
		}

		// 5. 处理业务逻辑
		log.Printf("Processing action '%s' for SessionID %s", action, session.SessionID)
		response := []byte("Action processed successfully: " + action)
		conn.WriteMessage(websocket.TextMessage, response)
	}
}

// activateSession 验证一次性 Ticket，并将会话状态更新为 ACTIVE
func activateSession(ticket string) (*Session, error) {
    // 使用 DynamoDB 的查询操作，通过 GSI (Global Secondary Index) 来查找 Ticket
    // 这里为了简化，我们假设可以通过某种方式查询到，实际生产环境需要为 Ticket 创建 GSI
    // 或者，更简单的方法是，让 ticket 就是 sessionID，这需要在认证服务做相应调整。
    // 我们采用一种折中的方式：用 UpdateItem + ConditionExpression 来原子性地验证和消耗 Ticket
    
    // 我们假设 ticket 是可预测的，或者可以从会话ID派生。此处为了演示，直接使用一个伪查询
    // 真实场景下，你需要在 `WebSocketSessions` 表上创建一个 GSI，索引 `Ticket`。
    // 以下代码使用 UpdateItem 来模拟这个过程，它更加健壮
    
    //
    // ... Find sessionID by ticket using a GSI ...
    // Let's assume we found the sessionID for the given ticket.
    //
    foundSessionID := "a-session-id-found-via-gsi" // Placeholder

	updateExpression := "SET #status = :active_status REMOVE Ticket, TicketExpiresAt"
	conditionExpression := "attribute_exists(SessionID) AND Ticket = :ticket AND TicketExpiresAt > :now"
	now := time.Now().Unix()

	updateInput := &dynamodb.UpdateItemInput{
		TableName: aws.String("WebSocketSessions"),
		Key: map[string]types.AttributeValue{
			"SessionID": &types.AttributeValueMemberS{Value: foundSessionID},
		},
		UpdateExpression:          aws.String(updateExpression),
		ConditionExpression:       aws.String(conditionExpression),
		ExpressionAttributeNames:  map[string]string{"#status": "Status"},
		ExpressionAttributeValues: map[string]types.AttributeValue{
			":active_status": &types.AttributeValueMemberS{Value: "ACTIVE"},
			":ticket":        &types.AttributeValueMemberS{Value: ticket},
			":now":           &types.AttributeValueMemberN{Value: fmt.Sprintf("%d", now)},
		},
		ReturnValues: types.ReturnValueAllNew,
	}

	result, err := dynamoClient.UpdateItem(context.TODO(), updateInput)
	if err != nil {
		if _, ok := err.(*types.ConditionalCheckFailedException); ok {
			return nil, fmt.Errorf("ticket is invalid, expired, or already used")
		}
		return nil, fmt.Errorf("dynamodb update error: %w", err)
	}

	var session Session
	err = attributevalue.UnmarshalMap(result.Attributes, &session)
	if err != nil {
		return nil, fmt.Errorf("failed to unmarshal session data: %w", err)
	}
	return &session, nil
}

// checkPermission 从 DynamoDB 获取最新的会话状态和权限
func checkPermission(sessionID, requiredPermission string) (bool, error) {
	getInput := &dynamodb.GetItemInput{
		TableName: aws.String("WebSocketSessions"),
		Key: map[string]types.AttributeValue{
			"SessionID": &types.AttributeValueMemberS{Value: sessionID},
		},
		// 只获取需要的字段，节省读取成本
		ProjectionExpression: aws.String("Permissions, #status"),
		ExpressionAttributeNames: map[string]string{"#status": "Status"},
	}

	result, err := dynamoClient.GetItem(context.TODO(), getInput)
	if err != nil {
		return false, fmt.Errorf("dynamodb get error: %w", err)
	}

	if result.Item == nil {
		return false, fmt.Errorf("session not found")
	}

	var partialSession struct {
		Permissions map[string]bool
		Status      string
	}
	if err := attributevalue.UnmarshalMap(result.Item, &partialSession); err != nil {
		return false, fmt.Errorf("unmarshal error: %w", err)
	}

	if partialSession.Status != "ACTIVE" {
		return false, fmt.Errorf("session is not active")
	}

	hasPerm, ok := partialSession.Permissions[requiredPermission]
	return ok && hasPerm, nil
}

这段代码的核心是 checkPermission 函数。它在每次处理消息前，都向 DynamoDB 发起一次 GetItem 请求，以获取最新的权限信息。这确保了任何在 DynamoDB 中的权限变更都能立即在下一次操作中生效。ProjectionExpression 的使用是一个重要的优化，它告诉 DynamoDB 只返回我们关心的字段（Permissions 和 Status），从而减少了读取容量单位（RCU）的消耗和网络传输的数据量。

架构的局限性与未来展望

此架构虽然提供了极高的安全性与灵活性，但也并非没有代价。

首先，性能开销是首要考量。每一次需要授权的消息都会触发一次对 DynamoDB 的网络调用。尽管 DynamoDB 的 P99 延迟通常在个位数毫秒，但在超高频、对延迟极度敏感的场景（如实时游戏的状态同步）下，这累积的延迟可能无法接受。此时，可能需要在 WebSocket 服务内存中引入一个短时（如1-2秒）的权限缓存，但这又是在安全性和性能之间做的另一个权衡。

其次，成本问题。如果每个 WebSocket 消息都需要一次 DynamoDB 读取，那么对于一个拥有大量活跃用户且消息频繁的应用来说，DynamoDB 的读取成本会相当可观。必须仔细评估业务模型，确定哪些操作是“关键操作”需要强校验，哪些可以放宽策略，以平衡安全与成本。

最后，依赖耦合。系统对 DynamoDB 产生了强依赖。尽管 DynamoDB 本身是高可用的分布式系统，但任何区域性的服务中断都将直接导致整个 WebSocket 服务的不可用。设计完善的降级策略，例如在 DynamoDB 不可用时暂时允许已建立连接继续通信（降级为信任模式），或直接断开所有连接，是生产环境中必须考虑的容错方案。

未来的一个演进方向可能是探索混合模型，例如引入一个像 Redis 这样的内存缓存层在 DynamoDB 之前，用于缓存权限信息，并通过 DynamoDB Streams 异步更新缓存，从而在保持高安全性的同时，显著降低对主数据库的读取压力和延迟。