Add the DTLS 1.3 reliable handshake and pack handshake flights into datagrams

Adds the ACK-driven reliable handshake of RFC 9147 sections 5.8 and 7, and packs
handshake flights into as few datagrams as the MTU allows.

The packing applies to DTLS 1.2 as well as 1.3, since that is what github #1487
asks for: on the existing aggregated-handshake test a client flight went from 5
datagrams to 3 for the same 1279 bytes. Only handshake records are packed, and a
non-handshake record flushes the buffer before it leaves, so write order is
preserved; that matters for the implicit change_cipher_spec, which must not
overtake the flight it follows.

The reliable handshake registers each written fragment against the record number
that carried it, retires fragments when an ACK arrives, retransmits only what is
outstanding, and emits ACKs on the RFC 9147 7.1 triggers. Inbound ACKs are
filtered by the epoch of the record carrying them, and both ACK emission and the
accumulated record-number list are bounded by what fits in one datagram.

DTLS 1.3 still cannot be negotiated, so none of the 1.3 paths are reachable yet.
DTLSReassembler.contributeFragment changed from void to boolean and gained
acceptsFragment and getNextExpectedOffset; the admission predicate is unchanged,
and DTLS 1.2 behaviour is unaffected.

Two test helpers, MinimalHandshakeAggregator and ServerHandshakeDropper, decided
what to do by inspecting only the first record of a datagram, which was exact
only while each datagram carried one record. Both now walk every record.

relates to github #1468.
closes github #1487.

Author: mondain

tls/src/main/java/org/bouncycastle/tls/DTLS13FlightTracker.java

