g3proxy: support smtp STARTTLS interception

g3proxy/src/inspect/error.rs

@@ -24,6 +24,8 @@ use crate::serve::ServerTaskError;
 pub(crate) enum InterceptionError {
     #[error("tls: {0}")]
     Tls(super::tls::TlsInterceptionError),
+    #[error("start tls: {0}")]
+    StartTls(super::tls::TlsInterceptionError),
     #[error("http1: {0}")]
     H1(super::http::H1InterceptionError),
     #[error("http2: {0}")]

g3proxy/src/inspect/mod.rs

@@ -42,6 +42,9 @@ pub(crate) mod stream;
 pub(crate) mod tls;
 use tls::TlsInterceptionContext;
 
+pub(crate) mod start_tls;
+use start_tls::StartTlsProtocol;
+
 pub(crate) mod http;
 mod websocket;
 
@@ -291,6 +294,7 @@ pub(crate) enum StreamInspection<SC: ServerConfig> {
     TlsModern(tls::TlsInterceptObject<SC>),
     #[cfg(feature = "vendored-tongsuo")]
     TlsTlcp(tls::TlsInterceptObject<SC>),
+    StartTls(start_tls::StartTlsInterceptObject<SC>),
     H1(http::H1InterceptObject<SC>),
     H2(http::H2InterceptObject<SC>),
     Websocket(websocket::H1WebsocketInterceptObject<SC>),

g3proxy/src/inspect/smtp/forward.rs

@@ -34,14 +34,16 @@ pub(super) enum ForwardNextAction {
 pub(super) struct Forward {
     local_ip: IpAddr,
     allow_odmr: bool,
+    allow_starttls: bool,
     auth_end: bool,
 }
 
 impl Forward {
-    pub(super) fn new(local_ip: IpAddr, allow_odmr: bool) -> Self {
+    pub(super) fn new(local_ip: IpAddr, allow_odmr: bool, allow_starttls: bool) -> Self {
         Forward {
             local_ip,
             allow_odmr,
+            allow_starttls,
             auth_end: false,
         }
     }
@@ -93,6 +95,11 @@ impl Forward {
                                     return Some(ResponseEncoder::AUTHENTICATION_REQUIRED);
                                 }
                             }
+                            Command::StartTls => {
+                                if !self.allow_starttls {
+                                    return Some(ResponseEncoder::COMMAND_NOT_IMPLEMENTED);
+                                }
+                            }
                             _ => {}
                         };
                         valid_cmd = cmd;

g3proxy/src/inspect/smtp/initiation.rs

@@ -31,12 +31,17 @@ use crate::serve::{ServerTaskError, ServerTaskResult};
 #[derive(Default)]
 pub(super) struct InitializedExtensions {
     odmr: bool,
+    starttls: bool,
 }
 
 impl InitializedExtensions {
     pub(super) fn allow_odmr(&self, config: &SmtpInterceptionConfig) -> bool {
         self.odmr && config.allow_on_demand_mail_relay
     }
+
+    pub(super) fn allow_starttls(&self) -> bool {
+        self.starttls
+    }
 }
 
 pub(super) struct Initiation<'a> {
@@ -219,7 +224,10 @@ impl<'a> Initiation<'a> {
                 // Enhanced-Status-Codes, RFC2034, add status code preface to response
                 "ENHANCEDSTATUSCODES" => false,
                 // STARTTLS, RFC3207, add STARTTLS command
-                "STARTTLS" => true,
+                "STARTTLS" => {
+                    self.server_ext.starttls = true;
+                    true
+                }
                 // No Soliciting, RFC3865, add a MAIL param key
                 "NO-SOLICITING" => true,
                 // Message Tracking, RFC3885, add a MAIL MTRK param key

g3proxy/src/inspect/smtp/mod.rs

@@ -19,12 +19,13 @@ use tokio::io::AsyncWriteExt;
 
 use g3_dpi::ProtocolInspectPolicy;
 use g3_io_ext::OnceBufReader;
-use g3_slog_types::{LtHost, LtUuid};
+use g3_slog_types::{LtHost, LtUpstreamAddr, LtUuid};
 use g3_smtp_proto::response::{ReplyCode, ResponseEncoder};
-use g3_types::net::Host;
+use g3_types::net::{Host, UpstreamAddr};
 
+use super::StartTlsProtocol;
 use crate::config::server::ServerConfig;
-use crate::inspect::{BoxAsyncRead, BoxAsyncWrite, StreamInspectContext};
+use crate::inspect::{BoxAsyncRead, BoxAsyncWrite, StreamInspectContext, StreamInspection};
 use crate::serve::{ServerTaskError, ServerTaskResult};
 
 mod ext;
@@ -48,7 +49,7 @@ macro_rules! intercept_log {
             "intercept_type" => "SMTP",
             "task_id" => LtUuid($obj.ctx.server_task_id()),
             "depth" => $obj.ctx.inspection_depth,
-            "upstream_host" => $obj.upstream_host.as_ref().map(LtHost),
+            "upstream" => LtUpstreamAddr(&$obj.upstream),
             "client_host" => $obj.client_host.as_ref().map(LtHost),
         )
     };
@@ -63,17 +64,20 @@ struct SmtpIo {
 
 pub(crate) struct SmtpInterceptObject<SC: ServerConfig> {
     io: Option<SmtpIo>,
-    pub(crate) ctx: StreamInspectContext<SC>,
-    upstream_host: Option<Host>,
+    ctx: StreamInspectContext<SC>,
+    upstream: UpstreamAddr,
     client_host: Option<Host>,
 }
 
-impl<SC: ServerConfig> SmtpInterceptObject<SC> {
-    pub(crate) fn new(ctx: StreamInspectContext<SC>) -> Self {
+impl<SC> SmtpInterceptObject<SC>
+where
+    SC: ServerConfig + Send + Sync + 'static,
+{
+    pub(crate) fn new(ctx: StreamInspectContext<SC>, upstream: UpstreamAddr) -> Self {
         SmtpInterceptObject {
             io: None,
             ctx,
-            upstream_host: None,
+            upstream,
             client_host: None,
         }
     }
@@ -94,19 +98,26 @@ impl<SC: ServerConfig> SmtpInterceptObject<SC> {
         self.io = Some(io);
     }
 
-    pub(crate) async fn intercept(mut self) -> ServerTaskResult<()> {
+    pub(crate) async fn intercept(mut self) -> ServerTaskResult<Option<StreamInspection<SC>>> {
         match self.ctx.smtp_inspect_policy() {
-            ProtocolInspectPolicy::Bypass => self.do_bypass().await,
-            ProtocolInspectPolicy::Intercept => {
-                if let Err(e) = self.do_intercept().await {
+            ProtocolInspectPolicy::Bypass => {
+                self.do_bypass().await?;
+                Ok(None)
+            }
+            ProtocolInspectPolicy::Intercept => match self.do_intercept().await {
+                Ok(obj) => {
+                    intercept_log!(self, "finished");
+                    Ok(obj)
+                }
+                Err(e) => {
                     intercept_log!(self, "{e}");
                     Err(e)
-                } else {
-                    intercept_log!(self, "finished");
-                    Ok(())
                 }
+            },
+            ProtocolInspectPolicy::Block => {
+                self.do_block().await?;
+                Ok(None)
             }
-            ProtocolInspectPolicy::Block => self.do_block().await,
         }
     }
 
@@ -151,7 +162,7 @@ impl<SC: ServerConfig> SmtpInterceptObject<SC> {
         EndWaitClient::new(local_ip).run_to_end(clt_r, clt_w).await
     }
 
-    async fn do_intercept(&mut self) -> ServerTaskResult<()> {
+    async fn do_intercept(&mut self) -> ServerTaskResult<Option<StreamInspection<SC>>> {
         let SmtpIo {
             clt_r,
             mut clt_w,
@@ -174,13 +185,16 @@ impl<SC: ServerConfig> SmtpInterceptObject<SC> {
             }
         };
         let (code, host) = greeting.into_parts();
-        self.upstream_host = Some(host);
+        self.upstream.set_host(host);
         if code == ReplyCode::NO_SERVICE {
             let timeout = interception_config.quit_wait_timeout;
             tokio::spawn(async move {
                 let _ = EndQuitServer::run_to_end(ups_r, ups_w, timeout).await;
             });
-            return EndWaitClient::new(local_ip).run_to_end(clt_r, clt_w).await;
+            return EndWaitClient::new(local_ip)
+                .run_to_end(clt_r, clt_w)
+                .await
+                .map(|_| None);
         }
 
         self.start_initiation(clt_r, clt_w, ups_r, ups_w).await
@@ -192,7 +206,7 @@ impl<SC: ServerConfig> SmtpInterceptObject<SC> {
         mut clt_w: BoxAsyncWrite,
         mut ups_r: BoxAsyncRead,
         mut ups_w: BoxAsyncWrite,
-    ) -> ServerTaskResult<()> {
+    ) -> ServerTaskResult<Option<StreamInspection<SC>>> {
         let local_ip = self.ctx.task_notes.server_addr.ip();
         let interception_config = self.ctx.smtp_interception();
 
@@ -205,13 +219,34 @@ impl<SC: ServerConfig> SmtpInterceptObject<SC> {
 
         let allow_odmr = server_ext.allow_odmr(interception_config);
 
-        let mut forward = Forward::new(local_ip, allow_odmr);
+        let mut forward = Forward::new(local_ip, allow_odmr, server_ext.allow_starttls());
         let next_action = forward
             .relay(&mut clt_r, &mut clt_w, &mut ups_r, &mut ups_w)
             .await?;
         match next_action {
             ForwardNextAction::StartTls => {
-                // TODO
+                return if let Some(tls_interception) = self.ctx.tls_interception() {
+                    let mut start_tls_obj = crate::inspect::start_tls::StartTlsInterceptObject::new(
+                        self.ctx.clone(),
+                        self.upstream.clone(),
+                        tls_interception,
+                        StartTlsProtocol::Smtp,
+                    );
+                    start_tls_obj.set_io(clt_r, clt_w, ups_r, ups_w);
+                    Ok(Some(StreamInspection::StartTls(start_tls_obj)))
+                } else {
+                    crate::inspect::stream::transit_transparent(
+                        clt_r,
+                        clt_w,
+                        ups_r,
+                        ups_w,
+                        &self.ctx.server_config,
+                        &self.ctx.server_quit_policy,
+                        self.ctx.user(),
+                    )
+                    .await
+                    .map(|_| None)
+                }
             }
             ForwardNextAction::ReverseConnection => {
                 return crate::inspect::stream::transit_transparent(
@@ -223,7 +258,8 @@ impl<SC: ServerConfig> SmtpInterceptObject<SC> {
                     &self.ctx.server_quit_policy,
                     self.ctx.user(),
                 )
-                .await;
+                .await
+                .map(|_| None);
             }
             ForwardNextAction::MailTransport(_param) => {
                 // TODO
@@ -240,5 +276,6 @@ impl<SC: ServerConfig> SmtpInterceptObject<SC> {
             self.ctx.user(),
         )
         .await
+        .map(|_| None)
     }
 }