External components are now using MINA. JM-1269

git-svn-id: http://svn.igniterealtime.org/svn/repos/openfire/trunk@9912 b35dd754-fafc-0310-a699-88a17e54d16e

src/java/org/jivesoftware/openfire/SessionManager.java

@@ -322,9 +322,9 @@ public class SessionManager extends BasicModule implements ClusterEventListener
         return session;
     }
 
-    public LocalComponentSession createComponentSession(JID address, Connection conn) throws UnauthorizedException {
+    public LocalComponentSession createComponentSession(JID address, Connection conn) {
         if (serverName == null) {
-            throw new UnauthorizedException("Server not initialized");
+            throw new IllegalStateException("Server not initialized");
         }
         StreamID id = nextStreamID();
         LocalComponentSession session = new LocalComponentSession(serverName, conn, id);

src/java/org/jivesoftware/openfire/net/ComponentSocketReader.java → src/java/org/jivesoftware/openfire/net/ComponentStanzaHandler.java

 /**
- * $RCSfile: ComponentSocketReader.java,v $
- * $Revision: 3174 $
- * $Date: 2005-12-08 17:41:00 -0300 (Thu, 08 Dec 2005) $
+ * $Revision: $
+ * $Date: $
  *
- * Copyright (C) 2007 Jive Software. All rights reserved.
+ * Copyright (C) 2008 Jive Software. All rights reserved.
  *
  * This software is published under the terms of the GNU Public License (GPL),
  * a copy of which is included in this distribution.
@@ -12,44 +11,51 @@
 package org.jivesoftware.openfire.net;
 
 import org.dom4j.Element;
+import org.jivesoftware.openfire.Connection;
 import org.jivesoftware.openfire.PacketRouter;
-import org.jivesoftware.openfire.RoutingTable;
 import org.jivesoftware.openfire.auth.UnauthorizedException;
 import org.jivesoftware.openfire.component.InternalComponentManager;
 import org.jivesoftware.openfire.session.ComponentSession;
 import org.jivesoftware.openfire.session.LocalComponentSession;
+import org.jivesoftware.openfire.session.Session;
 import org.jivesoftware.util.Log;
+import org.xmlpull.v1.XmlPullParser;
 import org.xmlpull.v1.XmlPullParserException;
 import org.xmpp.component.ComponentException;
+import org.xmpp.packet.IQ;
+import org.xmpp.packet.Message;
 import org.xmpp.packet.PacketError;
-
-import java.io.IOException;
-import java.net.Socket;
+import org.xmpp.packet.Presence;
 
 /**
- * A SocketReader specialized for component connections. This reader will be used when the open
- * stream contains a jabber:component:accept namespace.
+ * Handler of XML stanzas sent by external components connected directly to the server. Received packet will
+ * have their FROM attribute overriden to avoid spoofing.<p>
+ *
+ * This is an implementation of the XEP-114. In the future we will add support for XEP-225 now that
+ * we are using MINA things should be easier. Since we are now using MINA incoming traffic is handled
+ * by a set of worker threads.
  *
  * @author Gaston Dombiak
  */
-public class ComponentSocketReader extends SocketReader {
+public class ComponentStanzaHandler extends StanzaHandler {
 
-    public ComponentSocketReader(PacketRouter router, RoutingTable routingTable, String serverName,
-            Socket socket, SocketConnection connection, boolean useBlockingMode) {
-        super(router, routingTable, serverName, socket, connection, useBlockingMode);
+    public ComponentStanzaHandler(PacketRouter router, String serverName, Connection connection) {
+        super(router, serverName, connection);
     }
 
-    /**
-     * Only <tt>bind<tt> packets will be processed by this class to bind more domains
-     * to existing external components. Any other type of packet is unknown and thus
-     * rejected generating the connection to be closed.
-     *
-     * @param doc the unknown DOM element that was received
-     * @return false if packet is unknown otherwise true.
-     */
-    protected boolean processUnknowPacket(Element doc) {
-        // Handle subsequent bind packets
-        if ("bind".equals(doc.getName())) {
+    boolean processUnknowPacket(Element doc) throws UnauthorizedException {
+        String tag = doc.getName();
+        if ("handshake".equals(tag)) {
+            // External component is trying to authenticate
+            if (!((LocalComponentSession) session).authenticate(doc.getStringValue())) {
+                session.close();
+            }
+            return true;
+        } else if ("error".equals(tag) && "stream".equals(doc.getNamespacePrefix())) {
+            session.close();
+            return true;
+        } else if ("bind".equals(tag)) {
+            // Handle subsequent bind packets
             LocalComponentSession componentSession = (LocalComponentSession) session;
             // Get the external component of this session
             ComponentSession.ExternalComponent component = componentSession.getExternalComponent();
@@ -107,29 +113,76 @@ public class ComponentSocketReader extends SocketReader {
             }
             return true;
         }
-        // This is an unknown packet so return false (and close the connection)
         return false;
     }
 
-    boolean createSession(String namespace) throws UnauthorizedException, XmlPullParserException,
-            IOException {
-        if ("jabber:component:accept".equals(namespace)) {
-            // The connected client is a component so create a ComponentSession
-            session = LocalComponentSession.createSession(serverName, reader, connection);
-            return true;
+    protected void processIQ(IQ packet) throws UnauthorizedException {
+        if (session.getStatus() != Session.STATUS_AUTHENTICATED) {
+            // Session is not authenticated so return error
+            IQ reply = new IQ();
+            reply.setChildElement(packet.getChildElement().createCopy());
+            reply.setID(packet.getID());
+            reply.setTo(packet.getFrom());
+            reply.setFrom(packet.getTo());
+            reply.setError(PacketError.Condition.not_authorized);
+            session.process(reply);
+            return;
         }
-        return false;
+        super.processIQ(packet);
+    }
+
+    protected void processPresence(Presence packet) throws UnauthorizedException {
+        if (session.getStatus() != Session.STATUS_AUTHENTICATED) {
+            // Session is not authenticated so return error
+            Presence reply = new Presence();
+            reply.setID(packet.getID());
+            reply.setTo(packet.getFrom());
+            reply.setFrom(packet.getTo());
+            reply.setError(PacketError.Condition.not_authorized);
+            session.process(reply);
+            return;
+        }
+        super.processPresence(packet);
+    }
+
+    protected void processMessage(Message packet) throws UnauthorizedException {
+        if (session.getStatus() != Session.STATUS_AUTHENTICATED) {
+            // Session is not authenticated so return error
+            Message reply = new Message();
+            reply.setID(packet.getID());
+            reply.setTo(packet.getFrom());
+            reply.setFrom(packet.getTo());
+            reply.setError(PacketError.Condition.not_authorized);
+            session.process(reply);
+            return;
+        }
+        super.processMessage(packet);
+    }
+
+    void startTLS() throws Exception {
+        // TODO Finish implementation. We need to get the name of the CM if we want to validate certificates of the CM that requested TLS
+        connection.startTLS(false, "IMPLEMENT_ME", Connection.ClientAuth.disabled);
     }
 
     String getNamespace() {
         return "jabber:component:accept";
     }
 
-    String getName() {
-        return "Component SR - " + hashCode();
+    boolean validateHost() {
+        return false;
     }
 
-    boolean validateHost() {
+    boolean validateJIDs() {
+        return false;
+    }
+
+    boolean createSession(String namespace, String serverName, XmlPullParser xpp, Connection connection)
+            throws XmlPullParserException {
+        if (getNamespace().equals(namespace)) {
+            // The connected client is a connection manager so create a ConnectionMultiplexerSession
+            session = LocalComponentSession.createSession(serverName, xpp, connection);
+            return true;
+        }
         return false;
     }
 }

src/java/org/jivesoftware/openfire/net/StanzaHandler.java

@@ -22,12 +22,7 @@ import org.jivesoftware.util.Log;
 import org.jivesoftware.util.StringUtils;
 import org.xmlpull.v1.XmlPullParser;
 import org.xmlpull.v1.XmlPullParserException;
-import org.xmpp.packet.IQ;
-import org.xmpp.packet.Message;
-import org.xmpp.packet.PacketError;
-import org.xmpp.packet.Presence;
-import org.xmpp.packet.Roster;
-import org.xmpp.packet.StreamError;
+import org.xmpp.packet.*;
 
 import java.io.IOException;
 import java.io.StringReader;
@@ -69,7 +64,7 @@ public abstract class StanzaHandler {
     /**
      * Server name for which we are attending clients.
      */
-    private String serverName;
+    protected String serverName;
 
     /**
      * Router used to route incoming packets to the correct channels.

src/java/org/jivesoftware/openfire/nio/ComponentConnectionHandler.java

+/**
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2008 Jive Software. All rights reserved.
+ *
+ * This software is published under the terms of the GNU Public License (GPL),
+ * a copy of which is included in this distribution.
+ */
+
+package org.jivesoftware.openfire.nio;
+
+import org.apache.mina.common.IoSession;
+import org.jivesoftware.openfire.XMPPServer;
+import org.jivesoftware.openfire.net.ComponentStanzaHandler;
+import org.jivesoftware.openfire.net.StanzaHandler;
+import org.jivesoftware.util.JiveGlobals;
+
+/**
+ * ConnectionHandler that knows which subclass of {@link StanzaHandler} should
+ * be created and how to build and configure a {@link NIOConnection}.
+ *
+ * @author Gaston Dombiak
+ */
+public class ComponentConnectionHandler extends ConnectionHandler {
+    public ComponentConnectionHandler(String serverName) {
+        super(serverName);
+    }
+
+    NIOConnection createNIOConnection(IoSession session) {
+        return new NIOConnection(session, XMPPServer.getInstance().getPacketDeliverer());
+    }
+
+    StanzaHandler createStanzaHandler(NIOConnection connection) {
+        return new ComponentStanzaHandler(XMPPServer.getInstance().getPacketRouter(), serverName, connection);
+    }
+
+    int getMaxIdleTime() {
+        return JiveGlobals.getIntProperty("xmpp.component.idle", 6 * 60 * 1000) / 1000;
+    }
+}

src/java/org/jivesoftware/openfire/spi/ConnectionManagerImpl.java

@@ -28,6 +28,7 @@ import org.jivesoftware.openfire.container.PluginManagerListener;
 import org.jivesoftware.openfire.http.HttpBindManager;
 import org.jivesoftware.openfire.net.*;
 import org.jivesoftware.openfire.nio.ClientConnectionHandler;
+import org.jivesoftware.openfire.nio.ComponentConnectionHandler;
 import org.jivesoftware.openfire.nio.MultiplexerConnectionHandler;
 import org.jivesoftware.openfire.nio.XMPPCodecFactory;
 import org.jivesoftware.util.*;
@@ -53,7 +54,7 @@ public class ConnectionManagerImpl extends BasicModule implements ConnectionMana
 
     private SocketAcceptor socketAcceptor;
     private SocketAcceptor sslSocketAcceptor;
-    private SocketAcceptThread componentSocketThread;
+    private SocketAcceptor componentAcceptor;
     private SocketAcceptThread serverSocketThread;
     private SocketAcceptor multiplexerSocketAcceptor;
     private ArrayList<ServerPort> ports;
@@ -83,7 +84,7 @@ public class ConnectionManagerImpl extends BasicModule implements ConnectionMana
         // Create the port listener for Connections Multiplexers
         createConnectionManagerListener();
         // Create the port listener for external components
-        createComponentListener(localIPAddress);
+        createComponentListener();
         // Create the port listener for clients
         createClientListeners();
         // Create the port listener for secured clients
@@ -243,23 +244,22 @@ public class ConnectionManagerImpl extends BasicModule implements ConnectionMana
         }
     }
 
-    private void createComponentListener(String localIPAddress) {
+    private void createComponentListener() {
         // Start components socket unless it's been disabled.
         if (isComponentListenerEnabled()) {
-            int port = getComponentListenerPort();
-            try {
-                componentSocketThread = new SocketAcceptThread(this, new ServerPort(port,
-                        serverName, localIPAddress, false, null, ServerPort.Type.component));
-                ports.add(componentSocketThread.getServerPort());
-                componentSocketThread.setDaemon(true);
-                componentSocketThread.setPriority(Thread.MAX_PRIORITY);
+            // Create SocketAcceptor with correct number of processors
+            componentAcceptor = buildSocketAcceptor();
+            // Customize Executor that will be used by processors to process incoming stanzas
+            ExecutorThreadModel threadModel = ExecutorThreadModel.getInstance("component");
+            int eventThreads = JiveGlobals.getIntProperty("xmpp.component.processing.threads", 16);
+            ThreadPoolExecutor eventExecutor = (ThreadPoolExecutor)threadModel.getExecutor();
+            eventExecutor.setCorePoolSize(eventThreads + 1);
+            eventExecutor.setMaximumPoolSize(eventThreads + 1);
+            eventExecutor.setKeepAliveTime(60, TimeUnit.SECONDS);
 
-            }
-            catch (Exception e) {
-                System.err.println("Error starting component listener on port " + port + ": " +
-                        e.getMessage());
-                Log.error(LocaleUtils.getLocalizedString("admin.error.socket-setup"), e);
-            }
+            componentAcceptor.getDefaultConfig().setThreadModel(threadModel);
+            // Add the XMPP codec filter
+            componentAcceptor.getFilterChain().addFirst("xmpp", new ProtocolCodecFilter(new XMPPCodecFactory()));
         }
     }
 
@@ -268,10 +268,22 @@ public class ConnectionManagerImpl extends BasicModule implements ConnectionMana
         if (isComponentListenerEnabled()) {
             int port = getComponentListenerPort();
             try {
-                componentSocketThread.start();
+                // Listen on a specific network interface if it has been set.
+                String interfaceName = JiveGlobals.getXMLProperty("network.interface");
+                InetAddress bindInterface = null;
+                if (interfaceName != null) {
+                    if (interfaceName.trim().length() > 0) {
+                        bindInterface = InetAddress.getByName(interfaceName);
+                    }
+                }
+                // Start accepting connections
+                componentAcceptor
+                        .bind(new InetSocketAddress(bindInterface, port), new ComponentConnectionHandler(serverName));
+
+                ports.add(new ServerPort(port, serverName, localIPAddress, false, null, ServerPort.Type.component));
 
                 List<String> params = new ArrayList<String>();
-                params.add(Integer.toString(componentSocketThread.getPort()));
+                params.add(Integer.toString(port));
                 Log.info(LocaleUtils.getLocalizedString("startup.component", params));
             }
             catch (Exception e) {
@@ -283,10 +295,15 @@ public class ConnectionManagerImpl extends BasicModule implements ConnectionMana
     }
 
     private void stopComponentListener() {
-        if (componentSocketThread != null) {
-            componentSocketThread.shutdown();
-            ports.remove(componentSocketThread.getServerPort());
-            componentSocketThread = null;
+        if (componentAcceptor != null) {
+            componentAcceptor.unbindAll();
+            for (ServerPort port : ports) {
+                if (port.isComponentPort()) {
+                    ports.remove(port);
+                    break;
+                }
+            }
+            componentAcceptor = null;
         }
     }
 
@@ -488,12 +505,7 @@ public class ConnectionManagerImpl extends BasicModule implements ConnectionMana
 
     public SocketReader createSocketReader(Socket sock, boolean isSecure, ServerPort serverPort,
             boolean useBlockingMode) throws IOException {
-        if (serverPort.isComponentPort()) {
-            SocketConnection conn = new SocketConnection(deliverer, sock, isSecure);
-            return new ComponentSocketReader(router, routingTable, serverName, sock, conn,
-                    useBlockingMode);
-        }
-        else if (serverPort.isServerPort()) {
+        if (serverPort.isServerPort()) {
             SocketConnection conn = new SocketConnection(deliverer, sock, isSecure);
             return new ServerSocketReader(router, routingTable, serverName, sock, conn,
                     useBlockingMode);
@@ -578,7 +590,7 @@ public class ConnectionManagerImpl extends BasicModule implements ConnectionMana
         if (enabled) {
             JiveGlobals.setProperty("xmpp.component.socket.active", "true");
             // Start the port listener for external components
-            createComponentListener(localIPAddress);
+            createComponentListener();
             startComponentListener();
         }
         else {
@@ -692,7 +704,7 @@ public class ConnectionManagerImpl extends BasicModule implements ConnectionMana
         stopComponentListener();
         if (isComponentListenerEnabled()) {
             // Start the port listener for external components
-            createComponentListener(localIPAddress);
+            createComponentListener();
             startComponentListener();
         }
     }