@@ -0,0 +1,144 @@
+package org.bouncycastle.tls;
+
+import java.util.Enumeration;
+import java.util.Hashtable;
+import java.util.Vector;
+
+/**
+ * RFC 9147 7.2. Tracks which record carried which handshake fragment of the current outbound flight, so
+ * that an ACK retires exactly the fragments it covers and a retransmission resends only what is left.
+ * <p>
+ * A fragment may be registered more than once, under a different record number each time it is sent. It
+ * is acknowledged as soon as any one of those records is acknowledged.
+ * </p>
+ */
+class DTLS13FlightTracker
+{
+    /** One handshake fragment of the current outbound flight. */
+    static final class Fragment
+    {
+        private final int messageSeq;
+        private final int fragmentOffset;
+        private final int fragmentLength;
+
+        boolean acknowledged = false;
+
+        Fragment(int messageSeq, int fragmentOffset, int fragmentLength)
+        {
+            this.messageSeq = messageSeq;
+            this.fragmentOffset = fragmentOffset;
+            this.fragmentLength = fragmentLength;
+        }
+
+        int getMessageSeq()
+        {
+            return messageSeq;
+        }
+
+        int getFragmentOffset()
+        {
+            return fragmentOffset;
+        }
+
+        int getFragmentLength()
+        {
+            return fragmentLength;
+        }
+
+        private String key()
+        {
+            return messageSeq + ":" + fragmentOffset + ":" + fragmentLength;
+        }
+    }
+
+    // record number -> Fragment
+    private Hashtable carriers = new Hashtable();
+    // fragment key -> Fragment, so the same fragment sent twice is one entry
+    private Hashtable fragments = new Hashtable();
+    // fragments in registration order, for deterministic retransmission
+    private Vector order = new Vector();
+
+    void reset()
+    {
+        carriers = new Hashtable();
+        fragments = new Hashtable();
+        order = new Vector();
+    }
+
+    void register(DTLSRecordNumber recordNumber, int messageSeq, int fragmentOffset, int fragmentLength)
+    {
+        Fragment fragment = new Fragment(messageSeq, fragmentOffset, fragmentLength);
+        String key = fragment.key();
+
+        Fragment existing = (Fragment)fragments.get(key);
+        if (null == existing)
+        {
+            fragments.put(key, fragment);
+            order.addElement(fragment);
+            existing = fragment;
+        }
+
+        if (null != recordNumber)
+        {
+            carriers.put(recordNumber, existing);
+        }
+    }
+
+    void acknowledge(Vector recordNumbers)
+    {
+        for (int i = 0; i < recordNumbers.size(); ++i)
+        {
+            Fragment fragment = (Fragment)carriers.get(recordNumbers.elementAt(i));
+            if (null != fragment)
+            {
+                fragment.acknowledged = true;
+            }
+        }
+    }
+
+    boolean isEmpty()
+    {
+        return order.isEmpty();
+    }
+
+    /**
+     * @return true if fragments were registered and every one of them has been acknowledged.
+     */
+    boolean isComplete()
+    {
+        if (order.isEmpty())
+        {
+            return false;
+        }
+
+        Enumeration e = order.elements();
+        while (e.hasMoreElements())
+        {
+            if (!((Fragment)e.nextElement()).acknowledged)
+            {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    /**
+     * @return the fragments not yet acknowledged, in the order they were first registered.
+     */
+    Vector getOutstanding()
+    {
+        Vector outstanding = new Vector();
+
+        Enumeration e = order.elements();
+        while (e.hasMoreElements())
+        {
+            Fragment fragment = (Fragment)e.nextElement();
+            if (!fragment.acknowledged)
+            {
+                outstanding.addElement(fragment);
+            }
+        }
+
+        return outstanding;
+    }
+}

tls/src/main/java/org/bouncycastle/tls/DTLSAck.java

@@ -0,0 +1,77 @@
+package org.bouncycastle.tls;
+
+import java.io.IOException;
+import java.util.Vector;
+
+/**
+ * RFC 9147 7. The ACK message, carried in its own content type rather than as a handshake message so
+ * that it is not added to the handshake transcript.
+ * <pre>
+ * struct {
+ *     RecordNumber record_numbers&lt;0..2^16-1&gt;;
+ * } ACK;
+ * </pre>
+ */
+class DTLSAck
+{
+    /** The wire size of one RecordNumber: two uint64. */
+    static final int RECORD_NUMBER_LENGTH = 16;
+
+    /**
+     * Encode a list of {@link DTLSRecordNumber} as an ACK body. An empty list is valid (RFC 9147 7.1).
+     */
+    static byte[] encode(Vector recordNumbers) throws IOException
+    {
+        int count = recordNumbers.size();
+        int bodyLength = count * RECORD_NUMBER_LENGTH;
+        TlsUtils.checkUint16(bodyLength);
+
+        byte[] buf = new byte[2 + bodyLength];
+        TlsUtils.writeUint16(bodyLength, buf, 0);
+
+        int pos = 2;
+        for (int i = 0; i < count; ++i)
+        {
+            DTLSRecordNumber recordNumber = (DTLSRecordNumber)recordNumbers.elementAt(i);
+            TlsUtils.writeUint64(recordNumber.getEpoch(), buf, pos);
+            TlsUtils.writeUint64(recordNumber.getSequenceNumber(), buf, pos + 8);
+            pos += RECORD_NUMBER_LENGTH;
+        }
+
+        return buf;
+    }
+
+    /**
+     * Decode an ACK body.
+     *
+     * @return the record numbers in wire order, or null if the body is malformed. A malformed ACK is
+     *         discarded rather than failing the connection (RFC 9147 4.5.2).
+     */
+    static Vector decode(byte[] buf, int off, int len) throws IOException
+    {
+        if (len < 2)
+        {
+            return null;
+        }
+
+        int bodyLength = TlsUtils.readUint16(buf, off);
+        if (bodyLength != len - 2 || (bodyLength % RECORD_NUMBER_LENGTH) != 0)
+        {
+            return null;
+        }
+
+        Vector recordNumbers = new Vector();
+
+        int pos = off + 2;
+        int count = bodyLength / RECORD_NUMBER_LENGTH;
+        for (int i = 0; i < count; ++i)
+        {
+            long epoch = TlsUtils.readUint64(buf, pos);
+            long sequenceNumber = TlsUtils.readUint64(buf, pos + 8);
+            recordNumbers.addElement(new DTLSRecordNumber(epoch, sequenceNumber));
+            pos += RECORD_NUMBER_LENGTH;
+        }
+
+        return recordNumbers;
+    }
+}

tls/src/main/java/org/bouncycastle/tls/DTLSAckListener.java

@@ -0,0 +1,16 @@
+package org.bouncycastle.tls;
+
+import java.io.IOException;
+import java.util.Vector;
+
+/**
+ * RFC 9147 7. Receives ACK records from the record layer. ACK is a content type rather than a handshake
+ * message, so it does not reach the handshake through the normal message path.
+ */
+interface DTLSAckListener
+{
+    /**
+     * @param recordNumbers the acknowledged {@link DTLSRecordNumber}s, in the order they appeared.
+     */
+    void receivedAck(Vector recordNumbers) throws IOException;
+}

tls/src/main/java/org/bouncycastle/tls/DTLSReassembler.java

@@ -39,14 +39,43 @@ byte[] getBodyIfComplete()
         return missing.isEmpty() ? body : null;
     }
 
-    void contributeFragment(short msg_type, int length, byte[] buf, int off, int fragment_offset,
+    /**
+     * @return the offset of the first byte of this message that has not been received yet, or -1 once the
+     *         message is complete. Supports the RFC 9147 7.1 out-of-order ACK trigger: a fragment that does
+     *         not begin here is not the next piece of this message.
+     */
+    int getNextExpectedOffset()
+    {
+        return missing.isEmpty() ? -1 : ((Range)missing.firstElement()).start;
+    }
+
+    /**
+     * Whether a fragment with these parameters is one this reassembler could hold, which is the difference
+     * between a fragment that is rejected outright and one that is merely already held. RFC 9147 7.1
+     * acknowledges a record whose message was discarded because a previous copy had been received, but not
+     * one whose message was rejected, and only the reassembler knows which of the two happened.
+     *
+     * @return true if the fragment belongs to this message.
+     */
+    boolean acceptsFragment(short msg_type, int length, int fragment_offset, int fragment_length)
+    {
+        return this.msg_type == msg_type && this.body.length == length
+            && fragment_offset + fragment_length <= length;
+    }
+
+    /**
+     * @return true if the fragment contributed at least one byte (or completed an empty message), false if
+     *         it was ignored, whether because it was rejected or because every byte of it was already held.
+     *         {@link #acceptsFragment(short, int, int, int)} separates those two cases.
+     */
+    boolean contributeFragment(short msg_type, int length, byte[] buf, int off, int fragment_offset,
         int fragment_length)
     {
         int fragment_end = fragment_offset + fragment_length;
 
-        if (this.msg_type != msg_type || this.body.length != length || fragment_end > length)
+        if (!acceptsFragment(msg_type, length, fragment_offset, fragment_length))
         {
-            return;
+            return false;
         }
 
         // NOTE: Empty messages still require an empty fragment to complete it
@@ -55,10 +84,13 @@ void contributeFragment(short msg_type, int length, byte[] buf, int off, int fra
             if (fragment_offset == 0 && !missing.isEmpty() && ((Range)missing.firstElement()).end == 0)
             {
                 missing.removeElementAt(0);
+                return true;
             }
-            return;
+            return false;
         }
 
+        boolean contributed = false;
+
         for (int i = findStartIndex(fragment_offset); i < missing.size(); ++i)
         {
             Range range = (Range)missing.elementAt(i);
@@ -103,7 +135,10 @@ void contributeFragment(short msg_type, int length, byte[] buf, int off, int fra
             }
 
             System.arraycopy(buf, off + copyStart - fragment_offset, body, copyStart, copyLength);
+            contributed = true;
         }
+
+        return contributed;
     }
 
     /**