src/org/sonews/daemon/NNTPConnection.java
author František Kučera <franta-hg@frantovo.cz>
Tue Oct 25 10:39:57 2011 +0200 (2011-10-25)
changeset 108 fdc075324ef3
parent 101 d54786065fa3
child 113 a059aecd1794
permissions -rwxr-xr-x
SMTP: correct escaping of messages containing lines with single dot.
chris@1
     1
/*
chris@1
     2
 *   SONEWS News Server
chris@1
     3
 *   see AUTHORS for the list of contributors
chris@1
     4
 *
chris@1
     5
 *   This program is free software: you can redistribute it and/or modify
chris@1
     6
 *   it under the terms of the GNU General Public License as published by
chris@1
     7
 *   the Free Software Foundation, either version 3 of the License, or
chris@1
     8
 *   (at your option) any later version.
chris@1
     9
 *
chris@1
    10
 *   This program is distributed in the hope that it will be useful,
chris@1
    11
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
chris@1
    12
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
chris@1
    13
 *   GNU General Public License for more details.
chris@1
    14
 *
chris@1
    15
 *   You should have received a copy of the GNU General Public License
chris@1
    16
 *   along with this program.  If not, see <http://www.gnu.org/licenses/>.
chris@1
    17
 */
chris@1
    18
package org.sonews.daemon;
chris@1
    19
franta-hg@108
    20
import java.io.ByteArrayOutputStream;
chris@1
    21
import java.io.IOException;
chris@1
    22
import java.net.InetSocketAddress;
chris@3
    23
import java.net.SocketException;
chris@1
    24
import java.nio.ByteBuffer;
chris@1
    25
import java.nio.CharBuffer;
chris@1
    26
import java.nio.channels.ClosedChannelException;
chris@1
    27
import java.nio.channels.SelectionKey;
chris@1
    28
import java.nio.channels.SocketChannel;
chris@1
    29
import java.nio.charset.Charset;
chris@3
    30
import java.util.Arrays;
chris@1
    31
import java.util.Timer;
chris@1
    32
import java.util.TimerTask;
cli@49
    33
import java.util.logging.Level;
chris@3
    34
import org.sonews.daemon.command.Command;
chris@3
    35
import org.sonews.storage.Article;
cli@48
    36
import org.sonews.storage.Group;
cli@30
    37
import org.sonews.storage.StorageBackendException;
cli@25
    38
import org.sonews.util.Log;
chris@1
    39
import org.sonews.util.Stats;
franta-hg@108
    40
import org.sonews.util.io.CRLFOutputStream;
franta-hg@108
    41
import org.sonews.util.io.SMTPOutputStream;
chris@1
    42
chris@1
    43
/**
chris@1
    44
 * For every SocketChannel (so TCP/IP connection) there is an instance of
chris@1
    45
 * this class.
chris@1
    46
 * @author Christian Lins
chris@1
    47
 * @since sonews/0.5.0
chris@1
    48
 */
cli@49
    49
