273
|
1 /*
|
|
2 * Copyright (c) 2006-2013 Christian Plattner. All rights reserved.
|
|
3 * Please refer to the LICENSE.txt for licensing details.
|
|
4 */
|
|
5
|
|
6 package ch.ethz.ssh2.transport;
|
|
7
|
|
8 import java.io.IOException;
|
|
9 import java.io.InterruptedIOException;
|
305
d2b303406d63
remove extra override annotations that generate eclipse compiler errors
Carl Byington <carl@five-ten-sg.com>
diff
changeset
|
10 import java.net.Socket;
|
280
|
11 import java.security.KeyPair;
|
273
|
12 import java.util.ArrayList;
|
|
13 import java.util.List;
|
|
14
|
|
15 import ch.ethz.ssh2.ConnectionInfo;
|
|
16 import ch.ethz.ssh2.ConnectionMonitor;
|
|
17 import ch.ethz.ssh2.DHGexParameters;
|
|
18 import ch.ethz.ssh2.PacketTypeException;
|
|
19 import ch.ethz.ssh2.compression.Compressor;
|
|
20 import ch.ethz.ssh2.crypto.CryptoWishList;
|
|
21 import ch.ethz.ssh2.crypto.cipher.BlockCipher;
|
|
22 import ch.ethz.ssh2.crypto.digest.MAC;
|
|
23 import ch.ethz.ssh2.log.Logger;
|
|
24 import ch.ethz.ssh2.packets.PacketDisconnect;
|
|
25 import ch.ethz.ssh2.packets.Packets;
|
|
26 import ch.ethz.ssh2.packets.TypesReader;
|
|
27
|
|
28 /**
|
|
29 * Yes, the "standard" is a big mess. On one side, the say that arbitrary channel
|
|
30 * packets are allowed during kex exchange, on the other side we need to blindly
|
|
31 * ignore the next _packet_ if the KEX guess was wrong. Where do we know from that
|
|
32 * the next packet is not a channel data packet? Yes, we could check if it is in
|
|
33 * the KEX range. But the standard says nothing about this. The OpenSSH guys
|
|
34 * block local "normal" traffic during KEX. That's fine - however, they assume
|
|
35 * that the other side is doing the same. During re-key, if they receive traffic
|
|
36 * other than KEX, they become horribly irritated and kill the connection. Since
|
|
37 * we are very likely going to communicate with OpenSSH servers, we have to play
|
|
38 * the same game - even though we could do better.
|
|
39 *
|
|
40 * @author Christian Plattner
|
|
41 * @version $Id: TransportManager.java 161 2014-05-01 18:01:55Z dkocher@sudo.ch $
|
|
42 */
|
|
43 public abstract class TransportManager {
|
|
44 private static final Logger log = Logger.getLogger(TransportManager.class);
|
|
45
|
|
46 private static final class HandlerEntry {
|
|
47 MessageHandler mh;
|
|
48 int low;
|
|
49 int high;
|
|
50 }
|
|
51
|
|
52 /**
|
|
53 * Advertised maximum SSH packet size that the other side can send to us.
|
|
54 */
|
|
55 public static final int MAX_PACKET_SIZE = 64 * 1024;
|
|
56
|
|
57 private final List<AsynchronousEntry> asynchronousQueue
|
307
|
58 = new ArrayList<AsynchronousEntry>();
|
273
|
59
|
|
60 private Thread asynchronousThread = null;
|
|
61 private boolean asynchronousPending = false;
|
|
62
|
|
63 private Socket socket;
|
|
64
|
|
65 protected TransportManager(final Socket socket) {
|
|
66 this.socket = socket;
|
|
67 }
|
|
68
|
|
69 private static final class AsynchronousEntry {
|
|
70 public byte[] message;
|
|
71
|
|
72 public AsynchronousEntry(byte[] message) {
|
|
73 this.message = message;
|
|
74 }
|
|
75 }
|
|
76
|
|
77 private final class AsynchronousWorker implements Runnable {
|
|
78 public void run() {
|
307
|
79 while (true) {
|
273
|
80 final AsynchronousEntry item;
|
307
|
81
|
|
82 synchronized (asynchronousQueue) {
|
|
83 if (asynchronousQueue.size() == 0) {
|
273
|
84 // Only now we may reset the flag, since we are sure that all queued items
|
|
85 // have been sent (there is a slight delay between de-queuing and sending,
|
|
86 // this is why we need this flag! See code below. Sending takes place outside
|
|
87 // of this lock, this is why a test for size()==0 (from another thread) does not ensure
|
|
88 // that all messages have been sent.
|
|
89 asynchronousPending = false;
|
|
90 // Notify any senders that they can proceed, all async messages have been delivered
|
|
91 asynchronousQueue.notifyAll();
|
|
92
|
|
93 // After the queue is empty for about 2 seconds, stop this thread
|
|
94 try {
|
|
95 asynchronousQueue.wait(2000);
|
|
96 }
|
307
|
97 catch (InterruptedException ignore) {
|
273
|
98 //
|
|
99 }
|
307
|
100
|
|
101 if (asynchronousQueue.size() == 0) {
|
273
|
102 asynchronousThread = null;
|
|
103 return;
|
|
104 }
|
|
105 }
|
307
|
106
|
273
|
107 item = asynchronousQueue.remove(0);
|
|
108 }
|
307
|
109
|
273
|
110 try {
|
|
111 sendMessageImmediate(item.message);
|
|
112 }
|
307
|
113 catch (IOException e) {
|
273
|
114 // There is no point in handling it - it simply means that the connection has a problem and we should stop
|
|
115 // sending asynchronously messages. We do not need to signal that we have exited (asynchronousThread = null):
|
|
116 // further messages in the queue cannot be sent by this or any other thread.
|
|
117 // Other threads will sooner or later (when receiving or sending the next message) get the
|
|
118 // same IOException and get to the same conclusion.
|
|
119 log.warning(e.getMessage());
|
|
120 return;
|
|
121 }
|
|
122 }
|
|
123 }
|
|
124 }
|
|
125
|
|
126 private final Object connectionSemaphore = new Object();
|
|
127
|
|
128 private boolean flagKexOngoing;
|
|
129
|
|
130 private boolean connectionClosed;
|
|
131 private IOException reasonClosedCause;
|
|
132
|
|
133 private TransportConnection tc;
|
|
134 private KexManager km;
|
|
135
|
|
136 private final List<HandlerEntry> messageHandlers
|
307
|
137 = new ArrayList<HandlerEntry>();
|
273
|
138
|
|
139 private List<ConnectionMonitor> connectionMonitors
|
307
|
140 = new ArrayList<ConnectionMonitor>();
|
273
|
141
|
|
142 protected void init(TransportConnection tc, KexManager km) {
|
|
143 this.tc = tc;
|
|
144 this.km = km;
|
|
145 }
|
|
146
|
|
147 public int getPacketOverheadEstimate() {
|
|
148 return tc.getPacketOverheadEstimate();
|
|
149 }
|
|
150
|
|
151 public ConnectionInfo getConnectionInfo(int kexNumber) throws IOException {
|
|
152 return km.getOrWaitForConnectionInfo(kexNumber);
|
|
153 }
|
|
154
|
|
155 public IOException getReasonClosedCause() {
|
307
|
156 synchronized (connectionSemaphore) {
|
273
|
157 return reasonClosedCause;
|
|
158 }
|
|
159 }
|
|
160
|
|
161 public byte[] getSessionIdentifier() {
|
|
162 return km.sessionId;
|
|
163 }
|
|
164
|
|
165 public void close() {
|
|
166 // It is safe now to acquire the semaphore.
|
307
|
167 synchronized (connectionSemaphore) {
|
|
168 if (!connectionClosed) {
|
273
|
169 try {
|
|
170 tc.sendMessage(new PacketDisconnect(
|
307
|
171 PacketDisconnect.Reason.SSH_DISCONNECT_BY_APPLICATION, "").getPayload());
|
273
|
172 }
|
307
|
173 catch (IOException ignore) {
|
273
|
174 //
|
|
175 }
|
307
|
176
|
273
|
177 try {
|
|
178 socket.close();
|
|
179 }
|
307
|
180 catch (IOException ignore) {
|
273
|
181 //
|
|
182 }
|
307
|
183
|
273
|
184 connectionClosed = true;
|
307
|
185
|
|
186 synchronized (this) {
|
|
187 for (ConnectionMonitor cmon : connectionMonitors) {
|
273
|
188 cmon.connectionLost(reasonClosedCause);
|
|
189 }
|
|
190 }
|
|
191 }
|
307
|
192
|
273
|
193 connectionSemaphore.notifyAll();
|
|
194 }
|
|
195 }
|
|
196
|
|
197 public void close(IOException cause) {
|
|
198 // Do not acquire the semaphore, perhaps somebody is inside (and waits until
|
|
199 // the remote side is ready to accept new data
|
|
200 try {
|
|
201 socket.close();
|
|
202 }
|
307
|
203 catch (IOException ignore) {
|
273
|
204 }
|
307
|
205
|
273
|
206 // It is safe now to acquire the semaphore.
|
307
|
207 synchronized (connectionSemaphore) {
|
273
|
208 connectionClosed = true;
|
|
209 reasonClosedCause = cause;
|
|
210 connectionSemaphore.notifyAll();
|
|
211 }
|
307
|
212
|
|
213 synchronized (this) {
|
|
214 for (ConnectionMonitor cmon : connectionMonitors) {
|
273
|
215 cmon.connectionLost(reasonClosedCause);
|
|
216 }
|
|
217 }
|
|
218 }
|
|
219
|
|
220 protected void startReceiver() throws IOException {
|
|
221 final Thread receiveThread = new Thread(new Runnable() {
|
|
222 public void run() {
|
|
223 try {
|
|
224 receiveLoop();
|
|
225 // Can only exit with exception
|
|
226 }
|
307
|
227 catch (IOException e) {
|
273
|
228 close(e);
|
|
229 log.warning(e.getMessage());
|
307
|
230
|
273
|
231 // Tell all handlers that it is time to say goodbye
|
307
|
232 if (km != null) {
|
273
|
233 km.handleFailure(e);
|
|
234 }
|
307
|
235
|
|
236 for (HandlerEntry he : messageHandlers) {
|
273
|
237 he.mh.handleFailure(e);
|
|
238 }
|
|
239 }
|
307
|
240
|
|
241 if (log.isDebugEnabled()) {
|
273
|
242 log.debug("Receive thread: back from receiveLoop");
|
|
243 }
|
|
244 }
|
|
245 });
|
|
246 receiveThread.setName("Transport Manager");
|
|
247 receiveThread.setDaemon(true);
|
|
248 receiveThread.start();
|
|
249 }
|
|
250
|
|
251 public void registerMessageHandler(MessageHandler mh, int low, int high) {
|
|
252 HandlerEntry he = new HandlerEntry();
|
|
253 he.mh = mh;
|
|
254 he.low = low;
|
|
255 he.high = high;
|
|
256
|
307
|
257 synchronized (messageHandlers) {
|
273
|
258 messageHandlers.add(he);
|
|
259 }
|
|
260 }
|
|
261
|
|
262 public void removeMessageHandler(MessageHandler handler) {
|
307
|
263 synchronized (messageHandlers) {
|
|
264 for (int i = 0; i < messageHandlers.size(); i++) {
|
273
|
265 HandlerEntry he = messageHandlers.get(i);
|
307
|
266
|
|
267 if (he.mh == handler) {
|
273
|
268 messageHandlers.remove(i);
|
|
269 break;
|
|
270 }
|
|
271 }
|
|
272 }
|
|
273 }
|
|
274
|
|
275 public void sendKexMessage(byte[] msg) throws IOException {
|
307
|
276 synchronized (connectionSemaphore) {
|
|
277 if (connectionClosed) {
|
273
|
278 throw reasonClosedCause;
|
|
279 }
|
307
|
280
|
273
|
281 flagKexOngoing = true;
|
307
|
282
|
273
|
283 try {
|
|
284 tc.sendMessage(msg);
|
|
285 }
|
307
|
286 catch (IOException e) {
|
273
|
287 close(e);
|
|
288 throw e;
|
|
289 }
|
|
290 }
|
|
291 }
|
|
292
|
|
293 public void kexFinished() throws IOException {
|
307
|
294 synchronized (connectionSemaphore) {
|
273
|
295 flagKexOngoing = false;
|
|
296 connectionSemaphore.notifyAll();
|
|
297 }
|
|
298 }
|
|
299
|
|
300 /**
|
|
301 * @param cwl Crypto wishlist
|
|
302 * @param dhgex Diffie-hellman group exchange
|
|
303 * @param dsa may be null if this is a client connection
|
|
304 * @param rsa may be null if this is a client connection
|
|
305 * @throws IOException
|
|
306 */
|
301
|
307 public void forceKeyExchange(CryptoWishList cwl, DHGexParameters dhgex, KeyPair dsa, KeyPair rsa, KeyPair ec)
|
307
|
308 throws IOException {
|
|
309 synchronized (connectionSemaphore) {
|
|
310 if (connectionClosed) {
|
273
|
311 // Inform the caller that there is no point in triggering a new kex
|
|
312 throw reasonClosedCause;
|
|
313 }
|
|
314 }
|
307
|
315
|
301
|
316 km.initiateKEX(cwl, dhgex, dsa, rsa, ec);
|
273
|
317 }
|
|
318
|
|
319 public void changeRecvCipher(BlockCipher bc, MAC mac) {
|
|
320 tc.changeRecvCipher(bc, mac);
|
|
321 }
|
|
322
|
|
323 public void changeSendCipher(BlockCipher bc, MAC mac) {
|
|
324 tc.changeSendCipher(bc, mac);
|
|
325 }
|
|
326
|
|
327 public void changeRecvCompression(Compressor comp) {
|
|
328 tc.changeRecvCompression(comp);
|
|
329 }
|
|
330
|
|
331 public void changeSendCompression(Compressor comp) {
|
|
332 tc.changeSendCompression(comp);
|
|
333 }
|
|
334
|
|
335 public void sendAsynchronousMessage(byte[] msg) throws IOException {
|
307
|
336 synchronized (asynchronousQueue) {
|
273
|
337 asynchronousQueue.add(new AsynchronousEntry(msg));
|
|
338 asynchronousPending = true;
|
|
339
|
307
|
340 /* This limit should be flexible enough. We need this, otherwise the peer
|
273
|
341 * can flood us with global requests (and other stuff where we have to reply
|
307
|
342 * with an asynchronous message) and (if the server just sends data and does not
|
|
343 * read what we send) this will probably put us in a low memory situation
|
|
344 * (our send queue would grow and grow and...) */
|
273
|
345
|
307
|
346 if (asynchronousQueue.size() > 100) {
|
273
|
347 throw new IOException("The peer is not consuming our asynchronous replies.");
|
|
348 }
|
|
349
|
|
350 // Check if we have an asynchronous sending thread
|
307
|
351 if (asynchronousThread == null) {
|
273
|
352 asynchronousThread = new Thread(new AsynchronousWorker());
|
|
353 asynchronousThread.setDaemon(true);
|
|
354 asynchronousThread.start();
|
|
355 // The thread will stop after 2 seconds of inactivity (i.e., empty queue)
|
|
356 }
|
307
|
357
|
273
|
358 asynchronousQueue.notifyAll();
|
|
359 }
|
|
360 }
|
|
361
|
|
362 public void setConnectionMonitors(List<ConnectionMonitor> monitors) {
|
307
|
363 synchronized (this) {
|
273
|
364 connectionMonitors = new ArrayList<ConnectionMonitor>();
|
|
365 connectionMonitors.addAll(monitors);
|
|
366 }
|
|
367 }
|
|
368
|
|
369 /**
|
|
370 * Send a message but ensure that all queued messages are being sent first.
|
|
371 *
|
|
372 * @param msg Message
|
|
373 * @throws IOException
|
|
374 */
|
|
375 public void sendMessage(byte[] msg) throws IOException {
|
307
|
376 synchronized (asynchronousQueue) {
|
|
377 while (asynchronousPending) {
|
273
|
378 try {
|
|
379 asynchronousQueue.wait();
|
|
380 }
|
307
|
381 catch (InterruptedException e) {
|
273
|
382 throw new InterruptedIOException(e.getMessage());
|
|
383 }
|
|
384 }
|
|
385 }
|
307
|
386
|
273
|
387 sendMessageImmediate(msg);
|
|
388 }
|
|
389
|
|
390 /**
|
|
391 * Send message, ignore queued async messages that have not been delivered yet.
|
|
392 * Will be called directly from the asynchronousThread thread.
|
|
393 *
|
|
394 * @param msg Message
|
|
395 * @throws IOException
|
|
396 */
|
|
397 public void sendMessageImmediate(byte[] msg) throws IOException {
|
307
|
398 synchronized (connectionSemaphore) {
|
|
399 while (true) {
|
|
400 if (connectionClosed) {
|
273
|
401 throw reasonClosedCause;
|
|
402 }
|
307
|
403
|
|
404 if (!flagKexOngoing) {
|
273
|
405 break;
|
|
406 }
|
307
|
407
|
273
|
408 try {
|
|
409 connectionSemaphore.wait();
|
|
410 }
|
307
|
411 catch (InterruptedException e) {
|
273
|
412 throw new InterruptedIOException(e.getMessage());
|
|
413 }
|
|
414 }
|
|
415
|
|
416 try {
|
|
417 tc.sendMessage(msg);
|
|
418 }
|
307
|
419 catch (IOException e) {
|
273
|
420 close(e);
|
|
421 throw e;
|
|
422 }
|
|
423 }
|
|
424 }
|
|
425
|
|
426 private void receiveLoop() throws IOException {
|
307
|
427 while (true) {
|
273
|
428 final byte[] buffer = new byte[MAX_PACKET_SIZE];
|
|
429 final int length = tc.receiveMessage(buffer, 0, buffer.length);
|
|
430 final byte[] packet = new byte[length];
|
|
431 System.arraycopy(buffer, 0, packet, 0, length);
|
|
432 final int type = packet[0] & 0xff;
|
307
|
433
|
|
434 switch (type) {
|
273
|
435 case Packets.SSH_MSG_IGNORE:
|
|
436 break;
|
307
|
437
|
273
|
438 case Packets.SSH_MSG_DEBUG: {
|
307
|
439 TypesReader tr = new TypesReader(packet);
|
|
440 tr.readByte();
|
|
441 // always_display
|
|
442 tr.readBoolean();
|
|
443 String message = tr.readString();
|
|
444
|
|
445 if (log.isDebugEnabled()) {
|
|
446 log.debug(String.format("Debug message from remote: '%s'", message));
|
|
447 }
|
|
448
|
|
449 break;
|
273
|
450 }
|
307
|
451
|
273
|
452 case Packets.SSH_MSG_UNIMPLEMENTED:
|
|
453 throw new PacketTypeException(type);
|
307
|
454
|
273
|
455 case Packets.SSH_MSG_DISCONNECT: {
|
307
|
456 final PacketDisconnect disconnect = new PacketDisconnect(packet);
|
|
457 throw new DisconnectException(disconnect.getReason(), disconnect.getMessage());
|
|
458 }
|
|
459
|
273
|
460 case Packets.SSH_MSG_KEXINIT:
|
|
461 case Packets.SSH_MSG_NEWKEYS:
|
|
462 case Packets.SSH_MSG_KEXDH_INIT:
|
|
463 case Packets.SSH_MSG_KEXDH_REPLY:
|
|
464 case Packets.SSH_MSG_KEX_DH_GEX_REQUEST:
|
|
465 case Packets.SSH_MSG_KEX_DH_GEX_INIT:
|
|
466 case Packets.SSH_MSG_KEX_DH_GEX_REPLY:
|
|
467 // Is it a KEX Packet
|
|
468 km.handleMessage(packet);
|
|
469 break;
|
307
|
470
|
273
|
471 case Packets.SSH_MSG_USERAUTH_SUCCESS:
|
|
472 tc.startCompression();
|
307
|
473
|
|
474 // Continue with message handlers
|
273
|
475 default:
|
|
476 boolean handled = false;
|
307
|
477
|
|
478 for (HandlerEntry handler : messageHandlers) {
|
|
479 if ((handler.low <= type) && (type <= handler.high)) {
|
273
|
480 handler.mh.handleMessage(packet);
|
|
481 handled = true;
|
|
482 break;
|
|
483 }
|
|
484 }
|
307
|
485
|
|
486 if (!handled) {
|
273
|
487 throw new PacketTypeException(type);
|
|
488 }
|
307
|
489
|
273
|
490 break;
|
|
491 }
|
307
|
492
|
|
493 if (log.isDebugEnabled()) {
|
273
|
494 log.debug(String.format("Handled packet %d", type));
|
|
495 }
|
|
496 }
|
|
497 }
|
|
498 }
|