diff --git a/cspell.json b/cspell.json index 8ae512d8..58ecda7b 100644 --- a/cspell.json +++ b/cspell.json @@ -210,6 +210,7 @@ "extconfig", "fafbfc", "failmode", + "fallbacked", "Fapi", "Fbar", "fde", @@ -690,4 +691,4 @@ "аzureuser", "rооt" ] -} +} \ No newline at end of file diff --git a/proxy_agent/src/proxy_agent_status.rs b/proxy_agent/src/proxy_agent_status.rs index 422d8c50..e0baf8a5 100644 --- a/proxy_agent/src/proxy_agent_status.rs +++ b/proxy_agent/src/proxy_agent_status.rs @@ -34,8 +34,8 @@ use crate::shared_state::agent_status_wrapper::{AgentStatusModule, AgentStatusSh use crate::shared_state::connection_summary_wrapper::ConnectionSummarySharedState; use crate::shared_state::key_keeper_wrapper::KeyKeeperSharedState; use proxy_agent_shared::proxy_agent_aggregate_status::{ - GuestProxyAgentAggregateStatus, ModuleState, OverallState, ProxyAgentDetailStatus, - ProxyAgentStatus, + AuditFallback, GuestProxyAgentAggregateStatus, ModuleState, OverallState, + ProxyAgentDetailStatus, ProxyAgentStatus, }; use proxy_agent_shared::telemetry::{event_logger, Extension, OperationStatus}; use proxy_agent_shared::{current_info, misc_helpers}; @@ -233,6 +233,15 @@ impl ProxyAgentStatusTask { } else { OverallState::SUCCESS }; + let audit_fallback = + if let Ok(Some(reason)) = agent_status_shared_state.get_fallback_reason().await { + Some(AuditFallback { + fallbacked: true, + reason, + }) + } else { + None + }; ProxyAgentStatus { version: current_info::get_current_exe_version(), @@ -249,6 +258,7 @@ impl ProxyAgentStatusTask { telemetryLoggerStatus: agent_status_shared_state .get_module_status(AgentStatusModule::TelemetryLogger) .await, + auditFallback: audit_fallback, proxyConnectionsCount: match agent_status_shared_state.get_connection_count().await { Ok(count) => count, Err(e) => { diff --git a/proxy_agent/src/shared_state/agent_status_wrapper.rs b/proxy_agent/src/shared_state/agent_status_wrapper.rs index 3be23d0e..9f0cf27a 100644 --- a/proxy_agent/src/shared_state/agent_status_wrapper.rs +++ b/proxy_agent/src/shared_state/agent_status_wrapper.rs @@ -11,7 +11,7 @@ use crate::common::result::Result; use proxy_agent_shared::logger::LoggerLevel; use proxy_agent_shared::proxy_agent_aggregate_status::{ModuleState, ProxyAgentDetailStatus}; use proxy_agent_shared::telemetry::event_logger; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::{mpsc, oneshot, watch}; const MAX_STATUS_MESSAGE_LENGTH: usize = 1024; @@ -43,6 +43,20 @@ enum AgentStatusAction { IncreaseTcpConnectionCount { response: oneshot::Sender, }, + SetSoftAuditMode { + enable_soft_audit: bool, + response: oneshot::Sender, + }, + GetSoftAuditMode { + response: oneshot::Sender, + }, + SetFallbackReason { + reason: Option, + response: oneshot::Sender>, + }, + GetFallbackReason { + response: oneshot::Sender>, + }, } #[derive(Clone, Debug)] @@ -55,12 +69,50 @@ pub enum AgentStatusModule { ProxyAgentStatus, } +#[derive(Clone, Debug, PartialEq)] +pub struct RuntimeStatusSnapshot { + pub soft_audit_enabled: bool, + pub proxy_server_state: ModuleState, + pub redirector_state: ModuleState, +} + +impl Default for RuntimeStatusSnapshot { + fn default() -> Self { + Self { + soft_audit_enabled: false, + proxy_server_state: ModuleState::UNKNOWN, + redirector_state: ModuleState::UNKNOWN, + } + } +} + +impl RuntimeStatusSnapshot { + pub fn both_running(&self) -> bool { + self.proxy_server_state == ModuleState::RUNNING + && self.redirector_state == ModuleState::RUNNING + } + + pub fn either_stopped(&self) -> bool { + self.proxy_server_state == ModuleState::STOPPED + || self.redirector_state == ModuleState::STOPPED + } + + pub fn both_at_ultimate_state(&self) -> bool { + self.proxy_server_state.is_ultimate_state() && self.redirector_state.is_ultimate_state() + } +} + #[derive(Clone, Debug)] -pub struct AgentStatusSharedState(mpsc::Sender); +pub struct AgentStatusSharedState( + mpsc::Sender, + watch::Receiver, +); impl AgentStatusSharedState { pub fn start_new() -> Self { let (tx, mut rx) = mpsc::channel(100); + let (runtime_status_tx, runtime_status_rx) = + watch::channel(RuntimeStatusSnapshot::default()); tokio::spawn(async move { let mut key_keeper_state: ModuleState = ModuleState::UNKNOWN; let mut key_keeper_status_message: String = super::UNKNOWN_STATUS_MESSAGE.to_string(); @@ -74,6 +126,8 @@ impl AgentStatusSharedState { let mut proxy_server_status_message = super::UNKNOWN_STATUS_MESSAGE.to_string(); let mut proxy_agent_status_state = ModuleState::UNKNOWN; let mut proxy_agent_status_message = super::UNKNOWN_STATUS_MESSAGE.to_string(); + let mut soft_audit_mode_enabled: bool = false; + let mut fallback_reason: Option = None; // The proxied connection count for the listener let mut tcp_connection_count: u128 = 0; @@ -181,6 +235,19 @@ impl AgentStatusSharedState { proxy_agent_status_state = state.clone(); } } + runtime_status_tx.send_if_modified(|snapshot| { + let updated = RuntimeStatusSnapshot { + soft_audit_enabled: soft_audit_mode_enabled, + proxy_server_state: proxy_server_state.clone(), + redirector_state: redirector_state.clone(), + }; + if *snapshot == updated { + false + } else { + *snapshot = updated; + true + } + }); if let Err(state) = response.send(state) { logger::write_warning(format!("Failed to send response to AgentStatusAction::SetState '{state:?}' for module '{module:?}'")); } @@ -225,11 +292,81 @@ impl AgentStatusSharedState { )); } } + AgentStatusAction::SetSoftAuditMode { + enable_soft_audit, + response, + } => { + if soft_audit_mode_enabled != enable_soft_audit { + logger::write_warning(format!("enableSoftAudit value changed from {soft_audit_mode_enabled} to {enable_soft_audit} ")); + soft_audit_mode_enabled = enable_soft_audit; + runtime_status_tx.send_if_modified(|snapshot| { + let updated = RuntimeStatusSnapshot { + soft_audit_enabled: soft_audit_mode_enabled, + proxy_server_state: proxy_server_state.clone(), + redirector_state: redirector_state.clone(), + }; + if *snapshot == updated { + false + } else { + *snapshot = updated; + true + } + }); + } + if !soft_audit_mode_enabled && fallback_reason.is_some() { + logger::write_warning( + "Soft audit mode has been disabled, reset fallback reason." + .to_string(), + ); + fallback_reason = None; + } + if let Err(value) = response.send(soft_audit_mode_enabled) { + logger::write_warning(format!("Failed to send response to AgentStatusAction::SetSoftAuditMode with value '{value}'")); + } + } + AgentStatusAction::GetSoftAuditMode { response } => { + if let Err(value) = response.send(soft_audit_mode_enabled) { + logger::write_warning(format!("Failed to send response to AgentStatusAction::GetSoftAuditMode with value '{value}'")); + } + } + AgentStatusAction::SetFallbackReason { reason, response } => { + if !soft_audit_mode_enabled { + // telemetry events only and no actual fallback reason should be set. + logger::write_warning( + "SetFallbackReason called while soft audit mode is disabled." + .to_string(), + ); + if let Err(value) = response.send(None) { + logger::write_warning(format!("Failed to send response to AgentStatusAction::SetFallbackReason with value '{value:?}'")); + } + } else { + fallback_reason = reason; + if let Err(value) = response.send(fallback_reason.clone()) { + logger::write_warning(format!("Failed to send response to AgentStatusAction::SetFallbackReason with value '{value:?}'")); + } + } + } + AgentStatusAction::GetFallbackReason { response } => { + if let Err(value) = { + if !soft_audit_mode_enabled { + // if soft audit mode is disabled, always return None as the fallback reason. + response.send(None) + } else { + response.send(fallback_reason.clone()) + } + } { + logger::write_warning(format!("Failed to send response to AgentStatusAction::GetFallbackReason with value '{value:?}'")); + } + } } } }); - AgentStatusSharedState(tx) + AgentStatusSharedState(tx, runtime_status_rx) + } + + pub fn subscribe_runtime_status(&self) -> watch::Receiver { + self.1.clone() } async fn get_module_state(&self, module: AgentStatusModule) -> Result { @@ -436,6 +573,82 @@ impl AgentStatusSharedState { ) }) } + + pub async fn set_soft_audit_mode(&self, enabled: bool) -> Result<()> { + let (response_tx, response_rx) = oneshot::channel(); + self.0 + .send(AgentStatusAction::SetSoftAuditMode { + enable_soft_audit: enabled, + response: response_tx, + }) + .await + .map_err(|e| { + Error::SendError( + "AgentStatusAction::SetSoftAuditMode".to_string(), + e.to_string(), + ) + })?; + response_rx + .await + .map_err(|e| Error::RecvError("AgentStatusAction::SetSoftAuditMode".to_string(), e))?; + Ok(()) + } + + pub async fn get_soft_audit_mode(&self) -> Result { + let (response_tx, response_rx) = oneshot::channel(); + self.0 + .send(AgentStatusAction::GetSoftAuditMode { + response: response_tx, + }) + .await + .map_err(|e| { + Error::SendError( + "AgentStatusAction::GetSoftAuditMode".to_string(), + e.to_string(), + ) + })?; + response_rx + .await + .map_err(|e| Error::RecvError("AgentStatusAction::GetSoftAuditMode".to_string(), e)) + } + + pub async fn get_fallback_reason(&self) -> Result> { + let (response_tx, response_rx) = oneshot::channel(); + self.0 + .send(AgentStatusAction::GetFallbackReason { + response: response_tx, + }) + .await + .map_err(|e| { + Error::SendError( + "AgentStatusAction::GetFallbackReason".to_string(), + e.to_string(), + ) + })?; + response_rx + .await + .map_err(|e| Error::RecvError("AgentStatusAction::GetFallbackReason".to_string(), e)) + } + + pub async fn set_fallback_reason(&self, reason: Option) -> Result<()> { + let (response_tx, response_rx) = oneshot::channel(); + self.0 + .send(AgentStatusAction::SetFallbackReason { + reason, + response: response_tx, + }) + .await + .map_err(|e| { + Error::SendError( + "AgentStatusAction::SetFallbackReason".to_string(), + e.to_string(), + ) + })?; + response_rx + .await + .map_err(|e| Error::RecvError("AgentStatusAction::SetFallbackReason".to_string(), e))?; + Ok(()) + } } #[cfg(test)] @@ -447,6 +660,11 @@ mod tests { #[tokio::test] async fn test_agent_status_shared_state() { let agent_status_shared_state = AgentStatusSharedState::start_new(); + let mut runtime_status_rx = agent_status_shared_state.subscribe_runtime_status(); + assert_eq!( + RuntimeStatusSnapshot::default(), + runtime_status_rx.borrow().clone() + ); let modules = vec![ AgentStatusModule::KeyKeeper, @@ -490,6 +708,14 @@ mod tests { assert_eq!(state, get_state); assert_eq!(status_message, get_status_message); } + assert_eq!( + ModuleState::RUNNING, + runtime_status_rx.borrow().proxy_server_state.clone() + ); + assert_eq!( + ModuleState::RUNNING, + runtime_status_rx.borrow().redirector_state.clone() + ); let tcp_id = agent_status_shared_state .increase_tcp_connection_count() @@ -513,5 +739,90 @@ mod tests { .await .unwrap(); assert_eq!(2, connection_id); + + assert!(!agent_status_shared_state + .get_soft_audit_mode() + .await + .unwrap()); + agent_status_shared_state + .set_soft_audit_mode(true) + .await + .unwrap(); + runtime_status_rx.changed().await.unwrap(); + assert!(runtime_status_rx.borrow().soft_audit_enabled); + assert!(agent_status_shared_state + .get_soft_audit_mode() + .await + .unwrap()); + agent_status_shared_state + .set_soft_audit_mode(false) + .await + .unwrap(); + assert!(!agent_status_shared_state + .get_soft_audit_mode() + .await + .unwrap()); + + assert_eq!( + None, + agent_status_shared_state + .get_fallback_reason() + .await + .unwrap() + ); + agent_status_shared_state + .set_fallback_reason(Some("ignored while disabled".to_string())) + .await + .unwrap(); + assert_eq!( + None, + agent_status_shared_state + .get_fallback_reason() + .await + .unwrap() + ); + + agent_status_shared_state + .set_soft_audit_mode(true) + .await + .unwrap(); + agent_status_shared_state + .set_fallback_reason(Some("test fallback".to_string())) + .await + .unwrap(); + assert_eq!( + Some("test fallback".to_string()), + agent_status_shared_state + .get_fallback_reason() + .await + .unwrap() + ); + agent_status_shared_state + .set_fallback_reason(None) + .await + .unwrap(); + assert_eq!( + None, + agent_status_shared_state + .get_fallback_reason() + .await + .unwrap() + ); + + agent_status_shared_state + .set_fallback_reason(Some("reset on disable".to_string())) + .await + .unwrap(); + agent_status_shared_state + .set_soft_audit_mode(false) + .await + .unwrap(); + assert_eq!( + None, + agent_status_shared_state + .get_fallback_reason() + .await + .unwrap() + ); } } diff --git a/proxy_agent_extension/src/service_main.rs b/proxy_agent_extension/src/service_main.rs index 7b46ba03..75a004b9 100644 --- a/proxy_agent_extension/src/service_main.rs +++ b/proxy_agent_extension/src/service_main.rs @@ -1243,6 +1243,7 @@ mod tests { ebpfProgramStatus: make_detail_status(), proxyListenerStatus: make_detail_status(), telemetryLoggerStatus: make_detail_status(), + auditFallback: None, proxyConnectionsCount: 1, } } diff --git a/proxy_agent_shared/src/proxy_agent_aggregate_status.rs b/proxy_agent_shared/src/proxy_agent_aggregate_status.rs index ce744931..a0a601c8 100644 --- a/proxy_agent_shared/src/proxy_agent_aggregate_status.rs +++ b/proxy_agent_shared/src/proxy_agent_aggregate_status.rs @@ -26,10 +26,17 @@ pub fn get_proxy_agent_aggregate_status_folder() -> std::path::PathBuf { #[derive(Deserialize, Serialize, Debug, PartialEq, Clone)] pub enum ModuleState { UNKNOWN, + STARTING, RUNNING, STOPPED, } +impl ModuleState { + pub fn is_ultimate_state(&self) -> bool { + matches!(self, ModuleState::RUNNING | ModuleState::STOPPED) + } +} + #[derive(Deserialize, Serialize, Debug, PartialEq, Clone)] pub enum OverallState { SUCCESS, @@ -56,6 +63,7 @@ pub struct ProxyAgentStatus { pub ebpfProgramStatus: ProxyAgentDetailStatus, pub proxyListenerStatus: ProxyAgentDetailStatus, pub telemetryLoggerStatus: ProxyAgentDetailStatus, + pub auditFallback: Option, pub proxyConnectionsCount: u128, } @@ -63,6 +71,13 @@ fn default_address_family() -> String { "IPv4".to_string() } +#[derive(Serialize, Deserialize)] +#[allow(non_snake_case)] +pub struct AuditFallback { + pub fallbacked: bool, + pub reason: String, +} + #[derive(Serialize, Deserialize)] #[allow(non_snake_case)] pub struct ProxyConnectionSummary {