public final class NNTPConnection {
chris@1
    50
cli@37
    51
	public static final String NEWLINE = "\r\n";    // RFC defines this as newline
cli@37
    52
	public static final String MESSAGE_ID_PATTERN = "<[^>]+>";
cli@37
    53
	private static final Timer cancelTimer = new Timer(true); // Thread-safe? True for run as daemon
cli@37
    54
	/** SocketChannel is generally thread-safe */
cli@37
    55
	private SocketChannel channel = null;
cli@37
    56
	private Charset charset = Charset.forName("UTF-8");
cli@37
    57
	private Command command = null;
cli@37
    58
	private Article currentArticle = null;
cli@48
    59
	private Group currentGroup = null;
cli@37
    60
	private volatile long lastActivity = System.currentTimeMillis();
cli@37
    61
	private ChannelLineBuffers lineBuffers = new ChannelLineBuffers();
cli@37
    62
	private int readLock = 0;
cli@37
    63
	private final Object readLockGate = new Object();
cli@37
    64
	private SelectionKey writeSelKey = null;
franta-hg@101
    65
	
franta-hg@101
    66
	private String username;
franta-hg@101
    67
	private boolean userAuthenticated = false;
chris@1
    68
cli@37
    69
	public NNTPConnection(final SocketChannel channel)
cli@49
    70
			throws IOException {
cli@37
    71
		if (channel == null) {
cli@37
    72
			throw new IllegalArgumentException("channel is null");
cli@37
    73
		}
chris@1
    74
cli@37
    75
		this.channel = channel;
cli@37
    76
		Stats.getInstance().clientConnect();
cli@37
    77
	}
chris@1
    78
cli@37
    79
	/**
cli@37
    80
	 * Tries to get the read lock for this NNTPConnection. This method is Thread-
cli@37
    81
	 * safe and returns true of the read lock was successfully set. If the lock
cli@37
    82
	 * is still hold by another Thread the method returns false.
cli@37
    83
	 */
cli@49
    84
	boolean tryReadLock() {
cli@37
    85
		// As synchronizing simple types may cause deadlocks,
cli@37
    86
		// we use a gate object.
cli@37
    87
		synchronized (readLockGate) {
cli@37
    88
			if (readLock != 0) {
cli@37
    89
				return false;
cli@37
    90
			} else {
cli@37
    91
				readLock = Thread.currentThread().hashCode();
cli@37
    92
				return true;
cli@37
    93
			}
cli@37
    94
		}
cli@37
    95
	}
chris@3
    96
cli@37
    97
	/**
cli@37
    98
	 * Releases the read lock in a Thread-safe way.
cli@37
    99
	 * @throws IllegalMonitorStateException if a Thread not holding the lock
cli@37
   100
	 * tries to release it.
cli@37
   101
	 */
cli@49
   102
	void unlockReadLock() {
cli@37
   103
		synchronized (readLockGate) {
cli@37
   104
			if (readLock == Thread.currentThread().hashCode()) {
cli@37
   105
				readLock = 0;
cli@37
   106
			} else {
cli@37
   107
				throw new IllegalMonitorStateException();
cli@37
   108
			}
cli@37
   109
		}
cli@37
   110
	}
chris@1
   111
cli@37
   112
	/**
cli@37
   113
	 * @return Current input buffer of this NNTPConnection instance.
cli@37
   114
	 */
cli@49
   115
	public ByteBuffer getInputBuffer() {
cli@37
   116
		return this.lineBuffers.getInputBuffer();
cli@37
   117
	}
chris@1
   118
cli@37
   119
	/**
cli@37
   120
	 * @return Output buffer of this NNTPConnection which has at least one byte
cli@37
   121
	 * free storage.
cli@37
   122
	 */
cli@49
   123
	public ByteBuffer getOutputBuffer() {
cli@37
   124
		return this.lineBuffers.getOutputBuffer();
cli@37
   125
	}
cli@30
   126
cli@37
   127
	/**
cli@37
   128
	 * @return ChannelLineBuffers instance associated with this NNTPConnection.
cli@37
   129
	 */
cli@49
   130
	public ChannelLineBuffers getBuffers() {
cli@37
   131
		return this.lineBuffers;
cli@37
   132
	}
chris@1
   133
cli@37
   134
	/**
cli@37
   135
	 * @return true if this connection comes from a local remote address.
cli@37
   136
	 */
cli@49
   137
	public boolean isLocalConnection() {
cli@37
   138
		return ((InetSocketAddress) this.channel.socket().getRemoteSocketAddress()).getHostName().equalsIgnoreCase("localhost");
cli@37
   139
	}
chris@3
   140
cli@49
   141
	void setWriteSelectionKey(SelectionKey selKey) {
cli@37
   142
		this.writeSelKey = selKey;
cli@37
   143
	}
chris@3
   144
cli@49
   145
	public void shutdownInput() {
cli@37
   146
		try {
cli@37
   147
			// Closes the input line of the channel's socket, so no new data
cli@37
   148
			// will be received and a timeout can be triggered.
cli@37
   149
			this.channel.socket().shutdownInput();
cli@37
   150
		} catch (IOException ex) {
cli@37
   151
			Log.get().warning("Exception in NNTPConnection.shutdownInput(): " + ex);
cli@37
   152
		}
cli@37
   153
	}
chris@1
   154
cli@49
   155
	public void shutdownOutput() {
cli@49
   156
		cancelTimer.schedule(new TimerTask() {
cli@37
   157
			@Override
cli@49
   158
			public void run() {
cli@37
   159
				try {
cli@37
   160
					// Closes the output line of the channel's socket.
cli@37
   161
					channel.socket().shutdownOutput();
cli@37
   162
					channel.close();
cli@37
   163
				} catch (SocketException ex) {
cli@37
   164
					// Socket was already disconnected
cli@37
   165
					Log.get().info("NNTPConnection.shutdownOutput(): " + ex);
cli@37
   166
				} catch (Exception ex) {
cli@37
   167
					Log.get().warning("NNTPConnection.shutdownOutput(): " + ex);
cli@37
   168
				}
cli@37
   169
			}
cli@37
   170
		}, 3000);
cli@37
   171
	}
cli@37
   172
cli@49
   173
	public SocketChannel getSocketChannel() {
cli@37
   174
		return this.channel;
cli@37
   175
	}
cli@37
   176
cli@49
   177
	public Article getCurrentArticle() {
cli@37
   178
		return this.currentArticle;
cli@37
   179
	}
cli@37
   180
cli@49
   181
	public Charset getCurrentCharset() {
cli@37
   182
		return this.charset;
cli@37
   183
	}
cli@37
   184
cli@37
   185
	/**
cli@37
   186
	 * @return The currently selected communication channel (not SocketChannel)
cli@37
   187
	 */
cli@49
   188
	public Group getCurrentChannel() {
cli@37
   189
		return this.currentGroup;
cli@37
   190
	}
cli@37
   191
cli@49
   192
	public void setCurrentArticle(final Article article) {
cli@37
   193
		this.currentArticle = article;
cli@37
   194
	}
cli@37
   195
cli@49
   196
	public void setCurrentGroup(final Group group) {
cli@37
   197
		this.currentGroup = group;
cli@37
   198
	}
cli@37
   199
cli@49
   200
	public long getLastActivity() {
cli@37
   201
		return this.lastActivity;
cli@37
   202
	}
cli@37
   203
cli@37
   204
	/**
cli@37
   205
	 * Due to the readLockGate there is no need to synchronize this method.
cli@37
   206
	 * @param raw
cli@37
   207
	 * @throws IllegalArgumentException if raw is null.
cli@37
   208
	 * @throws IllegalStateException if calling thread does not own the readLock.
cli@37
   209
	 */
cli@49
   210
	void lineReceived(byte[] raw) {
cli@37
   211
		if (raw == null) {
cli@37
   212
			throw new IllegalArgumentException("raw is null");
cli@37
   213
		}
cli@37
   214
cli@37
   215
		if (readLock == 0 || readLock != Thread.currentThread().hashCode()) {
cli@37
   216
			throw new IllegalStateException("readLock not properly set");
cli@37
   217
		}
cli@37
   218
cli@37
   219
		this.lastActivity = System.currentTimeMillis();
cli@37
   220
cli@37
   221
		String line = new String(raw, this.charset);
cli@37
   222
cli@37
   223
		// There might be a trailing \r, but trim() is a bad idea
cli@37
   224
		// as it removes also leading spaces from long header lines.
cli@37
   225
		if (line.endsWith("\r")) {
cli@37
   226
			line = line.substring(0, line.length() - 1);
cli@37
   227
			raw = Arrays.copyOf(raw, raw.length - 1);
cli@37
   228
		}
cli@37
   229
cli@37
   230
		Log.get().fine("<< " + line);
cli@37
   231
cli@37
   232
		if (command == null) {
cli@37
   233
			command = parseCommandLine(line);
cli@37
   234
			assert command != null;
cli@37
   235
		}
cli@37
   236
cli@37
   237
		try {
cli@37
   238
			// The command object will process the line we just received
cli@37
   239
			try {
cli@37
   240
				command.processLine(this, line, raw);
cli@37
   241
			} catch (StorageBackendException ex) {
cli@37
   242
				Log.get().info("Retry command processing after StorageBackendException");
cli@37
   243
cli@37
   244
				// Try it a second time, so that the backend has time to recover
cli@37
   245
				command.processLine(this, line, raw);
cli@37
   246
			}
cli@37
   247
		} catch (ClosedChannelException ex0) {
cli@37
   248
			try {
cli@49
   249
				StringBuilder strBuf = new StringBuilder();
cli@49
   250
				strBuf.append("Connection to ");
cli@49
   251
				strBuf.append(channel.socket().getRemoteSocketAddress());
cli@49
   252
				strBuf.append(" closed: ");
cli@49
   253
				strBuf.append(ex0);
cli@49
   254
				Log.get().info(strBuf.toString());
cli@37
   255
			} catch (Exception ex0a) {
cli@37
   256
				ex0a.printStackTrace();
cli@37
   257
			}
cli@49
   258
		} catch (Exception ex1) { // This will catch a second StorageBackendException
cli@37
   259
			try {
cli@37
   260
				command = null;
cli@49
   261
				Log.get().log(Level.WARNING, ex1.getLocalizedMessage(), ex1);
cli@49
   262
				println("403 Internal server error");
cli@49
   263
cli@49
   264
				// Should we end the connection here?
cli@49
   265
				// RFC says we MUST return 400 before closing the connection
cli@49
   266
				shutdownInput();
cli@49
   267
				shutdownOutput();
cli@37
   268
			} catch (Exception ex2) {
cli@37
   269
				ex2.printStackTrace();
cli@37
   270
			}
cli@37
   271
		}
cli@37
   272
cli@37
   273
		if (command == null || command.hasFinished()) {
cli@37
   274
			command = null;
cli@37
   275
			charset = Charset.forName("UTF-8"); // Reset to default
cli@37
   276
		}
cli@37
   277
	}
cli@37
   278
cli@37
   279
	/**
cli@37
   280
	 * This method determines the fitting command processing class.
cli@37
   281
	 * @param line
cli@37
   282
	 * @return
cli@37
   283
	 */
cli@49
   284
	private Command parseCommandLine(String line) {
cli@37
   285
		String cmdStr = line.split(" ")[0];
cli@37
   286
		return CommandSelector.getInstance().get(cmdStr);
cli@37
   287
	}
cli@37
   288
cli@37
   289
	/**
cli@37
   290
	 * Puts the given line into the output buffer, adds a newline character
cli@37
   291
	 * and returns. The method returns immediately and does not block until
cli@37
   292
	 * the line was sent. If line is longer than 510 octets it is split up in
cli@37
   293
	 * several lines. Each line is terminated by \r\n (NNTPConnection.NEWLINE).
cli@37
   294
	 * @param line
cli@37
   295
	 */
cli@37
   296
	public void println(final CharSequence line, final Charset charset)
cli@49
   297
			throws IOException {
cli@37
   298
		writeToChannel(CharBuffer.wrap(line), charset, line);
cli@37
   299
		writeToChannel(CharBuffer.wrap(NEWLINE), charset, null);
cli@37
   300
	}
cli@37
   301
cli@37
   302
	/**
cli@37
   303
	 * Writes the given raw lines to the output buffers and finishes with
cli@37
   304
	 * a newline character (\r\n).
cli@37
   305
	 * @param rawLines
cli@37
   306
	 */
cli@37
   307
	public void println(final byte[] rawLines)
cli@49
   308
			throws IOException {
cli@37
   309
		this.lineBuffers.addOutputBuffer(ByteBuffer.wrap(rawLines));
cli@37
   310
		writeToChannel(CharBuffer.wrap(NEWLINE), charset, null);
cli@37
   311
	}
franta-hg@108
   312
	
franta-hg@108
   313
	/**
franta-hg@108
   314
	 * Same as {@link #println(byte[]) } but escapes lines containing single dot,
franta-hg@108
   315
	 * which has special meaning in protocol (end of message).
franta-hg@108
   316
	 * 
franta-hg@108
   317
	 * This method is safe to be used for writing messages – if message contains 
franta-hg@108
   318
	 * a line with single dot, it will be doubled and thus not interpreted 
franta-hg@108
   319
	 * by NNTP client as end of message
franta-hg@108
   320
	 * 
franta-hg@108
   321
	 * @param rawLines
franta-hg@108
   322
	 * @throws IOException 
franta-hg@108
   323
	 */
franta-hg@108
   324
	public void printlnEscapeDots(final byte[] rawLines) throws IOException {
franta-hg@108
   325
		// TODO: optimalizace
franta-hg@108
   326
		
franta-hg@108
   327
		ByteArrayOutputStream baos = new ByteArrayOutputStream(rawLines.length + 10);
franta-hg@108
   328
		CRLFOutputStream crlfStream = new CRLFOutputStream(baos);
franta-hg@108
   329
		SMTPOutputStream smtpStream = new SMTPOutputStream(crlfStream);
franta-hg@108
   330
		smtpStream.write(rawLines);
franta-hg@108
   331
		
franta-hg@108
   332
		println(baos.toByteArray());
franta-hg@108
   333
		
franta-hg@108
   334
		smtpStream.close();
franta-hg@108
   335
	}
cli@37
   336
cli@37
   337
	/**
cli@37
   338
	 * Encodes the given CharBuffer using the given Charset to a bunch of
cli@37
   339
	 * ByteBuffers (each 512 bytes large) and enqueues them for writing at the
cli@37
   340
	 * connected SocketChannel.
cli@37
   341
	 * @throws java.io.IOException
cli@37
   342
	 */
cli@37
   343
	private void writeToChannel(CharBuffer characters, final Charset charset,
cli@49
   344
			CharSequence debugLine)
cli@49
   345
			throws IOException {
cli@37
   346
		if (!charset.canEncode()) {
cli@37
   347
			Log.get().severe("FATAL: Charset " + charset + " cannot encode!");
cli@37
   348
			return;
cli@37
   349
		}
cli@37
   350
cli@37
   351
		// Write characters to output buffers
cli@37
   352
		LineEncoder lenc = new LineEncoder(characters, charset);
cli@37
   353
		lenc.encode(lineBuffers);
cli@37
   354
cli@37
   355
		enableWriteEvents(debugLine);
cli@37
   356
	}
cli@37
   357
cli@49
   358
	private void enableWriteEvents(CharSequence debugLine) {
cli@37
   359
		// Enable OP_WRITE events so that the buffers are processed
cli@37
   360
		try {
cli@37
   361
			this.writeSelKey.interestOps(SelectionKey.OP_WRITE);
cli@37
   362
			ChannelWriter.getInstance().getSelector().wakeup();
cli@37
   363
		} catch (Exception ex) // CancelledKeyException and ChannelCloseException
cli@37
   364
		{
cli@37
   365
			Log.get().warning("NNTPConnection.writeToChannel(): " + ex);
cli@37
   366
			return;
cli@37
   367
		}
cli@37
   368
cli@37
   369
		// Update last activity timestamp
cli@37
   370
		this.lastActivity = System.currentTimeMillis();
cli@37
   371
		if (debugLine != null) {
cli@37
   372
			Log.get().fine(">> " + debugLine);
cli@37
   373
		}
cli@37
   374
	}
cli@37
   375
cli@37
   376
	public void println(final CharSequence line)
cli@49
   377
			throws IOException {
cli@37
   378
		println(line, charset);
cli@37
   379
	}
cli@37
   380
cli@37
   381
	public void print(final String line)
cli@49
   382
			throws IOException {
cli@37
   383
		writeToChannel(CharBuffer.wrap(line), charset, line);
cli@37
   384
	}
cli@37
   385
cli@49
   386
	public void setCurrentCharset(final Charset charset) {
cli@37
   387
		this.charset = charset;
cli@37
   388
	}
cli@37
   389
cli@49
   390
	void setLastActivity(long timestamp) {
cli@37
   391
		this.lastActivity = timestamp;
cli@37
   392
	}
franta-hg@101
   393
franta-hg@101
   394
	/**
franta-hg@101
   395
	 * @return Current username. 
franta-hg@101
   396
	 * But user may not have been authenticated yet.
franta-hg@101
   397
	 * You must check {@link #isUserAuthenticated()}
franta-hg@101
   398
	 */
franta-hg@101
   399
	public String getUsername() {
franta-hg@101
   400
		return username;
franta-hg@101
   401
	}
franta-hg@101
   402
franta-hg@101
   403
	/**
franta-hg@101
   404
	 * This method is to be called from AUTHINFO USER Command implementation.
franta-hg@101
   405
	 * @param username username from AUTHINFO USER username.
franta-hg@101
   406
	 */
franta-hg@101
   407
	public void setUsername(String username) {
franta-hg@101
   408
		this.username = username;
franta-hg@101
   409
	}
franta-hg@101
   410
franta-hg@101
   411
	/**
franta-hg@101
   412
	 * @return true if current user (see {@link #getUsername()}) has been succesfully authenticated.
franta-hg@101
   413
	 */
franta-hg@101
   414
	public boolean isUserAuthenticated() {
franta-hg@101
   415
		return userAuthenticated;
franta-hg@101
   416
	}
franta-hg@101
   417
franta-hg@101
   418
	/**
franta-hg@101
   419
	 * This method is to be called from AUTHINFO PASS Command implementation.
franta-hg@101
   420
	 * @param userAuthenticated true if user has provided right password in AUTHINFO PASS password.
franta-hg@101
   421
	 */
franta-hg@101
   422
	public void setUserAuthenticated(boolean userAuthenticated) {
franta-hg@101
   423
		this.userAuthenticated = userAuthenticated;
franta-hg@101
   424
	}
chris@1
   425
}