g3proxy: use remote socket address for udpdump

g3proxy/CHANGELOG

@@ -6,6 +6,8 @@ v1.9.8:
  - Feature: allow to write event logs to stdout
  - Feature: add stream detour support to audit
  - Feature: only enable anonymous user if ingress network ACL matched
+ - Feature: add stream_id and match_uint tag to exported_pdu frame when do udp dump
+ - Optimization: use real remote peer address in exported_pdu when do udp dump
  - Optimization: check user ingress network ACL early and reply auth failed if denied
  - Optimization: ignore more clean connection close error in interception
 

g3proxy/doc/protocol/setup/wireshark_udpdump.rst

@@ -43,4 +43,51 @@ Example:
 
   tshark -i udpdump -o extcap.udpdump.payload:exported_pdu -o extcap.udpdump.port:5555 <...>
 
+Packet Layout
+-------------
+
+You will see two exported_pdu frame for each packet in the saved capture file.
+
+The first frame is added by `udpdump`, it will contain:
+
+- UDP Socket Address of g3proxy
+
+  The src ip may be in tag `EXP_PDU_TAG_IPV4_SRC` or `EXP_PDU_TAG_IPV6_SRC`.
+  The src port will be in tag `EXP_PDU_TAG_SRC_PORT`.
+
+- UDP Socket Address of udpdump
+
+  The dst ip may be in tag `EXP_PDU_TAG_IPV4_DST` or `EXP_PDU_TAG_IPV6_DST`.
+  The dst port will be in tag `EXP_PDU_TAG_DST_PORT`.
+
+The second frame is generated by `g3proxy`, it will contain:
+
+- Socket Address
+
+  The address used here will include:
+
+    * client socket address for the client connection
+    * remote socket address for the remote connection
+
+  The src ip may be in tag `EXP_PDU_TAG_IPV4_SRC` or `EXP_PDU_TAG_IPV6_SRC`.
+  The src port will be in tag `EXP_PDU_TAG_SRC_PORT`.
+  The dst ip may be in tag `EXP_PDU_TAG_IPV4_DST` or `EXP_PDU_TAG_IPV6_DST`.
+  The dst port will be in tag `EXP_PDU_TAG_DST_PORT`.
+
+- Port Type
+
+  It will be in tag `EXP_PDU_TAG_PORT_TYPE`,  and the value will be EXP_PDU_PT_TCP for stream based connections.
+
+- Stream ID
+
+  It will be in tag `EXP_PDU_TAG_TCP_STREAM_ID`, and will only be set for stream based connections.
+
+- Match Uint
+
+  It will be in tag `EXP_PDU_TAG_MATCH_UINT`, and currently the value should be the same as the remote address port.
+
+To identify an unique stream, you need to use all of the following values:
 
+- src ip + src port in the first frame
+- src ip + src port + dst ip + dst port + port type in the second frame
+- stream id in the second frame

g3proxy/src/inspect/mod.rs

@@ -73,6 +73,7 @@ pub(crate) struct StreamInspectTaskNotes {
     task_id: Uuid,
     pub(crate) client_addr: SocketAddr,
     pub(crate) server_addr: SocketAddr,
+    pub(crate) remote_addr: SocketAddr,
     worker_id: Option<usize>,
     user_ctx: Option<StreamInspectUserContext>,
 }
@@ -92,14 +93,13 @@ impl StreamInspectTaskNotes {
     pub(crate) fn task_id(&self) -> &Uuid {
         &self.task_id
     }
-}
 
