Feature process rule

leaf/Cargo.toml

@@ -246,6 +246,10 @@ warp = { version = "0.3", default-features = false, optional = true }
 # Auto reload
 notify = { version = "6", optional = true }
 
+# Process
+netstat2 = "0.11.0"
+sysinfo = "0.33.0"
+
 [target.'cfg(target_os = "android")'.dependencies]
 jni = "0.21"
 

leaf/src/app/mod.rs

@@ -24,6 +24,13 @@ pub mod api;
 ))]
 pub mod fake_dns;
 
+#[cfg(any(
+    target_os = "windows",
+    target_os = "macos",
+    target_os = "linux"
+))]
+pub mod process_finder;
+
 pub type SyncDnsClient = Arc<RwLock<dns_client::DnsClient>>;
 
 #[cfg(feature = "stat")]

leaf/src/app/process_finder.rs

@@ -0,0 +1,89 @@
+use std::net::IpAddr;
+use std::sync::{Arc, Mutex};
+
+use lazy_static::lazy_static;
+use netstat2::{get_sockets_info, AddressFamilyFlags, ProtocolFlags, SocketInfo};
+use sysinfo::{Pid, ProcessRefreshKind, ProcessesToUpdate, System, UpdateKind};
+
+lazy_static! {
+    static ref SYSTEM: Arc<Mutex<System>> = {
+        let system = System::new_all();
+        Arc::new(Mutex::new(system))
+    };
+    static ref CACHE: Mutex<Option<Vec<SocketInfo>>> = Mutex::new(None);
+}
+
+#[derive(Debug)]
+pub struct ProcessInfo {
+    pub name: String,
+    pub pid: u32,
+    pub process_path: String,
+}
+
+impl ProcessInfo {
+    fn from_socket_info(socket_info: SocketInfo) -> Option<Self> {
+        let pid = socket_info.associated_pids.first()?.to_owned();
+        let mut system = SYSTEM.lock().ok()?;
+        let the_pid = Pid::from(pid.to_owned() as usize);
+        let mut process = system.process(the_pid);
+        if process.is_none() {
+            system.refresh_processes_specifics(
+                ProcessesToUpdate::Some(&[the_pid]),
+                true,
+                ProcessRefreshKind::nothing().with_exe(UpdateKind::Always),
+            );
+            process = system.process(the_pid);
+        }
+        let process = process?;
+        let name = process.name().to_string_lossy().to_string();
+        let process_path = process.exe()?.to_string_lossy().to_string();
+        Some(ProcessInfo {
+            name,
+            pid,
+            process_path,
+        })
+    }
+}
+
+fn get_socket_info(protocol: &str, ip: &IpAddr, port: u16) -> Option<SocketInfo> {
+    let af_flags = match ip {
+        IpAddr::V4(_) => AddressFamilyFlags::IPV4,
+        IpAddr::V6(_) => AddressFamilyFlags::IPV6,
+    };
+
+    let proto_flags = match protocol {
+        "udp" => ProtocolFlags::UDP,
+        "tcp" => ProtocolFlags::TCP,
+        _ => ProtocolFlags::from_bits(0).unwrap(),
+    };
+
+    let mut cache = CACHE.lock().unwrap();
+    let sockets = cache.clone().unwrap_or_else(|| {
+        let new_sockets = get_sockets_info(af_flags, proto_flags).unwrap_or_default();
+        *cache = Some(new_sockets.clone());
+        new_sockets
+    });
+
+    sockets
+        .into_iter()
+        .find(|p| p.local_addr() == ip.to_owned() && p.local_port() == port)
+        .or_else(|| {
+            let new_sockets = get_sockets_info(af_flags, proto_flags).unwrap_or_default();
+            *cache = Some(new_sockets.clone());
+            new_sockets
+                .into_iter()
+                .find(|p| p.local_addr() == ip.to_owned() && p.local_port() == port)
+        })
+}
+
+pub fn find_pid(protocol: &str, ip: &IpAddr, port: u16) -> Option<u32> {
+    let socket_info = get_socket_info(protocol, ip, port);
+    socket_info.map(|s| s.associated_pids.first().unwrap().to_owned())
+}
+
+pub fn find_process(protocol: &str, ip: &IpAddr, port: u16) -> Option<ProcessInfo> {
+    let socket_info = get_socket_info(protocol, ip, port);
+    socket_info
+        .map(|s| ProcessInfo::from_socket_info(s))
+        .flatten()
+}

