From a092e09bcdf25f299e82ec25f4dd553d41928199 Mon Sep 17 00:00:00 2001 From: Zhidong Peng Date: Thu, 20 Aug 2026 21:25:23 +0000 Subject: [PATCH 1/6] localIPBindMonitorOnly --- ebpf/redirect.bpf.c | 58 +++++-- linux-ebpf/ebpf_cgroup.c | 65 ++++++-- proxy_agent/config/GuestProxyAgent.linux.json | 1 + .../config/GuestProxyAgent.windows.json | 1 + proxy_agent/src/common/config.rs | 18 ++- proxy_agent/src/redirector.rs | 141 +++++++++++++----- proxy_agent/src/redirector/linux.rs | 76 ++++++++++ proxy_agent/src/redirector/shared_ebpf.rs | 11 +- proxy_agent/src/redirector/windows/bpf_api.rs | 13 ++ .../src/redirector/windows/bpf_prog.rs | 74 +++++++++ shared-ebpf/include/gpa_audit_event.h | 14 +- 11 files changed, 399 insertions(+), 73 deletions(-) diff --git a/ebpf/redirect.bpf.c b/ebpf/redirect.bpf.c index 1acee7c7..81a151e7 100644 --- a/ebpf/redirect.bpf.c +++ b/ebpf/redirect.bpf.c @@ -12,6 +12,13 @@ struct bpf_map_def policy_map = { .value_size = sizeof(destination_entry_t), .max_entries = 10}; +#pragma clang section data = "maps" +struct bpf_map_def config_map = { + .type = BPF_MAP_TYPE_HASH, + .key_size = sizeof(uint32_t), + .value_size = sizeof(struct gpa_config_entry), + .max_entries = 1}; + #pragma clang section data = "maps" struct bpf_map_def skip_process_map = { .type = BPF_MAP_TYPE_HASH, @@ -26,6 +33,13 @@ struct bpf_map_def audit_map = { .value_size = sizeof(sock_addr_audit_entry_t), .max_entries = 1000}; +#pragma clang section data = "maps" +struct bpf_map_def audit_only_map = { + .type = BPF_MAP_TYPE_LRU_HASH, + .key_size = sizeof(sock_addr_audit_key_t), + .value_size = sizeof(sock_addr_audit_entry_t), + .max_entries = 1000}; + /* check the current pid in the skip_process map. return 1 if found, otherwise return 0. @@ -41,13 +55,21 @@ check_skip_process_map_entry(uint32_t pid) return (skip_entry != NULL) ? 1 : 0; } +inline __attribute__((always_inline)) int +local_ip_bind_monitor_only_enabled(void) +{ + uint32_t key = GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY; + struct gpa_config_entry *entry = bpf_map_lookup_elem(&config_map, &key); + return entry != NULL && entry->enabled != 0; +} + /* update audit map entry if not skip redirecting. return 0 if the entry is updated, otherwise return 1 if pid found in the skip_process_map. */ inline __attribute__((always_inline)) int -update_audit_map_entry(bpf_sock_addr_t *ctx) +update_audit_map_entry(bpf_sock_addr_t *ctx, int audit_only) { uint64_t pid_tip = bpf_get_current_pid_tgid(); uint32_t pid = (uint32_t)(pid_tip >> 32); @@ -77,6 +99,19 @@ update_audit_map_entry(bpf_sock_addr_t *ctx) entry.destination_ipv4 = ctx->user_ip4; // we only support ipv4 so far. entry.destination_port = ctx->user_port; uint16_t source_port = ctx->msg_src_port; + if (audit_only) + { + sock_addr_audit_key_t key = {0}; + key.protocol = ctx->protocol; + key.source_port = source_port != 0 ? source_port : pid; + uint64_t ret = bpf_map_update_elem(&audit_only_map, &key, &entry, 0); + if (ret != 0) + { + bpf_printk("Failed to update audit-only map with results: %u.", ret); + } + return 0; + } + if (source_port == 0) { int32_t result = bpf_sock_addr_set_redirect_context(ctx, &entry, sizeof(sock_addr_audit_entry_t)); @@ -121,23 +156,22 @@ authorize_v4(bpf_sock_addr_t *ctx) { bpf_printk("Found v4 proxy entry value: %u, %u", policy->destination_ip.ipv4, policy->destination_port); + uint32_t source_ip = ctx->msg_src_ip4; + int audit_only = local_ip_bind_monitor_only_enabled() && // check the config map for localIPBindMonitorOnly + source_ip != 0 && (source_ip & 0xff) != 0x7f; // check if the source ip is set and not loopback + // update to the audit map before changing the destination ip and port. - if (update_audit_map_entry(ctx) == 1) + if (update_audit_map_entry(ctx, audit_only) == 1) { bpf_printk("Found skip process entry, skip the redirection."); return BPF_SOCK_ADDR_VERDICT_PROCEED_SOFT; } - // if (ctx->msg_src_ip4 == 0) - // { - // bpf_printk("Local/source ip is not set, redirect to loopback ip."); - // ctx->user_ip4 = policy->destination_ip.ipv4; - // } - // else - // { - // ctx->user_ip4 = ctx->msg_src_ip4; - // bpf_printk("Local/source ip is set, redirect to source ip:%u.", ctx->user_ip4); - // } + if (audit_only) + { + bpf_printk("Source address is explicitly bound, audit without redirecting."); + return BPF_SOCK_ADDR_VERDICT_PROCEED_SOFT; + } bpf_printk("redirecting to destination loopback ip."); ctx->user_ip4 = policy->destination_ip.ipv4; diff --git a/linux-ebpf/ebpf_cgroup.c b/linux-ebpf/ebpf_cgroup.c index a58f82f7..98136bbc 100644 --- a/linux-ebpf/ebpf_cgroup.c +++ b/linux-ebpf/ebpf_cgroup.c @@ -9,6 +9,7 @@ #include #include #include +#include #include "socket.h" @@ -27,6 +28,13 @@ struct { __uint(max_entries, 10); } policy_map SEC(".maps"); +struct { + __uint(type, BPF_MAP_TYPE_HASH); + __type(key, __u32); + __type(value, struct gpa_config_entry); + __uint(max_entries, 1); +} config_map SEC(".maps"); + struct { __uint(type, BPF_MAP_TYPE_LRU_HASH); __type(key, struct gpa_audit_key); // source port and protocol @@ -34,6 +42,13 @@ struct { __uint(max_entries, 200); // LRU evicts oldest on overflow } audit_map SEC(".maps"); +struct { + __uint(type, BPF_MAP_TYPE_LRU_HASH); + __type(key, struct gpa_audit_key); + __type(value, struct gpa_audit_event); + __uint(max_entries, 200); +} audit_only_map SEC(".maps"); + struct { __uint(type, BPF_MAP_TYPE_LRU_HASH); __type(key, __u64); // pid-tgid or socket cookie @@ -57,13 +72,21 @@ check_skip_process_map_entry(__u32 pid) return (skip_entry != NULL) ? 1 : 0; } +static __always_inline int +local_ip_bind_monitor_only_enabled(void) +{ + __u32 key = GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY; + struct gpa_config_entry *entry = bpf_map_lookup_elem(&config_map, &key); + return entry != NULL && entry->enabled != 0; +} + /* update audit map entry if not skip redirecting. return 0 if the entry is updated, otherwise return 1 if pid found in the skip_process_map. */ static __always_inline int -update_local_map_entry(struct bpf_sock_addr *ctx) +update_local_map_entry(struct bpf_sock_addr *ctx, __u32 audit_only) { __u64 pid_tip = bpf_get_current_pid_tgid(); __u32 pid = (__u32)(pid_tip >> 32); @@ -81,6 +104,7 @@ update_local_map_entry(struct bpf_sock_addr *ctx) entry.destination_ipv4 = ctx->user_ip4; // we only support ipv4 so far. entry.destination_port = ctx->user_port; entry.protocol = ctx->protocol; + entry.audit_only = audit_only; __u64 ret = bpf_map_update_elem(&local_map, &pid_tip, &entry, 0); if (ret != 0) @@ -109,27 +133,30 @@ authorize_v4(struct bpf_sock_addr *ctx) { bpf_printk("authorize_v4: Found v4 proxy entry value: %u, %u", policy->destination_ip.ipv4, policy->destination_port); + // At connect4, msg_src_ip4 is not valid; it is only populated for + // UDP sendmsg hooks. A concrete address set by bind(2) is available + // from the socket before TCP performs automatic source selection. + __u32 source_ip = ctx->sk != NULL ? ctx->sk->src_ip4 : 0; + __u32 source_ip_host = bpf_ntohl(source_ip); + __u32 audit_only = local_ip_bind_monitor_only_enabled() && + source_ip != 0 && + (source_ip_host & 0xff000000) != 0x7f000000; + // update to the audit map before changing the destination ip and port. - if (update_local_map_entry(ctx) == 1) + if (update_local_map_entry(ctx, audit_only) == 1) { bpf_printk("authorize_v4: Found skip process entry, skip the redirection."); return BPF_SOCK_ADDR_VERDICT_PROCEED; } - // TODO: check if the local ip is set. - // __u32 local_ip; - // __u64 read = bpf_probe_read_kernel(&local_ip, sizeof(__u32), &ctx->msg_src_ip4); - // if (read == 0 && local_ip != 0) - // { - // // read the local ip from the msg_src_ip4 successfully and ip is set. - // ctx->user_ip4 = local_ip; - // bpf_printk("authorize_v4: Local/source ip is set, redirect to source ip:%u.", local_ip); - // } - // else + if (audit_only) { - ctx->user_ip4 = policy->destination_ip.ipv4; - bpf_printk("authorize_v4: Local/source ip is not set, redirect to loopback ip."); + bpf_printk("authorize_v4: Source address is explicitly bound, audit without redirecting."); + return BPF_SOCK_ADDR_VERDICT_PROCEED; } + + ctx->user_ip4 = policy->destination_ip.ipv4; + bpf_printk("authorize_v4: Local/source ip is not set, redirect to loopback ip."); ctx->user_port = policy->destination_port; } @@ -157,7 +184,15 @@ update_audit_map_entry_sk(__u32 local_port, struct gpa_sock_addr_local_entry *lo entry.destination_ipv4 = local_entry->destination_ipv4; entry.destination_port = local_entry->destination_port; - __u64 ret = bpf_map_update_elem(&audit_map, &key, &entry, 0); + __u64 ret; + if (local_entry->audit_only) + { + ret = bpf_map_update_elem(&audit_only_map, &key, &entry, 0); + } + else + { + ret = bpf_map_update_elem(&audit_map, &key, &entry, 0); + } if (ret != 0) { bpf_printk("update_audit_map_entry_sk: Failed to update audit map entry with results:%u.", ret); diff --git a/proxy_agent/config/GuestProxyAgent.linux.json b/proxy_agent/config/GuestProxyAgent.linux.json index 90b6d0b5..a4693012 100644 --- a/proxy_agent/config/GuestProxyAgent.linux.json +++ b/proxy_agent/config/GuestProxyAgent.linux.json @@ -10,5 +10,6 @@ "fileLogLevel": "Trace", "fileLogLevelForEvents": "Info", "fileLogLevelForSystemEvents": "Info", + "localIPBindMonitorOnly": true, "canonicalRequestMode": "Shadow" } \ No newline at end of file diff --git a/proxy_agent/config/GuestProxyAgent.windows.json b/proxy_agent/config/GuestProxyAgent.windows.json index b43e5768..ba8960e8 100644 --- a/proxy_agent/config/GuestProxyAgent.windows.json +++ b/proxy_agent/config/GuestProxyAgent.windows.json @@ -9,5 +9,6 @@ "fileLogLevel": "Trace", "fileLogLevelForEvents": "Info", "fileLogLevelForSystemEvents": "Info", + "localIPBindMonitorOnly": true, "canonicalRequestMode": "Shadow" } \ No newline at end of file diff --git a/proxy_agent/src/common/config.rs b/proxy_agent/src/common/config.rs index 4365f148..2f1e2efa 100644 --- a/proxy_agent/src/common/config.rs +++ b/proxy_agent/src/common/config.rs @@ -79,6 +79,10 @@ pub fn get_enable_http_proxy_trace() -> bool { SYSTEM_CONFIG.enableHttpProxyTrace.unwrap_or(false) } +pub fn get_local_ip_bind_monitor_only() -> bool { + SYSTEM_CONFIG.get_local_ip_bind_monitor_only() +} + /// Rollout flag for the Innovation 2.1 canonical request pipeline. /// /// Read from the optional `canonicalRequestMode` key in the GPA config @@ -115,6 +119,8 @@ pub struct Config { /// This is an optional config, mainly for manual debugging purpose #[serde(skip_serializing_if = "Option::is_none")] enableHttpProxyTrace: Option, + #[serde(skip_serializing_if = "Option::is_none")] + localIPBindMonitorOnly: Option, /// Innovation 2.1 canonical request rollout flag. /// Optional; absent or unparseable values resolve to /// [`crate::proxy::canonical::CanonicalMode::Off`] so production @@ -230,6 +236,10 @@ impl Config { None } + pub fn get_local_ip_bind_monitor_only(&self) -> bool { + self.localIPBindMonitorOnly.unwrap_or(false) + } + /// Resolve the canonical-request rollout flag. /// /// Returns [`crate::proxy::canonical::CanonicalMode::Off`] when the @@ -278,7 +288,7 @@ mod tests { Err(err) => panic!("Failed to create folder: {}", err), } let config_file_path = temp_test_path.join("test_config.json"); - let config = create_config_file(config_file_path); + let mut config = create_config_file(config_file_path); assert_eq!( r#"C:\logFolderName"#, @@ -331,6 +341,10 @@ mod tests { ); } + assert!(config.get_local_ip_bind_monitor_only()); + config.localIPBindMonitorOnly = None; + assert!(!config.get_local_ip_bind_monitor_only()); + assert_eq!( proxy_agent_shared::logger::LoggerLevel::Info, config.get_file_log_level_for_events().unwrap(), @@ -364,6 +378,7 @@ mod tests { "hostGAPluginSupport": 1, "imdsSupport": 1, "ebpfProgramName": "ebpfProgramName", + "localIPBindMonitorOnly": true, "fileLogLevelForEvents": "Info", "fileLogLevelForSystemEvents": "Info" }"# @@ -378,6 +393,7 @@ mod tests { "hostGAPluginSupport": 1, "imdsSupport": 1, "ebpfProgramName": "ebpfProgramName", + "localIPBindMonitorOnly": true, "fileLogLevelForEvents": "Info", "fileLogLevelForSystemEvents": "Info" }"# diff --git a/proxy_agent/src/redirector.rs b/proxy_agent/src/redirector.rs index e94f1ee1..e0c1f86c 100644 --- a/proxy_agent/src/redirector.rs +++ b/proxy_agent/src/redirector.rs @@ -53,15 +53,12 @@ use crate::common::helpers; use crate::common::result::Result; use crate::common::{config, logger}; use crate::provision; -use crate::shared_state::access_control_wrapper::AccessControlSharedState; +use crate::proxy::Claims; use crate::shared_state::agent_status_wrapper::{AgentStatusModule, AgentStatusSharedState}; -use crate::shared_state::connection_summary_wrapper::ConnectionSummarySharedState; -use crate::shared_state::key_keeper_wrapper::KeyKeeperSharedState; -use crate::shared_state::provision_wrapper::ProvisionSharedState; +use crate::shared_state::proxy_server_wrapper::ProxyServerSharedState; use crate::shared_state::redirector_wrapper::RedirectorSharedState; use crate::shared_state::EventThreadsSharedState; use crate::shared_state::SharedState; -use proxy_agent_shared::common_state::CommonState; use proxy_agent_shared::logger::LoggerLevel; use proxy_agent_shared::misc_helpers; use proxy_agent_shared::proxy_agent_aggregate_status::ModuleState; @@ -109,28 +106,14 @@ impl AuditEntry { pub struct Redirector { local_port: u16, - redirector_shared_state: RedirectorSharedState, - key_keeper_shared_state: KeyKeeperSharedState, - agent_status_shared_state: AgentStatusSharedState, - cancellation_token: CancellationToken, - common_state: CommonState, - provision_shared_state: ProvisionSharedState, - access_control_shared_state: AccessControlSharedState, - connection_summary_shared_state: ConnectionSummarySharedState, + shared_state: SharedState, } impl Redirector { pub fn new(local_port: u16, shared_state: &SharedState) -> Self { Redirector { local_port, - cancellation_token: shared_state.get_cancellation_token(), - key_keeper_shared_state: shared_state.get_key_keeper_shared_state(), - common_state: shared_state.get_common_state(), - provision_shared_state: shared_state.get_provision_shared_state(), - agent_status_shared_state: shared_state.get_agent_status_shared_state(), - redirector_shared_state: shared_state.get_redirector_shared_state(), - access_control_shared_state: shared_state.get_access_control_shared_state(), - connection_summary_shared_state: shared_state.get_connection_summary_shared_state(), + shared_state: shared_state.clone(), } } @@ -140,7 +123,8 @@ impl Redirector { pub async fn start(&self) { let message = "eBPF redirector is starting"; if let Err(e) = self - .agent_status_shared_state + .shared_state + .get_agent_status_shared_state() .set_module_status_message(message.to_string(), AgentStatusModule::Redirector) .await { @@ -193,6 +177,13 @@ impl Redirector { logger::write_information(format!( "Success updated bpf skip_process map with pid={pid}." )); + let monitor_only = config::get_local_ip_bind_monitor_only(); + if monitor_only { + bpf_object.update_local_ip_bind_monitor_only(true)?; + } + logger::write_information(format!( + "Updated eBPF localIPBindMonitorOnly={monitor_only}." + )); // Do not update redirect policy map here, it will be updated by provision module // When provision is finished, it will call update_xxx_redirect_policy functions to update the redirect policy maps. @@ -202,14 +193,23 @@ impl Redirector { logger::write_information("Success attached bpf prog.".to_string()); if let Err(e) = self - .redirector_shared_state + .shared_state + .get_redirector_shared_state() .update_bpf_object(Arc::new(Mutex::new(bpf_object))) .await { logger::write_error(format!("Failed to update bpf object in shared state: {e}")); } + if monitor_only { + tokio::spawn(poll_audit_only( + self.shared_state.get_redirector_shared_state(), + self.shared_state.get_proxy_server_shared_state(), + self.shared_state.get_cancellation_token(), + )); + } if let Err(e) = self - .redirector_shared_state + .shared_state + .get_redirector_shared_state() .set_local_port(self.local_port) .await { @@ -222,7 +222,8 @@ impl Redirector { logger::AGENT_LOGGER_KEY, ); if let Err(e) = self - .agent_status_shared_state + .shared_state + .get_agent_status_shared_state() .set_module_status_message(message.to_string(), AgentStatusModule::Redirector) .await { @@ -231,7 +232,8 @@ impl Redirector { )); } if let Err(e) = self - .agent_status_shared_state + .shared_state + .get_agent_status_shared_state() .set_module_state(ModuleState::RUNNING, AgentStatusModule::Redirector) .await { @@ -239,23 +241,14 @@ impl Redirector { } // report redirector ready for provision - provision::redirector_ready(EventThreadsSharedState { - cancellation_token: self.cancellation_token.clone(), - common_state: self.common_state.clone(), - access_control_shared_state: self.access_control_shared_state.clone(), - redirector_shared_state: self.redirector_shared_state.clone(), - key_keeper_shared_state: self.key_keeper_shared_state.clone(), - provision_shared_state: self.provision_shared_state.clone(), - agent_status_shared_state: self.agent_status_shared_state.clone(), - connection_summary_shared_state: self.connection_summary_shared_state.clone(), - }) - .await; + provision::redirector_ready(EventThreadsSharedState::new(&self.shared_state)).await; Ok(()) } async fn get_status_message(&self) -> String { - self.agent_status_shared_state + self.shared_state + .get_agent_status_shared_state() .get_module_status(AgentStatusModule::Redirector) .await .message @@ -263,7 +256,8 @@ impl Redirector { async fn set_error_status(&self, message: String) { if let Err(e) = self - .agent_status_shared_state + .shared_state + .get_agent_status_shared_state() .set_module_status_message(message.to_string(), AgentStatusModule::Redirector) .await { @@ -274,6 +268,75 @@ impl Redirector { } } +async fn poll_audit_only( + redirector_shared_state: RedirectorSharedState, + proxy_server_shared_state: ProxyServerSharedState, + cancellation_token: CancellationToken, +) { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(1)); + loop { + tokio::select! { + _ = cancellation_token.cancelled() => return, + _ = interval.tick() => { + let Some(bpf_object) = redirector_shared_state + .get_bpf_object() + .await + .ok() + .flatten() + else { + continue; + }; + let records = bpf_object.lock().unwrap().drain_audit_only(); + match records { + Ok(records) => { + for entry in records { + let destination_ip = entry.destination_ipv4_addr(); + let destination_port = entry.destination_port_in_host_byte_order(); + let message = match Claims::from_audit_entry( + &entry, + std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), // not used for audit-only, so just use unspecified + 0, // not used for audit-only, so just use 0 + proxy_server_shared_state.clone(), + ) + .await + { + Ok(claims) => format!( + "eBPF audit-only connection: userName={}, processId={}, processName={}, processFullPath={}, processCmdLine={}, runAsElevated={}, destination={}:{}", + claims.userName, + claims.processId, + claims.processName.to_string_lossy(), + claims.processFullPath.display(), + claims.processCmdLine, + claims.runAsElevated, + destination_ip, + destination_port, + ), + Err(err) => format!( + "eBPF audit-only connection: userId={}, processId={}, processDetails=unavailable ({err}), destination={}:{}", + entry.logon_id, + entry.process_id, + destination_ip, + destination_port, + ), + }; + event_logger::write_event( + LoggerLevel::Warn, + message, + "poll_audit_only", + "redirector", + logger::AGENT_LOGGER_KEY, + ); + } + } + Err(err) => logger::write_warning(format!( + "Failed to drain eBPF audit-only map: {err}" + )), + } + } + } + } +} + #[cfg(windows)] pub fn get_audit_from_stream_socket(raw_socket_id: usize) -> Result { windows::get_audit_from_redirect_context(raw_socket_id) diff --git a/proxy_agent/src/redirector/linux.rs b/proxy_agent/src/redirector/linux.rs index 36571e4e..2a2e8ab0 100644 --- a/proxy_agent/src/redirector/linux.rs +++ b/proxy_agent/src/redirector/linux.rs @@ -8,6 +8,7 @@ use crate::common::{ }; use crate::redirector::shared_ebpf::linux_types::{ destination_entry, sock_addr_audit_entry, sock_addr_audit_key, sock_addr_skip_process_entry, + GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY, }; use crate::redirector::{ip_to_string, AuditEntry}; use crate::shared_state::redirector_wrapper::RedirectorSharedState; @@ -91,6 +92,40 @@ impl BpfObject { Ok(()) } + pub fn update_local_ip_bind_monitor_only(&mut self, enabled: bool) -> Result<()> { + let config_map_name = "config_map"; + match self.0.map_mut(config_map_name) { + Some(map) => match HashMap::<&mut MapData, u32, [u32; 1]>::try_from(map) { + Ok(mut config_map) => config_map + .insert( + GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY, + [u32::from(enabled)], + 0, + ) + .map_err(|err| { + Error::Bpf(BpfErrorType::UpdateBpfMapHashMap( + config_map_name.to_string(), + "localIPBindMonitorOnly".to_string(), + err.to_string(), + )) + })?, + Err(err) => { + return Err(Error::Bpf(BpfErrorType::LoadBpfMapHashMap( + config_map_name.to_string(), + err.to_string(), + ))); + } + }, + None => { + return Err(Error::Bpf(BpfErrorType::GetBpfMap( + config_map_name.to_string(), + "Map does not exist".to_string(), + ))); + } + } + Ok(()) + } + pub fn update_policy_elem_bpf_map( &mut self, endpoint_name: &str, @@ -381,6 +416,47 @@ impl BpfObject { } Ok(()) } + + pub fn drain_audit_only(&mut self) -> Result> { + let audit_map_name = "audit_only_map"; + match self.0.map_mut(audit_map_name) { + Some(map) => { + let mut audit_map = HashMap::<&mut MapData, [u32; 2], [u32; 5]>::try_from(map) + .map_err(|err| { + Error::Bpf(BpfErrorType::LoadBpfMapHashMap( + audit_map_name.to_string(), + err.to_string(), + )) + })?; + let mut records = Vec::new(); + for item in audit_map.iter() { + let (key, value) = item.map_err(|err| { + Error::Bpf(BpfErrorType::MapLookupElem( + audit_map_name.to_string(), + err.to_string(), + )) + })?; + records.push(( + key, + sock_addr_audit_entry::from_array(value).to_audit_entry(), + )); + } + for (key, _) in &records { + audit_map.remove(key).map_err(|err| { + Error::Bpf(BpfErrorType::MapDeleteElem( + audit_map_name.to_string(), + err.to_string(), + )) + })?; + } + Ok(records.into_iter().map(|(_, entry)| entry).collect()) + } + None => Err(Error::Bpf(BpfErrorType::GetBpfMap( + audit_map_name.to_string(), + "Map does not exist".to_string(), + ))), + } + } } // Redirector implementation for Linux platform diff --git a/proxy_agent/src/redirector/shared_ebpf.rs b/proxy_agent/src/redirector/shared_ebpf.rs index 15bee538..63125dbc 100644 --- a/proxy_agent/src/redirector/shared_ebpf.rs +++ b/proxy_agent/src/redirector/shared_ebpf.rs @@ -68,6 +68,7 @@ pub type destination_entry = _destination_entry; pub const IPPROTO_TCP: u32 = 6; #[allow(dead_code)] pub const IPPROTO_UDP: u32 = 17; +pub const GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY: u32 = 0; #[repr(C)] pub struct sock_addr_skip_process_entry { @@ -90,7 +91,7 @@ impl sock_addr_skip_process_entry { } #[repr(C)] -#[derive(Debug)] +#[derive(Clone, Copy, Debug)] pub struct sock_addr_audit_key { pub protocol: u32, pub source_port: u32, @@ -361,15 +362,17 @@ impl AuditValueEntry { #[cfg(not(windows))] pub mod linux_types { pub use super::{ - destination_entry, sock_addr_audit_entry, sock_addr_audit_key, sock_addr_skip_process_entry, + destination_entry, sock_addr_audit_entry, sock_addr_audit_key, + sock_addr_skip_process_entry, GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY, }; } #[cfg(windows)] pub mod windows_types { pub use super::{ - destination_entry as destination_entry_t, sock_addr_audit_key as sock_addr_audit_key_t, - sock_addr_skip_process_entry, + destination_entry as destination_entry_t, sock_addr_audit_entry, + sock_addr_audit_key as sock_addr_audit_key_t, sock_addr_skip_process_entry, + GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY, }; } diff --git a/proxy_agent/src/redirector/windows/bpf_api.rs b/proxy_agent/src/redirector/windows/bpf_api.rs index 55b52bcb..04578c59 100644 --- a/proxy_agent/src/redirector/windows/bpf_api.rs +++ b/proxy_agent/src/redirector/windows/bpf_api.rs @@ -184,6 +184,8 @@ type BpfMapLookupElem = unsafe extern "C" fn(map_fd: c_int, key: *const c_void, value: *mut c_void) -> c_int; type BpfMapDeleteElem = unsafe extern "C" fn(map_fd: c_int, key: *const c_void) -> c_int; +type BpfMapGetNextKey = + unsafe extern "C" fn(map_fd: c_int, key: *const c_void, next_key: *mut c_void) -> c_int; type LibBpfGetError = unsafe extern "C" fn(no_use_ptr: *const c_void) -> c_long; @@ -304,3 +306,14 @@ pub fn bpf_map_delete_elem(map_fd: c_int, key: *const c_void) -> Result { get_ebpf_api_fun(ebpf_api, "bpf_map_delete_elem\0")?; Ok(unsafe { map_delete_elem(map_fd, key) }) } + +pub fn bpf_map_get_next_key( + map_fd: c_int, + key: *const c_void, + next_key: *mut c_void, +) -> Result { + let ebpf_api = get_ebpf_api()?; + let map_get_next_key: Symbol = + get_ebpf_api_fun(ebpf_api, "bpf_map_get_next_key\0")?; + Ok(unsafe { map_get_next_key(map_fd, key, next_key) }) +} diff --git a/proxy_agent/src/redirector/windows/bpf_prog.rs b/proxy_agent/src/redirector/windows/bpf_prog.rs index 3a0fb5e9..11a8b732 100644 --- a/proxy_agent/src/redirector/windows/bpf_prog.rs +++ b/proxy_agent/src/redirector/windows/bpf_prog.rs @@ -9,6 +9,9 @@ use crate::common::{ error::{BpfErrorType, Error}, result::Result, }; +use crate::redirector::shared_ebpf::windows_types::{ + sock_addr_audit_entry, GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY, +}; use crate::redirector::AuditEntry; use proxy_agent_shared::misc_helpers; use std::ffi::c_void; @@ -334,6 +337,77 @@ impl BpfObject { Ok(()) } + pub fn update_local_ip_bind_monitor_only(&self, enabled: bool) -> Result<()> { + let map_name = "config_map"; + let map_fd = self.get_bpf_map_fd(map_name)?; + let key = GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY; + let value = [u32::from(enabled)]; + + let result = bpf_map_update_elem( + map_fd, + &key as *const u32 as *const c_void, + value.as_ptr() as *const c_void, + 0, + ) + .map_err(|e| { + Error::Bpf(BpfErrorType::UpdateBpfMapHashMap( + map_name.to_string(), + "localIPBindMonitorOnly".to_string(), + e.to_string(), + )) + })?; + if result != 0 { + return Err(Error::Bpf(BpfErrorType::UpdateBpfMapHashMap( + map_name.to_string(), + "localIPBindMonitorOnly".to_string(), + format!("bpf_map_update_elem returned error code {result}"), + ))); + } + Ok(()) + } + + pub fn drain_audit_only(&self) -> Result> { + let map_name = "audit_only_map"; + let map_fd = self.get_bpf_map_fd(map_name)?; + let mut keys = Vec::new(); + let mut previous_key: Option = None; + + loop { + let mut next_key = sock_addr_audit_key_t::from_array([0; 2]); + let previous_key_ptr = previous_key + .as_ref() + .map_or(std::ptr::null(), |key| key as *const _ as *const c_void); + let result = bpf_map_get_next_key( + map_fd, + previous_key_ptr, + &mut next_key as *mut sock_addr_audit_key_t as *mut c_void, + )?; + if result != 0 { + break; + } + previous_key = Some(next_key); + keys.push(next_key); + } + + let mut records = Vec::with_capacity(keys.len()); + for key in keys { + let mut value = sock_addr_audit_entry::empty(); + let result = bpf_map_lookup_elem( + map_fd, + &key as *const sock_addr_audit_key_t as *const c_void, + &mut value as *mut sock_addr_audit_entry as *mut c_void, + )?; + if result == 0 { + records.push(value.to_audit_entry()); + let _ = bpf_map_delete_elem( + map_fd, + &key as *const sock_addr_audit_key_t as *const c_void, + )?; + } + } + Ok(records) + } + /** Routine Description: This routine delete element from policy_map. diff --git a/shared-ebpf/include/gpa_audit_event.h b/shared-ebpf/include/gpa_audit_event.h index 667cd36f..816450a7 100644 --- a/shared-ebpf/include/gpa_audit_event.h +++ b/shared-ebpf/include/gpa_audit_event.h @@ -12,6 +12,8 @@ #pragma once +#define GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY 0 + // IP address - union allows IPv4 (first element) or IPv6 (all 4 elements) // Size: 16 bytes (4 x u32) - matches Rust _ip_address { ip: [u32; 4] } struct gpa_ip_address @@ -63,8 +65,14 @@ struct gpa_skip_process_entry __u32 pid; }; +// Runtime configuration passed from GPA user mode to the eBPF program. +struct gpa_config_entry +{ + __u32 enabled; +}; + // Local address entry - tracks current connection state in the local_map -// Size: 24 bytes (6 x u32) +// Size: 28 bytes (7 x u32) struct gpa_sock_addr_local_entry { __u32 logon_id; // uid @@ -73,6 +81,7 @@ struct gpa_sock_addr_local_entry __u32 destination_ipv4; __u32 destination_port; __u32 protocol; + __u32 audit_only; }; // Compile-time layout assertions to guarantee binary compatibility with Rust loader. @@ -82,4 +91,5 @@ _Static_assert(sizeof(struct gpa_destination_entry) == 24, "destination_entry mu _Static_assert(sizeof(struct gpa_audit_key) == 8, "audit_key must be 8 bytes ([u32; 2])"); _Static_assert(sizeof(struct gpa_audit_event) == 20, "audit_event must be 20 bytes ([u32; 5])"); _Static_assert(sizeof(struct gpa_skip_process_entry) == 4, "skip_process_entry must be 4 bytes ([u32; 1])"); -_Static_assert(sizeof(struct gpa_sock_addr_local_entry) == 24, "sock_addr_local_entry must be 24 bytes ([u32; 6])"); +_Static_assert(sizeof(struct gpa_config_entry) == 4, "config_entry must be 4 bytes ([u32; 1])"); +_Static_assert(sizeof(struct gpa_sock_addr_local_entry) == 28, "sock_addr_local_entry must be 28 bytes ([u32; 7])"); From ea503eb13471d156fea576099c574d00ce767180 Mon Sep 17 00:00:00 2001 From: Zhidong Peng Date: Thu, 20 Aug 2026 21:30:11 +0000 Subject: [PATCH 2/6] fix naming issue caught by clippy::wrong-self-convention --- proxy_agent/src/redirector/linux.rs | 14 +++++++------- proxy_agent/src/redirector/shared_ebpf.rs | 16 ++++++++-------- 2 files changed, 15 insertions(+), 15 deletions(-) diff --git a/proxy_agent/src/redirector/linux.rs b/proxy_agent/src/redirector/linux.rs index 2a2e8ab0..33c3e9f5 100644 --- a/proxy_agent/src/redirector/linux.rs +++ b/proxy_agent/src/redirector/linux.rs @@ -64,7 +64,7 @@ impl BpfObject { Ok(mut skip_process_map) => { let key = sock_addr_skip_process_entry::from_pid(pid); let value = sock_addr_skip_process_entry::from_pid(pid); - match skip_process_map.insert(key.to_array(), value.to_array(), 0) { + match skip_process_map.insert(key.as_array(), value.as_array(), 0) { Ok(_) => logger::write(format!("skip_process_map updated with {pid}")), Err(err) => { return Err(Error::Bpf(BpfErrorType::UpdateBpfMapHashMap( @@ -140,7 +140,7 @@ impl BpfObject { let local_ip = super::string_to_ip(constants::PROXY_AGENT_IP); let key = destination_entry::from_ipv4(dest_ipv4, dest_port); let value = destination_entry::from_ipv4(local_ip, local_port); - match policy_map.insert(key.to_array(), value.to_array(), 0) { + match policy_map.insert(key.as_array(), value.as_array(), 0) { Ok(_) => { logger::write(format!("policy_map updated for {endpoint_name}")); } @@ -277,7 +277,7 @@ impl BpfObject { Some(map) => match HashMap::try_from(map) { Ok(audit_map) => { let key = sock_addr_audit_key::from_source_port(source_port); - match audit_map.get(&key.to_array(), 0) { + match audit_map.get(&key.as_array(), 0) { Ok(value) => { let audit_value = sock_addr_audit_entry::from_array(value); Ok(AuditEntry { @@ -319,7 +319,7 @@ impl BpfObject { Ok(mut policy_map) => { let key = destination_entry::from_ipv4(dest_ipv4, dest_port); if !redirect { - match policy_map.remove(&key.to_array()) { + match policy_map.remove(&key.as_array()) { Ok(_) => { event_logger::write_event( LoggerLevel::Info, @@ -351,7 +351,7 @@ impl BpfObject { ); let local_ip: u32 = super::string_to_ip(&local_ip); let value = destination_entry::from_ipv4(local_ip, local_port); - match policy_map.insert(key.to_array(), value.to_array(), 0) { + match policy_map.insert(key.as_array(), value.as_array(), 0) { Ok(_) => { event_logger::write_event( LoggerLevel::Info, @@ -393,7 +393,7 @@ impl BpfObject { Some(map) => match HashMap::<&mut MapData, [u32; 2], [u32; 5]>::try_from(map) { Ok(mut audit_map) => { let key = sock_addr_audit_key::from_source_port(source_port); - audit_map.remove(&key.to_array()).map_err(|err| { + audit_map.remove(&key.as_array()).map_err(|err| { Error::Bpf(BpfErrorType::MapDeleteElem( source_port.to_string(), format!("Error: {err}"), @@ -657,7 +657,7 @@ mod tests { ) .unwrap(); audit_map - .insert(key.to_array(), value.to_array(), 0) + .insert(key.as_array(), value.as_array(), 0) .unwrap(); } let audit = bpf.lookup_audit(source_port); diff --git a/proxy_agent/src/redirector/shared_ebpf.rs b/proxy_agent/src/redirector/shared_ebpf.rs index 63125dbc..d22e1c49 100644 --- a/proxy_agent/src/redirector/shared_ebpf.rs +++ b/proxy_agent/src/redirector/shared_ebpf.rs @@ -55,7 +55,7 @@ impl _destination_entry { entry } - pub fn to_array(&self) -> [u32; 6] { + pub fn as_array(&self) -> [u32; 6] { let mut array: [u32; 6] = [0; 6]; array[..4].copy_from_slice(&self.destination_ip.ip); array[4] = self.destination_port; @@ -85,7 +85,7 @@ impl sock_addr_skip_process_entry { entry } - pub fn to_array(&self) -> [u32; 1] { + pub fn as_array(&self) -> [u32; 1] { [self.pid] } } @@ -114,7 +114,7 @@ impl sock_addr_audit_key { } } - pub fn to_array(&self) -> [u32; 2] { + pub fn as_array(&self) -> [u32; 2] { [self.protocol, self.source_port] } @@ -157,7 +157,7 @@ impl sock_addr_audit_entry { } #[allow(dead_code)] - pub fn to_array(&self) -> [u32; 5] { + pub fn as_array(&self) -> [u32; 5] { [ self.logon_id, self.process_id, @@ -383,7 +383,7 @@ mod tests { #[test] fn destination_entry_ipv4_roundtrip_array_shape() { let entry = destination_entry::from_ipv4(0x1081_3FA8, 80); - let array = entry.to_array(); + let array = entry.as_array(); assert_eq!( array[0], 0x1081_3FA8, @@ -400,7 +400,7 @@ mod tests { #[test] fn audit_key_array_roundtrip() { let key = sock_addr_audit_key::from_source_port(1234); - let array = key.to_array(); + let array = key.as_array(); let rebuilt = sock_addr_audit_key::from_array(array); assert_eq!(rebuilt.protocol, IPPROTO_TCP, "protocol mismatch"); @@ -422,7 +422,7 @@ mod tests { let key = sock_addr_skip_process_entry::from_pid(pid); assert_eq!( - key.to_array(), + key.as_array(), [pid], "pid should roundtrip through the map key layout" ); @@ -438,7 +438,7 @@ mod tests { destination_port: 5, }; - let rebuilt = sock_addr_audit_entry::from_array(canonical.to_array()); + let rebuilt = sock_addr_audit_entry::from_array(canonical.as_array()); assert_eq!(rebuilt.logon_id, canonical.logon_id); assert_eq!(rebuilt.process_id, canonical.process_id); From 1af41c519d55fbbe882402b0b7c52726fb640ac4 Mon Sep 17 00:00:00 2001 From: Zhidong Peng Date: Mon, 31 Aug 2026 16:36:45 +0000 Subject: [PATCH 3/6] use ring-buffer to send audit events to user space. --- build.cmd | 4 +- ebpf/redirect.bpf.c | 19 +-- linux-ebpf/ebpf_cgroup.c | 17 +-- packages.config | 4 +- proxy_agent/src/redirector.rs | 58 +++++----- proxy_agent/src/redirector/linux.rs | 95 ++++++++------- proxy_agent/src/redirector/shared_ebpf.rs | 72 +++++++++++- proxy_agent/src/redirector/windows/bpf_api.rs | 33 ++++-- proxy_agent/src/redirector/windows/bpf_obj.rs | 5 + .../src/redirector/windows/bpf_prog.rs | 108 ++++++++++++------ shared-ebpf/include/gpa_audit_event.h | 10 ++ 11 files changed, 289 insertions(+), 136 deletions(-) diff --git a/build.cmd b/build.cmd index bba3d99f..cc8e8eee 100644 --- a/build.cmd +++ b/build.cmd @@ -27,8 +27,8 @@ echo out_dir=%out_dir% REM Set the path to the eBPF-for-Windows binaries and include files, REM We build ARM64 binaries on x64 machine, so we need to set the path to the x64 binaries -SET eBPF_for_Windows_bin_path=%root_path%packages\eBPF-for-Windows.x64.1.0.0-rc1\build\native\bin -SET eBPF_for_Windows_inc_path=%root_path%packages\eBPF-for-Windows.%eBPF_Platform%.1.0.0-rc1\build\native\include +SET eBPF_for_Windows_bin_path=%root_path%packages\eBPF-for-Windows.x64.1.5.0\build\native\bin +SET eBPF_for_Windows_inc_path=%root_path%packages\eBPF-for-Windows.%eBPF_Platform%.1.5.0\build\native\include SET bin_skim_path=%root_path%packages\Microsoft.CodeAnalysis.BinSkim.1.9.5\tools\netcoreapp3.1\win-x64 if "%CleanBuild%"=="clean" ( diff --git a/ebpf/redirect.bpf.c b/ebpf/redirect.bpf.c index 81a151e7..e4478596 100644 --- a/ebpf/redirect.bpf.c +++ b/ebpf/redirect.bpf.c @@ -35,10 +35,10 @@ struct bpf_map_def audit_map = { #pragma clang section data = "maps" struct bpf_map_def audit_only_map = { - .type = BPF_MAP_TYPE_LRU_HASH, - .key_size = sizeof(sock_addr_audit_key_t), - .value_size = sizeof(sock_addr_audit_entry_t), - .max_entries = 1000}; + .type = BPF_MAP_TYPE_RINGBUF, + .key_size = 0, + .value_size = 0, + .max_entries = 256 * 1024}; /* check the current pid in the skip_process map. @@ -101,13 +101,14 @@ update_audit_map_entry(bpf_sock_addr_t *ctx, int audit_only) uint16_t source_port = ctx->msg_src_port; if (audit_only) { - sock_addr_audit_key_t key = {0}; - key.protocol = ctx->protocol; - key.source_port = source_port != 0 ? source_port : pid; - uint64_t ret = bpf_map_update_elem(&audit_only_map, &key, &entry, 0); + struct gpa_audit_only_event event = {0}; + event.kernel_timestamp_ns = bpf_ktime_get_ns(); + event.local_ipv4 = ctx->msg_src_ip4; + event.audit = entry; + uint64_t ret = bpf_ringbuf_output(&audit_only_map, &event, sizeof(event), 0); if (ret != 0) { - bpf_printk("Failed to update audit-only map with results: %u.", ret); + bpf_printk("Failed to emit audit-only event with results: %u.", ret); } return 0; } diff --git a/linux-ebpf/ebpf_cgroup.c b/linux-ebpf/ebpf_cgroup.c index 98136bbc..eadb5e0f 100644 --- a/linux-ebpf/ebpf_cgroup.c +++ b/linux-ebpf/ebpf_cgroup.c @@ -43,10 +43,8 @@ struct { } audit_map SEC(".maps"); struct { - __uint(type, BPF_MAP_TYPE_LRU_HASH); - __type(key, struct gpa_audit_key); - __type(value, struct gpa_audit_event); - __uint(max_entries, 200); + __uint(type, BPF_MAP_TYPE_RINGBUF); + __uint(max_entries, 256 * 1024); } audit_only_map SEC(".maps"); struct { @@ -171,7 +169,7 @@ int connect4(struct bpf_sock_addr *ctx) } static __always_inline int -update_audit_map_entry_sk(__u32 local_port, struct gpa_sock_addr_local_entry *local_entry) +update_audit_map_entry_sk(__u32 local_port, __u32 local_ipv4, struct gpa_sock_addr_local_entry *local_entry) { struct gpa_audit_key key = {0}; key.protocol = local_entry->protocol; @@ -187,7 +185,11 @@ update_audit_map_entry_sk(__u32 local_port, struct gpa_sock_addr_local_entry *lo __u64 ret; if (local_entry->audit_only) { - ret = bpf_map_update_elem(&audit_only_map, &key, &entry, 0); + struct gpa_audit_only_event event = {0}; + event.kernel_timestamp_ns = bpf_ktime_get_ns(); + event.local_ipv4 = local_ipv4; + event.audit = entry; + ret = bpf_ringbuf_output(&audit_only_map, &event, sizeof(event), 0); } else { @@ -222,6 +224,7 @@ trace_v4(struct pt_regs *ctx, struct sock *sk) return 0; } __be32 skc_daddr = BPF_CORE_READ(sk, __sk_common.skc_daddr); + __be32 skc_rcv_saddr = BPF_CORE_READ(sk, __sk_common.skc_rcv_saddr); __be16 skc_dport = BPF_CORE_READ(sk, __sk_common.skc_dport); __u16 skc_num = BPF_CORE_READ(sk, __sk_common.skc_num); @@ -237,7 +240,7 @@ trace_v4(struct pt_regs *ctx, struct sock *sk) struct gpa_sock_addr_local_entry *local_entry = bpf_map_lookup_elem(&local_map, &pid_tgid); if (local_entry != NULL) { - update_audit_map_entry_sk(skc_num, local_entry); + update_audit_map_entry_sk(skc_num, skc_rcv_saddr, local_entry); __u64 ret = bpf_map_delete_elem(&local_map, &pid_tgid); if (ret != 0) { diff --git a/packages.config b/packages.config index 93bf3b9a..e053bdc0 100644 --- a/packages.config +++ b/packages.config @@ -1,7 +1,7 @@ - - + + diff --git a/proxy_agent/src/redirector.rs b/proxy_agent/src/redirector.rs index e0c1f86c..39df48d9 100644 --- a/proxy_agent/src/redirector.rs +++ b/proxy_agent/src/redirector.rs @@ -84,6 +84,13 @@ pub struct AuditEntry { pub destination_port: u16, // in network byte order } +pub struct AuditOnlyRecord { + pub entry: AuditEntry, + pub kernel_timestamp_ns: u64, + pub timestamp_utc_ns: i128, + pub local_ipv4: u32, +} + impl AuditEntry { pub fn empty() -> Self { AuditEntry { @@ -192,6 +199,12 @@ impl Redirector { self.attach_bpf_prog(&mut bpf_object)?; logger::write_information("Success attached bpf prog.".to_string()); + let audit_only_receiver = if monitor_only { + Some(bpf_object.subscribe_audit_only(self.shared_state.get_cancellation_token())?) + } else { + None + }; + if let Err(e) = self .shared_state .get_redirector_shared_state() @@ -200,9 +213,9 @@ impl Redirector { { logger::write_error(format!("Failed to update bpf object in shared state: {e}")); } - if monitor_only { - tokio::spawn(poll_audit_only( - self.shared_state.get_redirector_shared_state(), + if let Some(receiver) = audit_only_receiver { + tokio::spawn(process_audit_only_events( + receiver, self.shared_state.get_proxy_server_shared_state(), self.shared_state.get_cancellation_token(), )); @@ -268,28 +281,17 @@ impl Redirector { } } -async fn poll_audit_only( - redirector_shared_state: RedirectorSharedState, +async fn process_audit_only_events( + mut receiver: tokio::sync::mpsc::UnboundedReceiver, proxy_server_shared_state: ProxyServerSharedState, cancellation_token: CancellationToken, ) { - let mut interval = tokio::time::interval(std::time::Duration::from_secs(1)); loop { tokio::select! { _ = cancellation_token.cancelled() => return, - _ = interval.tick() => { - let Some(bpf_object) = redirector_shared_state - .get_bpf_object() - .await - .ok() - .flatten() - else { - continue; - }; - let records = bpf_object.lock().unwrap().drain_audit_only(); - match records { - Ok(records) => { - for entry in records { + record = receiver.recv() => { + let Some(record) = record else { return; }; + let entry = record.entry; let destination_ip = entry.destination_ipv4_addr(); let destination_port = entry.destination_port_in_host_byte_order(); let message = match Claims::from_audit_entry( @@ -301,7 +303,10 @@ async fn poll_audit_only( .await { Ok(claims) => format!( - "eBPF audit-only connection: userName={}, processId={}, processName={}, processFullPath={}, processCmdLine={}, runAsElevated={}, destination={}:{}", + "eBPF audit-only connection: timestampUtcNs={}, kernelTimestampNs={}, localIp={}, userName={}, processId={}, processName={}, processFullPath={}, processCmdLine={}, runAsElevated={}, destination={}:{}", + record.timestamp_utc_ns, + record.kernel_timestamp_ns, + Ipv4Addr::from_bits(record.local_ipv4.to_be()), claims.userName, claims.processId, claims.processName.to_string_lossy(), @@ -312,7 +317,10 @@ async fn poll_audit_only( destination_port, ), Err(err) => format!( - "eBPF audit-only connection: userId={}, processId={}, processDetails=unavailable ({err}), destination={}:{}", + "eBPF audit-only connection: timestampUtcNs={}, kernelTimestampNs={}, localIp={}, userId={}, processId={}, processDetails=unavailable ({err}), destination={}:{}", + record.timestamp_utc_ns, + record.kernel_timestamp_ns, + Ipv4Addr::from_bits(record.local_ipv4.to_be()), entry.logon_id, entry.process_id, destination_ip, @@ -322,16 +330,10 @@ async fn poll_audit_only( event_logger::write_event( LoggerLevel::Warn, message, - "poll_audit_only", + "process_audit_only_events", "redirector", logger::AGENT_LOGGER_KEY, ); - } - } - Err(err) => logger::write_warning(format!( - "Failed to drain eBPF audit-only map: {err}" - )), - } } } } diff --git a/proxy_agent/src/redirector/linux.rs b/proxy_agent/src/redirector/linux.rs index 33c3e9f5..13e03c0b 100644 --- a/proxy_agent/src/redirector/linux.rs +++ b/proxy_agent/src/redirector/linux.rs @@ -7,14 +7,14 @@ use crate::common::{ result::Result, }; use crate::redirector::shared_ebpf::linux_types::{ - destination_entry, sock_addr_audit_entry, sock_addr_audit_key, sock_addr_skip_process_entry, - GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY, + audit_only_event, destination_entry, sock_addr_audit_entry, sock_addr_audit_key, + sock_addr_skip_process_entry, GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY, }; -use crate::redirector::{ip_to_string, AuditEntry}; +use crate::redirector::{ip_to_string, AuditEntry, AuditOnlyRecord}; use crate::shared_state::redirector_wrapper::RedirectorSharedState; use aya::programs::{CgroupSockAddr, KProbe}; use aya::{ - maps::{HashMap, MapData}, + maps::{HashMap, MapData, RingBuf}, programs::CgroupAttachMode, }; use aya::{Btf, Ebpf, EbpfLoader}; @@ -417,45 +417,60 @@ impl BpfObject { Ok(()) } - pub fn drain_audit_only(&mut self) -> Result> { + pub fn subscribe_audit_only( + &mut self, + cancellation_token: tokio_util::sync::CancellationToken, + ) -> Result> { let audit_map_name = "audit_only_map"; - match self.0.map_mut(audit_map_name) { - Some(map) => { - let mut audit_map = HashMap::<&mut MapData, [u32; 2], [u32; 5]>::try_from(map) - .map_err(|err| { - Error::Bpf(BpfErrorType::LoadBpfMapHashMap( - audit_map_name.to_string(), - err.to_string(), - )) - })?; - let mut records = Vec::new(); - for item in audit_map.iter() { - let (key, value) = item.map_err(|err| { - Error::Bpf(BpfErrorType::MapLookupElem( - audit_map_name.to_string(), - err.to_string(), - )) - })?; - records.push(( - key, - sock_addr_audit_entry::from_array(value).to_audit_entry(), - )); - } - for (key, _) in &records { - audit_map.remove(key).map_err(|err| { - Error::Bpf(BpfErrorType::MapDeleteElem( - audit_map_name.to_string(), - err.to_string(), - )) - })?; - } - Ok(records.into_iter().map(|(_, entry)| entry).collect()) - } - None => Err(Error::Bpf(BpfErrorType::GetBpfMap( + let map = self.0.take_map(audit_map_name).ok_or_else(|| { + Error::Bpf(BpfErrorType::GetBpfMap( audit_map_name.to_string(), "Map does not exist".to_string(), - ))), - } + )) + })?; + let ring = RingBuf::try_from(map).map_err(|err| { + Error::Bpf(BpfErrorType::LoadBpfMapHashMap( + audit_map_name.to_string(), + err.to_string(), + )) + })?; + let mut async_ring = tokio::io::unix::AsyncFd::new(ring).map_err(|err| { + Error::Bpf(BpfErrorType::LoadBpfMapHashMap( + audit_map_name.to_string(), + err.to_string(), + )) + })?; + let (sender, receiver) = tokio::sync::mpsc::unbounded_channel(); + tokio::spawn(async move { + loop { + tokio::select! { + _ = cancellation_token.cancelled() => return, + readiness = async_ring.readable_mut() => { + let Ok(mut guard) = readiness else { return; }; + while let Some(item) = guard.get_inner_mut().next() { + match audit_only_event::from_bytes(&item) { + Ok(event) => { + let record = AuditOnlyRecord { + entry: event.to_audit_entry(), + kernel_timestamp_ns: event.kernel_timestamp_ns, + timestamp_utc_ns: proxy_agent_shared::misc_helpers::get_date_time_unix_nano(), + local_ipv4: event.local_ipv4, + }; + if sender.send(record).is_err() { + return; + } + } + Err(err) => logger::write_warning(format!( + "Failed to decode eBPF audit-only event: {err}" + )), + } + } + guard.clear_ready(); + } + } + } + }); + Ok(receiver) } } diff --git a/proxy_agent/src/redirector/shared_ebpf.rs b/proxy_agent/src/redirector/shared_ebpf.rs index d22e1c49..88303e90 100644 --- a/proxy_agent/src/redirector/shared_ebpf.rs +++ b/proxy_agent/src/redirector/shared_ebpf.rs @@ -135,6 +135,44 @@ pub struct sock_addr_audit_entry { pub destination_ipv4: u32, pub destination_port: u32, } + +#[repr(C)] +#[derive(Clone, Copy, Debug)] +pub struct audit_only_event { + pub kernel_timestamp_ns: u64, + pub local_ipv4: u32, + pub logon_id: u32, + pub process_id: u32, + pub is_root: u32, + pub destination_ipv4: u32, + pub destination_port: u32, +} + +impl audit_only_event { + pub fn from_bytes(bytes: &[u8]) -> Result { + if bytes.len() != std::mem::size_of::() { + return Err(Error::Bpf(BpfErrorType::MapLookupElem( + "audit_only_map".to_string(), + format!( + "Invalid ring-buffer record size: {}, expected {}", + bytes.len(), + std::mem::size_of::() + ), + ))); + } + Ok(unsafe { std::ptr::read_unaligned(bytes.as_ptr() as *const Self) }) + } + + pub fn to_audit_entry(self) -> crate::redirector::AuditEntry { + crate::redirector::AuditEntry { + logon_id: u64::from(self.logon_id), + process_id: self.process_id, + is_admin: self.is_root as i32, + destination_ipv4: self.destination_ipv4, + destination_port: self.destination_port as u16, + } + } +} impl sock_addr_audit_entry { pub fn empty() -> Self { sock_addr_audit_entry { @@ -362,7 +400,7 @@ impl AuditValueEntry { #[cfg(not(windows))] pub mod linux_types { pub use super::{ - destination_entry, sock_addr_audit_entry, sock_addr_audit_key, + audit_only_event, destination_entry, sock_addr_audit_entry, sock_addr_audit_key, sock_addr_skip_process_entry, GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY, }; } @@ -370,7 +408,7 @@ pub mod linux_types { #[cfg(windows)] pub mod windows_types { pub use super::{ - destination_entry as destination_entry_t, sock_addr_audit_entry, + audit_only_event, destination_entry as destination_entry_t, sock_addr_audit_key as sock_addr_audit_key_t, sock_addr_skip_process_entry, GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY, }; @@ -465,6 +503,36 @@ mod tests { assert_eq!(audit.destination_port, 8080u16.to_be()); } + #[test] + fn audit_only_event_binary_layout_and_decode() { + let event = audit_only_event { + kernel_timestamp_ns: 123, + local_ipv4: 0x0A00_0001, + logon_id: 42, + process_id: 1000, + is_root: 1, + destination_ipv4: 0x1081_3FA8, + destination_port: u32::from(80u16.to_be()), + }; + assert_eq!(std::mem::size_of::(), 32); + + let bytes = unsafe { + std::slice::from_raw_parts( + &event as *const audit_only_event as *const u8, + std::mem::size_of::(), + ) + }; + let decoded = audit_only_event::from_bytes(bytes).expect("record should decode"); + let audit = decoded.to_audit_entry(); + + assert_eq!(decoded.kernel_timestamp_ns, 123); + assert_eq!(decoded.local_ipv4, 0x0A00_0001); + assert_eq!(audit.logon_id, 42); + assert_eq!(audit.process_id, 1000); + assert_eq!(audit.destination_port, 80u16.to_be()); + assert!(audit_only_event::from_bytes(&bytes[..bytes.len() - 1]).is_err()); + } + #[test] fn audit_entry_legacy_to_agent_audit_entry() { let legacy = sock_addr_audit_entry_legacy { diff --git a/proxy_agent/src/redirector/windows/bpf_api.rs b/proxy_agent/src/redirector/windows/bpf_api.rs index 04578c59..e1e0d6d6 100644 --- a/proxy_agent/src/redirector/windows/bpf_api.rs +++ b/proxy_agent/src/redirector/windows/bpf_api.rs @@ -184,8 +184,11 @@ type BpfMapLookupElem = unsafe extern "C" fn(map_fd: c_int, key: *const c_void, value: *mut c_void) -> c_int; type BpfMapDeleteElem = unsafe extern "C" fn(map_fd: c_int, key: *const c_void) -> c_int; -type BpfMapGetNextKey = - unsafe extern "C" fn(map_fd: c_int, key: *const c_void, next_key: *mut c_void) -> c_int; +type RingBufferSampleFn = unsafe extern "C" fn(*mut c_void, *mut c_void, usize) -> c_int; +type RingBufferNew = + unsafe extern "C" fn(c_int, RingBufferSampleFn, *mut c_void, *const c_void) -> *mut ring_buffer; +type RingBufferPoll = unsafe extern "C" fn(*mut ring_buffer, c_int) -> c_int; +type RingBufferFree = unsafe extern "C" fn(*mut ring_buffer); type LibBpfGetError = unsafe extern "C" fn(no_use_ptr: *const c_void) -> c_long; @@ -307,13 +310,25 @@ pub fn bpf_map_delete_elem(map_fd: c_int, key: *const c_void) -> Result { Ok(unsafe { map_delete_elem(map_fd, key) }) } -pub fn bpf_map_get_next_key( +pub fn ring_buffer__new( map_fd: c_int, - key: *const c_void, - next_key: *mut c_void, -) -> Result { + callback: RingBufferSampleFn, + context: *mut c_void, +) -> Result<*mut ring_buffer> { + let ebpf_api = get_ebpf_api()?; + let new_ring: Symbol = get_ebpf_api_fun(ebpf_api, "ring_buffer__new\0")?; + Ok(unsafe { new_ring(map_fd, callback, context, std::ptr::null()) }) +} + +pub fn ring_buffer__poll(ring: *mut ring_buffer, timeout_ms: c_int) -> Result { + let ebpf_api = get_ebpf_api()?; + let poll_ring: Symbol = get_ebpf_api_fun(ebpf_api, "ring_buffer__poll\0")?; + Ok(unsafe { poll_ring(ring, timeout_ms) }) +} + +pub fn ring_buffer__free(ring: *mut ring_buffer) -> Result<()> { let ebpf_api = get_ebpf_api()?; - let map_get_next_key: Symbol = - get_ebpf_api_fun(ebpf_api, "bpf_map_get_next_key\0")?; - Ok(unsafe { map_get_next_key(map_fd, key, next_key) }) + let free_ring: Symbol = get_ebpf_api_fun(ebpf_api, "ring_buffer__free\0")?; + unsafe { free_ring(ring) }; + Ok(()) } diff --git a/proxy_agent/src/redirector/windows/bpf_obj.rs b/proxy_agent/src/redirector/windows/bpf_obj.rs index 3b70f8b4..c77a6f96 100644 --- a/proxy_agent/src/redirector/windows/bpf_obj.rs +++ b/proxy_agent/src/redirector/windows/bpf_obj.rs @@ -15,6 +15,11 @@ pub type ebpf_handle_t = i64; pub type ebpf_program_type_t = uuid::Uuid; pub type ebpf_attach_type_t = uuid::Uuid; +#[repr(C)] +pub struct ring_buffer { + _private: [u8; 0], +} + // Type aliases used by libbpf headers. pub type __s32 = i32; pub type __s64 = i64; diff --git a/proxy_agent/src/redirector/windows/bpf_prog.rs b/proxy_agent/src/redirector/windows/bpf_prog.rs index 11a8b732..bbf79184 100644 --- a/proxy_agent/src/redirector/windows/bpf_prog.rs +++ b/proxy_agent/src/redirector/windows/bpf_prog.rs @@ -10,14 +10,39 @@ use crate::common::{ result::Result, }; use crate::redirector::shared_ebpf::windows_types::{ - sock_addr_audit_entry, GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY, + audit_only_event, GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY, }; -use crate::redirector::AuditEntry; +use crate::redirector::{AuditEntry, AuditOnlyRecord}; use proxy_agent_shared::misc_helpers; use std::ffi::c_void; use std::mem::size_of_val; use std::path::Path; +unsafe extern "C" fn audit_only_callback( + context: *mut c_void, + data: *mut c_void, + size: usize, +) -> i32 { + let sender = &*(context as *const tokio::sync::mpsc::UnboundedSender); + let bytes = std::slice::from_raw_parts(data as *const u8, size); + match audit_only_event::from_bytes(bytes) { + Ok(event) => { + let record = AuditOnlyRecord { + entry: event.to_audit_entry(), + kernel_timestamp_ns: event.kernel_timestamp_ns, + timestamp_utc_ns: proxy_agent_shared::misc_helpers::get_date_time_unix_nano(), + local_ipv4: event.local_ipv4, + }; + if sender.send(record).is_ok() { + 0 + } else { + -1 + } + } + Err(_) => -1, + } +} + // This module contains the logic to interact with the windows eBPF program & maps. impl BpfObject { fn is_null(&self) -> bool { @@ -366,46 +391,55 @@ impl BpfObject { Ok(()) } - pub fn drain_audit_only(&self) -> Result> { + pub fn subscribe_audit_only( + &self, + cancellation_token: tokio_util::sync::CancellationToken, + ) -> Result> { let map_name = "audit_only_map"; let map_fd = self.get_bpf_map_fd(map_name)?; - let mut keys = Vec::new(); - let mut previous_key: Option = None; - - loop { - let mut next_key = sock_addr_audit_key_t::from_array([0; 2]); - let previous_key_ptr = previous_key - .as_ref() - .map_or(std::ptr::null(), |key| key as *const _ as *const c_void); - let result = bpf_map_get_next_key( - map_fd, - previous_key_ptr, - &mut next_key as *mut sock_addr_audit_key_t as *mut c_void, - )?; - if result != 0 { - break; + let (sender, receiver) = tokio::sync::mpsc::unbounded_channel(); + let context = Box::into_raw(Box::new(sender)); + let ring = match ring_buffer__new(map_fd, audit_only_callback, context.cast()) { + Ok(ring) => ring, + Err(err) => { + unsafe { drop(Box::from_raw(context)) }; + return Err(err); } - previous_key = Some(next_key); - keys.push(next_key); + }; + if ring.is_null() { + unsafe { drop(Box::from_raw(context)) }; + return Err(Error::Bpf(BpfErrorType::LoadBpfMapHashMap( + map_name.to_string(), + "ring_buffer__new returned null".to_string(), + ))); } - - let mut records = Vec::with_capacity(keys.len()); - for key in keys { - let mut value = sock_addr_audit_entry::empty(); - let result = bpf_map_lookup_elem( - map_fd, - &key as *const sock_addr_audit_key_t as *const c_void, - &mut value as *mut sock_addr_audit_entry as *mut c_void, - )?; - if result == 0 { - records.push(value.to_audit_entry()); - let _ = bpf_map_delete_elem( - map_fd, - &key as *const sock_addr_audit_key_t as *const c_void, - )?; + let ring_address = ring as usize; + let context_address = context as usize; + tokio::task::spawn_blocking(move || { + let ring = ring_address as *mut ring_buffer; + while !cancellation_token.is_cancelled() { + match ring_buffer__poll(ring, 250) { + Ok(result) if result >= 0 => {} + Ok(result) => { + logger::write_warning(format!( + "ring_buffer__poll failed with result {result}" + )); + break; + } + Err(err) => { + logger::write_warning(format!("ring_buffer__poll failed: {err}")); + break; + } + } } - } - Ok(records) + let _ = ring_buffer__free(ring); + unsafe { + drop(Box::from_raw( + context_address as *mut tokio::sync::mpsc::UnboundedSender, + )); + } + }); + Ok(receiver) } /** diff --git a/shared-ebpf/include/gpa_audit_event.h b/shared-ebpf/include/gpa_audit_event.h index 816450a7..ecc9432d 100644 --- a/shared-ebpf/include/gpa_audit_event.h +++ b/shared-ebpf/include/gpa_audit_event.h @@ -58,6 +58,15 @@ struct gpa_audit_event __u32 destination_port; // Destination port (stored as u32) }; +// Audit-only ring-buffer record. The kernel timestamp is monotonic; user mode +// adds UTC at receipt because eBPF does not expose a UTC clock on all platforms. +struct gpa_audit_only_event +{ + __u64 kernel_timestamp_ns; + __u32 local_ipv4; + struct gpa_audit_event audit; +}; + // Skip process entry - processes in this map bypass audit/redirect // Size: 4 bytes - matches Rust sock_addr_skip_process_entry -> [u32; 1] struct gpa_skip_process_entry @@ -90,6 +99,7 @@ _Static_assert(sizeof(struct gpa_ip_address) == 16, "ip_address must be 16 bytes _Static_assert(sizeof(struct gpa_destination_entry) == 24, "destination_entry must be 24 bytes ([u32; 6])"); _Static_assert(sizeof(struct gpa_audit_key) == 8, "audit_key must be 8 bytes ([u32; 2])"); _Static_assert(sizeof(struct gpa_audit_event) == 20, "audit_event must be 20 bytes ([u32; 5])"); +_Static_assert(sizeof(struct gpa_audit_only_event) == 32, "audit_only_event must be 32 bytes"); _Static_assert(sizeof(struct gpa_skip_process_entry) == 4, "skip_process_entry must be 4 bytes ([u32; 1])"); _Static_assert(sizeof(struct gpa_config_entry) == 4, "config_entry must be 4 bytes ([u32; 1])"); _Static_assert(sizeof(struct gpa_sock_addr_local_entry) == 28, "sock_addr_local_entry must be 28 bytes ([u32; 7])"); From 9c4f572239eb5d96534168d86c29cd568cbe18e4 Mon Sep 17 00:00:00 2001 From: "Zhidong Peng (HE/HIM)" Date: Mon, 31 Aug 2026 17:13:24 -0700 Subject: [PATCH 4/6] fix format --- proxy_agent/src/redirector/shared_ebpf.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/proxy_agent/src/redirector/shared_ebpf.rs b/proxy_agent/src/redirector/shared_ebpf.rs index 75983d33..e767e2d5 100644 --- a/proxy_agent/src/redirector/shared_ebpf.rs +++ b/proxy_agent/src/redirector/shared_ebpf.rs @@ -176,7 +176,7 @@ impl audit_only_event { is_admin: self.is_root as i32, destination_ipv4: self.destination_ipv4, destination_port: self.destination_port as u16, - address_family: crate::redirector::AddressFamily::IPv4, //TODO: audit_only_event does not include address_family, so we assume IPv4 for now. + address_family: crate::redirector::AddressFamily::IPv4, //TODO: audit_only_event does not include address_family, so we assume IPv4 for now. } } } From 5d481b3bc602763bc4735f44b5e0ea6f7acd17c6 Mon Sep 17 00:00:00 2001 From: "Zhidong Peng (HE/HIM)" Date: Mon, 31 Aug 2026 17:16:04 -0700 Subject: [PATCH 5/6] fix clippy --- proxy_agent/src/redirector/shared_ebpf.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/proxy_agent/src/redirector/shared_ebpf.rs b/proxy_agent/src/redirector/shared_ebpf.rs index e767e2d5..924e88ba 100644 --- a/proxy_agent/src/redirector/shared_ebpf.rs +++ b/proxy_agent/src/redirector/shared_ebpf.rs @@ -118,7 +118,7 @@ impl sock_addr_audit_key { } } - pub fn to_array(&self) -> AuditMapKey { + pub fn as_array(&self) -> AuditMapKey { [self.protocol, self.source_port] } From d7694e129c2515c9f4efc0f97041ce6292d332bb Mon Sep 17 00:00:00 2001 From: Zhidong Peng Date: Tue, 1 Sep 2026 00:29:19 +0000 Subject: [PATCH 6/6] fix build --- linux-ebpf/ebpf_cgroup.c | 526 +++++++++++----------- proxy_agent/src/redirector/linux.rs | 2 +- proxy_agent/src/redirector/shared_ebpf.rs | 2 +- 3 files changed, 263 insertions(+), 267 deletions(-) diff --git a/linux-ebpf/ebpf_cgroup.c b/linux-ebpf/ebpf_cgroup.c index 657d80cc..beb4bb97 100644 --- a/linux-ebpf/ebpf_cgroup.c +++ b/linux-ebpf/ebpf_cgroup.c @@ -40,293 +40,289 @@ struct struct { - struct - { - __uint(type, BPF_MAP_TYPE_LRU_HASH); - __type(key, struct gpa_audit_key); // source port and protocol - __type(value, struct gpa_audit_event); // audit event (canonical struct) - __uint(max_entries, 200); // LRU evicts oldest on overflow - } audit_map SEC(".maps"); + __uint(type, BPF_MAP_TYPE_LRU_HASH); + __type(key, struct gpa_audit_key); // source port and protocol + __type(value, struct gpa_audit_event); // audit event (canonical struct) + __uint(max_entries, 200); // LRU evicts oldest on overflow +} audit_map SEC(".maps"); - struct - { - __uint(type, BPF_MAP_TYPE_RINGBUF); - __uint(max_entries, 256 * 1024); - } audit_only_map SEC(".maps"); +struct +{ + __uint(type, BPF_MAP_TYPE_RINGBUF); + __uint(max_entries, 256 * 1024); +} audit_only_map SEC(".maps"); + +struct +{ + __uint(type, BPF_MAP_TYPE_LRU_HASH); + __type(key, __u64); // pid-tgid or socket cookie + __type(value, struct gpa_sock_addr_local_entry); + __uint(max_entries, 200); +} local_map SEC(".maps"); + +/* + check the current pid in the skip_process map. + return 1 if found, otherwise return 0. +*/ +static __always_inline int +check_skip_process_map_entry(__u32 pid) +{ + struct gpa_skip_process_entry key = {0}; + key.pid = pid; + + // Find the entry in the skip_process map. + struct gpa_skip_process_entry *skip_entry = bpf_map_lookup_elem(&skip_process_map, &key); + return (skip_entry != NULL) ? 1 : 0; +} - struct +static __always_inline int +local_ip_bind_monitor_only_enabled(void) +{ + __u32 key = GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY; + struct gpa_config_entry *entry = bpf_map_lookup_elem(&config_map, &key); + return entry != NULL && entry->enabled != 0; +} + +/* + update audit map entry if not skip redirecting. + return 0 if the entry is updated, otherwise + return 1 if pid found in the skip_process_map. +*/ +static __always_inline int +update_local_map_entry(struct bpf_sock_addr *ctx, __u32 audit_only, __be32 destination_ipv4, __u32 address_family) +{ + __u64 pid_tip = bpf_get_current_pid_tgid(); + __u32 pid = (__u32)(pid_tip >> 32); + + if (check_skip_process_map_entry(pid) == 1) { - struct - { - __uint(type, BPF_MAP_TYPE_LRU_HASH); - __type(key, __u64); // pid-tgid or socket cookie - __type(value, struct gpa_sock_addr_local_entry); - __uint(max_entries, 200); - } local_map SEC(".maps"); - - /* - check the current pid in the skip_process map. - return 1 if found, otherwise return 0. - */ - static __always_inline int - check_skip_process_map_entry(__u32 pid) - { - struct gpa_skip_process_entry key = {0}; - key.pid = pid; + return 1; + } + + struct gpa_sock_addr_local_entry entry = {0}; + entry.process_id = pid; + __u32 uid = (__u32)(bpf_get_current_uid_gid() >> 32); + entry.logon_id = uid; + entry.is_root = (uid == 0) ? 1 : 0; // root uid is 0. + entry.destination_ipv4 = destination_ipv4; + entry.destination_port = ctx->user_port; + entry.protocol = ctx->protocol; + entry.audit_only = audit_only; + entry.address_family = address_family; + + __u64 ret = bpf_map_update_elem(&local_map, &pid_tip, &entry, 0); + if (ret != 0) + { + bpf_printk("update_local_map_entry: Failed to update local map entry with results:%u.", ret); + } + else + { + bpf_printk("update_local_map_entry: Updated local map entry with key:%u.", pid_tip); + } - // Find the entry in the skip_process map. - struct gpa_skip_process_entry *skip_entry = bpf_map_lookup_elem(&skip_process_map, &key); - return (skip_entry != NULL) ? 1 : 0; - } + return 0; +} - static __always_inline int - local_ip_bind_monitor_only_enabled(void) +static __always_inline int +authorize_v4(struct bpf_sock_addr *ctx) +{ + struct gpa_destination_entry entry = {0}; + entry.destination_ip.ipv4 = ctx->user_ip4; + entry.destination_port = ctx->user_port; + entry.protocol = ctx->protocol; + + // Find the entry in the policy map. + struct gpa_destination_entry *policy = bpf_map_lookup_elem(&policy_map, &entry); + if (policy != NULL) + { + bpf_printk("authorize_v4: Found v4 proxy entry value: %u, %u", policy->destination_ip.ipv4, policy->destination_port); + + // At connect4, msg_src_ip4 is not valid; it is only populated for + // UDP sendmsg hooks. A concrete address set by bind(2) is available + // from the socket before TCP performs automatic source selection. + __u32 source_ip = ctx->sk != NULL ? ctx->sk->src_ip4 : 0; + __u32 source_ip_host = bpf_ntohl(source_ip); + __u32 audit_only = local_ip_bind_monitor_only_enabled() && + source_ip != 0 && + (source_ip_host & 0xff000000) != 0x7f000000; + + // update to the audit map before changing the destination ip and port. + if (update_local_map_entry(ctx, audit_only, ctx->user_ip4, GPA_ADDRESS_FAMILY_IPV4) == 1) { - __u32 key = GPA_CONFIG_LOCAL_IP_BIND_MONITOR_ONLY; - struct gpa_config_entry *entry = bpf_map_lookup_elem(&config_map, &key); - return entry != NULL && entry->enabled != 0; + bpf_printk("authorize_v4: Found skip process entry, skip the redirection."); + return BPF_SOCK_ADDR_VERDICT_PROCEED; } - /* - update audit map entry if not skip redirecting. - return 0 if the entry is updated, otherwise - return 1 if pid found in the skip_process_map. - */ - static __always_inline int - update_local_map_entry(struct bpf_sock_addr *ctx, __u32 audit_only, __be32 destination_ipv4, __u32 address_family) + if (audit_only) { - __u64 pid_tip = bpf_get_current_pid_tgid(); - __u32 pid = (__u32)(pid_tip >> 32); - - if (check_skip_process_map_entry(pid) == 1) - { - return 1; - } - - struct gpa_sock_addr_local_entry entry = {0}; - entry.process_id = pid; - __u32 uid = (__u32)(bpf_get_current_uid_gid() >> 32); - entry.logon_id = uid; - entry.is_root = (uid == 0) ? 1 : 0; // root uid is 0. - entry.destination_ipv4 = destination_ipv4; - entry.destination_port = ctx->user_port; - entry.protocol = ctx->protocol; - entry.audit_only = audit_only; - entry.address_family = address_family; - - __u64 ret = bpf_map_update_elem(&local_map, &pid_tip, &entry, 0); - if (ret != 0) - { - bpf_printk("update_local_map_entry: Failed to update local map entry with results:%u.", ret); - } - else - { - bpf_printk("update_local_map_entry: Updated local map entry with key:%u.", pid_tip); - } - - return 0; + bpf_printk("authorize_v4: Source address is explicitly bound, audit without redirecting."); + return BPF_SOCK_ADDR_VERDICT_PROCEED; } - static __always_inline int - authorize_v4(struct bpf_sock_addr *ctx) - { - struct gpa_destination_entry entry = {0}; - entry.destination_ip.ipv4 = ctx->user_ip4; - entry.destination_port = ctx->user_port; - entry.protocol = ctx->protocol; - - // Find the entry in the policy map. - struct gpa_destination_entry *policy = bpf_map_lookup_elem(&policy_map, &entry); - if (policy != NULL) - { - bpf_printk("authorize_v4: Found v4 proxy entry value: %u, %u", policy->destination_ip.ipv4, policy->destination_port); - - // At connect4, msg_src_ip4 is not valid; it is only populated for - // UDP sendmsg hooks. A concrete address set by bind(2) is available - // from the socket before TCP performs automatic source selection. - __u32 source_ip = ctx->sk != NULL ? ctx->sk->src_ip4 : 0; - __u32 source_ip_host = bpf_ntohl(source_ip); - __u32 audit_only = local_ip_bind_monitor_only_enabled() && - source_ip != 0 && - (source_ip_host & 0xff000000) != 0x7f000000; - - // update to the audit map before changing the destination ip and port. - if (update_local_map_entry(ctx, audit_only, ctx->user_ip4, GPA_ADDRESS_FAMILY_IPV4) == 1) - { - bpf_printk("authorize_v4: Found skip process entry, skip the redirection."); - return BPF_SOCK_ADDR_VERDICT_PROCEED; - } - - if (audit_only) - { - bpf_printk("authorize_v4: Source address is explicitly bound, audit without redirecting."); - return BPF_SOCK_ADDR_VERDICT_PROCEED; - } - - ctx->user_ip4 = policy->destination_ip.ipv4; - bpf_printk("authorize_v4: Local/source ip is not set, redirect to loopback ip."); - ctx->user_port = policy->destination_port; - } + ctx->user_ip4 = policy->destination_ip.ipv4; + bpf_printk("authorize_v4: Local/source ip is not set, redirect to loopback ip."); + ctx->user_port = policy->destination_port; + } - return BPF_SOCK_ADDR_VERDICT_PROCEED; - } + return BPF_SOCK_ADDR_VERDICT_PROCEED; +} - SEC("cgroup/connect4") - int connect4(struct bpf_sock_addr *ctx) - { - return authorize_v4(ctx); - } +SEC("cgroup/connect4") +int connect4(struct bpf_sock_addr *ctx) +{ + return authorize_v4(ctx); +} + +/// @brief Extract the IPv4 address from an IPv4-mapped IPv6 address. +/// @param ctx The socket address context containing the IPv6 address. +/// @param destination_ipv4 Pointer to store the extracted IPv4 address. +/// @return 1 if the address is IPv4-mapped, 0 otherwise. +static __always_inline int +get_ipv4_mapped_address(struct bpf_sock_addr *ctx, __be32 *destination_ipv4) +{ + if (ctx->user_ip6[0] != 0 || + ctx->user_ip6[1] != 0 || + ctx->user_ip6[2] != bpf_htonl(0x0000ffff)) + { + return 0; + } - /// @brief Extract the IPv4 address from an IPv4-mapped IPv6 address. - /// @param ctx The socket address context containing the IPv6 address. - /// @param destination_ipv4 Pointer to store the extracted IPv4 address. - /// @return 1 if the address is IPv4-mapped, 0 otherwise. - static __always_inline int - get_ipv4_mapped_address(struct bpf_sock_addr *ctx, __be32 *destination_ipv4) - { - if (ctx->user_ip6[0] != 0 || - ctx->user_ip6[1] != 0 || - ctx->user_ip6[2] != bpf_htonl(0x0000ffff)) - { - return 0; - } - - *destination_ipv4 = ctx->user_ip6[3]; - return 1; - } + *destination_ipv4 = ctx->user_ip6[3]; + return 1; +} - SEC("cgroup/connect6") - int connect6(struct bpf_sock_addr *ctx) - { - __be32 destination_ipv4; - if (get_ipv4_mapped_address(ctx, &destination_ipv4) == 0) - { - // Native IPv6 destinations are not supported yet and must remain unchanged. - return BPF_SOCK_ADDR_VERDICT_PROCEED; - } - - struct gpa_destination_entry entry = {0}; - entry.destination_ip.ipv4 = destination_ipv4; - entry.destination_port = ctx->user_port; - entry.protocol = ctx->protocol; - - struct gpa_destination_entry *policy = bpf_map_lookup_elem(&policy_map, &entry); - if (policy != NULL) - { - bpf_printk("connect6: Found IPv4-mapped proxy entry."); - // TODO: check bind to IPv4 mapped address, if so, skip the redirection and update the audit map. - __u32 audit_only = 0; - if (update_local_map_entry(ctx, audit_only, destination_ipv4, GPA_ADDRESS_FAMILY_IPV6) == 1) - { - bpf_printk("connect6: Found skip process entry, skip the redirection."); - return BPF_SOCK_ADDR_VERDICT_PROCEED; - } - - // Keep the socket in AF_INET6 and redirect it to IPv4-mapped loopback. - ctx->user_ip6[0] = 0; - ctx->user_ip6[1] = 0; - ctx->user_ip6[2] = bpf_htonl(0x0000ffff); - ctx->user_ip6[3] = policy->destination_ip.ipv4; - ctx->user_port = policy->destination_port; - } +SEC("cgroup/connect6") +int connect6(struct bpf_sock_addr *ctx) +{ + __be32 destination_ipv4; + if (get_ipv4_mapped_address(ctx, &destination_ipv4) == 0) + { + // Native IPv6 destinations are not supported yet and must remain unchanged. + return BPF_SOCK_ADDR_VERDICT_PROCEED; + } - return BPF_SOCK_ADDR_VERDICT_PROCEED; - } + struct gpa_destination_entry entry = {0}; + entry.destination_ip.ipv4 = destination_ipv4; + entry.destination_port = ctx->user_port; + entry.protocol = ctx->protocol; - static __always_inline int - update_audit_map_entry_sk(__u32 local_port, __u32 local_ipv4, struct gpa_sock_addr_local_entry *local_entry) + struct gpa_destination_entry *policy = bpf_map_lookup_elem(&policy_map, &entry); + if (policy != NULL) + { + bpf_printk("connect6: Found IPv4-mapped proxy entry."); + // TODO: check bind to IPv4 mapped address, if so, skip the redirection and update the audit map. + __u32 audit_only = 0; + if (update_local_map_entry(ctx, audit_only, destination_ipv4, GPA_ADDRESS_FAMILY_IPV6) == 1) { - struct gpa_audit_key key = {0}; - key.protocol = local_entry->protocol; - key.source_port = local_port; - - struct gpa_audit_event entry = {0}; - entry.process_id = local_entry->process_id; - entry.logon_id = local_entry->logon_id; - entry.is_root = local_entry->is_root; - entry.destination_ipv4 = local_entry->destination_ipv4; - entry.destination_port = local_entry->destination_port; - entry.address_family = local_entry->address_family; - - __u64 ret; - if (local_entry->audit_only) - { - struct gpa_audit_only_event event = {0}; - event.kernel_timestamp_ns = bpf_ktime_get_ns(); - event.local_ipv4 = local_ipv4; - event.audit = entry; - ret = bpf_ringbuf_output(&audit_only_map, &event, sizeof(event), 0); - } - else - { - ret = bpf_map_update_elem(&audit_map, &key, &entry, 0); - } - if (ret != 0) - { - bpf_printk("update_audit_map_entry_sk: Failed to update audit map entry with results:%u.", ret); - } - else - { - bpf_printk("update_audit_map_entry_sk: Updated audit map entry with local port:%u.", key.source_port); - } - - return 0; + bpf_printk("connect6: Found skip process entry, skip the redirection."); + return BPF_SOCK_ADDR_VERDICT_PROCEED; } - static __always_inline int - trace_tcp_connect(struct sock *sk) + // Keep the socket in AF_INET6 and redirect it to IPv4-mapped loopback. + ctx->user_ip6[0] = 0; + ctx->user_ip6[1] = 0; + ctx->user_ip6[2] = bpf_htonl(0x0000ffff); + ctx->user_ip6[3] = policy->destination_ip.ipv4; + ctx->user_port = policy->destination_port; + } + + return BPF_SOCK_ADDR_VERDICT_PROCEED; +} + +static __always_inline int +update_audit_map_entry_sk(__u32 local_port, __u32 local_ipv4, struct gpa_sock_addr_local_entry *local_entry) +{ + struct gpa_audit_key key = {0}; + key.protocol = local_entry->protocol; + key.source_port = local_port; + + struct gpa_audit_event entry = {0}; + entry.process_id = local_entry->process_id; + entry.logon_id = local_entry->logon_id; + entry.is_root = local_entry->is_root; + entry.destination_ipv4 = local_entry->destination_ipv4; + entry.destination_port = local_entry->destination_port; + entry.address_family = local_entry->address_family; + + __u64 ret; + if (local_entry->audit_only) + { + struct gpa_audit_only_event event = {0}; + event.kernel_timestamp_ns = bpf_ktime_get_ns(); + event.local_ipv4 = local_ipv4; + event.audit = entry; + ret = bpf_ringbuf_output(&audit_only_map, &event, sizeof(event), 0); + } + else + { + ret = bpf_map_update_elem(&audit_map, &key, &entry, 0); + } + if (ret != 0) + { + bpf_printk("update_audit_map_entry_sk: Failed to update audit map entry with results:%u.", ret); + } + else + { + bpf_printk("update_audit_map_entry_sk: Updated audit map entry with local port:%u.", key.source_port); + } + + return 0; +} + +static __always_inline int +trace_tcp_connect(struct sock *sk) +{ + // CO-RE relocatable reads of kernel struct sock fields. + // BPF_CORE_READ relocates each field offset on the KERNEL-side type + // (struct sock, which carries preserve_access_index in socket.h) to the + // running kernel's layout at load time. The destinations below are plain + // local scalars (no preserve_access_index), so their offsets are NOT + // relocated - this is required, otherwise the verifier rejects writes that + // would land outside our local stack copy. + __u16 skc_family = BPF_CORE_READ(sk, __sk_common.skc_family); + if (skc_family != AF_INET) + { + // Only support IPv4. + return 0; + } + __be32 skc_daddr = BPF_CORE_READ(sk, __sk_common.skc_daddr); + __be32 skc_rcv_saddr = BPF_CORE_READ(sk, __sk_common.skc_rcv_saddr); + __be16 skc_dport = BPF_CORE_READ(sk, __sk_common.skc_dport); + __u16 skc_num = BPF_CORE_READ(sk, __sk_common.skc_num); + + __u64 pid_tgid = bpf_get_current_pid_tgid(); + __u32 pid = (__u32)(pid_tgid >> 32); + if (check_skip_process_map_entry(pid) == 1) + { + bpf_printk("trace_tcp_connect: Found skip process entry %u, skip the trace.", pid); + return 0; + } + + // Find the entry in the local map. + struct gpa_sock_addr_local_entry *local_entry = bpf_map_lookup_elem(&local_map, &pid_tgid); + if (local_entry != NULL) + { + update_audit_map_entry_sk(skc_num, skc_rcv_saddr, local_entry); + __u64 ret = bpf_map_delete_elem(&local_map, &pid_tgid); + if (ret != 0) { - // CO-RE relocatable reads of kernel struct sock fields. - // BPF_CORE_READ relocates each field offset on the KERNEL-side type - // (struct sock, which carries preserve_access_index in socket.h) to the - // running kernel's layout at load time. The destinations below are plain - // local scalars (no preserve_access_index), so their offsets are NOT - // relocated - this is required, otherwise the verifier rejects writes that - // would land outside our local stack copy. - __u16 skc_family = BPF_CORE_READ(sk, __sk_common.skc_family); - if (skc_family != AF_INET) - { - // Only support IPv4. - return 0; - } - __be32 skc_daddr = BPF_CORE_READ(sk, __sk_common.skc_daddr); - __be32 skc_rcv_saddr = BPF_CORE_READ(sk, __sk_common.skc_rcv_saddr); - __be16 skc_dport = BPF_CORE_READ(sk, __sk_common.skc_dport); - __u16 skc_num = BPF_CORE_READ(sk, __sk_common.skc_num); - - __u64 pid_tgid = bpf_get_current_pid_tgid(); - __u32 pid = (__u32)(pid_tgid >> 32); - if (check_skip_process_map_entry(pid) == 1) - { - bpf_printk("trace_tcp_connect: Found skip process entry %u, skip the trace.", pid); - return 0; - } - - // Find the entry in the local map. - struct gpa_sock_addr_local_entry *local_entry = bpf_map_lookup_elem(&local_map, &pid_tgid); - if (local_entry != NULL) - { - update_audit_map_entry_sk(skc_num, skc_rcv_saddr, local_entry); - __u64 ret = bpf_map_delete_elem(&local_map, &pid_tgid); - if (ret != 0) - { - bpf_printk("trace_tcp_connect: Failed to delete local map entry with results:%u.", ret); - } - else - { - bpf_printk("trace_tcp_connect: Deleted local map entry with key:%u.", pid_tgid); - } - return 0; - } - - return 0; + bpf_printk("trace_tcp_connect: Failed to delete local map entry with results:%u.", ret); } - - SEC("kprobe/tcp_connect") // ELF program type/section metadata - int BPF_KPROBE(tcp_connect_probe, // eBPF program name used by Aya - struct sock *sk) + else { - return trace_tcp_connect(sk); + bpf_printk("trace_tcp_connect: Deleted local map entry with key:%u.", pid_tgid); } + return 0; + } + + return 0; +} + +SEC("kprobe/tcp_connect") // ELF program type/section metadata +int BPF_KPROBE(tcp_connect_probe, // eBPF program name used by Aya + struct sock *sk) +{ + return trace_tcp_connect(sk); +} - char _license[] SEC("license") = "GPL"; \ No newline at end of file +char _license[] SEC("license") = "GPL"; \ No newline at end of file diff --git a/proxy_agent/src/redirector/linux.rs b/proxy_agent/src/redirector/linux.rs index 77c4b52a..d0b7827e 100644 --- a/proxy_agent/src/redirector/linux.rs +++ b/proxy_agent/src/redirector/linux.rs @@ -679,7 +679,7 @@ mod tests { ) .unwrap(); audit_map - .insert(key.as_array(), value.as_array(), 0) + .insert(key.as_array(), value.to_array(), 0) .unwrap(); } let audit = bpf.lookup_audit(source_port); diff --git a/proxy_agent/src/redirector/shared_ebpf.rs b/proxy_agent/src/redirector/shared_ebpf.rs index 924e88ba..12e53e2e 100644 --- a/proxy_agent/src/redirector/shared_ebpf.rs +++ b/proxy_agent/src/redirector/shared_ebpf.rs @@ -500,7 +500,7 @@ mod tests { reserved: 0, }; - let rebuilt = sock_addr_audit_entry::from_array(canonical.as_array()); + let rebuilt = sock_addr_audit_entry::from_array(canonical.to_array()); assert_eq!(rebuilt.logon_id, canonical.logon_id); assert_eq!(rebuilt.process_id, canonical.process_id);