SocketReader.java 19.7 KB
Newer Older
1 2 3 4 5
/**
 * $RCSfile$
 * $Revision: 3187 $
 * $Date: 2005-12-11 13:34:34 -0300 (Sun, 11 Dec 2005) $
 *
6
 * Copyright (C) 2005-2008 Jive Software. All rights reserved.
7 8
 *
 * This software is published under the terms of the GNU Public License (GPL),
9 10
 * a copy of which is included in this distribution, or a commercial license
 * agreement with Jive.
11 12
 */

13
package org.jivesoftware.openfire.net;
14 15 16

import org.dom4j.Element;
import org.dom4j.io.XMPPPacketReader;
17 18 19 20
import org.jivesoftware.openfire.Connection;
import org.jivesoftware.openfire.PacketRouter;
import org.jivesoftware.openfire.RoutingTable;
import org.jivesoftware.openfire.auth.UnauthorizedException;
21
import org.jivesoftware.openfire.session.LocalSession;
22
import org.jivesoftware.openfire.session.Session;
Gaston Dombiak's avatar
Gaston Dombiak committed
23 24 25
import org.jivesoftware.util.LocaleUtils;
import org.jivesoftware.util.Log;
import org.jivesoftware.util.StringUtils;
26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50
import org.xmlpull.v1.XmlPullParser;
import org.xmlpull.v1.XmlPullParserException;
import org.xmlpull.v1.XmlPullParserFactory;
import org.xmpp.packet.*;

import java.io.IOException;
import java.net.Socket;

/**
 * A SocketReader creates the appropriate {@link Session} based on the defined namespace in the
 * stream element and will then keep reading and routing the received packets.
 *
 * @author Gaston Dombiak
 */
public abstract class SocketReader implements Runnable {

    /**
     * The utf-8 charset for decoding and encoding Jabber packet streams.
     */
    private static String CHARSET = "UTF-8";
    /**
     * Reuse the same factory for all the connections.
     */
    private static XmlPullParserFactory factory = null;

Gaston Dombiak's avatar
Gaston Dombiak committed
51 52 53
    /**
     * Session associated with the socket reader.
     */
54
    protected LocalSession session;
Gaston Dombiak's avatar
Gaston Dombiak committed
55 56 57
    /**
     * Reference to the physical connection.
     */
58
    protected SocketConnection connection;
Gaston Dombiak's avatar
Gaston Dombiak committed
59 60 61
    /**
     * Server name for which we are attending clients.
     */
62 63 64 65 66 67
    protected String serverName;

    /**
     * Router used to route incoming packets to the correct channels.
     */
    private PacketRouter router;
Gaston Dombiak's avatar
Gaston Dombiak committed
68 69 70 71 72 73 74
    /**
     * Routing table used for checking whether a domain is known or not.
     */
    private RoutingTable routingTable;
    /**
     * Specifies whether the socket is using blocking or non-blocking connections.
     */
75
    private SocketReadingMode readingMode;
76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91
    XMPPPacketReader reader = null;
    protected boolean open;

    static {
        try {
            factory = XmlPullParserFactory.newInstance(MXParser.class.getName(), null);
        }
        catch (XmlPullParserException e) {
            Log.error("Error creating a parser factory", e);
        }
    }

    /**
     * Creates a dedicated reader for a socket.
     *
     * @param router the router for sending packets that were read.
Gaston Dombiak's avatar
Gaston Dombiak committed
92
     * @param routingTable the table that keeps routes to registered services.
93 94 95
     * @param serverName the name of the server this socket is working for.
     * @param socket the socket to read from.
     * @param connection the connection being read.
96
     * @param useBlockingMode true means that the server will use a thread per connection.
97
     */
Gaston Dombiak's avatar
Gaston Dombiak committed
98 99
    public SocketReader(PacketRouter router, RoutingTable routingTable, String serverName,
            Socket socket, SocketConnection connection, boolean useBlockingMode) {
100 101
        this.serverName = serverName;
        this.router = router;
Gaston Dombiak's avatar
Gaston Dombiak committed
102
        this.routingTable = routingTable;
103 104 105
        this.connection = connection;

        connection.setSocketReader(this);
106 107 108 109 110 111

        // Reader is associated with a new XMPPPacketReader
        reader = new XMPPPacketReader();
        reader.setXPPFactory(factory);

        // Set the blocking reading mode to use
112
        readingMode = new BlockingReadingMode(socket, this);
113 114 115 116 117 118 119
    }

