SSL transport support. Issue #22

src/main/java/com/corundumstudio/socketio/AuthorizeHandler.java

@@ -147,7 +147,7 @@ public class AuthorizeHandler extends SimpleChannelUpstreamHandler implements Di
         client.send(new Packet(PacketType.CONNECT));
 
         Namespace ns = namespacesHub.get(Namespace.DEFAULT_NAME);
-        SocketIOClient nsClient = client.getClient(ns);
+        SocketIOClient nsClient = client.getChildClient(ns);
         namespacesHub.get(ns.getName()).onConnect(nsClient);
     }
 

src/main/java/com/corundumstudio/socketio/Configuration.java

@@ -41,6 +41,9 @@ public class Configuration {
     private String hostname;
     private int port;
 
+    private String keyStorePath;
+    private String keyStorePassword;
+
     private JsonSupport jsonSupport = new JacksonJsonSupport(this);
 
     public Configuration() {
@@ -65,6 +68,8 @@ public class Configuration {
         setAllowCustomRequests(conf.isAllowCustomRequests());
         setPollingDuration(conf.getPollingDuration());
         setJsonTypeFieldName(conf.getJsonTypeFieldName());
+        setKeyStorePassword(conf.getKeyStorePassword());
+        setKeyStorePath(conf.getKeyStorePath());
     }
 
     public String getJsonTypeFieldName() {
@@ -208,4 +213,18 @@ public class Configuration {
         this.pollingDuration = pollingDuration;
     }
 
+    public void setKeyStorePassword(String keyStorePassword) {
+        this.keyStorePassword = keyStorePassword;
+    }
+    public String getKeyStorePassword() {
+        return keyStorePassword;
+    }
+
+    public void setKeyStorePath(String keyStorePath) {
+        this.keyStorePath = keyStorePath;
+    }
+    public String getKeyStorePath() {
+        return keyStorePath;
+    }
+
 }

src/main/java/com/corundumstudio/socketio/PacketHandler.java

@@ -61,7 +61,7 @@ public class PacketHandler extends SimpleChannelUpstreamHandler {
             while (content.readable()) {
                 Packet packet = decoder.decodePackets(content);
                 Namespace ns = namespacesHub.get(packet.getEndpoint());
-                SocketIOClient client = message.getClient().getClient(ns);
+                SocketIOClient client = message.getClient().getChildClient(ns);
                 sendAck(packet, client);
                 packetListener.onPacket(packet, client);
             }

src/main/java/com/corundumstudio/socketio/SocketIOPipelineFactory.java

@@ -17,11 +17,20 @@ package com.corundumstudio.socketio;
 
 import static org.jboss.netty.channel.Channels.pipeline;
 
+import java.io.FileInputStream;
+import java.security.KeyStore;
+import java.security.Security;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+
 import org.jboss.netty.channel.ChannelPipeline;
 import org.jboss.netty.channel.ChannelPipelineFactory;
 import org.jboss.netty.handler.codec.http.HttpChunkAggregator;
 import org.jboss.netty.handler.codec.http.HttpRequestDecoder;
 import org.jboss.netty.handler.codec.http.HttpResponseEncoder;
+import org.jboss.netty.handler.ssl.SslHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,6 +53,7 @@ public class SocketIOPipelineFactory implements ChannelPipelineFactory, Disconne
     protected static final String HTTP_ENCODER = "encoder";
     protected static final String HTTP_AGGREGATOR = "aggregator";
     protected static final String HTTP_REQUEST_DECODER = "decoder";
+    protected static final String SSL_HANDLER = "ssl";
 
     private final Logger log = LoggerFactory.getLogger(getClass());
 
@@ -60,6 +70,7 @@ public class SocketIOPipelineFactory implements ChannelPipelineFactory, Disconne
 
     private PacketHandler packetHandler;
     private HeartbeatHandler heartbeatHandler;
+    private SSLContext sslContext;
 
     public void start(Configuration configuration, NamespacesHub namespacesHub) {
         scheduler = new CancelableScheduler(configuration.getHeartbeatThreadPoolSize());
@@ -74,10 +85,19 @@ public class SocketIOPipelineFactory implements ChannelPipelineFactory, Disconne
 
         String connectPath = configuration.getContext() + "/" + protocol + "/";
 
+        boolean isSsl = configuration.getKeyStorePath() != null;
+        if (isSsl) {
+            try {
+                sslContext = createSSLContext(configuration.getKeyStorePath(), configuration.getKeyStorePassword());
+            } catch (Exception e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
         packetHandler = new PacketHandler(packetListener, decoder, namespacesHub);
         authorizeHandler = new AuthorizeHandler(connectPath, scheduler, configuration, namespacesHub);
         xhrPollingTransport = new XHRPollingTransport(connectPath, ackManager, this, scheduler, authorizeHandler, configuration);
-        webSocketTransport = new WebSocketTransport(connectPath, ackManager, this, authorizeHandler, heartbeatHandler);
+        webSocketTransport = new WebSocketTransport(connectPath, isSsl, ackManager, this, authorizeHandler, heartbeatHandler);
         socketIOEncoder = new SocketIOEncoder(encoder);
     }
 
@@ -90,6 +110,12 @@ public class SocketIOPipelineFactory implements ChannelPipelineFactory, Disconne
     public ChannelPipeline getPipeline() throws Exception {
         ChannelPipeline pipeline = pipeline();
 
+        if (sslContext != null) {
+            SSLEngine engine = sslContext.createSSLEngine();
+            engine.setUseClientMode(false);
+            pipeline.addLast(SSL_HANDLER, new SslHandler(engine));
+        }
+
         pipeline.addLast(HTTP_REQUEST_DECODER, new HttpRequestDecoder());
         pipeline.addLast(HTTP_AGGREGATOR, new HttpChunkAggregator(65536));
         pipeline.addLast(HTTP_ENCODER, new HttpResponseEncoder());
@@ -105,6 +131,24 @@ public class SocketIOPipelineFactory implements ChannelPipelineFactory, Disconne
         return pipeline;
     }
 
+    private SSLContext createSSLContext(String keyStoreFilePath, String keyStoreFilePassword) throws Exception {
+        String algorithm = Security.getProperty("ssl.KeyManagerFactory.algorithm");
+        if (algorithm == null) {
+            algorithm = "SunX509";
+        }
+
+        KeyStore ks = KeyStore.getInstance("JKS");
+        FileInputStream fin = new FileInputStream(keyStoreFilePath);
+        ks.load(fin, keyStoreFilePassword.toCharArray());
+
+        KeyManagerFactory kmf = KeyManagerFactory.getInstance(algorithm);
+        kmf.init(ks, keyStoreFilePassword.toCharArray());
+
+        SSLContext serverContext = SSLContext.getInstance("TLS");
+        serverContext.init(kmf.getKeyManagers(), null, null);
+        return serverContext;
+    }
+
     public void onDisconnect(BaseClient client) {
         log.debug("Client with sessionId: {} disconnected", client.getSessionId());
         heartbeatHandler.onDisconnect(client);

src/main/java/com/corundumstudio/socketio/transport/BaseClient.java

@@ -57,14 +57,14 @@ public abstract class BaseClient {
 
     public abstract ChannelFuture send(Packet packet);
 
-    public void removeClient(SocketIOClient client) {
+    public void removeChildClient(SocketIOClient client) {
         namespaceClients.remove((Namespace)client.getNamespace());
         if (namespaceClients.isEmpty()) {
             disconnectable.onDisconnect(this);
         }
     }
 
-    public SocketIOClient getClient(Namespace namespace) {
+    public SocketIOClient getChildClient(Namespace namespace) {
         SocketIOClient client = namespaceClients.get(namespace);
         if (client == null) {
             client = new NamespaceClient(this, namespace);
@@ -76,12 +76,12 @@ public abstract class BaseClient {
         return client;
     }
 
-    public Collection<SocketIOClient> getAllClients() {
+    public Collection<SocketIOClient> getAllChildClients() {
         return namespaceClients.values();
     }
 
     public void onChannelDisconnect() {
-        for (SocketIOClient client : getAllClients()) {
+        for (SocketIOClient client : getAllChildClients()) {
             ((NamespaceClient) client).onDisconnect();
         }
     }

src/main/java/com/corundumstudio/socketio/transport/NamespaceClient.java

@@ -103,13 +103,12 @@ public class NamespaceClient implements SocketIOClient {
 
     public void onDisconnect() {
         namespace.onDisconnect(this);
-        baseClient.removeClient(this);
+        baseClient.removeChildClient(this);
     }
 
     @Override
     public void disconnect() {
         send(new Packet(PacketType.DISCONNECT));
-        // TODO disconnect on channel close
         onDisconnect();
     }
 

src/main/java/com/corundumstudio/socketio/transport/WebSocketTransport.java

@@ -63,11 +63,13 @@ public class WebSocketTransport extends SimpleChannelUpstreamHandler implements
     private final AuthorizeHandler authorizeHandler;
     private final DisconnectableHub disconnectableHub;
     private final String path;
+    private final boolean isSsl;
 
 
-    public WebSocketTransport(String connectPath, AckManager ackManager, DisconnectableHub disconnectable,
+    public WebSocketTransport(String connectPath, boolean isSsl, AckManager ackManager, DisconnectableHub disconnectable,
             AuthorizeHandler authorizeHandler, HeartbeatHandler heartbeatHandler) {
         this.path = connectPath + "websocket";
+        this.isSsl = isSsl;
         this.authorizeHandler = authorizeHandler;
         this.ackManager = ackManager;
         this.disconnectableHub = disconnectable;
@@ -153,7 +155,11 @@ public class WebSocketTransport extends SimpleChannelUpstreamHandler implements
     }
 
     private String getWebSocketLocation(HttpRequest req) {
-        return "ws://" + req.getHeader(HttpHeaders.Names.HOST) + path;
+        String protocol = "ws://";
+        if (isSsl) {
+            protocol = "wss://";
+        }
+        return protocol + req.getHeader(HttpHeaders.Names.HOST) + path;
     }
 
     @Override
@@ -170,7 +176,7 @@ public class WebSocketTransport extends SimpleChannelUpstreamHandler implements
         Collection<WebSocketClient> clients = sessionId2Client.values();
         List<Iterable<SocketIOClient>> allClients = new ArrayList<Iterable<SocketIOClient>>(clients.size());
         for (WebSocketClient client : sessionId2Client.values()) {
-            allClients.add(client.getAllClients());
+            allClients.add(client.getAllChildClients());
         }
         return new CompositeIterable<SocketIOClient>(allClients);
     }

src/main/java/com/corundumstudio/socketio/transport/XHRPollingTransport.java

@@ -220,7 +220,7 @@ public class XHRPollingTransport extends SimpleChannelUpstreamHandler implements
         Collection<XHRPollingClient> clients = sessionId2Client.values();
         List<Iterable<SocketIOClient>> allClients = new ArrayList<Iterable<SocketIOClient>>(clients.size());
         for (XHRPollingClient client : sessionId2Client.values()) {
-            allClients.add(client.getAllClients());
+            allClients.add(client.getAllChildClients());
         }
         return new CompositeIterable<SocketIOClient>(allClients);
     }