NextJS经NGINX连接Scala（Akka-HTTP）WebSocket失败，求排查及安全认证方案

Background #

我有一个基于Scala（搭配akka-http）的Web服务器运行在NGINX之后，同时还有一个NextJS应用也部署在同一台NGINX后面。

我的目标是让NextJS应用与Web服务器建立一个（安全的）ws连接。我已经遵循了官方文档和指南，以下是当前的实现：

Web服务器端实现 #

val responseQueue: mutable.Queue[ApiResponse] = mutable.Queue()

private def webSocketFlow: Flow[Message, Message, _] = {
  Flow[Message].mapAsync(1) { _ =>
    if (responseQueue.nonEmpty) {
      system.log.info("Flushing responseQueue")
      val response = responseQueue.dequeue()
      val protobufMessage = ByteString(response.toByteArray)

      Future.successful(BinaryMessage(protobufMessage))
    } else {
      system.log.warn("Response queue empty")
      Future.successful(BinaryMessage(ByteString.empty))
    }
  }
}

private def websocketRoute: Route = {
  pathPrefix("ws") {
    pathEndOrSingleSlash {
      extractRequest { req =>
        // extract auth token from header
        val tokenOpt = req.headers.collectFirst {
          case header if header.name() == "Sec-WebSocket-Protocol" =>
            OAuth2BearerToken(header.value()) // Extract the value of the Sec-WebSocket-Protocol header
        }

        system.log.info(s"Handling ws auth:${tokenOpt.toString}")

        // run it through token verification
        extractUri { uri =>
          val callingURI = uri.toRelative.path.dropChars(1).toString

          system.log.info(s"===== $callingURI")

          oAuthAuthenticator(Credentials(tokenOpt), handleWebSocketMessages(webSocketFlow), callingURI).get
        }
      }
    }
  }
}

private def oAuthAuthenticator(credentials: Credentials, protectedRoutes: Route, callingURI: String): Option[Route] =
  credentials match {
    case p@Credentials.Provided(_) =>
      system.log.info("Credentials provided")
      val user = loggedInUsers.find(user => p.verify(user.oAuthToken.access_token))
      if (user.isDefined) {
        system.log.info(s"User found:${user.head.toString}")
        val userPermissions = Permissions.valueOf(user.head.user.username.toUpperCase) // <- get permissions for this user

        if (userPermissions.getAllowedRoutes.contains(callingURI)) {
          system.log.info(s"User has permission for route: $callingURI")
          if (user.head.oneTime) loggedInUsers -= user.head // remove token if it's a one-time use token
          Option(protectedRoutes)
        } else {
          system.log.error(s"User does not have permission for route: $callingURI")
          Option(complete(ApiResponse().withStatusResponse(HydraStatusCodes.UNAUTHORISED_ROUTE.getStatusResponse)))
        }
      } else {
        system.log.error("We did not distribute this token or its expired")
        Option(complete(ApiResponse().withStatusResponse(HydraStatusCodes.INVALID_AUTH_TOKEN.getStatusResponse)))
      }
    case _ =>
      system.log.error(s"No credentials provided: ${credentials.toString}")
      Option(complete(ApiResponse().withStatusResponse(HydraStatusCodes.MISSING_CREDENTIALS.getStatusResponse)))
  }

服务器的目标是当队列中有新的ApiResponse时通知网页。

从服务器日志可以确认授权部分正常工作：

INFO[typed-system-actor-akka.actor.default-dispatcher-11] ActorSystem - Handling ws auth:Some(Bearer 61704059-2e51-4d0f-b574-bdcebf3aeae3)
INFO[typed-system-actor-akka.actor.default-dispatcher-11] ActorSystem - ===== ws
INFO[typed-system-actor-akka.actor.default-dispatcher-11] ActorSystem - Credentials provided
INFO[typed-system-actor-akka.actor.default-dispatcher-11] ActorSystem - User found:LoggedInUser(database.objects.User@bb81dcf5, username: admin, password: 517ffce87ad701f071040b32ddaa7f4b7b0bb6774b02ff45bf2eef3f2fc1a549,AuthToken(61704059-2e51-4d0f-b574-bdcebf3aeae3,bearer,3600),2025-02-21T16:59:06.343277,false)
INFO[typed-system-actor-akka.actor.default-dispatcher-11] ActorSystem - User has permission for route: ws

NextJS客户端实现 #

const [pageToken, setPageToken] = useState("")
const router = useRouter()

useEffect(() => {
    if (pageToken) {
        const ws = new WebSocket("/ws", pageToken);

        ws.onopen = () => {
            console.log("Connected to WebSocket server");
        };

        ws.onmessage = (event) => {
            try {
                if (event.data.byteLength === 0) {
                    console.log("No message to decode, queue was empty.");
                    return; // Ignore empty messages
                }

                // Deserialize the Protobuf message
                const buffer = new Uint8Array(event.data);
                const decodedMessage = ApiResponse.deserializeBinary(buffer)

                console.log(decodedMessage)
            } catch (err) {
                console.error("Failed to decode Protobuf message:", err);
            }
        };

        ws.onerror = (err) => {
            console.error("WebSocket error:", err);
        };

        return () => {
            if (ws) {
                ws.close();
            }
        };
    }
}, [pageToken]); // <-- listen to token changes

const fetcher = (url) => fetchWithErrors(url, {}, (error) => {
    if (error.status === 401) {
        setToken(null) //<-- for anything that still might be using token
        setPageToken(null)
        router.push("/login");
    } else {
        errorToast("Unknown Internal Error:" + "[" + error.status + "]" + error.message);
    }
})
    .then(data => {
        console.log("Fetched token data:", data)
        if (data.access_token) {
            if (pageToken !== data.access_token) {
                setPageToken(data.access_token);
                setToken(data.access_token);
            }
        }
        return data
    });

// read the token every second to ensure it has not expired
const {data, mutate} = useSWR(
    "/api/gettoken",
    fetcher,
    {
        refreshInterval: 100
    }
);

useEffect(() => {
    if (data) return;
    // mark as stale
    mutate()
}, [data]);

这段代码的逻辑：

• 从自身的服务器路由获取token（从cookie中提取）并全局设置
• 当token变化时，建立WebSocket连接

NGINX配置 #

location /ws {
        proxy_pass http://localhost:8080;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection 'upgrade';
        proxy_set_header Host $host;
}

问题 #

在NextJS端，我遇到了错误：Page.jsx:19 WebSocket connection to 'wss://192.168.0.11/ws' failed:

我已经捕获了HAR输出用于调试。

更新 #

我尝试了一些调试步骤：

1. 移除了服务器端的认证流程——没有变化
2. 移除了连接ws时发送的token（同时去掉了服务器端的认证）——连接成功了

我有点困惑，我看到的所有WebSocket认证指南都说应该用这种方式（因为通过URI发送token不安全）。那我该如何验证连接WebSocket的用户是正确的呢？

备注：内容来源于stack exchange，提问作者Kris Rice