NTCP: Refactor EstablishState into interface/base/inbound/outbound,

in prep for NTCP2

history.txt

+2018-06-02 zzz
+ * Console: Sort tunnels within pools by expiration (ticket #2232)
+ * NTCP: Refactor EstablishState in prep for NTCP2
+
+2018-06-01 zzz
+ * SusiDNS: Fix deleting notes (ticket #1433)
+
 2018-05-31 zzz
  * Console:
    - Fix CSS preventing ordered lists (ticket #2075)

router/java/src/net/i2p/router/RouterVersion.java

@@ -18,7 +18,7 @@ public class RouterVersion {
     /** deprecated */
     public final static String ID = "Monotone";
     public final static String VERSION = CoreVersion.VERSION;
-    public final static long BUILD = 16;
+    public final static long BUILD = 17;
 
     /** for example "-test" */
     public final static String EXTRA = "";

router/java/src/net/i2p/router/transport/ntcp/EstablishBase.java

+package net.i2p.router.transport.ntcp;
+
+import java.nio.ByteBuffer;
+
+import net.i2p.router.Router;
+import net.i2p.router.RouterContext;
+import net.i2p.router.transport.crypto.DHSessionKeyBuilder;
+import net.i2p.util.Log;
+import net.i2p.util.SimpleByteCache;
+
+/**
+ * Handle the 4-phase establishment, which is as follows:
+ *
+ * <pre>
+ *
+ * Alice                   contacts                      Bob
+ * =========================================================
+ *
+ * Message 1 (Session Request):
+ *  X+(H(X) xor Bob.identHash)-----------------------------&gt;
+ *
+ * Message 2 (Session Created):
+ *  &lt;----------------------------------------Y+E(H(X+Y)+tsB, sk, Y[239:255])
+ *
+ * Message 3 (Session Confirm A):
+ *  E(sz+Alice.identity+tsA+padding+S(X+Y+Bob.identHash+tsA+tsB), sk, hX_xor_Bob.identHash[16:31])---&gt;
+ *
+ * Message 4 (Session Confirm B):
+ *  &lt;----------------------E(S(X+Y+Alice.identHash+tsA+tsB)+padding, sk, prev)
+ *
+ *  Key:
+ *
+ *    X, Y: 256 byte DH keys
+ *    H(): 32 byte SHA256 Hash
+ *    E(data, session key, IV): AES256 Encrypt
+ *    S(): 40 byte DSA Signature
+ *    tsA, tsB: timestamps (4 bytes, seconds since epoch)
+ *    sk: 32 byte Session key
+ *    sz: 2 byte size of Alice identity to follow
+ *
+ * </pre>
+ *
+ *
+ * Alternately, when Bob receives a connection, it could be a
+ * check connection (perhaps prompted by Bob asking for someone
+ * to verify his listener).  check connections are formatted per
+ * isCheckInfo()
+ * NOTE: Check info is unused.
+ *
+ */
+abstract class EstablishBase implements EstablishState {
+    
+    public static final VerifiedEstablishState VERIFIED = new VerifiedEstablishState();
+    public static final FailedEstablishState FAILED = new FailedEstablishState();
+    
+    protected final RouterContext _context;
+    protected final Log _log;
+
+    // bob receives (and alice sends)
+    protected final byte _X[];
+    protected final byte _hX_xor_bobIdentHash[];
+    // alice receives (and bob sends)
+    protected final byte _Y[];
+    protected final byte _e_hXY_tsB[];
+    /** Bob's timestamp in seconds, this is in message #2, *before* _tsA */
+    protected transient long _tsB;
+    /** Alice's timestamp in seconds, this is in message #3, *after* _tsB
+     *  Only saved for outbound. For inbound, see verifyInbound().
+     */
+    protected transient long _tsA;
+    /**
+     *  OUR clock minus HIS clock, in seconds
+     *
+     *  Inbound: tsB - tsA - rtt/2
+     *  Outbound: tsA - tsB - rtt/2
+     */
+    protected transient long _peerSkew;
+    protected transient byte _e_bobSig[];
+
+    /** previously received encrypted block (or the IV) */
+    protected byte _prevEncrypted[];
+    /** decryption buffer */
+    protected final byte _curDecrypted[];
+
+    /** bytes received so far */
+    protected int _received;
+    private byte _extra[];
+
+    protected final DHSessionKeyBuilder _dh;
+
+    protected final NTCPTransport _transport;
+    protected final NTCPConnection _con;
+    /** error causing the corruption */
+    private String _err;
+    /** exception causing the error */
+    private Exception _e;
+    private boolean _failedBySkew;
+    
+    protected static final int MIN_RI_SIZE = 387;
+    protected static final int MAX_RI_SIZE = 3072;
+
+    protected static final int AES_SIZE = 16;
+    protected static final int XY_SIZE = 256;
+    protected static final int HXY_SIZE = 32;  //Hash.HASH_LENGTH;
+    protected static final int HXY_TSB_PAD_SIZE = HXY_SIZE + 4 + 12;  // 48
+
+    protected static final Object _stateLock = new Object();
+    protected State _state;
+
+    protected enum State {
+        OB_INIT,
+        /** sent 1 */
+        OB_SENT_X,
+        /** sent 1, got 2 partial */
+        OB_GOT_Y,
+        /** sent 1, got 2 */
+        OB_GOT_HXY,
+        /** sent 1, got 2, sent 3 */
+        OB_SENT_RI,
+        /** sent 1, got 2, sent 3, got 4 */
+        OB_GOT_SIG,
+
+        IB_INIT,
+        /** got 1 partial */
+        IB_GOT_X,
+        /** got 1 */
+        IB_GOT_HX,
+        /** got 1, sent 2 */
+        IB_SENT_Y,
+        /** got 1, sent 2, got partial 3 */
+        IB_GOT_RI_SIZE,
+        /** got 1, sent 2, got 3 */
+        IB_GOT_RI,
+
+        /** OB: got and verified 4; IB: got and verified 3 and sent 4 */
+        VERIFIED,
+        CORRUPT
+    }
+
+    private EstablishBase() {
+        _context = null;
+        _log = null;
+        _X = null;
+        _Y = null;
+        _hX_xor_bobIdentHash = null;
+        _curDecrypted = null;
+        _dh = null;
+        _transport = null;
+        _con = null;
+        _e_hXY_tsB = null;
+    }
+
+    protected EstablishBase(RouterContext ctx, NTCPTransport transport, NTCPConnection con) {
+        _context = ctx;
+        _log = ctx.logManager().getLog(getClass());
+        _transport = transport;
+        _con = con;
+        _dh = _transport.getDHBuilder();
+        _hX_xor_bobIdentHash = SimpleByteCache.acquire(HXY_SIZE);
+        if (_con.isInbound()) {
+            _X = SimpleByteCache.acquire(XY_SIZE);
+            _Y = _dh.getMyPublicValueBytes();
+        } else {
+            _X = _dh.getMyPublicValueBytes();
+            _Y = SimpleByteCache.acquire(XY_SIZE);
+        }
+
+        _e_hXY_tsB = new byte[HXY_TSB_PAD_SIZE];
+        _curDecrypted = SimpleByteCache.acquire(AES_SIZE);
+    }
+
+    /** @since 0.9.16 */
+    protected void changeState(State state) {
+        synchronized (_stateLock) {
+            _state = state;
+        }
+    }
+
+    /**
+     * parse the contents of the buffer as part of the handshake.  if the
+     * handshake is completed and there is more data remaining, the data are
+     * copieed out so that the next read will be the (still encrypted) remaining
+     * data (available from getExtraBytes)
+     *
+     * All data must be copied out of the buffer as Reader.processRead()
+     * will return it to the pool.
+     */
+    public synchronized void receive(ByteBuffer src) {
+        synchronized(_stateLock) {    
+            if (_state == State.VERIFIED || _state == State.CORRUPT)
+                throw new IllegalStateException(prefix() + "received unexpected data on " + _con);
+        }
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug(prefix() + "Receiving: " + src.remaining() + " Received: " + _received);
+    }
+
+    /**
+     * Does nothing. Outbound (Alice) must override.
+     * We are establishing an outbound connection, so prepare ourselves by
+     * queueing up the write of the first part of the handshake
+     */
+    public void prepareOutbound() {}
+
+    /**
+     *  Was this connection failed because of clock skew?
+     */
+    public synchronized boolean getFailedBySkew() { return _failedBySkew; }
+
+    /** did the handshake fail for some reason? */
+    public boolean isCorrupt() {
+        synchronized(_stateLock) {
+            return _state == State.CORRUPT;
+        }
+    }
+
+    /**
+     *  If synchronized on this, fails with
+     *  deadlocks from all over via CSFI.isEstablished().
+     *  Also CSFI.getFramedAveragePeerClockSkew().
+     *
+     *  @return is the handshake complete and valid?
+     */
+    public boolean isComplete() {
+        synchronized(_stateLock) {
+            return _state == State.VERIFIED;
+        }
+    }
+
+    /** Anything left over in the byte buffer after verification is extra
+     *
+     *  All data must be copied out of the buffer as Reader.processRead()
+     *  will return it to the pool.
+     *
+     *  State must be VERIFIED.
+     *  Caller must synch.
+     */
+    protected void prepareExtra(ByteBuffer buf) {
+        int remaining = buf.remaining();
+        if (remaining > 0) {
+            _extra = new byte[remaining];
+            buf.get(_extra);
+            _received += remaining;
+        }
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug(prefix() + "prepare extra " + remaining + " (total received: " + _received + ")");
+    }
+
+    /**
+     * if complete, this will contain any bytes received as part of the
+     * handshake that were after the actual handshake.  This may return null.
+     */
+    public synchronized byte[] getExtraBytes() { return _extra; }
+
+    /**
+     *  Release resources on timeout.
+     *  @param e may be null
+     *  @since 0.9.16
+     */
+    public synchronized void close(String reason, Exception e) {
+        fail(reason, e);
+    }
+
+    /** Caller must synch. */
+    protected void fail(String reason) { fail(reason, null); }
+
+    /** Caller must synch. */
+    protected void fail(String reason, Exception e) { fail(reason, e, false); }
+
+    /** Caller must synch. */
+    protected void fail(String reason, Exception e, boolean bySkew) {
+        synchronized(_stateLock) {    
+            if (_state == State.CORRUPT || _state == State.VERIFIED)
+                return;
+            changeState(State.CORRUPT);
+        }
+        _failedBySkew = bySkew;
+        _err = reason;
+        _e = e;
+        if (_log.shouldLog(Log.WARN))
+            _log.warn(prefix()+"Failed to establish: " + _err, e);
+        releaseBufs(false);
+    }
+
+    /**
+     *  Only call once. Caller must synch.
+     *  @since 0.9.16
+     */
+    protected void releaseBufs(boolean isVerified) {
+        // null or longer for OB
+        if (_prevEncrypted != null && _prevEncrypted.length == AES_SIZE)
+            SimpleByteCache.release(_prevEncrypted);
+        SimpleByteCache.release(_curDecrypted);
+        SimpleByteCache.release(_hX_xor_bobIdentHash);
+        if (_dh.getPeerPublicValue() == null)
+            _transport.returnUnused(_dh);
+    }
+
+    public synchronized String getError() { return _err; }
+
+    public synchronized Exception getException() { return _e; }
+    
+    /**
+     *  XOR a into b. Modifies b. a is unmodified.
+     *  @param a 32 bytes
+     *  @param b 32 bytes
+     *  @since 0.9.12
+     */
+    protected static void xor32(byte[] a, byte[] b) {
+        for (int i = 0; i < 32; i++) {
+            b[i] ^= a[i];
+        }
+    }
+
+    protected String prefix() { return toString(); }
+
+    @Override
+    public String toString() {
+        StringBuilder buf = new StringBuilder(64);
+        if (_con.isInbound())
+            buf.append("IBES ");
+        else
+            buf.append("OBES ");
+        buf.append(System.identityHashCode(this));
+        buf.append(' ').append(_state);
+        if (_con.isEstablished()) buf.append(" established");
+        buf.append(": ");
+        return buf.toString();
+    }
+
+    /**
+     *  @since 0.9.8
+     */
+    private static class VerifiedEstablishState extends EstablishBase {
+
+        public VerifiedEstablishState() {
+            super();
+            _state = State.VERIFIED;
+        }
+
+        @Override public void prepareOutbound() {
+            Log log =RouterContext.getCurrentContext().logManager().getLog(VerifiedEstablishState.class);
+            log.warn("prepareOutbound() on verified state, doing nothing!");
+        }
+
+        @Override public String toString() { return "VerifiedEstablishState: ";}
+    }
+
+    /**
+     *  @since 0.9.16
+     */
+    private static class FailedEstablishState extends EstablishBase {
+
+        public FailedEstablishState() {
+            super();
+            _state = State.CORRUPT;
+        }
+
+        @Override public void prepareOutbound() {
+            Log log =RouterContext.getCurrentContext().logManager().getLog(VerifiedEstablishState.class);
+            log.warn("prepareOutbound() on verified state, doing nothing!");
+        }
+
+        @Override public String toString() { return "FailedEstablishState: ";}
+    }
+
+    /**
+     *  Mark a string for extraction by xgettext and translation.
+     *  Use this only in static initializers.
+     *  It does not translate!
+     *  @return s
+     */
+    protected static final String _x(String s) {
+        return s;
+    }
+
+}

router/java/src/net/i2p/router/transport/ntcp/NTCPConnection.java

@@ -190,7 +190,7 @@ public class NTCPConnection implements Closeable {
         _isInbound = true;
         _decryptBlockBuf = new byte[BLOCK_SIZE];
         _curReadState = new ReadState();
-        _establishState = new EstablishState(ctx, transport, this);
+        _establishState = new InboundEstablishState(ctx, transport, this);
         _conKey = key;
         _conKey.attach(this);
         _inboundListener = new InboundListener();
@@ -216,7 +216,7 @@ public class NTCPConnection implements Closeable {
         //_outbound = new CoDelPriorityBlockingQueue(ctx, "NTCP-Connection", 32);
         _outbound = new PriBlockingQueue<OutNetMessage>(ctx, "NTCP-Connection", 32);
         _isInbound = false;
-        _establishState = new EstablishState(ctx, transport, this);
+        _establishState = new OutboundEstablishState(ctx, transport, this);
         _decryptBlockBuf = new byte[BLOCK_SIZE];
         _curReadState = new ReadState();
         _inboundListener = new InboundListener();
@@ -309,7 +309,7 @@ public class NTCPConnection implements Closeable {
         NTCPConnection rv = _transport.inboundEstablished(this);
         _nextMetaTime = _establishedOn + (META_FREQUENCY / 2) + _context.random().nextInt(META_FREQUENCY);
         _nextInfoTime = _establishedOn + (INFO_FREQUENCY / 2) + _context.random().nextInt(INFO_FREQUENCY);
-        _establishState = EstablishState.VERIFIED;
+        _establishState = EstablishBase.VERIFIED;
         return rv;
     }
 
@@ -427,7 +427,7 @@ public class NTCPConnection implements Closeable {
     private synchronized NTCPConnection locked_close(boolean allowRequeue) {
         if (_chan != null) try { _chan.close(); } catch (IOException ioe) { }
         if (_conKey != null) _conKey.cancel();
-        _establishState = EstablishState.FAILED;
+        _establishState = EstablishBase.FAILED;
         NTCPConnection old = _transport.removeCon(this);
         _transport.getReader().connectionClosed(this);
         _transport.getWriter().connectionClosed(this);
@@ -532,7 +532,7 @@ public class NTCPConnection implements Closeable {
             _log.debug("Outbound established, prevWriteEnd: " + Base64.encode(prevWriteEnd) + " prevReadEnd: " + Base64.encode(prevReadEnd));
 
         _establishedOn = _context.clock().now();
-        _establishState = EstablishState.VERIFIED;
+        _establishState = EstablishBase.VERIFIED;
         _transport.markReachable(getRemotePeer().calculateHash(), false);
         boolean msgs = !_outbound.isEmpty();
         _nextMetaTime = _establishedOn + (META_FREQUENCY / 2) + _context.random().nextInt(META_FREQUENCY);

router/java/src/net/i2p/router/transport/ntcp/OutboundEstablishState.java

+package net.i2p.router.transport.ntcp;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.nio.ByteBuffer;
+
+import net.i2p.crypto.SigType;
+import net.i2p.data.DataFormatException;
+import net.i2p.data.DataHelper;
+import net.i2p.data.Hash;
+import net.i2p.data.router.RouterIdentity;
+import net.i2p.data.Signature;
+import net.i2p.router.Router;
+import net.i2p.router.RouterContext;
+import net.i2p.router.transport.crypto.DHSessionKeyBuilder;
+import net.i2p.util.Log;
+import net.i2p.util.SimpleByteCache;
+
+/**
+ *
+ *  We are Alice
+ *
+ */
+class OutboundEstablishState extends EstablishBase {
+    
+    public OutboundEstablishState(RouterContext ctx, NTCPTransport transport, NTCPConnection con) {
+        super(ctx, transport, con);
+        _state = State.OB_INIT;
+        ctx.sha().calculateHash(_X, 0, XY_SIZE, _hX_xor_bobIdentHash, 0);
+        xor32(con.getRemotePeer().calculateHash().getData(), _hX_xor_bobIdentHash);
+        // _prevEncrypted will be created later
+    }
+
+    /**
+     * parse the contents of the buffer as part of the handshake.  if the
+     * handshake is completed and there is more data remaining, the data are
+     * copieed out so that the next read will be the (still encrypted) remaining
+     * data (available from getExtraBytes)
+     *
+     * All data must be copied out of the buffer as Reader.processRead()
+     * will return it to the pool.
+     */
+    @Override
+    public synchronized void receive(ByteBuffer src) {
+        super.receive(src);
+        if (!src.hasRemaining())
+            return; // nothing to receive
+        receiveOutbound(src);
+    }
+
+    /**
+     *  We are Alice, so receive these bytes as part of an outbound connection.
+     *  This method receives messages 2 and 4, and sends message 3.
+     *
+     *  All data must be copied out of the buffer as Reader.processRead()
+     *  will return it to the pool.
+     *
+     *  Caller must synch.
+     *
+     *  FIXME none of the _state comparisons use _stateLock, but whole thing
+     *  is synchronized, should be OK. See isComplete()
+     */
+    private void receiveOutbound(ByteBuffer src) {
+
+        // recv Y+E(H(X+Y)+tsB, sk, Y[239:255])
+        // Read in Y, which is the first part of message #2
+        while (_state == State.OB_SENT_X && src.hasRemaining()) {
+            byte c = src.get();
+            _Y[_received++] = c;
+            if (_received >= XY_SIZE) {
+                try {
+                    _dh.setPeerPublicValue(_Y);
+                    _dh.getSessionKey(); // force the calc
+                    if (_log.shouldLog(Log.DEBUG))
+                        _log.debug(prefix()+"DH session key calculated (" + _dh.getSessionKey().toBase64() + ")");
+                    changeState(State.OB_GOT_Y);
+                } catch (DHSessionKeyBuilder.InvalidPublicParameterException e) {
+                    _context.statManager().addRateData("ntcp.invalidDH", 1);
+                    fail("Invalid X", e);
+                    return;
+                }
+            }
+        }
+
+        // Read in Y, which is the first part of message #2
+        // Read in the rest of message #2
+        while (_state == State.OB_GOT_Y && src.hasRemaining()) {
+            int i = _received-XY_SIZE;
+            _received++;
+            byte c = src.get();
+            _e_hXY_tsB[i] = c;
+            if (i+1 >= HXY_TSB_PAD_SIZE) {
+                if (_log.shouldLog(Log.DEBUG)) _log.debug(prefix() + "received _e_hXY_tsB fully");
+                byte hXY_tsB[] = new byte[HXY_TSB_PAD_SIZE];
+                _context.aes().decrypt(_e_hXY_tsB, 0, hXY_tsB, 0, _dh.getSessionKey(), _Y, XY_SIZE-AES_SIZE, HXY_TSB_PAD_SIZE);
+                byte XY[] = new byte[XY_SIZE + XY_SIZE];
+                System.arraycopy(_X, 0, XY, 0, XY_SIZE);
+                System.arraycopy(_Y, 0, XY, XY_SIZE, XY_SIZE);
+                byte[] h = SimpleByteCache.acquire(HXY_SIZE);
+                _context.sha().calculateHash(XY, 0, XY_SIZE + XY_SIZE, h, 0);
+                if (!DataHelper.eq(h, 0, hXY_tsB, 0, HXY_SIZE)) {
+                    SimpleByteCache.release(h);
+                    _context.statManager().addRateData("ntcp.invalidHXY", 1);
+                    fail("Invalid H(X+Y) - mitm attack attempted?");
+                    return;
+                }
+                SimpleByteCache.release(h);
+                changeState(State.OB_GOT_HXY);
+                // their (Bob's) timestamp in seconds
+                _tsB = DataHelper.fromLong(hXY_tsB, HXY_SIZE, 4);
+                long now = _context.clock().now();
+                // rtt from sending #1 to receiving #2
+                long rtt = now - _con.getCreated();
+                // our (Alice's) timestamp in seconds
+                _tsA = (now + 500) / 1000;
+                _peerSkew = (now - (_tsB * 1000) - (rtt / 2) + 500) / 1000; 
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug(prefix()+"h(X+Y) is correct, skew = " + _peerSkew);
+
+                // the skew is not authenticated yet, but it is certainly fatal to
+                // the establishment, so fail hard if appropriate
+                long diff = 1000*Math.abs(_peerSkew);
+                if (!_context.clock().getUpdatedSuccessfully()) {
+                    // Adjust the clock one time in desperation
+                    // We are Alice, he is Bob, adjust to match Bob
+                    _context.clock().setOffset(1000 * (0 - _peerSkew), true);
+                    _peerSkew = 0;
+                    if (diff != 0)
+                        _log.logAlways(Log.WARN, "NTP failure, NTCP adjusting clock by " + DataHelper.formatDuration(diff));
+                } else if (diff >= Router.CLOCK_FUDGE_FACTOR) {
+                    _context.statManager().addRateData("ntcp.invalidOutboundSkew", diff);
+                    _transport.markReachable(_con.getRemotePeer().calculateHash(), false);
+                    // Only banlist if we know what time it is
+                    _context.banlist().banlistRouter(DataHelper.formatDuration(diff),
+                                                       _con.getRemotePeer().calculateHash(),
+                                                       _x("Excessive clock skew: {0}"));
+                    _transport.setLastBadSkew(_peerSkew);
+                    fail("Clocks too skewed (" + diff + " ms)", null, true);
+                    return;
+                } else if (_log.shouldLog(Log.DEBUG)) {
+                    _log.debug(prefix()+"Clock skew: " + diff + " ms");
+                }
+
+                // now prepare and send our response
+                // send E(#+Alice.identity+tsA+padding+S(X+Y+Bob.identHash+tsA+tsB), sk, hX_xor_Bob.identHash[16:31])
+                int sigSize = XY_SIZE + XY_SIZE + HXY_SIZE + 4+4;//+12;
+                byte preSign[] = new byte[sigSize];
+                System.arraycopy(_X, 0, preSign, 0, XY_SIZE);
+                System.arraycopy(_Y, 0, preSign, XY_SIZE, XY_SIZE);
+                System.arraycopy(_con.getRemotePeer().calculateHash().getData(), 0, preSign, XY_SIZE + XY_SIZE, HXY_SIZE);
+                DataHelper.toLong(preSign, XY_SIZE + XY_SIZE + HXY_SIZE, 4, _tsA);
+                DataHelper.toLong(preSign, XY_SIZE + XY_SIZE + HXY_SIZE + 4, 4, _tsB);
+                // hXY_tsB has 12 bytes of padding (size=48, tsB=4 + hXY=32)
+                Signature sig = _context.dsa().sign(preSign, _context.keyManager().getSigningPrivateKey());
+
+                byte ident[] = _context.router().getRouterInfo().getIdentity().toByteArray();
+                // handle variable signature size
+                int min = 2 + ident.length + 4 + sig.length();
+                int rem = min % AES_SIZE;
+                int padding = 0;
+                if (rem > 0)
+                    padding = AES_SIZE - rem;
+                byte preEncrypt[] = new byte[min+padding];
+                DataHelper.toLong(preEncrypt, 0, 2, ident.length);
+                System.arraycopy(ident, 0, preEncrypt, 2, ident.length);
+                DataHelper.toLong(preEncrypt, 2+ident.length, 4, _tsA);
+                if (padding > 0)
+                    _context.random().nextBytes(preEncrypt, 2 + ident.length + 4, padding);
+                System.arraycopy(sig.getData(), 0, preEncrypt, 2+ident.length+4+padding, sig.length());
+
+                _prevEncrypted = new byte[preEncrypt.length];
+                _context.aes().encrypt(preEncrypt, 0, _prevEncrypted, 0, _dh.getSessionKey(),
+                                       _hX_xor_bobIdentHash, _hX_xor_bobIdentHash.length-AES_SIZE, preEncrypt.length);
+
+                changeState(State.OB_SENT_RI);
+                _transport.getPumper().wantsWrite(_con, _prevEncrypted);
+            }
+        }
+
+        // Read in message #4
+        if (_state == State.OB_SENT_RI && src.hasRemaining()) {
+            // we are receiving their confirmation
+
+            // recv E(S(X+Y+Alice.identHash+tsA+tsB)+padding, sk, prev)
+            int off = 0;
+            if (_e_bobSig == null) {
+                // handle variable signature size
+                int siglen = _con.getRemotePeer().getSigningPublicKey().getType().getSigLen();
+                int rem = siglen % AES_SIZE;
+                int padding;
+                if (rem > 0)
+                    padding = AES_SIZE - rem;
+                else
+                    padding = 0;
+                _e_bobSig = new byte[siglen + padding];
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug(prefix() + "receiving E(S(X+Y+Alice.identHash+tsA+tsB)+padding, sk, prev) (remaining? " +
+                               src.hasRemaining() + ")");
+            } else {
+                off = _received - XY_SIZE - HXY_TSB_PAD_SIZE;
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug(prefix() + "continuing to receive E(S(X+Y+Alice.identHash+tsA+tsB)+padding, sk, prev) (remaining? " +
+                               src.hasRemaining() + " off=" + off + " recv=" + _received + ")");
+            }
+            while (_state == State.OB_SENT_RI && src.hasRemaining()) {
+                _e_bobSig[off++] = src.get();
+                _received++;
+
+                if (off >= _e_bobSig.length) {
+                    changeState(State.OB_GOT_SIG);
+                    byte bobSig[] = new byte[_e_bobSig.length];
+                    _context.aes().decrypt(_e_bobSig, 0, bobSig, 0, _dh.getSessionKey(),
+                                           _e_hXY_tsB, HXY_TSB_PAD_SIZE - AES_SIZE, _e_bobSig.length);
+                    // ignore the padding
+                    // handle variable signature size
+                    SigType type = _con.getRemotePeer().getSigningPublicKey().getType();
+                    int siglen = type.getSigLen();
+                    byte bobSigData[] = new byte[siglen];
+                    System.arraycopy(bobSig, 0, bobSigData, 0, siglen);
+                    Signature sig = new Signature(type, bobSigData);
+
+                    byte toVerify[] = new byte[XY_SIZE + XY_SIZE + HXY_SIZE +4+4];
+                    int voff = 0;
+                    System.arraycopy(_X, 0, toVerify, voff, XY_SIZE); voff += XY_SIZE;
+                    System.arraycopy(_Y, 0, toVerify, voff, XY_SIZE); voff += XY_SIZE;
+                    System.arraycopy(_context.routerHash().getData(), 0, toVerify, voff, HXY_SIZE); voff += HXY_SIZE;
+                    DataHelper.toLong(toVerify, voff, 4, _tsA); voff += 4;
+                    DataHelper.toLong(toVerify, voff, 4, _tsB); voff += 4;
+
+                    boolean ok = _context.dsa().verifySignature(sig, toVerify, _con.getRemotePeer().getSigningPublicKey());
+                    if (!ok) {
+                        _context.statManager().addRateData("ntcp.invalidSignature", 1);
+                        fail("Signature was invalid - attempt to spoof " + _con.getRemotePeer().calculateHash().toBase64() + "?");
+                    } else {
+                        if (_log.shouldLog(Log.DEBUG))
+                            _log.debug(prefix() + "signature verified from Bob.  done!");
+                        prepareExtra(src);
+                        byte nextWriteIV[] = SimpleByteCache.acquire(AES_SIZE);
+                        System.arraycopy(_prevEncrypted, _prevEncrypted.length-AES_SIZE, nextWriteIV, 0, AES_SIZE);
+                        // this does not copy the nextWriteIV, do not release to cache
+                        // We are Alice, he is Bob, clock skew is Bob - Alice
+                        _con.finishOutboundEstablishment(_dh.getSessionKey(), _peerSkew, nextWriteIV, _e_bobSig); // skew in seconds
+                        releaseBufs(true);
+                        // if socket gets closed this will be null - prevent NPE
+                        InetAddress ia = _con.getChannel().socket().getInetAddress();
+                        if (ia != null)
+                            _transport.setIP(_con.getRemotePeer().calculateHash(), ia.getAddress());
+                        changeState(State.VERIFIED);
+                    }
+                    return;
+                }
+            }
+        }
+    }
+
+    /**
+     * We are Alice.
+     * We are establishing an outbound connection, so prepare ourselves by
+     * queueing up the write of the first part of the handshake
+     * This method sends message #1 to Bob.
+     */
+    public synchronized void prepareOutbound() {
+        boolean shouldSend;
+        synchronized(_stateLock) {    
+            shouldSend = _state == State.OB_INIT;
+        }
+        if (shouldSend) {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug(prefix() + "send X");
+            byte toWrite[] = new byte[XY_SIZE + _hX_xor_bobIdentHash.length];
+            System.arraycopy(_X, 0, toWrite, 0, XY_SIZE);
+            System.arraycopy(_hX_xor_bobIdentHash, 0, toWrite, XY_SIZE, _hX_xor_bobIdentHash.length);
+            changeState(State.OB_SENT_X);
+            _transport.getPumper().wantsWrite(_con, toWrite);
+        } else {
+            if (_log.shouldLog(Log.WARN))
+                _log.warn(prefix() + "unexpected prepareOutbound()");
+        }
+    }
+
+    /**
+     *  Only call once. Caller must synch.
+     *  @since 0.9.16
+     */
+    @Override
+    protected void releaseBufs(boolean isVerified) {
+        super.releaseBufs(isVerified);
+        SimpleByteCache.release(_Y);
+    }
+}