    /**
     * A dedicated thread loop for reading the stream and sending incoming
     * packets to the appropriate router.
     */
    public void run() {
120 121
        readingMode.run();
    }
122

123 124 125 126
    protected void process(Element doc) throws Exception {
        if (doc == null) {
            return;
        }
127

128 129 130
        String tag = doc.getName();
        if ("message".equals(tag)) {
            Message packet;
131
            try {
132
                packet = new Message(doc);
133
            }
134
            catch(IllegalArgumentException e) {
135
                Log.debug("SocketReader: Rejecting packet. JID malformed", e);
136 137 138 139 140 141 142 143
                // The original packet contains a malformed JID so answer with an error.
                Message reply = new Message();
                reply.setID(doc.attributeValue("id"));
                reply.setTo(session.getAddress());
                reply.getElement().addAttribute("from", doc.attributeValue("to"));
                reply.setError(PacketError.Condition.jid_malformed);
                session.process(reply);
                return;
144
            }
145
            processMessage(packet);
146
        }
147 148 149 150
        else if ("presence".equals(tag)) {
            Presence packet;
            try {
                packet = new Presence(doc);
151
            }
152
            catch (IllegalArgumentException e) {
153
                Log.debug("SocketReader: Rejecting packet. JID malformed", e);
154 155 156 157 158 159 160
                // The original packet contains a malformed JID so answer an error
                Presence reply = new Presence();
                reply.setID(doc.attributeValue("id"));
                reply.setTo(session.getAddress());
                reply.getElement().addAttribute("from", doc.attributeValue("to"));
                reply.setError(PacketError.Condition.jid_malformed);
                session.process(reply);
161 162
                return;
            }
163 164 165
            // Check that the presence type is valid. If not then assume available type
            try {
                packet.getType();
166
            }
167 168 169 170 171
            catch (IllegalArgumentException e) {
                Log.warn("Invalid presence type", e);
                // The presence packet contains an invalid presence type so replace it with
                // an available presence type
                packet.setType(null);
172
            }
173 174 175
            // Check that the presence show is valid. If not then assume available show value
            try {
                packet.getShow();
176
            }
177 178 179 180 181
            catch (IllegalArgumentException e) {
                Log.warn("Invalid presence show", e);
                // The presence packet contains an invalid presence show so replace it with
                // an available presence show
                packet.setShow(null);
182
            }
183 184 185 186 187 188
            if (session.getStatus() == Session.STATUS_CLOSED && packet.isAvailable()) {
                // Ignore available presence packets sent from a closed session. A closed
                // session may have buffered data pending to be processes so we want to ignore
                // just Presences of type available
                Log.warn("Ignoring available presence packet of closed session: " + packet);
                return;
189
            }
190 191 192 193 194 195
            processPresence(packet);
        }
        else if ("iq".equals(tag)) {
            IQ packet;
            try {
                packet = getIQ(doc);
196
            }
197
            catch(IllegalArgumentException e) {
198
                Log.debug("SocketReader: Rejecting packet. JID malformed", e);
199 200 201 202
                // The original packet contains a malformed JID so answer an error
                IQ reply = new IQ();
                if (!doc.elements().isEmpty()) {
                    reply.setChildElement(((Element) doc.elements().get(0)).createCopy());
203
                }
204 205 206 207
                reply.setID(doc.attributeValue("id"));
                reply.setTo(session.getAddress());
                if (doc.attributeValue("to") != null) {
                    reply.getElement().addAttribute("from", doc.attributeValue("to"));
208
                }
209 210 211
                reply.setError(PacketError.Condition.jid_malformed);
                session.process(reply);
                return;
212
            }
213
            processIQ(packet);
214
        }
215 216 217 218 219 220
        else
        {
            if (!processUnknowPacket(doc)) {
                Log.warn(LocaleUtils.getLocalizedString("admin.error.packet.tag") +
                        doc.asXML());
                open = false;
221 222 223 224 225 226
            }
        }
    }

    /**
     * Process the received IQ packet. Registered
227
     * {@link org.jivesoftware.openfire.interceptor.PacketInterceptor} will be invoked before
228 229 230 231 232 233 234
     * and after the packet was routed.<p>
     *
     * Subclasses may redefine this method for different reasons such as modifying the sender
     * of the packet to avoid spoofing, rejecting the packet or even process the packet in
     * another thread.
     *
     * @param packet the received packet.
235
     * @throws UnauthorizedException if the connection required security but was not secured.
236 237 238 239 240 241 242 243
     */
    protected void processIQ(IQ packet) throws UnauthorizedException {
        // Ensure that connection was secured if TLS was required
        if (connection.getTlsPolicy() == Connection.TLSPolicy.required &&
                !connection.isSecure()) {
            closeNeverSecuredConnection();
            return;
        }
244 245
        router.route(packet);
        session.incrementClientPacketCount();
246 247 248 249
    }