leaf/src/app/router.rs

@@ -9,6 +9,8 @@ use maxminddb::geoip2::Country;
 use maxminddb::Mmap;
 use tracing::{debug, warn};
 
+#[cfg(any(target_os = "windows", target_os = "macos", target_os = "linux"))]
+use crate::app::process_finder;
 use crate::app::SyncDnsClient;
 use crate::config;
 use crate::session::{Network, Session, SocksAddr};
@@ -354,6 +356,106 @@ impl Condition for DomainMatcher {
     }
 }
 
+#[cfg(any(target_os = "windows", target_os = "macos", target_os = "linux"))]
+struct ProcessPidMatcher {
+    value: String,
+}
+
+#[cfg(any(target_os = "windows", target_os = "macos", target_os = "linux"))]
+impl ProcessPidMatcher {
+    fn new(value: String) -> Self {
+        ProcessPidMatcher { value }
+    }
+}
+
+#[cfg(any(target_os = "windows", target_os = "macos", target_os = "linux"))]
+impl Condition for ProcessPidMatcher {
+    fn apply(&self, sess: &Session) -> bool {
+        let process_id = process_finder::find_pid(
+            &sess.network.to_string(),
+            &sess.source.ip(),
+            sess.source.port(),
+        );
+        if let Some(pid) = process_id {
+            if pid.to_string() == self.value {
+                debug!("{} matches process id [{}]", pid, self.value);
+                return true;
+            }
+        }
+        false
+    }
+}
+
+#[cfg(any(target_os = "windows", target_os = "macos", target_os = "linux"))]
+struct ProcessNameMatcher {
+    value: String,
+}
+
+#[cfg(any(target_os = "windows", target_os = "macos", target_os = "linux"))]
+impl ProcessNameMatcher {
+    fn new(value: String) -> Self {
+        ProcessNameMatcher { value }
+    }
+}
+
+#[cfg(any(target_os = "windows", target_os = "macos", target_os = "linux"))]
+impl Condition for ProcessNameMatcher {
+    fn apply(&self, sess: &Session) -> bool {
+        let process_info = process_finder::find_process(
+            &sess.network.to_string(),
+            &sess.source.ip(),
+            sess.source.port(),
+        );
+        if let Some(process) = process_info {
+            if process
+                .process_path
+                .to_lowercase()
+                .contains(&self.value.to_lowercase())
+            {
+                debug!(
+                    "[{}] matches process name [{}]",
+                    process.process_path, &self.value
+                );
+                return true;
+            }
+        }
+        false
+    }
+}
+
+#[cfg(any(target_os = "windows", target_os = "macos", target_os = "linux"))]
+struct ProcessMatcher {
+    condition: Box<dyn Condition>,
+}
+
+#[cfg(any(target_os = "windows", target_os = "macos", target_os = "linux"))]
+impl ProcessMatcher {
+    fn new(processes: &mut Vec<config::router::rule::Process>) -> Self {
+        let mut cond_or = ConditionOr::new();
+        for rr_process in processes.iter_mut() {
+            let filter = std::mem::take(&mut rr_process.value);
+            match rr_process.type_.unwrap() {
+                config::router::rule::process::Type::PID => {
+                    cond_or.add(Box::new(ProcessPidMatcher::new(filter)));
+                }
+                config::router::rule::process::Type::NAME => {
+                    cond_or.add(Box::new(ProcessNameMatcher::new(filter)));
+                }
+            }
+        }
+        ProcessMatcher {
+            condition: Box::new(cond_or),
+        }
+    }
+}
+
+#[cfg(any(target_os = "windows", target_os = "macos", target_os = "linux"))]
+impl Condition for ProcessMatcher {
+    fn apply(&self, sess: &Session) -> bool {
+        self.condition.apply(sess)
+    }
+}
+
 struct ConditionAnd {
     conditions: Vec<Box<dyn Condition>>,
 }
