webscoket 鉴权方式修改

cim-boot-server/pom.xml

@@ -67,7 +67,7 @@
         <dependency>
             <groupId>com.farsunset</groupId>
             <artifactId>cim-server-sdk-netty</artifactId>
-            <version>4.2.6</version>
+            <version>4.2.7-SNAPSHOT</version>
         </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>

cim-boot-server/src/main/java/com/farsunset/cim/component/predicate/AuthPredicate.java

@@ -0,0 +1,28 @@
+package com.farsunset.cim.component.predicate;
+
+
+import com.farsunset.cim.auth.AuthPredicateInfo;
+import org.springframework.stereotype.Component;
+
+import java.util.function.Predicate;
+
+/**
+ * WS 鉴权验证
+ */
+@Component
+public class AuthPredicate implements Predicate<AuthPredicateInfo> {
+
+    @Override
+    public boolean test(AuthPredicateInfo auth) {
+        /*
+            可通过header或者uri传递参数
+            String token = auth.getHeader("token");
+            String token = auth.getParameter("token");
+            User user = doAuth(token);
+            auth.getCtx().attr(AttributeKey.valueOf("user_id")).set(user.getId());
+         */
+
+
+        return true;
+    }
+}

cim-boot-server/src/main/java/com/farsunset/cim/component/predicate/HandshakePredicate.java

@@ -30,6 +30,7 @@ import java.util.function.Predicate;
 /**
  * WS链接握手鉴权验证
  */
