g3proxy: fix smtp interception

g3proxy/examples/inspect_http_proxy/g3proxy.yaml

@@ -19,6 +19,12 @@ server:
     listen:
       address: "[::]:8080"
     tls_client: {}
+  - name: socks
+    escaper: default
+    auditor: default
+    type: socks_proxy
+    listen:
+      address: "[::]:1080"
 
 resolver:
   - name: default

g3proxy/src/inspect/smtp/ending.rs

@@ -143,7 +143,10 @@ impl EndWaitClient {
             };
 
             let cmd = match Command::parse_line(line) {
-                Ok(cmd) => cmd,
+                Ok(cmd) => {
+                    recv_buf.consume_line();
+                    cmd
+                }
                 Err(e) => {
                     let _ = ResponseEncoder::from(&e).write(&mut clt_w).await;
                     return Err(ServerTaskError::ClientAppError(anyhow!(
@@ -164,8 +167,6 @@ impl EndWaitClient {
                     .await
                     .map_err(ServerTaskError::ClientTcpWriteFailed)?;
             }
-
-            recv_buf.consume_line();
         }
 
         let _ = clt_w.shutdown().await;

g3proxy/src/inspect/smtp/forward.rs

@@ -77,6 +77,7 @@ impl<'a> Forward<'a> {
     {
         loop {
             let mut valid_cmd = Command::NoOperation;
+            buf.cmd_recv_buf.consume_line();
             let Some(_cmd_line) = buf
                 .cmd_recv_buf
                 .recv_cmd_and_relay(
@@ -183,6 +184,7 @@ impl<'a> Forward<'a> {
     {
         let mut rsp = ResponseParser::default();
         loop {
+            buf.rsp_recv_buf.consume_line();
             let line = buf
                 .rsp_recv_buf
                 .read_rsp_line_with_feedback(

g3proxy/src/inspect/smtp/greeting.rs

@@ -61,6 +61,7 @@ impl Greeting {
         let mut recv_buf = LineRecvBuf::<{ ResponseParser::MAX_LINE_SIZE }>::default();
 
         loop {
+            recv_buf.consume_line();
             let line = recv_buf.read_line(&mut ups_r).await.map_err(|e| match e {
                 RecvLineError::IoError(e) => GreetingError::UpstreamReadFailed(e),
                 RecvLineError::IoClosed => GreetingError::UpstreamClosed,
@@ -99,8 +100,6 @@ impl Greeting {
                 }
                 c => return Err(GreetingError::UnexpectedReplyCode(c)),
             }
-
-            recv_buf.consume_line();
         }
     }
 

g3proxy/src/inspect/smtp/initiation.rs

@@ -98,6 +98,7 @@ impl<'a> Initiation<'a> {
         let mut rsp_recv_buf = LineRecvBuf::<{ ResponseParser::MAX_LINE_SIZE }>::default();
 
         loop {
+            cmd_recv_buf.consume_line();
             let Some(_cmd_line) = cmd_recv_buf
                 .recv_cmd_and_relay(
                     self.config.command_wait_timeout,
@@ -144,6 +145,7 @@ impl<'a> Initiation<'a> {
     {
         let mut rsp = ResponseParser::default();
         loop {
+            rsp_recv_buf.consume_line();
             let line = rsp_recv_buf
                 .read_rsp_line_with_feedback(
                     self.config.response_wait_timeout,

g3proxy/src/inspect/smtp/transaction/mod.rs

@@ -129,6 +129,7 @@ impl<'a, SC: ServerConfig> Transaction<'a, SC> {
         loop {
             let mut valid_cmd = Command::NoOperation;
 
+            buf.cmd_recv_buf.consume_line();
             let Some(cmd_line) = buf
                 .cmd_recv_buf
                 .recv_cmd_and_relay(
@@ -138,7 +139,11 @@ impl<'a, SC: ServerConfig> Transaction<'a, SC> {
                     ups_w,
                     |cmd| {
                         match &cmd {
-                            Command::Recipient(_) => {}
+                            Command::Recipient(_) => {
+                                if in_chunking {
+                                    return Some(ResponseEncoder::BAD_SEQUENCE_OF_COMMANDS);
+                                }
+                            }
                             Command::Data => {
                                 if in_chunking {
                                     return Some(ResponseEncoder::BAD_SEQUENCE_OF_COMMANDS);
@@ -264,6 +269,7 @@ impl<'a, SC: ServerConfig> Transaction<'a, SC> {
     {
         let mut rsp = ResponseParser::default();
         loop {
+            buf.rsp_recv_buf.consume_line();
             let line = buf
                 .rsp_recv_buf
                 .read_rsp_line_with_feedback(recv_timeout, ups_r, clt_w, self.local_ip)

lib/g3-io-ext/src/io/line_recv_buf.rs

@@ -51,8 +51,8 @@ impl<const MAX_LINE_SIZE: usize> Default for LineRecvBuf<MAX_LINE_SIZE> {
 }
 
 impl<const MAX_LINE_SIZE: usize> LineRecvBuf<MAX_LINE_SIZE> {
-    pub async fn read_line_with_timeout<'a, R>(
-        &'a mut self,
+    pub async fn read_line_with_timeout<R>(
+        &mut self,
         reader: &mut R,
         timeout: Duration,
     ) -> Result<&[u8], RecvLineError>
@@ -64,7 +64,7 @@ impl<const MAX_LINE_SIZE: usize> LineRecvBuf<MAX_LINE_SIZE> {
             .map_err(|_| RecvLineError::Timeout)?
     }
 
-    pub async fn read_line<'a, R>(&'a mut self, reader: &mut R) -> Result<&[u8], RecvLineError>
+    pub async fn read_line<R>(&mut self, reader: &mut R) -> Result<&[u8], RecvLineError>
     where
         R: AsyncRead + Unpin,
     {

lib/g3-smtp-proto/src/command/mod.rs

@@ -69,7 +69,7 @@ pub enum Command {
 }
 
 impl Command {
-    pub const MAX_LINE_SIZE: usize = 512;
+    pub const MAX_LINE_SIZE: usize = 2048;
     pub const MAX_CONTINUE_LINE_SIZE: usize = 12288; // for AUTH continue line
 
     pub fn parse_line(line: &[u8]) -> Result<Self, CommandLineError> {

lib/g3-smtp-proto/src/io/data.rs

@@ -20,7 +20,7 @@ use std::task::{ready, Context, Poll};
 
 use tokio::io::{AsyncRead, ReadBuf};
 
-const END_CHARS: &[u8] = b"\r\n.\r\n\r\n";
+const END_CHARS: &[u8] = b"\r\n.\r\n";
 const END_SIZE: usize = 16; // big enough for END_CHARS
 
 struct EndChecker {