conf: support process rules

leaf/Cargo.toml

@@ -232,6 +232,10 @@ warp = { version = "0.3", default-features = false, optional = true }
 # Auto reload
 notify = { version = "5.0.0-pre.13", optional = true }
 
+# Process
+netstat2 = "0.9.1"
+sysinfo = "0.28.4"
+
 [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,76 @@
+use std::net;
+use std::net::IpAddr;
+use log::debug;
+use netstat2::{AddressFamilyFlags, get_sockets_info, ProtocolFlags, ProtocolSocketInfo, TcpState};
+use sysinfo::{ProcessExt, System, SystemExt, Pid, PidExt};
+
+#[derive(Debug)]
+pub struct PortInfo {
+    pub address: net::IpAddr,
+    pub port: u16,
+    pub protocol: String,
+    pub process_info: Option<ProcessInfo>,
+}
+
+#[derive(Debug)]
+pub struct ProcessInfo {
+    pub name: String,
+    pub pid: u32,
+    pub process_path: String,
+}
+
+impl From<netstat2::SocketInfo> for PortInfo {
+    fn from(socket_info: netstat2::SocketInfo) -> Self {
+        let protocol = match socket_info.protocol_socket_info {
+            ProtocolSocketInfo::Tcp(_) => "TCP",
+            ProtocolSocketInfo::Udp(_) => "UDP",
+        };
+        let system = System::new_all();
+        // system.refresh_system();
+        let pid = socket_info.associated_pids.first().unwrap();
+        let process_info = system
+            .process(Pid::from(pid.to_owned() as usize))
+            .map(|p| ProcessInfo {
+                name: p.name().to_owned(),
+                pid: p.pid().as_u32(),
+                process_path: p.exe().to_string_lossy().to_string(),
+            });
+        Self {
+            address: socket_info.local_addr(),
+            port: socket_info.local_port(),
+            protocol: protocol.to_string(),
+            process_info,
+        }
+    }
+}
+
+pub fn find_process(protocol: &str, ip: IpAddr, port: u16) -> Option<PortInfo> {
+    let mut af_flags: AddressFamilyFlags = AddressFamilyFlags::from_bits(0).unwrap();
+    if ip.is_ipv6() {
+        af_flags |= AddressFamilyFlags::IPV6;
+    }
+    if ip.is_ipv4() {
+        af_flags |= AddressFamilyFlags::IPV4;
+    }
+
+    let mut proto_flags: ProtocolFlags = ProtocolFlags::from_bits(0).unwrap();
+    if protocol == "udp" {
+        proto_flags |= ProtocolFlags::UDP;
+    }
+    if protocol == "tcp" {
+        proto_flags |= ProtocolFlags::TCP;
+    }
+    let sockets = get_sockets_info(af_flags, proto_flags).unwrap_or_default();
+    let mut ports = sockets
+        .into_iter()
+        .filter(|socket_info| match &socket_info.protocol_socket_info {
+            ProtocolSocketInfo::Tcp(tcp) => tcp.state != TcpState::Closed,
+            ProtocolSocketInfo::Udp(_) => true,
+        })
+        .map(|socket_info| PortInfo::from(socket_info));
+    let port = ports.find(|p| p.port == port);
+    if let Some(ref p) = port {
+        debug!("find process port {:?}", p);
+    }
+    return port;
+}

leaf/src/app/router.rs

@@ -13,6 +13,13 @@ use crate::app::SyncDnsClient;
 use crate::config;
 use crate::session::{Network, Session, SocksAddr};
 
+#[cfg(any(
+    target_os = "windows",
+    target_os = "macos",
+    target_os = "linux"
+))]
+use crate::app::process_finder;
+
 pub trait Condition: Send + Sync + Unpin {
     fn apply(&self, sess: &Session) -> bool;
 }
@@ -354,6 +361,133 @@ 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 port_info = process_finder::find_process(&sess.network.to_string(),
+                                                     sess.source.ip(), sess.source.port());
+        if let Some(port) = port_info {
+            if let Some(process_info) = port.process_info {
+                if process_info.pid.to_string() == self.value {
+                    debug!("{} matches process id [{}]", process_info.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 port_info = process_finder::find_process(&sess.network.to_string(),
+                                                     sess.source.ip(), sess.source.port());
+        if let Some(port) = port_info {
+            if let Some(info) = port.process_info {
+                if info.process_path.to_lowercase().contains(&self.value.to_lowercase()) {
+                    debug!("[{}] matches process name [{}]", info.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 +601,15 @@ impl Router {
                 cond_and.add(Box::new(InboundTagMatcher::new(&mut rr.inbound_tags)));
             }
 
+            #[cfg(any(
+                target_os = "windows",
+                target_os = "macos",
+                target_os = "linux"
+            ))]
+            if rr.processes.len() > 0 {
+                cond_and.add(Box::new(ProcessMatcher::new(&mut rr.processes)));
+            }
+
             if cond_and.is_empty() {
                 warn!("empty rule at target {}", rr.target_tag);
                 continue;

leaf/src/config/json/config.rs

@@ -232,6 +232,10 @@ pub struct Rule {
     pub domain_keyword: Option<Vec<String>>,
     #[serde(rename = "domainSuffix")]
     pub domain_suffix: Option<Vec<String>>,
+    #[serde(rename = "processPid")]
+    pub process_pid: Option<Vec<String>>,
+    #[serde(rename = "processName")]
+    pub process_name: Option<Vec<String>>,
     pub geoip: Option<Vec<String>>,
     pub external: Option<Vec<String>>,
     #[serde(rename = "portRange")]
@@ -967,6 +971,26 @@ pub fn to_internal(json: &mut Config) -> Result<internal::Config> {
                         rule.domains.push(domain);
                     }
                 }
+                if let Some(ext_process_pids) = ext_rule.process_pid.as_mut() {
+                    for ext_process_pid in ext_process_pids.drain(0..) {
+                        let mut process = internal::router::rule::Process::new();
+                        process.type_ = protobuf::EnumOrUnknown::new(
+                            internal::router::rule::process::Type::PID,
+                        );
+                        process.value = ext_process_pid;
+                        rule.processes.push(process);
+                    }
+                }
+                if let Some(ext_process_names) = ext_rule.process_name.as_mut() {
+                    for ext_process_name in ext_process_names.drain(0..) {
+                        let mut process = internal::router::rule::Process::new();
+                        process.type_ = protobuf::EnumOrUnknown::new(
+                            internal::router::rule::process::Type::NAME,
+                        );
+                        process.value = ext_process_name;
+                        rule.processes.push(process);
+                    }
+                }
                 if let Some(ext_geoips) = ext_rule.geoip.as_mut() {
                     for ext_geoip in ext_geoips.drain(0..) {
                         let mut mmdb = internal::router::rule::Mmdb::new();