frag_send: crafted IP fragments, so ordering can be tested and not just delivery

Letting the kernel fragment an oversized datagram answers one question — do
fragments get through. It cannot answer the more interesting one, because the
kernel always emits them in order, first one first.

The classic middlebox fault is exactly about that ordering. Only the first
fragment carries the UDP header, and therefore the ports; a stateful firewall
or NAT that has not seen it has no flow to match the rest against, and many
drop them. That is invisible to any in-order test and shows up in the field as
"large DNS answers fail on this network" or "the tunnel breaks when the MTU
drops" — it works until the network reorders, then fails intermittently, which
is the hardest kind of fault to chase.

So the server now builds the fragments itself (raw socket, IP_HDRINCL) and
controls their order: in_order as a baseline, reversed, and first-fragment-last.
The datagram is assembled and signed whole before being cut up, so what the
client reassembles is indistinguishable from an ordinary packet — otherwise it
would be measuring our sender rather than the path.

Two details that would silently produce wrong answers:
  - The UDP checksum is computed rather than left zero. A zero-checksum datagram
    is dropped by some middleboxes, and that drop would be recorded as a
    fragmentation failure, which is the wrong conclusion entirely.
  - Fragment offsets are in 8-byte units, so non-final fragments are rounded to
    a multiple of 8. A 100-byte fragment is not an error, it is a datagram no
    host will ever reassemble.

frag-send is advertised only when a raw socket can actually be opened — checked
by opening one, since a permission model has more ways to say no than a
capability bit has to say yes.

Fragment header arithmetic is unit-tested (reassembly coverage, MF flags, shared
IP ID, 8-byte offsets, checksum verification), cross-compiled and run on Linux
since the code is build-tagged.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

echolot-app/core-engine/src/main/kotlin/app/echo_lot/engine/DownstreamMeasurement.kt

