fix: return remote exec errors for unsupported types

Author: rvql

AGENTS.md

@@ -44,6 +44,8 @@
 
 ### Rust
 - Format code with rustfmt.
+- While still satisfying rustfmt, prefer code shapes that minimize indentation.
+- Prefer captured formatting arguments such as `format!("{argument}")` over positional forms such as `format!("{}", argument)`.
 - `@agent` uses the `stable` Rust toolchain via `@agent/rust-toolchain`.
 - Prefer placing new Rust code in `@agent/crates`.
 - Dependencies in any `Cargo.toml` must be sorted in alphabet order.

agent/src/rpc/remote_exec.rs

@@ -326,15 +326,8 @@ impl Interior {
                 if message.exec_type.is_none() {
                     continue;
                 }
-                match pb::ExecutionType::try_from(message.exec_type.unwrap()) {
-                    Ok(t) => debug!("received {:?} command from server", t),
-                    Err(_) => {
-                        warn!(
-                            "unsupported remote exec type id {}",
-                            message.exec_type.unwrap()
-                        );
-                        continue;
-                    }
+                if let Ok(t) = pb::ExecutionType::try_from(message.exec_type.unwrap()) {
+                    debug!("received {:?} command from server", t);
                 }
                 if sender.send(message).await.is_err() {
                     debug!("responser channel closed");
@@ -902,30 +895,41 @@ impl Responser {
         match self.msg_recv.poll_recv(ctx) {
             // sender closed, terminate the current stream
             Poll::Ready(None) => ControlFlow::Return(None),
-            Poll::Ready(Some(msg)) => match pb::ExecutionType::try_from(msg.exec_type.unwrap()) {
-                Ok(pb::ExecutionType::ListCommand) => {
-                    let commands = Self::generate_command_list();
-                    debug!("list command returning {} entries", commands.len());
-                    ControlFlow::Return(Some(pb::RemoteExecResponse {
-                        agent_id: Some(self.agent_id.read().deref().into()),
-                        request_id: msg.request_id,
-                        commands,
-                        ..Default::default()
-                    }))
-                }
-                Ok(pb::ExecutionType::ListNamespace) => {
-                    trace!("pending list namespace");
-                    self.pending_lsns = Some((msg.request_id, Box::pin(ls_netns())));
-                    ControlFlow::Continue
-                }
-                Ok(pb::ExecutionType::RunCommand) => self.handle_run_command_message(msg),
-                #[cfg(feature = "enterprise")]
-                Ok(pb::ExecutionType::DryReplayPcap) => self.handle_dry_replay_pcap_message(msg),
-                _ => {
-                    warn!("unsupported execution type: {:?}", msg.exec_type.unwrap());
-                    ControlFlow::Fallthrough
+            Poll::Ready(Some(msg)) => {
+                let raw_exec_type = msg.exec_type.unwrap();
+                match pb::ExecutionType::try_from(raw_exec_type) {
+                    Ok(pb::ExecutionType::ListCommand) => {
+                        let commands = Self::generate_command_list();
+                        debug!("list command returning {} entries", commands.len());
+                        ControlFlow::Return(Some(pb::RemoteExecResponse {
+                            agent_id: Some(self.agent_id.read().deref().into()),
+                            request_id: msg.request_id,
+                            commands,
+                            ..Default::default()
+                        }))
+                    }
+                    Ok(pb::ExecutionType::ListNamespace) => {
+                        trace!("pending list namespace");
+                        self.pending_lsns = Some((msg.request_id, Box::pin(ls_netns())));
+                        ControlFlow::Continue
+                    }
+                    Ok(pb::ExecutionType::RunCommand) => self.handle_run_command_message(msg),
+                    #[cfg(feature = "enterprise")]
+                    Ok(pb::ExecutionType::DryReplayPcap) => {
+                        self.handle_dry_replay_pcap_message(msg)
+                    }
+                    Ok(exec_type) => ControlFlow::Return(self.command_failed_helper(
+                        msg.request_id,
+                        None,
+                        format!("unsupported execution type: {exec_type:?}"),
+                    )),
+                    Err(_) => ControlFlow::Return(self.command_failed_helper(
+                        msg.request_id,
+                        None,
+                        format!("unsupported execution type: {raw_exec_type}"),
+                    )),
                 }
-            },
+            }
             _ => ControlFlow::Fallthrough,
         }
     }
@@ -1418,3 +1422,79 @@ async fn kubectl_log(namespace: String, pod: String, previous: bool) -> Result<O
         stderr: vec![],
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use std::{
+        net::{IpAddr, Ipv4Addr},
+        sync::Arc,
+    };
+
+    use arc_swap::{access::Map, ArcSwap};
+    use futures::StreamExt;
+    use parking_lot::RwLock;
+
+    use super::*;
+    use crate::config::ModuleConfig;
+
+    fn test_config() -> Config {
+        let current_config = Arc::new(ArcSwap::from_pointee(ModuleConfig::default()));
+        Config {
+            flow: Map::new(current_config.clone(), |config| &config.flow),
+            log_parser: Map::new(current_config, |config| &config.log_parser),
+        }
+    }
+
+    fn test_agent_id() -> Arc<RwLock<AgentId>> {
+        Arc::new(RwLock::new(AgentId {
+            ipmac: (IpAddr::V4(Ipv4Addr::LOCALHOST), Default::default()).into(),
+            ..Default::default()
+        }))
+    }
+
+    async fn next_response_for(exec_type: i32) -> pb::RemoteExecResponse {
+        let (sender, receiver) = mpsc::channel(1);
+        let mut responser = Responser::new(test_agent_id(), receiver, test_config());
+        sender
+            .send(pb::RemoteExecRequest {
+                request_id: Some(42),
+                exec_type: Some(exec_type),
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        responser.next().await.unwrap()
+    }
+
+    #[tokio::test]
+    async fn unsupported_exec_type_returns_error_response() {
+        let response = next_response_for(99).await;
+
+        assert_eq!(response.request_id, Some(42));
+        assert_eq!(
+            response.errmsg.as_deref(),
+            Some("unsupported execution type: 99")
+        );
+        assert!(response.agent_id.is_some());
+        assert!(response.command_result.is_some());
+        assert!(response.commands.is_empty());
+        assert!(response.linux_namespaces.is_empty());
+    }
+
+    #[cfg(not(feature = "enterprise"))]
+    #[tokio::test]
+    async fn dry_replay_pcap_returns_error_without_enterprise_feature() {
+        let response = next_response_for(pb::ExecutionType::DryReplayPcap as i32).await;
+
+        assert_eq!(response.request_id, Some(42));
+        assert_eq!(
+            response.errmsg.as_deref(),
+            Some("unsupported execution type: DryReplayPcap")
+        );
+        assert!(response.agent_id.is_some());
+        assert!(response.command_result.is_some());
+        assert!(response.commands.is_empty());
+        assert!(response.linux_namespaces.is_empty());
+    }
+}