面试题：网络编程之Java WebSocket API安全与可靠性

Question

在使用Java WebSocket API构建关键业务的实时Web应用时，安全性和可靠性至关重要。请详细说明如何确保WebSocket连接的安全性，防止诸如中间人攻击、恶意连接等安全威胁，同时阐述如何保证消息传输的可靠性，例如处理网络波动、消息重传等情况，需要结合具体的技术方案和代码示例进行说明。

Answer 1

确保WebSocket连接安全性

1. 使用TLS/SSL加密
   • 技术方案：通过HTTPS协议来建立WebSocket连接，这样在传输层就对数据进行了加密，防止中间人窃听和篡改数据。在Java中，对于基于Servlet容器的Web应用，可以配置Tomcat等服务器启用HTTPS。
   • 代码示例（以Tomcat配置为例）：
     • 生成SSL证书，例如使用keytool工具：

       keytool -genkeypair -alias tomcat -keyalg RSA -keysize 2048 -storetype PKCS12 -keystore keystore.p12 -validity 3650
       

     • 在Tomcat的conf/server.xml文件中配置SSL连接器：

       <Connector
           protocol="org.apache.coyote.http11.Http11NioProtocol"
           port="8443" maxThreads="200"
           scheme="https" secure="true" SSLEnabled="true">
           <SSLHostConfig>
               <Certificate certificateFile="conf/localhost.crt"
                            certificateKeyFile="conf/localhost.key"
                            type="RSA" />
           </SSLHostConfig>
       </Connector>
       

   • 在WebSocket客户端连接时，使用wss://协议：

     WebSocketContainer container = ContainerProvider.getWebSocketContainer();
     WebSocketClientEndpoint clientEndpoint = new WebSocketClientEndpoint(new URI("wss://your - server - address:8443/websocket - endpoint"));
     container.connectToServer(clientEndpoint);
     

2. 身份验证和授权
   • 技术方案：在建立WebSocket连接前，通过HTTP的身份验证机制（如Basic认证、OAuth等）进行用户身份验证。成功验证后，将相关的身份信息传递到WebSocket连接中，在服务端进行授权检查，确保只有合法用户可以建立WebSocket连接。
   • 代码示例（使用Servlet过滤器进行基本身份验证）：
     • 创建一个过滤器类：

       import javax.servlet.Filter;
       import javax.servlet.FilterChain;
       import javax.servlet.FilterConfig;
       import javax.servlet.ServletException;
       import javax.servlet.ServletRequest;
       import javax.servlet.ServletResponse;
       import javax.servlet.annotation.WebFilter;
       import javax.servlet.http.HttpServletRequest;
       import javax.servlet.http.HttpServletResponse;
       import java.io.IOException;
       import java.util.Base64;
       
       @WebFilter("/websocket - endpoint")
       public class BasicAuthFilter implements Filter {
           private static final String REALM = "My Realm";
           private static final String USERNAME = "admin";
           private static final String PASSWORD = "password";
       
           @Override
           public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
               HttpServletRequest httpRequest = (HttpServletRequest) request;
               HttpServletResponse httpResponse = (HttpServletResponse) response;
               String authHeader = httpRequest.getHeader("Authorization");
               if (authHeader != null && authHeader.startsWith("Basic ")) {
                   String base64Credentials = authHeader.substring("Basic ".length());
                   String credentials = new String(Base64.getDecoder().decode(base64Credentials));
                   String[] values = credentials.split(":", 2);
                   String username = values[0];
                   String password = values[1];
                   if (USERNAME.equals(username) && PASSWORD.equals(password)) {
                       chain.doFilter(request, response);
                       return;
                   }
               }
               httpResponse.setHeader("WWW - Authenticate", "Basic realm=\"" + REALM + "\"");
               httpResponse.sendError(HttpServletResponse.SC_UNAUTHORIZED, "Unauthorized");
           }
       
           @Override
           public void init(FilterConfig filterConfig) throws ServletException {}
       
           @Override
           public void destroy() {}
       }
       