@@ -467,6 +569,13 @@ impl Router {
                 cond_and.add(Box::new(InboundTagMatcher::new(&mut rr.inbound_tags)));
             }
 
+            if rr.processes.len() > 0 {
+                #[cfg(any(target_os = "windows", target_os = "macos", target_os = "linux"))]
+                cond_and.add(Box::new(ProcessMatcher::new(&mut rr.processes)));
+                #[cfg(not(any(target_os = "windows", target_os = "macos", target_os = "linux")))]
+                warn!("The 'process' rule is not applicable to the current operating system");
+            }
+
             if cond_and.is_empty() {
                 warn!("empty rule at target {}", rr.target_tag);
                 continue;

leaf/src/config/conf/config.rs

@@ -594,7 +594,7 @@
                         group.cache_timeout = i;
                     }
                     "last-resort" => {
                         let i = if let Ok(i) = v.parse::<String>() {
                             Some(i)
                         } else {
                             None
@@ -643,7 +643,7 @@
                         group.delay_base = i;
                     }
                     "method" => {
                         let i = if let Ok(i) = v.parse::<String>() {
                             Some(i)
                         } else {
                             None
@@ -703,7 +703,8 @@
         rule.target = params[2].to_string();
 
         match rule.type_field.as_str() {
-            "IP-CIDR" | "DOMAIN" | "DOMAIN-SUFFIX" | "DOMAIN-KEYWORD" | "GEOIP" | "EXTERNAL"
+            "IP-CIDR" | "DOMAIN" | "DOMAIN-SUFFIX" | "DOMAIN-KEYWORD"
+            | "PROCESS-PID" | "PROCESS-NAME" | "GEOIP" | "EXTERNAL"
             | "PORT-RANGE" | "NETWORK" | "INBOUND-TAG" => {
                 rule.filter = Some(params[1].to_string());
             }
@@ -1410,6 +1411,20 @@
                     domain.value = ext_filter;
                     rule.domains.push(domain);
                 }
+                "PROCESS-PID" => {
+                    let mut process = internal::router::rule::Process::new();
+                    process.type_ =
+                        protobuf::EnumOrUnknown::new(internal::router::rule::process::Type::PID);
+                    process.value = ext_filter;
+                    rule.processes.push(process);
+                }
+                "PROCESS-NAME" => {
+                    let mut process = internal::router::rule::Process::new();
+                    process.type_ =
+                        protobuf::EnumOrUnknown::new(internal::router::rule::process::Type::NAME);
+                    process.value = ext_filter;
+                    rule.processes.push(process);
+                }
                 "GEOIP" => {
                     let mut mmdb = internal::router::rule::Mmdb::new();
 

leaf/src/config/internal/config.proto

@@ -249,13 +249,24 @@ message Router {
 			string country_code = 2;
 		}
 
+		message Process {
+			enum Type {
+				PID = 0;
+				NAME = 1;
+			}
+
+			Type type = 1;
+			string value = 2;
+		}
+
 		string target_tag = 1;
 		repeated Domain domains = 2;
 		repeated string ip_cidrs = 3;
 		repeated Mmdb mmdbs = 4;
 		repeated string port_ranges = 5;
 		repeated string networks = 6;
 		repeated string inbound_tags = 7;
+		repeated Process processes = 8;
 	}
 
 	repeated Rule rules = 1;