    /**
     * Process the received Presence packet. Registered
250
     * {@link org.jivesoftware.openfire.interceptor.PacketInterceptor} will be invoked before
251 252 253 254 255 256 257
     * and after the packet was routed.<p>
     *
     * Subclasses may redefine this method for different reasons such as modifying the sender
     * of the packet to avoid spoofing, rejecting the packet or even process the packet in
     * another thread.
     *
     * @param packet the received packet.
258
     * @throws UnauthorizedException if the connection required security but was not secured.
259 260 261 262 263 264 265 266
     */
    protected void processPresence(Presence packet) throws UnauthorizedException {
        // Ensure that connection was secured if TLS was required
        if (connection.getTlsPolicy() == Connection.TLSPolicy.required &&
                !connection.isSecure()) {
            closeNeverSecuredConnection();
            return;
        }
267 268
        router.route(packet);
        session.incrementClientPacketCount();
269 270 271 272
    }

    /**
     * Process the received Message packet. Registered
273
     * {@link org.jivesoftware.openfire.interceptor.PacketInterceptor} will be invoked before
274 275 276 277 278 279 280
     * and after the packet was routed.<p>
     *
     * Subclasses may redefine this method for different reasons such as modifying the sender
     * of the packet to avoid spoofing, rejecting the packet or even process the packet in
     * another thread.
     *
     * @param packet the received packet.
281
     * @throws UnauthorizedException if the connection required security but was not secured.
282 283 284 285 286 287 288 289
     */
    protected void processMessage(Message packet) throws UnauthorizedException {
        // Ensure that connection was secured if TLS was required
        if (connection.getTlsPolicy() == Connection.TLSPolicy.required &&
                !connection.isSecure()) {
            closeNeverSecuredConnection();
            return;
        }
290 291
        router.route(packet);
        session.incrementClientPacketCount();
292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313
    }

    /**
     * Returns true if a received packet of an unkown type (i.e. not a Message, Presence
     * or IQ) has been processed. If the packet was not processed then an exception will
     * be thrown which will make the thread to stop processing further packets.
     *
     * @param doc the DOM element of an unkown type.
     * @return  true if a received packet has been processed.
     */
    abstract boolean processUnknowPacket(Element doc);

    /**
     * Returns the last time a full Document was read or a heartbeat was received. Hearbeats
     * are represented as whitespaces received while a Document is not being parsed.
     *
     * @return the time in milliseconds when the last document or heartbeat was received.
     */
    long getLastActive() {
        return reader.getLastActive();
    }

314 315 316 317 318 319 320
    /**
     * Returns a name that identifies the type of reader and the unique instance.
     *
     * @return a name that identifies the type of reader and the unique instance.
     */
    abstract String getName();

321 322 323 324
    /**
     * Close the connection since TLS was mandatory and the entity never negotiated TLS. Before
     * closing the connection a stream error will be sent to the entity.
     */
325
    void closeNeverSecuredConnection() {
326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354
        // Set the not_authorized error
        StreamError error = new StreamError(StreamError.Condition.not_authorized);
        // Deliver stanza
        connection.deliverRawText(error.toXML());
        // Close the underlying connection
        connection.close();
        // Log a warning so that admins can track this case from the server side
        Log.warn("TLS was required by the server and connection was never secured. " +
                "Closing connection : " + connection);
    }

    private IQ getIQ(Element doc) {
        Element query = doc.element("query");
        if (query != null && "jabber:iq:roster".equals(query.getNamespaceURI())) {
            return new Roster(doc);
        }
        else {
            return new IQ(doc);
        }
    }