3. 防止恶意连接
   • 技术方案：使用IP地址过滤、连接频率限制等手段。IP地址过滤可以阻止来自恶意IP的连接；连接频率限制可以防止恶意用户通过大量快速连接来耗尽服务器资源。
   • 代码示例（使用Guava的RateLimiter进行连接频率限制）：
     • 引入Guava依赖：

       <dependency>
           <groupId>com.google.guava</groupId>
           <artifactId>guava</artifactId>
           <version>31.1 - jre</version>
       </dependency>
       

     • 在WebSocket服务端添加频率限制逻辑：

       import com.google.common.util.concurrent.RateLimiter;
       import javax.websocket.EndpointConfig;
       import javax.websocket.OnOpen;
       import javax.websocket.Session;
       import javax.websocket.server.ServerEndpoint;
       import java.io.IOException;
       
       @ServerEndpoint("/websocket - endpoint")
       public class WebSocketEndpoint {
           private static final RateLimiter rateLimiter = RateLimiter.create(10); // 每秒允许10个连接
       
           @OnOpen
           public void onOpen(Session session, EndpointConfig config) {
               if (rateLimiter.tryAcquire()) {
                   try {
                       session.getBasicRemote().sendText("Connected successfully");
                   } catch (IOException e) {
                       e.printStackTrace();
                   }
               } else {
                   try {
                       session.close();
                   } catch (IOException e) {
                       e.printStackTrace();
                   }
               }
           }
       }
       

保证消息传输的可靠性

