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.
This commit is contained in:
glm-5.3-flash committed 2026-09-28 09:15:05 +00:00
1 parent 086aeb4135
commit 31fb11cf83
2 files changed
+197 -31

No files matched your search

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