+@Deprecated
 @Component
 public class HandshakePredicate implements Predicate<HandshakeEvent> {
 

cim-boot-server/src/main/java/com/farsunset/cim/config/CIMConfig.java

@@ -26,7 +26,7 @@ import com.farsunset.cim.acceptor.WebsocketAcceptor;
 import com.farsunset.cim.acceptor.config.SocketConfig;
 import com.farsunset.cim.acceptor.config.WebsocketConfig;
 import com.farsunset.cim.component.handler.annotation.CIMHandler;
-import com.farsunset.cim.component.predicate.HandshakePredicate;
+import com.farsunset.cim.component.predicate.AuthPredicate;
 import com.farsunset.cim.config.properties.APNsProperties;
 import com.farsunset.cim.config.properties.CIMAppSocketProperties;
 import com.farsunset.cim.config.properties.CIMWebsocketProperties;
@@ -76,9 +76,9 @@ public class CIMConfig implements CIMRequestHandler, ApplicationListener<Applica
 
 	@Bean(destroyMethod = "destroy",initMethod = "bind")
 	@ConditionalOnProperty(name = {"cim.websocket.enable"},matchIfMissing = true)
-	public WebsocketAcceptor websocketAcceptor(CIMWebsocketProperties properties, HandshakePredicate handshakePredicate) {
+	public WebsocketAcceptor websocketAcceptor(CIMWebsocketProperties properties, AuthPredicate authPredicate) {
 		WebsocketConfig config = new WebsocketConfig();
-		config.setHandshakePredicate(handshakePredicate);
+		config.setAuthPredicate(authPredicate);
 		config.setPath(properties.getPath());
 		config.setPort(properties.getPort());
 		config.setProtocol(properties.getProtocol());

cim-server-sdk/pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.farsunset</groupId>
     <artifactId>cim-server-sdk-netty</artifactId>
-    <version>4.2.6</version>
+    <version>4.2.7-SNAPSHOT</version>
     <packaging>jar</packaging>
 
     <name>${project.groupId}:${project.artifactId}</name>

cim-server-sdk/src/main/java/com/farsunset/cim/acceptor/WebsocketAcceptor.java

@@ -22,13 +22,13 @@
 package com.farsunset.cim.acceptor;
 
 import com.farsunset.cim.acceptor.config.WebsocketConfig;
+import com.farsunset.cim.auth.AuthHandler;
 import com.farsunset.cim.coder.json.TextMessageDecoder;
 import com.farsunset.cim.coder.json.TextMessageEncoder;
 import com.farsunset.cim.coder.protobuf.WebMessageDecoder;
 import com.farsunset.cim.coder.protobuf.WebMessageEncoder;
 import com.farsunset.cim.constant.WebsocketProtocol;
 import com.farsunset.cim.handler.IllegalRequestHandler;
-import com.farsunset.cim.handshake.HandshakeHandler;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandler;
@@ -69,14 +69,14 @@ public class WebsocketAcceptor extends NioSocketAcceptor {
 
 	private final WebsocketConfig config;
 
-	private final HandshakeHandler handshakeHandler;
+	private final AuthHandler authHandler;
 
 	private final ChannelHandler illegalRequestHandler = new IllegalRequestHandler();
 
 	public WebsocketAcceptor(WebsocketConfig config){
 		super(config);
 		this.config = config;
-		this.handshakeHandler = new HandshakeHandler(config.getHandshakePredicate());
+		this.authHandler = new AuthHandler(config.getAuthPredicate());
 	}
 
 	/**
@@ -97,8 +97,8 @@ public class WebsocketAcceptor extends NioSocketAcceptor {
 				ch.pipeline().addLast(new HttpServerCodec());
 				ch.pipeline().addLast(new ChunkedWriteHandler());
 				ch.pipeline().addLast(new HttpObjectAggregator(4 * 1024));
+				ch.pipeline().addLast(authHandler);
 				ch.pipeline().addLast(new WebSocketServerProtocolHandler(config.getPath(),true));
-				ch.pipeline().addLast(handshakeHandler);
 				if (config.getProtocol() == WebsocketProtocol.JSON){
 					ch.pipeline().addLast(new TextMessageDecoder());
 					ch.pipeline().addLast(new TextMessageEncoder());

cim-server-sdk/src/main/java/com/farsunset/cim/acceptor/config/WebsocketConfig.java

@@ -21,6 +21,7 @@
  */
 package com.farsunset.cim.acceptor.config;
 
+import com.farsunset.cim.auth.AuthPredicateInfo;
 import com.farsunset.cim.constant.WebsocketProtocol;
 import com.farsunset.cim.handshake.HandshakeEvent;
 
@@ -50,8 +51,14 @@ public class WebsocketConfig extends SocketConfig{
     /**
      * websocket鉴权实现
      */
+    @Deprecated
     private Predicate<HandshakeEvent> handshakePredicate;
 
+    /**
+     * 鉴权实现
+     */
+    private Predicate<AuthPredicateInfo> authPredicate;
+
 
     @Override
     public Integer getPort() {
@@ -83,4 +90,11 @@ public class WebsocketConfig extends SocketConfig{
         this.handshakePredicate = handshakePredicate;
     }
 
+    public void setAuthPredicate(Predicate<AuthPredicateInfo> authPredicate) {
+        this.authPredicate = authPredicate;
+    }
+
+    public Predicate<AuthPredicateInfo> getAuthPredicate() {
+        return authPredicate;
+    }
 }

cim-server-sdk/src/main/java/com/farsunset/cim/auth/AuthHandler.java

@@ -0,0 +1,56 @@
+package com.farsunset.cim.auth;
+
+import com.farsunset.cim.constant.CIMConstant;
+import com.farsunset.cim.model.ReplyBody;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.function.Predicate;
+
+/**
+ * 鉴权处理器
+ */
+@ChannelHandler.Sharable
+public class AuthHandler extends ChannelInboundHandlerAdapter {
+
+    /*
+     *认证失败，返回replyBody给客户端
+     */
+    private final ReplyBody failedBody = ReplyBody.make(CIMConstant.CLIENT_HANDSHAKE,
+            HttpResponseStatus.UNAUTHORIZED.code(),
+            HttpResponseStatus.UNAUTHORIZED.reasonPhrase());
+
+
+    private final Predicate<AuthPredicateInfo> authPredicate;
+
+
+    public AuthHandler(Predicate<AuthPredicateInfo> authPredicate) {
+        this.authPredicate = authPredicate;
+    }
+
+
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        if (msg instanceof FullHttpRequest) {
+            FullHttpRequest request = (FullHttpRequest) msg;
+
+            /*
+             * 鉴权不通过，发送响应体并关闭链接
+             */
+            if (!authPredicate.test(AuthPredicateInfo.of(request, ctx))) {
+                ctx.channel().writeAndFlush(failedBody).addListener(ChannelFutureListener.CLOSE);
+                return;
+            }
+
+            //鉴权通过后移除本handler
+            ctx.pipeline().remove(this);
+        }
+
+        //other protocols
+        super.channelRead(ctx, msg);
+    }
+
+}

cim-server-sdk/src/main/java/com/farsunset/cim/auth/AuthPredicateInfo.java

@@ -0,0 +1,64 @@
+package com.farsunset.cim.auth;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.QueryStringDecoder;
+
+import java.util.List;
+
+/**
+ * 鉴权相关信息
+ */
+
+public class AuthPredicateInfo {
+
+    private final String uri;
+
+    private final HttpHeaders header;
+
+    private final ChannelHandlerContext ctx;
+
+    public AuthPredicateInfo(String uri, HttpHeaders header, ChannelHandlerContext ctx) {
+        this.uri = uri;
+        this.header = header;
+        this.ctx = ctx;
+    }
+
+    public String getHeader(String name) {
+        return header.get(name);
+    }
+
+    public List<String> getHeaders(String name) {
+        return header.getAll(name);
+    }
+
+    public Integer getIntHeader(String name) {
+        return header.getInt(name);
+    }
+
+    public String getParameter(String name) {
+        QueryStringDecoder decoder = new QueryStringDecoder(uri);
+        List<String> valueList = decoder.parameters().get(name);
+        return valueList == null || valueList.isEmpty() ? null : valueList.get(0);
+    }
+
+    public List<String> getParameters(String name) {
+        QueryStringDecoder decoder = new QueryStringDecoder(uri);
+        return decoder.parameters().get(name);
+    }
+
+    public String getUri() {
+        return this.uri;
+    }
+
+    public ChannelHandlerContext getCtx() {
+        return this.ctx;
+    }
+
+    public static AuthPredicateInfo of(FullHttpRequest request, ChannelHandlerContext context) {
+        return new AuthPredicateInfo(request.uri(), request.headers(), context);
+    }
+
+
+}

cim-server-sdk/src/main/java/com/farsunset/cim/handshake/HandshakeEvent.java

@@ -31,6 +31,7 @@ import java.util.List;
  * websocket客户端握手请求信息
  * 用于在握手阶段鉴权
  */
+@Deprecated
 public class HandshakeEvent {
 
     private final String uri;

cim-server-sdk/src/main/java/com/farsunset/cim/handshake/HandshakeHandler.java

@@ -35,6 +35,7 @@ import java.util.function.Predicate;
 /**
  * WS握手时鉴权
  */
+@Deprecated
 @ChannelHandler.Sharable
 public class HandshakeHandler extends ChannelInboundHandlerAdapter {
     /*