door: the peer-pin fix — the UDP front door's same-IP lock + reply-port refresh (OQ-SK2-09 option 2)

- serve_on pins the client's source IP (parsed datagrams only —
  protocol garbage can never move the pin) and sends replies to the
  latest-seen port from that IP: a NAT port rebind under a stable IP
  keeps receiving (the review 003 S-1 residual, now fixed)
- the RFC 1928 MUST implemented: datagrams from a source IP other
  than the pinned one are dropped (previously every parsed datagram
  was forwarded regardless of source); the spoof primitive shrinks
  from any-source to same-IP off-path
- the residual (a full client IP change mid-association) stays the
  RFC's expected-peer posture — revisitable per OQ-SK2-09 (the
  control-conn-derived expected IP is the complete fix, ADR-shaped)
- two tests pin the new behavior: the NAT port-rebind follow (the
  old port stops receiving, the new one takes over) and the
  foreign-IP drop (no reply to the spoofer, the association healthy)

Verification: cargo test --features local (15 suites green, door-udp
38 tests), clippy --all-targets --all-features -D warnings clean,
fmt clean.

src/door.rs

@@ -831,13 +831,20 @@ impl UdpFrontDoor {
     /// Serve the association over an existing relay socket (the
     /// test/assembly seam — the same pump without the bind).
     ///
-    /// Posture (accepted residual, review 003 S-1): the client's
-    /// source address is learned from the first inbound datagram and
-    /// pinned for the association's lifetime — a wildcard-bind
-    /// deployment accepts the off-path-first-datagram shape (the
-    /// RFC's own per-association peer expectation; the loopback bind
-    /// is the documented posture). A client whose source port
-    /// rebinds mid-association (NAT) stops receiving replies.
+    /// Posture (the peer-pin shape, OQ-SK2-09): the client's source
+    /// **IP** is learned from the first inbound datagram and pinned for
+    /// the association's lifetime; replies go to the **latest-seen
+    /// port** from that IP (a NAT port rebind under a stable IP keeps
+    /// receiving). Datagrams from a source IP other than the pinned
+    /// one are dropped — RFC 1928's MUST ("MUST drop any datagrams
+    /// arriving from any source IP address other than the one recorded
+    /// for the particular association"). The residual (a full client
+    /// IP change mid-association — a NAT *address* rebind) stops
+    /// receiving replies by design: the RFC's expected-peer posture,
+    /// revisitable per OQ-SK2-09. The loopback bind is the documented
+    /// posture; a wildcard-bind deployment accepts the same-IP
+    /// off-path-first-datagram shape (a much weaker primitive than the
+    /// pre-fix any-source pin).
     pub async fn serve_on(
         self,
         socket: UdpSocket,
@@ -865,18 +872,26 @@ impl UdpFrontDoor {
             return Ok(());
         };
         let socket = Arc::new(socket);
-        // The client's source address — learned from the first inbound
-        // datagram and used for every reply (the relay's single-client
-        // shape; the door serves one association per socket).
-        let client_addr: Arc<std::sync::Mutex<Option<SocketAddr>>> =
-            Arc::new(std::sync::Mutex::new(None));
+        // The peer pin (OQ-SK2-09): the client's source IP — learned
+        // from the first inbound datagram and pinned; the reply port
+        // refreshes to the latest-seen source port from that IP (the
+        // NAT-port-rebind shape). A source IP mismatch is the RFC's
+        // drop case (the datagram is dropped, not forwarded).
+        let client_peer: Arc<std::sync::RwLock<Option<std::net::IpAddr>>> =
+            Arc::new(std::sync::RwLock::new(None));
+        let reply_port: Arc<std::sync::RwLock<Option<u16>>> =
+            Arc::new(std::sync::RwLock::new(None));
 
         // socket→channel: decode the RFC header (the codec boundary),
-        // refuse FRAG ≠ 0 (drop the datagram — the v1 posture), send
-        // `{addr, data}` in-band. Each datagram resets the idle window.
+        // refuse FRAG ≠ 0 (drop the datagram — the v1 posture), pin
+        // the peer IP + enforce the RFC's same-IP drop rule (the pin
+        // reacts to *parsed* datagrams only — protocol garbage can
+        // never move it), refresh the reply port, send `{addr, data}`
+        // in-band. Each datagram resets the idle window.
         let s2c = {
             let socket = Arc::clone(&socket);
-            let client_addr = Arc::clone(&client_addr);
+            let client_peer = Arc::clone(&client_peer);
+            let reply_port = Arc::clone(&reply_port);
             async move {
                 let mut scratch = vec![0u8; 65535];
                 loop {
@@ -900,11 +915,22 @@ impl UdpFrontDoor {
                         continue;
                     }
                     {
-                        let mut known = client_addr.lock().unwrap_or_else(|e| e.into_inner());
-                        if known.is_none() {
-                            *known = Some(from);
+                        let mut known = client_peer.write().unwrap_or_else(|e| e.into_inner());
+                        match *known {
+                            None => *known = Some(from.ip()),
+                            Some(ip) if ip != from.ip() => {
+                                tracing::debug!(
+                                    peer = %from,
+                                    pinned = %ip,
+                                    "udp front door: datagram from a non-expected \
+                                     source IP dropped (RFC 1928 peer posture)"
+                                );
+                                continue;
+                            }
+                            Some(_) => {}
                         }
                     }
+                    *reply_port.write().unwrap_or_else(|e| e.into_inner()) = Some(from.port());
                     if let Err(err) = send_inband(&mut channel_write, &addr, payload).await {
                         return UdpDoorExit::Session(err.to_string());
                     }
@@ -914,15 +940,16 @@ impl UdpFrontDoor {
 
         // channel→socket: receive the in-band `{addr, data}` frame,
         // encode the RFC reply header (the codec boundary), send the
-        // RFC-header'd datagram back to the client's source address.
-        // The association's end (the channel EOF) is this pump's clean
-        // exit. The decode state (the incremental reader + the
-        // batched-datagram pending queue) persists across pump
-        // iterations — the bridge owns it for the association's
-        // lifetime.
+        // RFC-header'd datagram back to the pinned peer IP's
+        // latest-seen reply port. The association's end (the channel
+        // EOF) is this pump's clean exit. The decode state (the
+        // incremental reader + the batched-datagram pending queue)
+        // persists across pump iterations — the bridge owns it for the
+        // association's lifetime.
         let c2s = {
             let socket = Arc::clone(&socket);
-            let client_addr = Arc::clone(&client_addr);
+            let client_peer = Arc::clone(&client_peer);
+            let reply_port = Arc::clone(&reply_port);
             async move {
                 let mut bridge = InbandBridge::new();
                 loop {
@@ -936,14 +963,12 @@ impl UdpFrontDoor {
                         Err(err) => return UdpDoorExit::Session(err.to_string()),
                     };
                     header.extend_from_slice(&payload);
-                    let to = client_addr
-                        .lock()
-                        .unwrap_or_else(|e| e.into_inner())
-                        .as_ref()
-                        .copied();
-                    let Some(to) = to else {
+                    let ip = *client_peer.read().unwrap_or_else(|e| e.into_inner());
+                    let port = *reply_port.read().unwrap_or_else(|e| e.into_inner());
+                    let (Some(ip), Some(port)) = (ip, port) else {
                         continue;
                     };
+                    let to = SocketAddr::new(ip, port);
                     if let Err(err) = socket.send_to(&header, to).await {
                         return UdpDoorExit::Io(err);
                     }

tests/door-udp.rs

@@ -501,3 +501,144 @@ async fn udp_door_serve_wrapper_round_trips() {
     let (_, payload) = parse_rfc_datagram(&rx[..n]);
     assert_eq!(payload, b"wrapped");
 }
+
+/// The NAT port-rebind shape (OQ-SK2-09's option-2 fix): the client's
+/// source IP pins on the first datagram, and a later datagram from the
+/// same IP on a *new source port* moves the reply path — the rebind
+/// keeps receiving replies. The RFC drop rule's complement: replies go
+/// to the latest-seen port from the pinned IP.
+#[tokio::test]
+async fn nat_port_rebind_keeps_receiving_replies() {
+    let echo_addr = udp_echo_server().await;
+    let topo = harness::wire_consumer_side_with(
+        echo_dial(),
+        Some(alksocks::udp_egress()),
+        None,
+        Some(harness::consumer_identity()),
+    )
+    .await;
+
+    let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
+    let relay_addr = relay.local_addr().expect("relay addr");
+    let door = UdpFrontDoor::builder()
+        .bind(relay_addr)
+        .build()
+        .expect("the door builds");
+    let _serve = tokio::spawn({
+        let client = Arc::clone(&topo.consumer);
+        async move { door.serve_on(relay, client).await.expect("the door serves") }
+    });
+
+    // Association born on socket A (its source port pins the reply
+    // path's IP; the port rides the latest-seen shape).
+    let client_a = UdpSocket::bind("127.0.0.1:0").await.expect("client A");
+    client_a
+        .send_to(&rfc_datagram(echo_addr, b"before"), relay_addr)
+        .await
+        .expect("send from A");
+    let mut rx = vec![0u8; 65535];
+    let (n, _) = timeout(BOUND, client_a.recv_from(&mut rx))
+        .await
+        .expect("reply within bound")
+        .expect("recv");
+    assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"before");
+
+    // The "NAT rebind": a fresh socket, same IP, new source port. The
+    // datagram is accepted (same IP) and the reply goes to the *new*
+    // port — the rebind keeps receiving.
+    let client_b = UdpSocket::bind("127.0.0.1:0").await.expect("client B");
+    client_b
+        .send_to(&rfc_datagram(echo_addr, b"after"), relay_addr)
+        .await
+        .expect("send from B");
+    let (n, _) = timeout(BOUND, client_b.recv_from(&mut rx))
+        .await
+        .expect("rebound reply within bound")
+        .expect("recv");
+    assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"after");
+
+    // A (the old port) no longer receives: the reply path followed
+    // the rebind.
+    let mut probe = vec![0u8; 65535];
+    assert!(
+        timeout(Duration::from_millis(300), client_a.recv_from(&mut probe))
+            .await
+            .is_err(),
+        "the reply followed the rebind (the old port stopped receiving)"
+    );
+}
+
+/// The RFC 1928 peer posture's drop rule (OQ-SK2-09's option-2 fix):
+/// a datagram from a *different source IP* is dropped — never
+/// egressed — and the association stays healthy for the pinned peer.
+/// (Loopback-only IPs are used: the pinned peer keeps receiving, the
+/// foreign-IP sender's datagram disappears.)
+#[tokio::test]
+async fn foreign_source_ip_datagrams_dropped_and_association_stays_healthy() {
+    init_trace();
+    let echo_addr = udp_echo_server().await;
+    let topo = harness::wire_consumer_side_with(
+        echo_dial(),
+        Some(alksocks::udp_egress()),
+        None,
+        Some(harness::consumer_identity()),
+    )
+    .await;
+
+    let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
+    let relay_addr = relay.local_addr().expect("relay addr");
+    let door = UdpFrontDoor::builder()
+        .bind(relay_addr)
+        .build()
+        .expect("the door builds");
+    let _serve = tokio::spawn({
+        let client = Arc::clone(&topo.consumer);
+        async move { door.serve_on(relay, client).await.expect("the door serves") }
+    });
+
+    // The real client pins the peer on loopback (127.0.0.1).
+    let client = UdpSocket::bind("127.0.0.1:0").await.expect("client");
+    client
+        .send_to(&rfc_datagram(echo_addr, b"pinned"), relay_addr)
+        .await
+        .expect("send pinned");
+    let mut rx = vec![0u8; 65535];
+    let (n, _) = timeout(BOUND, client.recv_from(&mut rx))
+        .await
+        .expect("reply within bound")
+        .expect("recv");
+    assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"pinned");
+
+    // A "foreign-IP" sender: 127.0.0.2 (loopback, a different IP —
+    // bindable without privileges). Its datagram is dropped at the
+    // door's peer gate — no echo comes back to it, and nothing is
+    // egressed to the echo target on its behalf.
+    let foreign = UdpSocket::bind("127.0.0.2:0").await.expect("foreign bind");
+    foreign
+        .send_to(&rfc_datagram(echo_addr, b"hijack"), relay_addr)
+        .await
+        .expect("send foreign");
+    let mut probe = vec![0u8; 65535];
+    assert!(
+        timeout(Duration::from_millis(400), foreign.recv_from(&mut probe))
+            .await
+            .is_err(),
+        "the foreign-IP datagram was dropped (no reply to the spoofer)"
+    );
+
+    // The pinned peer is unaffected: its next datagram round-trips
+    // (the drop left the association healthy — the gate never evicts).
+    client
+        .send_to(&rfc_datagram(echo_addr, b"still"), relay_addr)
+        .await
+        .expect("send still");
+    let (n, _) = timeout(BOUND, client.recv_from(&mut rx))
+        .await
+        .expect("still-reply within bound")
+        .expect("recv");
+    assert_eq!(
+        parse_rfc_datagram(&rx[..n]).1,
+        b"still",
+        "the association survived the foreign-IP drop"
+    );
+}