@@ -37,6 +37,120 @@ class DownstreamMeasurement(private val ids: IdSource) {
     /** How long to wait for a granted burst after the server accepts the action. */
     private val collectWindowMs = 4_000L
 
+    /**
+     * Shorter, but long enough to cover the first_last mode's deliberate 250 ms hold plus a
+     * reassembly. A fragment burst is one datagram: it is here quickly or not at all.
+     */
+    private val fragWindowMs = 1_500L
+
+    /**
+     * Asks the server to send one deliberately-fragmented datagram per ordering, and reports
+     * which orderings survive the path.
+     *
+     * Kernel fragmentation always emits fragments in order, first one first, so an oversized
+     * datagram can only answer "do fragments get through at all". The interesting fault is about
+     * ordering: only the *first* fragment carries the UDP ports, so a stateful firewall that has
+     * not seen it has nothing to match the rest against, and many drop them. That failure is
+     * invisible to every in-order test and shows up in the field as "large DNS answers fail here"
+     * or "the tunnel breaks when the MTU drops".
+     */
+    fun fragmentOrdering(
+        credential: String,
+        sessionId: String,
+        control: ControlClient,
+        probe: ProbeSession,
+        sessionRef: String,
+        sizeBytes: Int = 2000,
+        fragBytes: Int = 576,
+    ): Pair<Test, List<Finding>> {
+        val testId = ids.uuid()
+        val started = ids.monoNs()
+        val delivered = LinkedHashMap<String, Boolean>()
+        val fragmentCounts = LinkedHashMap<String, Int>()
+        var unsupported = false
+
+        for (mode in FRAG_MODES) {
+            val reply = runCatching {
+                control.action(
+                    credential, sessionId,
+                    """{"action":"frag_send","size_bytes":$sizeBytes,"mode":"$mode","frag_bytes":$fragBytes}""",
+                )
+            }
+            if (reply.isFailure) {
+                // A server without a raw socket says so; that is a missing capability, not a
+                // property of the network, and must not be recorded as a failed delivery.
+                unsupported = true
+                break
+            }
+            parseInt(reply.getOrNull(), "fragments")?.let { fragmentCounts[mode] = it }
+            // The burst is already on the wire when the action returns (it is sent
+            // synchronously), so anything that survived is either here or lost.
+            val got = probe.collectGranted(fragWindowMs).any { it.type == Wire.TYPE_FRAG_DATA }
+            delivered[mode] = got
+        }
+
+        if (unsupported) {
+            return Test(
+                id = testId, type = TestType.MTU_FRAG_ORDERING, sessionRef = sessionRef, tier = Tier.APP,
+                startedMonoNs = started, endedMonoNs = ids.monoNs(),
+                status = TestStatus.UNSUPPORTED,
+                error = TestError("no_raw_socket", "this server cannot craft fragments"),
+            ) to emptyList()
+        }
+
+        val metrics = json.encodeToJsonElement(
+            FragOrderingMetrics(
+                sizeBytes = sizeBytes,
+                fragBytes = fragBytes,
+                fragmentsPerBurst = fragmentCounts,
+                deliveredByMode = delivered,
+                inOrderDelivered = delivered[FRAG_IN_ORDER] == true,
+                reorderedDelivered = delivered[FRAG_REVERSED] == true,
+                delayedFirstDelivered = delivered[FRAG_FIRST_LAST] == true,
+            ),
+        ) as JsonObject
+
+        val findings = ArrayList<Finding>()
+        val inOrder = delivered[FRAG_IN_ORDER] == true
+        val reversed = delivered[FRAG_REVERSED] == true
+        val firstLast = delivered[FRAG_FIRST_LAST] == true
+
+        if (!inOrder) {
+            findings.add(
+                finding(
+                    "mtu.fragments_blocked", Category.MTU, Severity.MEDIUM, testId,
+                    "IP fragments do not reach this device",
+                    "A fragmented datagram sent in the normal order never arrived. Anything that " +
+                        "relies on fragmentation — large DNS answers over UDP, some VPN traffic — " +
+                        "will fail here rather than slow down.",
+                ),
+            )
+        } else if (!reversed || !firstLast) {
+            // The precise and useful finding: fragments work, but only if they arrive tidily.
+            val which = buildList {
+                if (!reversed) add("out of order")
+                if (!firstLast) add("with the first fragment delayed")
+            }.joinToString(" or ")
+            findings.add(
+                finding(
+                    "mtu.fragment_reorder_sensitive", Category.MTU, Severity.LOW, testId,
+                    "Fragments are dropped when they arrive $which",
+                    "In-order fragments are delivered, but the same datagram sent $which is not. " +
+                        "Something on the path only reassembles when the first fragment (the one " +
+                        "carrying the UDP ports) arrives first — typical of a stateful firewall " +
+                        "or NAT. It works until the network reorders, then fails intermittently, " +
+                        "which is the hardest kind of fault to chase.",
+                ),
+            )
+        }
+        return Test(
+            id = testId, type = TestType.MTU_FRAG_ORDERING, sessionRef = sessionRef, tier = Tier.APP,
+            startedMonoNs = started, endedMonoNs = ids.monoNs(),
+            status = if (inOrder) TestStatus.OK else TestStatus.PARTIAL,
+            metrics = metrics,
+        ) to findings
+    }
+
     /**
      * Runs all three against an already-primed session.
      *
@@ -66,6 +180,16 @@ class DownstreamMeasurement(private val ids: IdSource) {
 
         tests.add(df.test); tests.add(frag.test); tests.add(train.test)
 
+        // Fragment ordering only makes sense once we know fragments arrive at all; when they do
+        // not, the ordering variants would all report "not delivered" and read as three faults
+        // instead of one.
+        if (frag.largestDelivered != null) {
+            val (fragTest, fragFindings) =
+                fragmentOrdering(credential, sessionId, control, probe, sessionRef)
+            tests.add(fragTest)
+            findings.addAll(fragFindings)
+        }
+
         // A downstream MTU below the classic 1500-byte Ethernet payload is worth saying out loud:
         // it is the usual cause of "small requests work, large responses hang".
         val pathMtu = df.largestDelivered
@@ -310,6 +434,11 @@ class DownstreamMeasurement(private val ids: IdSource) {
         /** IPv4 (20) + UDP (8). The v6 case is 48; reported per-family once v6 sessions land. */
         const val IP_UDP_OVERHEAD4 = 28
 
+        const val FRAG_IN_ORDER = "in_order"
+        const val FRAG_REVERSED = "reversed"
+        const val FRAG_FIRST_LAST = "first_last"
+        val FRAG_MODES = listOf(FRAG_IN_ORDER, FRAG_REVERSED, FRAG_FIRST_LAST)
+
         /** Straddles the usual suspects: 1500 Ethernet, 1492 PPPoE, 1400-ish tunnels. */
         val DEFAULT_SIZES = listOf(600, 1200, 1372, 1400, 1450, 1472, 1500, 2000, 4000)
 
@@ -330,6 +459,18 @@ data class BigSendMetrics(
     @SerialName("path_mtu_bytes") val pathMtuBytes: Int? = null,
 )
 
+/** Metrics for mtu.frag_ordering. */
+@Serializable
+data class FragOrderingMetrics(
+    @SerialName("size_bytes") val sizeBytes: Int,
+    @SerialName("frag_bytes") val fragBytes: Int,
+    @SerialName("fragments_per_burst") val fragmentsPerBurst: Map<String, Int>,
+    @SerialName("delivered_by_mode") val deliveredByMode: Map<String, Boolean>,
+    @SerialName("in_order_delivered") val inOrderDelivered: Boolean,
+    @SerialName("reordered_delivered") val reorderedDelivered: Boolean,
+    @SerialName("delayed_first_delivered") val delayedFirstDelivered: Boolean,
+)
+
 /** Metrics for train.udp_downstream. */
 @Serializable
 data class DownTrainMetrics(

echolot-app/core-measurement/src/main/kotlin/app/echo_lot/measurement/Test.kt

@@ -77,6 +77,8 @@ object TestType {
     const val MTU_BLACKHOLE = "mtu.blackhole"
     const val MTU_MSS_OBSERVED = "mtu.mss_observed"
     const val MTU_FRAG_DELIVERY = "mtu.frag_delivery"
+    /** Whether fragments survive arriving out of order, not merely whether they survive. */
+    const val MTU_FRAG_ORDERING = "mtu.frag_ordering"
     // nat
     const val NAT_STUN_5780 = "nat.stun_5780"
     const val NAT_MAPPING_LIFETIME_UDP = "nat.mapping_lifetime_udp"

echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/Wire.kt

@@ -32,6 +32,12 @@ object Wire {
     const val TYPE_DOWNTRAIN_DATA: Int = 0x06
     const val TYPE_BIG_SEND: Int = 0x0C
 
+    /**
+     * A datagram the server deliberately fragmented. Its arrival IS the measurement: it can only
+     * be delivered if every fragment survived the path and the local stack reassembled them.
+     */
+    const val TYPE_FRAG_DATA: Int = 0x0D
+
     /** The 8-byte on-the-wire prefix = first 16 hex chars of the session id, decoded. */
     fun wirePrefix(sessionId: String): ByteArray {
         require(sessionId.length >= 16) { "session id too short" }