-impl From<&ServerTaskNotes> for StreamInspectTaskNotes {
-    fn from(task_notes: &ServerTaskNotes) -> Self {
+    fn new(task_notes: &ServerTaskNotes, remote_addr: SocketAddr) -> Self {
         StreamInspectTaskNotes {
             task_id: task_notes.id,
             client_addr: task_notes.client_addr(),
             server_addr: task_notes.server_addr(),
+            remote_addr,
             worker_id: task_notes.worker_id(),
             user_ctx: task_notes.user_ctx().map(|ctx| StreamInspectUserContext {
                 raw_user_name: ctx.raw_user_name().cloned(),
@@ -143,6 +143,7 @@ impl<SC: ServerConfig> StreamInspectContext<SC> {
         server_stats: ArcServerStats,
         server_quit_policy: Arc<ServerQuitPolicy>,
         task_notes: &ServerTaskNotes,
+        remote_addr: SocketAddr,
     ) -> Self {
         let mut task_max_idle_count = server_config.task_max_idle_count();
         if let Some(user_ctx) = task_notes.user_ctx() {
@@ -154,7 +155,7 @@ impl<SC: ServerConfig> StreamInspectContext<SC> {
             server_config,
             server_stats,
             server_quit_policy,
-            task_notes: StreamInspectTaskNotes::from(task_notes),
+            task_notes: StreamInspectTaskNotes::new(task_notes, remote_addr),
             inspection_depth: 0,
             task_max_idle_count,
         }

g3proxy/src/inspect/start_tls/mod.rs

@@ -287,7 +287,7 @@ where
             if stream_dumper.client_side() {
                 let (clt_r, clt_w) = stream_dumper.wrap_client_io(
                     self.ctx.task_notes.client_addr,
-                    self.ctx.task_notes.server_addr,
+                    self.ctx.task_notes.remote_addr,
                     dissector_hint,
                     clt_r,
                     clt_w,
@@ -296,7 +296,7 @@ where
             } else {
                 let (ups_r, ups_w) = stream_dumper.wrap_remote_io(
                     self.ctx.task_notes.client_addr,
-                    self.ctx.task_notes.server_addr,
+                    self.ctx.task_notes.remote_addr,
                     dissector_hint,
                     ups_r,
                     ups_w,

g3proxy/src/inspect/tls/mod.rs

@@ -215,7 +215,7 @@ where
             if stream_dumper.client_side() {
                 let (clt_r, clt_w) = stream_dumper.wrap_client_io(
                     self.ctx.task_notes.client_addr,
-                    self.ctx.task_notes.server_addr,
+                    self.ctx.task_notes.remote_addr,
                     dissector_hint,
                     clt_r,
                     clt_w,
@@ -224,7 +224,7 @@ where
             } else {
                 let (ups_r, ups_w) = stream_dumper.wrap_remote_io(
                     self.ctx.task_notes.client_addr,
-                    self.ctx.task_notes.server_addr,
+                    self.ctx.task_notes.remote_addr,
                     dissector_hint,
                     ups_r,
                     ups_w,

g3proxy/src/serve/http_proxy/task/connect/task.rs

@@ -438,12 +438,18 @@ impl HttpProxyConnectTask {
                 .unwrap_or_else(|| audit_handle.do_task_audit());
 
             if audit_task {
+                let Some(remote_addr) = self.tcp_notes.next else {
+                    return Err(ServerTaskError::InternalServerError(
+                        "no remote peer address set for connected socket",
+                    ));
+                };
                 let ctx = StreamInspectContext::new(
                     audit_handle.clone(),
                     self.ctx.server_config.clone(),
                     self.ctx.server_stats.clone(),
                     self.ctx.server_quit_policy.clone(),
                     &self.task_notes,
+                    remote_addr,
                 );
                 return crate::inspect::stream::transit_with_inspection(
                     clt_r,

g3proxy/src/serve/sni_proxy/task/relay/task.rs

@@ -181,12 +181,18 @@ impl TcpStreamTask {
         clt_w.reset_stats(clt_w_stats);
 
         if let Some(audit_handle) = self.ctx.audit_handle.take() {
+            let Some(remote_addr) = self.tcp_notes.next else {
+                return Err(ServerTaskError::InternalServerError(
+                    "no remote peer address set for connected socket",
+                ));
+            };
             let ctx = StreamInspectContext::new(
                 audit_handle,
                 self.ctx.server_config.clone(),
                 self.ctx.server_stats.clone(),
                 self.ctx.server_quit_policy.clone(),
                 &self.task_notes,
+                remote_addr,
             );
             let protocol_inspector = ctx.protocol_inspector(None);
             match self.protocol {

g3proxy/src/serve/socks_proxy/task/tcp_connect/task.rs

@@ -377,12 +377,18 @@ impl SocksProxyTcpConnectTask {
                 .unwrap_or_else(|| audit_handle.do_task_audit());
 
             if audit_task {
+                let Some(remote_addr) = self.tcp_notes.next else {
+                    return Err(ServerTaskError::InternalServerError(
+                        "no remote peer address set for connected socket",
+                    ));
+                };
                 let ctx = StreamInspectContext::new(
                     audit_handle.clone(),
                     self.ctx.server_config.clone(),
                     self.ctx.server_stats.clone(),
                     self.ctx.server_quit_policy.clone(),
                     &self.task_notes,
+                    remote_addr,
                 );
                 return crate::inspect::stream::transit_with_inspection(
                     clt_r,

g3proxy/src/serve/tcp_stream/task.rs

@@ -173,12 +173,18 @@ impl TcpStreamTask {
         UW: AsyncWrite + Send + Sync + Unpin + 'static,
     {
         if let Some(audit_handle) = self.ctx.audit_handle.take() {
+            let Some(remote_addr) = self.tcp_notes.next else {
+                return Err(ServerTaskError::InternalServerError(
+                    "no remote peer address set for connected socket",
+                ));
+            };
             let ctx = StreamInspectContext::new(
                 audit_handle,
                 self.ctx.server_config.clone(),
                 self.ctx.server_stats.clone(),
                 self.ctx.server_quit_policy.clone(),
                 &self.task_notes,
+                remote_addr,
             );
             crate::inspect::stream::transit_with_inspection(
                 clt_r,

g3proxy/src/serve/tcp_tproxy/task.rs

@@ -142,12 +142,18 @@ impl TProxyStreamTask {
         let (clt_r, clt_w) = self.split_clt(clt_stream);
 
         if let Some(audit_handle) = self.ctx.audit_handle.take() {
+            let Some(remote_addr) = self.tcp_notes.next else {
+                return Err(ServerTaskError::InternalServerError(
+                    "no remote peer address set for connected socket",
+                ));
+            };
             let ctx = StreamInspectContext::new(
                 audit_handle,
                 self.ctx.server_config.clone(),
                 self.ctx.server_stats.clone(),
                 self.ctx.server_quit_policy.clone(),
                 &self.task_notes,
+                remote_addr,
             );
             crate::inspect::stream::transit_with_inspection(
                 clt_r,

g3proxy/src/serve/tls_stream/task.rs

@@ -162,12 +162,18 @@ impl TlsStreamTask {
         let (clt_r, clt_w) = self.split_clt(clt_stream);
 
         if let Some(audit_handle) = self.ctx.audit_handle.take() {
+            let Some(remote_addr) = self.tcp_notes.next else {
+                return Err(ServerTaskError::InternalServerError(
+                    "no remote peer address set for connected socket",
+                ));
+            };
             let ctx = StreamInspectContext::new(
                 audit_handle,
                 self.ctx.server_config.clone(),
                 self.ctx.server_stats.clone(),
                 self.ctx.server_quit_policy.clone(),
                 &self.task_notes,
+                remote_addr,
             );
             crate::inspect::stream::transit_with_inspection(
                 clt_r,