1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
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
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
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
355
356
357
358
359
360
361
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
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
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
464
465
466
467
468
469
470
471
472
473
/**
* $RCSfile$
* $Revision: 3187 $
* $Date: 2005-12-11 13:34:34 -0300 (Sun, 11 Dec 2005) $
*
* Copyright (C) 2007 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.wildfire.net;
import org.dom4j.Element;
import org.dom4j.io.XMPPPacketReader;
import org.jivesoftware.util.LocaleUtils;
import org.jivesoftware.util.Log;
import org.jivesoftware.util.StringUtils;
import org.jivesoftware.wildfire.Connection;
import org.jivesoftware.wildfire.PacketRouter;
import org.jivesoftware.wildfire.RoutableChannelHandler;
import org.jivesoftware.wildfire.RoutingTable;
import org.jivesoftware.wildfire.auth.UnauthorizedException;
import org.jivesoftware.wildfire.server.OutgoingSessionPromise;
import org.jivesoftware.wildfire.session.Session;
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;
/**
* Session associated with the socket reader.
*/
protected Session session;
/**
* Reference to the physical connection.
*/
protected SocketConnection connection;
/**
* Server name for which we are attending clients.
*/
protected String serverName;
/**
* Router used to route incoming packets to the correct channels.
*/
private PacketRouter router;
/**
* 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.
*/
private SocketReadingMode readingMode;
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.
* @param routingTable the table that keeps routes to registered services.
* @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.
* @param useBlockingMode true means that the server will use a thread per connection.
*/
public SocketReader(PacketRouter router, RoutingTable routingTable, String serverName,
Socket socket, SocketConnection connection, boolean useBlockingMode) {
this.serverName = serverName;
this.router = router;
this.routingTable = routingTable;
this.connection = connection;
connection.setSocketReader(this);
// Reader is associated with a new XMPPPacketReader
reader = new XMPPPacketReader();
reader.setXPPFactory(factory);
// Set the blocking reading mode to use
readingMode = new BlockingReadingMode(socket, this);
}
/**
* A dedicated thread loop for reading the stream and sending incoming
* packets to the appropriate router.
*/
public void run() {
readingMode.run();
}
protected void process(Element doc) throws Exception {
if (doc == null) {
return;
}
String tag = doc.getName();
if ("message".equals(tag)) {
Message packet;
try {
packet = new Message(doc);
}
catch(IllegalArgumentException e) {
Log.debug("Rejecting packet. JID malformed", e);
// 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;
}
processMessage(packet);
}
else if ("presence".equals(tag)) {
Presence packet;
try {
packet = new Presence(doc);
}
catch (IllegalArgumentException e) {
Log.debug("Rejecting packet. JID malformed", e);
// 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);
return;
}
// Check that the presence type is valid. If not then assume available type
try {
packet.getType();
}
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);
}
// Check that the presence show is valid. If not then assume available show value
try {
packet.getShow();
}
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);
}
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;
}
processPresence(packet);
}
else if ("iq".equals(tag)) {
IQ packet;
try {
packet = getIQ(doc);
}
catch(IllegalArgumentException e) {
Log.debug("Rejecting packet. JID malformed", e);
// 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());
}
reply.setID(doc.attributeValue("id"));
reply.setTo(session.getAddress());
if (doc.attributeValue("to") != null) {
reply.getElement().addAttribute("from", doc.attributeValue("to"));
}
reply.setError(PacketError.Condition.jid_malformed);
session.process(reply);
return;
}
processIQ(packet);
}
else
{
if (!processUnknowPacket(doc)) {
Log.warn(LocaleUtils.getLocalizedString("admin.error.packet.tag") +
doc.asXML());
open = false;
}
}
}
/**
* Process the received IQ packet. Registered
* {@link org.jivesoftware.wildfire.interceptor.PacketInterceptor} will be invoked before
* 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.
*/
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;
}
router.route(packet);
session.incrementClientPacketCount();
}
/**
* Process the received Presence packet. Registered
* {@link org.jivesoftware.wildfire.interceptor.PacketInterceptor} will be invoked before
* 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.
*/
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;
}
router.route(packet);
session.incrementClientPacketCount();
}
/**
* Process the received Message packet. Registered
* {@link org.jivesoftware.wildfire.interceptor.PacketInterceptor} will be invoked before
* 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.
*/
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;
}
router.route(packet);
session.incrementClientPacketCount();
}
/**
* 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();
}
/**
* 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();
/**
* 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.
*/
void closeNeverSecuredConnection() {
// 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.
*/
protected void createSession()
throws UnauthorizedException, XmlPullParserException, IOException {
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
Log.warn("Closing session due to incorrect hostname in stream header. Host: " + host +
". Connection: " + connection);
}
// 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
else if (!createSession(xpp.getNamespace(null))) {
// 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
Log.warn("Closing session due to bad_namespace_prefix in stream header. Prefix: " +
xpp.getNamespace(null) + ". Connection: " + connection);
}
}
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
RoutableChannelHandler route = routingTable.getRoute(new JID(host));
if (route == null || route instanceof OutgoingSessionPromise) {
return true;
}
else {
return false;
}
}
/**
* 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() {
}
/**
* Creates the appropriate {@link org.jivesoftware.wildfire.session.Session} subclass based on the specified namespace.
*
* @param namespace the namespace sent in the stream element. eg. jabber:client.
* @return the created session or null.
* @throws UnauthorizedException
* @throws XmlPullParserException
* @throws IOException
*/
abstract boolean createSession(String namespace) throws UnauthorizedException,
XmlPullParserException, IOException;
}