Stabilize PTCP relay sessions

README.md

@@ -36,11 +36,18 @@ Arguments:
 
 Options:
   -p, --port <[bind_address:]port:remote_port>
-          Bind address, port and remote port. Default: 127.0.0.1:1554:554
+          Bind address, port and remote port. Repeat for multiple mappings
   -h, --help
           Print help
 ```
 
+When `--port` is omitted, the default mapping is `127.0.0.1:1554:554`.
+Multiple device ports can share one P2P session by repeating the option:
+
+```bash
+dh-p2p --port 127.0.0.1:18080:80 --port 127.0.0.1:1554:554 [CAMERA_SERIAL]
+```
+
 ## Python implementation
 
 The Python implementation of DH-P2P is a simple and straightforward approach. It is used for drafting and testing purposes due to its quick and easy-to-write nature. Additionally, the implementation is more linear and follows a top-down execution flow, making it easier to understand. Python, being a popular programming language, further contributes to its accessibility and familiarity among developers.

src/dh.rs

@@ -106,10 +106,20 @@ pub async fn p2p_handshake(
 
     let mut res = socket.dh_read_raw().await;
 
-    if res.code == 100 {
+    // A busy P2P server can emit more than one provisional response before
+    // returning the device NAT information. Keep reading a small, bounded
+    // number of responses instead of trying to parse a 100 body as device data.
+    for _ in 0..4 {
+        if res.code >= 200 {
+            break;
+        }
         res = socket.dh_read_raw().await;
     }
 
+    if res.code < 200 {
+        panic!("Too many provisional responses while creating P2P channel");
+    }
+
     if res.code >= 400 {
         if res.code == 403 {
             println!("Device requires authentication when creating P2P channel.");
@@ -143,7 +153,22 @@ pub async fn p2p_handshake(
     let mut session = PTCPSession::new();
 
     socket2.ptcp_request(session.send(PTCPBody::Sync)).await;
-    session.recv(socket2.ptcp_read().await);
+    loop {
+        let packet = session.recv(socket2.ptcp_read().await);
+
+        if packet.is_duplicate() {
+            if !matches!(&packet.body, PTCPBody::Empty) {
+                socket2.ptcp_request(session.send(PTCPBody::Empty)).await;
+            }
+            continue;
+        }
+
+        match packet.body {
+            PTCPBody::Sync => break,
+            PTCPBody::Empty => {}
+            _ => socket2.ptcp_request(session.send(PTCPBody::Empty)).await,
+        }
+    }
 
     if relay_mode {
         return (socket2, session);
@@ -154,15 +179,23 @@ pub async fn p2p_handshake(
             b"\x17\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00".to_vec(),
         )))
         .await;
-    let mut res = session.recv(socket2.ptcp_read().await);
+    let sign = loop {
+        let packet = session.recv(socket2.ptcp_read().await);
 
-    while let PTCPBody::Empty = res.body {
-        res = session.recv(socket2.ptcp_read().await);
-    }
+        if packet.is_duplicate() {
+            if !matches!(&packet.body, PTCPBody::Empty) {
+                socket2.ptcp_request(session.send(PTCPBody::Empty)).await;
+            }
+            continue;
+        }
 
-    let sign = match res.body {
-        PTCPBody::Command(ref c) => &c[12..],
-        _ => panic!("Invalid response"),
+        match packet.body {
+            PTCPBody::Command(c) if c.first() == Some(&0x18) && c.len() >= 16 => {
+                break c[12..].to_vec()
+            }
+            PTCPBody::Empty => {}
+            _ => socket2.ptcp_request(session.send(PTCPBody::Empty)).await,
+        }
     };
 
     println!(
@@ -263,40 +296,72 @@ pub async fn p2p_handshake(
     let mut session = PTCPSession::new();
 
     socket.ptcp_request(session.send(PTCPBody::Sync)).await;
-    let mut res = session.recv(socket.ptcp_read().await);
-    assert!(matches!(res.body, PTCPBody::Sync), "Invalid response");
+    loop {
+        let packet = session.recv(socket.ptcp_read().await);
+
+        if packet.is_duplicate() {
+            if !matches!(&packet.body, PTCPBody::Empty) {
+                socket.ptcp_request(session.send(PTCPBody::Empty)).await;
+            }
+            continue;
+        }
+
+        match packet.body {
+            PTCPBody::Sync => break,
+            PTCPBody::Empty => {}
+            _ => socket.ptcp_request(session.send(PTCPBody::Empty)).await,
+        }
+    }
 
     socket
         .ptcp_request(
             session.send(PTCPBody::Command(
                 [
                     b"\x19\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00".to_vec(),
-                    sign.to_vec(),
+                    sign,
                 ]
                 .concat(),
             )),
         )
         .await;
 
-    res = session.recv(socket.ptcp_read().await);
-    while let PTCPBody::Empty = res.body {
-        res = session.recv(socket.ptcp_read().await);
-    }
-    match res.body {
-        PTCPBody::Command(ref c) => {
-            assert_eq!(c[0], 0x1A, "Invalid response");
+    loop {
+        let packet = session.recv(socket.ptcp_read().await);
+
+        if packet.is_duplicate() {
+            if !matches!(&packet.body, PTCPBody::Empty) {
+                socket.ptcp_request(session.send(PTCPBody::Empty)).await;
+            }
+            continue;
+        }
+
+        match packet.body {
+            PTCPBody::Command(c) if c.first() == Some(&0x1A) => break,
+            PTCPBody::Empty => {}
+            _ => socket.ptcp_request(session.send(PTCPBody::Empty)).await,
         }
-        _ => panic!("Invalid response"),
     }
 
     socket
         .ptcp_request(session.send(PTCPBody::Command(
             b"\x1b\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00".to_vec(),
         )))
         .await;
-    res = session.recv(socket.ptcp_read().await);
+    loop {
+        let packet = session.recv(socket.ptcp_read().await);
 
-    assert!(matches!(res.body, PTCPBody::Empty), "Invalid response");
+        if packet.is_duplicate() {
+            if !matches!(&packet.body, PTCPBody::Empty) {
+                socket.ptcp_request(session.send(PTCPBody::Empty)).await;
+            }
+            continue;
+        }
+
+        match packet.body {
+            PTCPBody::Empty => break,
+            _ => socket.ptcp_request(session.send(PTCPBody::Empty)).await,
+        }
+    }
 
     (socket, session)
 }