Bound and concurrentify SYN queue to hopefully prevent explosion

apps/streaming/java/src/net/i2p/client/streaming/ConnectionHandler.java

@@ -1,5 +1,7 @@
 package net.i2p.client.streaming;
 
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -10,24 +12,36 @@ import net.i2p.util.SimpleTimer;
 
 /**
  * Receive new connection attempts
+ *
+ * Use a bounded queue to limit the damage from SYN floods,
+ * router overload, or a slow client
+ *
+ * @author zzz modded to use concurrent and bound queue size
  */
 class ConnectionHandler {
     private I2PAppContext _context;
     private Log _log;
     private ConnectionManager _manager;
-    private List _synQueue;
+    private LinkedBlockingQueue<Packet> _synQueue;
     private boolean _active;
     private int _acceptTimeout;
     
     /** max time after receiveNewSyn() and before the matched accept() */
     private static final int DEFAULT_ACCEPT_TIMEOUT = 3*1000;
 
+    /**
+     *  This is both SYNs and subsequent packets, and with an initial window size of 12,
+     *  this is a backlog of 5 to 64 Syns, which seems like plenty for now
+     *  Don't make this too big because the removal by all the TimeoutSyns is O(n**2) - sortof.
+     */
+    private static final int MAX_QUEUE_SIZE = 64;
+    
     /** Creates a new instance of ConnectionHandler */
     public ConnectionHandler(I2PAppContext context, ConnectionManager mgr) {
         _context = context;
         _log = context.logManager().getLog(ConnectionHandler.class);
         _manager = mgr;
-        _synQueue = new ArrayList(5);
+        _synQueue = new LinkedBlockingQueue(MAX_QUEUE_SIZE);
         _active = false;
         _acceptTimeout = DEFAULT_ACCEPT_TIMEOUT;
     }
@@ -35,9 +49,11 @@ class ConnectionHandler {
     public void setActive(boolean active) { 
         if (_log.shouldLog(Log.DEBUG))
             _log.debug("setActive(" + active + ") called");
-        synchronized (_synQueue) {
         _active = active; 
-            _synQueue.notifyAll(); // so we break from the accept()
+        if (!active) {
+            try {
+                _synQueue.put(new PoisonPacket()); // so we break from the accept() - waits until space is available
+            } catch (InterruptedException ie) {}
         }
     }
     public boolean getActive() { return _active; }
@@ -45,6 +61,11 @@ class ConnectionHandler {
     /**
      * Non-SYN packets with a zero SendStreamID may also be queued here so 
      * that they don't get thrown away while the SYN packet before it is queued.
+     *
+     * Additional overload protection may be required here...
+     * We don't have a 3-way handshake, so the SYN fully opens a connection.
+     * Does that make us more or less vulnerable to SYN flooding?
+     *
      */
     public void receiveNewSyn(Packet packet) {
         if (!_active) {
@@ -55,10 +76,15 @@ class ConnectionHandler {
         }
         if (_log.shouldLog(Log.DEBUG))
             _log.debug("Receive new SYN: " + packet + ": timeout in " + _acceptTimeout);
+        // also check if expiration of the head is long past for overload detection with peek() ?
+        boolean success = _synQueue.offer(packet); // fail immediately if full
+        if (success) {
             SimpleScheduler.getInstance().addEvent(new TimeoutSyn(packet), _acceptTimeout);
-        synchronized (_synQueue) {
-            _synQueue.add(packet);
-            _synQueue.notifyAll();
+        } else {
+            if (_log.shouldLog(Log.WARN))
+                _log.warn("Dropping new SYN request, as the queue is full");
+            if (packet.isFlagSet(Packet.FLAG_SYNCHRONIZE))
+                sendReset(packet);
         }
     }
     
@@ -82,41 +108,44 @@ class ConnectionHandler {
                 return null;
             if (!_active) {
                 // fail all the ones we had queued up
-                synchronized (_synQueue) {
-                    for (int i = 0; i < _synQueue.size(); i++) {
-                        Packet packet = (Packet)_synQueue.get(i);
+                while(true) {
+                    Packet packet = _synQueue.poll(); // fails immediately if empty
+                    if (packet == null || packet.getOptionalDelay() == PoisonPacket.MAX_DELAY_REQUEST)
+                        break;
                     sendReset(packet);
                 }
-                    _synQueue.clear();
-                }
                 return null;
             }
             
             Packet syn = null;
-            synchronized (_synQueue) {
-                while ( _active && (_synQueue.size() <= 0) ) {
+            while ( _active && syn == null) {
                 if (_log.shouldLog(Log.DEBUG))
                     _log.debug("Accept("+ timeoutMs+"): active=" + _active + " queue: " 
                                + _synQueue.size());
                 if (timeoutMs <= 0) {
-                        try { _synQueue.wait(); } catch (InterruptedException ie) {}
+                    try {
+                       syn = _synQueue.take(); // waits forever
+                    } catch (InterruptedException ie) {}
                 } else {
                     long remaining = expiration - _context.clock().now();
+                    // (dont think this applies anymore for LinkedBlockingQueue)
                     // BUGFIX
                     // The specified amount of real time has elapsed, more or less. 
                     // If timeout is zero, however, then real time is not taken into consideration 
                     // and the thread simply waits until notified.
                     if (remaining < 1)
                         break;
-                        try { _synQueue.wait(remaining); } catch (InterruptedException ie) {}
-                    }
-                }
-                if (_active && _synQueue.size() > 0) {
-                    syn = (Packet)_synQueue.remove(0);
+                    try {
+                        syn = _synQueue.poll(remaining, TimeUnit.MILLISECONDS); // waits the specified time max
+                    } catch (InterruptedException ie) {}
+                    break;
                 }
             }
 
             if (syn != null) {
+                if (syn.getOptionalDelay() == PoisonPacket.MAX_DELAY_REQUEST)
+                    return null;
+
                 // deal with forged / invalid syn packets
 
                 // Handle both SYN and non-SYN packets in the queue
@@ -179,10 +208,7 @@ class ConnectionHandler {
         }
         
         public void timeReached() {
-            boolean removed = false;
-            synchronized (_synQueue) {
-                removed = _synQueue.remove(_synPacket);
-            }
+            boolean removed = _synQueue.remove(_synPacket);
             
             if (removed) {
                 if (_synPacket.isFlagSet(Packet.FLAG_SYNCHRONIZE))
@@ -196,4 +222,17 @@ class ConnectionHandler {
             }
         }
     }
+
+    /**
+     * Simple end-of-queue marker.
+     * The standard class limits the delay to MAX_DELAY_REQUEST so
+     * an evil user can't use this to shut us down
+     */
+    private static class PoisonPacket extends Packet {
+        public static final int MAX_DELAY_REQUEST = Packet.MAX_DELAY_REQUEST + 1;
+
+        public PoisonPacket() {
+            setOptionalDelay(MAX_DELAY_REQUEST);
+        }
+    }
 }