1. 处理网络波动
   • 技术方案：客户端和服务端都设置心跳机制，通过定期发送心跳消息来检测连接是否正常。如果在一定时间内没有收到心跳响应，则认为连接出现问题并尝试重新连接。
   • 代码示例（客户端心跳）：

     import javax.websocket.ClientEndpoint;
     import javax.websocket.CloseReason;
     import javax.websocket.ContainerProvider;
     import javax.websocket.OnClose;
     import javax.websocket.OnMessage;
     import javax.websocket.OnOpen;
     import javax.websocket.Session;
     import javax.websocket.WebSocketContainer;
     import java.net.URI;
     import java.util.concurrent.Executors;
     import java.util.concurrent.ScheduledExecutorService;
     import java.util.concurrent.TimeUnit;
     
     @ClientEndpoint
     public class WebSocketClientEndpoint {
         private Session session;
         private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
     
         public WebSocketClientEndpoint(URI endpointURI) {
             try {
                 WebSocketContainer container = ContainerProvider.getWebSocketContainer();
                 container.connectToServer(this, endpointURI);
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }
     
         @OnOpen
         public void onOpen(Session session) {
             this.session = session;
             scheduler.scheduleAtFixedRate(() -> {
                 try {
                     session.getBasicRemote().sendText("ping");
                 } catch (IOException e) {
                     e.printStackTrace();
                 }
             }, 0, 10, TimeUnit.SECONDS);
         }
     
         @OnMessage
         public void onMessage(String message) {
             if ("pong".equals(message)) {
                 // 收到心跳响应，连接正常
             }
         }
     
         @OnClose
         public void onClose(Session session, CloseReason closeReason) {
             scheduler.shutdown();
         }
     }
     

   • 服务端心跳示例：

     import javax.websocket.OnMessage;
     import javax.websocket.OnOpen;
     import javax.websocket.Session;
     import javax.websocket.server.ServerEndpoint;
     import java.io.IOException;
     import java.util.concurrent.Executors;
     import java.util.concurrent.ScheduledExecutorService;
     import java.util.concurrent.TimeUnit;
     
     @ServerEndpoint("/websocket - endpoint")
     public class WebSocketEndpoint {
         private Session session;
         private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
     
         @OnOpen
         public void onOpen(Session session) {
             this.session = session;
             scheduler.scheduleAtFixedRate(() -> {
                 try {
                     session.getBasicRemote().sendText("ping");
                 } catch (IOException e) {
                     e.printStackTrace();
                 }
             }, 0, 10, TimeUnit.SECONDS);
         }
     
         @OnMessage
         public void onMessage(String message, Session session) {
             if ("ping".equals(message)) {
                 try {
                     session.getBasicRemote().sendText("pong");
                 } catch (IOException e) {
                     e.printStackTrace();
                 }
             }
         }
     }
     

2. 消息重传
   • 技术方案：为每个发送的消息分配一个唯一的ID，服务端接收到消息后返回确认消息（包含消息ID）。如果客户端在一定时间内没有收到确认消息，则重传该消息。
   • 代码示例（客户端消息重传）：

     import javax.websocket.ClientEndpoint;
     import javax.websocket.CloseReason;
     import javax.websocket.ContainerProvider;
     import javax.websocket.OnClose;
     import javax.websocket.OnMessage;
     import javax.websocket.OnOpen;
     import javax.websocket.Session;
     import javax.websocket.WebSocketContainer;
     import java.net.URI;
     import java.util.HashMap;
     import java.util.Map;
     import java.util.concurrent.ConcurrentHashMap;
     import java.util.concurrent.Executors;
     import java.util.concurrent.ScheduledExecutorService;
     import java.util.concurrent.TimeUnit;
     
     @ClientEndpoint
     public class WebSocketClientEndpoint {
         private Session session;
         private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
         private final Map<String, Long> messageSendTimes = new ConcurrentHashMap<>();
         private static final long RETRY_TIMEOUT = 5000; // 5秒后重传
     
         public WebSocketClientEndpoint(URI endpointURI) {
             try {
                 WebSocketContainer container = ContainerProvider.getWebSocketContainer();
                 container.connectToServer(this, endpointURI);
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }
     
         @OnOpen
         public void onOpen(Session session) {
             this.session = session;
             scheduler.scheduleAtFixedRate(() -> {
                 for (Map.Entry<String, Long> entry : messageSendTimes.entrySet()) {
                     if (System.currentTimeMillis() - entry.getValue() > RETRY_TIMEOUT) {
                         try {
                             session.getBasicRemote().sendText("retry:" + entry.getKey());
                             messageSendTimes.put(entry.getKey(), System.currentTimeMillis());
                         } catch (IOException e) {
                             e.printStackTrace();
                         }
                     }
                 }
             }, 0, 1, TimeUnit.SECONDS);
         }
     
         public void sendMessage(String message) {
             String messageId = java.util.UUID.randomUUID().toString();
             try {
                 session.getBasicRemote().sendText(messageId + ":" + message);
                 messageSendTimes.put(messageId, System.currentTimeMillis());
             } catch (IOException e) {
                 e.printStackTrace();
             }
         }
     
         @OnMessage
         public void onMessage(String message) {
             if (message.startsWith("ack:")) {
                 String messageId = message.substring(4);
                 messageSendTimes.remove(messageId);
             }
         }
     
         @OnClose
         public void onClose(Session session, CloseReason closeReason) {
             scheduler.shutdown();
         }
     }
     

   • 服务端确认消息示例：

     import javax.websocket.OnMessage;
     import javax.websocket.OnOpen;
     import javax.websocket.Session;
     import javax.websocket.server.ServerEndpoint;
     import java.io.IOException;
     
     @ServerEndpoint("/websocket - endpoint")
     public class WebSocketEndpoint {
         @OnMessage
         public void onMessage(String message, Session session) {
             if (message.contains(":")) {
                 String[] parts = message.split(":", 2);
                 String messageId = parts[0];
                 try {
                     session.getBasicRemote().sendText("ack:" + messageId);
                 } catch (IOException e) {
                     e.printStackTrace();
                 }
             }
         }
     }