    /**
     * Uses the XPP to grab the opening stream tag and create an active session
     * object. The session to create will depend on the sent namespace. In all
     * cases, the method obtains the opening stream tag, checks for errors, and
     * either creates a session or returns an error and kills the connection.
     * If the connection remains open, the XPP will be set to be ready for the
     * first packet. A call to next() should result in an START_TAG state with
     * the first packet in the stream.
355 356 357 358
     *
     * @throws UnauthorizedException if the connection required security but was not secured.
     * @throws XmlPullParserException if there was an XML error while creating the session.
     * @throws IOException if an IO error occured while creating the session.
359
     */
360 361
    protected void createSession()
            throws UnauthorizedException, XmlPullParserException, IOException {
362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390
        XmlPullParser xpp = reader.getXPPParser();
        for (int eventType = xpp.getEventType(); eventType != XmlPullParser.START_TAG;) {
            eventType = xpp.next();
        }

        // Check that the TO attribute of the stream header matches the server name or a valid
        // subdomain. If the value of the 'to' attribute is not valid then return a host-unknown
        // error and close the underlying connection.
        String host = reader.getXPPParser().getAttributeValue("", "to");
        if (validateHost() && isHostUnknown(host)) {
            StringBuilder sb = new StringBuilder(250);
            sb.append("<?xml version='1.0' encoding='");
            sb.append(CHARSET);
            sb.append("'?>");
            // Append stream header
            sb.append("<stream:stream ");
            sb.append("from=\"").append(serverName).append("\" ");
            sb.append("id=\"").append(StringUtils.randomString(5)).append("\" ");
            sb.append("xmlns=\"").append(xpp.getNamespace(null)).append("\" ");
            sb.append("xmlns:stream=\"").append(xpp.getNamespace("stream")).append("\" ");
            sb.append("version=\"1.0\">");
            // Set the host_unknown error
            StreamError error = new StreamError(StreamError.Condition.host_unknown);
            sb.append(error.toXML());
            // Deliver stanza
            connection.deliverRawText(sb.toString());
            // Close the underlying connection
            connection.close();
            // Log a warning so that admins can track this cases from the server side
391 392
            Log.warn("Closing session due to incorrect hostname in stream header. Host: " + host +
                    ". Connection: " + connection);
393 394 395 396 397
        }

        // Create the correct session based on the sent namespace. At this point the server
        // may offer the client to secure the connection. If the client decides to secure
        // the connection then a <starttls> stanza should be received
398
        else if (!createSession(xpp.getNamespace(null))) {
399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418
            // No session was created because of an invalid namespace prefix so answer a stream
            // error and close the underlying connection
            StringBuilder sb = new StringBuilder(250);
            sb.append("<?xml version='1.0' encoding='");
            sb.append(CHARSET);
            sb.append("'?>");
            // Append stream header
            sb.append("<stream:stream ");
            sb.append("from=\"").append(serverName).append("\" ");
            sb.append("id=\"").append(StringUtils.randomString(5)).append("\" ");
            sb.append("xmlns=\"").append(xpp.getNamespace(null)).append("\" ");
            sb.append("xmlns:stream=\"").append(xpp.getNamespace("stream")).append("\" ");
            sb.append("version=\"1.0\">");
            // Include the bad-namespace-prefix in the response
            StreamError error = new StreamError(StreamError.Condition.bad_namespace_prefix);
            sb.append(error.toXML());
            connection.deliverRawText(sb.toString());
            // Close the underlying connection
            connection.close();
            // Log a warning so that admins can track this cases from the server side
419 420
            Log.warn("Closing session due to bad_namespace_prefix in stream header. Prefix: " +
                    xpp.getNamespace(null) + ". Connection: " + connection);
421 422 423 424 425 426 427 428 429 430 431 432 433 434
        }
    }

    private boolean isHostUnknown(String host) {
        if (host == null) {
            // Answer false since when using server dialback the stream header will not
            // have a TO attribute
            return false;
        }
        if (serverName.equals(host)) {
            // requested host matched the server name
            return false;
        }
        // Check if the host matches a subdomain of this host
Gaston Dombiak's avatar
Gaston Dombiak committed
435
        return !routingTable.hasComponentRoute(new JID(host));
436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463
    }

    /**
     * Returns the stream namespace. (E.g. jabber:client, jabber:server, etc.).
     *
     * @return the stream namespace.
     */
    abstract String getNamespace();

    /**
     * Returns true if the value of the 'to' attribute in the stream header should be
     * validated. If the value of the 'to' attribute is not valid then a host-unknown error
     * will be returned and the underlying connection will be closed.
     *
     * @return true if the value of the 'to' attribute in the initial stream header should be
     *         validated.
     */
    abstract boolean validateHost();

    /**
     * Notification message indicating that the SocketReader is shutting down. The thread will
     * stop reading and processing new requests. Subclasses may want to redefine this message
     * for releasing any resource they might need.
     */
    protected void shutdown() {
    }

    /**
464
     * Creates the appropriate {@link org.jivesoftware.openfire.session.Session} subclass based on the specified namespace.
465 466 467
     *
     * @param namespace the namespace sent in the stream element. eg. jabber:client.
     * @return the created session or null.
468 469 470
     * @throws UnauthorizedException if the connection required security but was not secured.
     * @throws XmlPullParserException if there was an XML error while creating the session.
     * @throws IOException if an IO error occured while creating the session.
471
     */
472
    abstract boolean createSession(String namespace) throws UnauthorizedException,
473 474
            XmlPullParserException, IOException;
}