Add: dns leak detection

crates/wtfnet-platform-linux/src/lib.rs

@@ -2,10 +2,12 @@ use async_trait::async_trait;
 use network_interface::{Addr, NetworkInterface, NetworkInterfaceConfig};
 use sha2::Digest;
 use std::collections::HashMap;
+use std::net::{IpAddr, SocketAddr};
 use std::sync::Arc;
 use wtfnet_core::ErrorCode;
 use wtfnet_platform::{
-    CertProvider, ConnSocket, DnsConfigSnapshot, ListenSocket, NeighborEntry, NeighProvider,
+    CertProvider, ConnSocket, DnsConfigSnapshot, FlowOwner, FlowOwnerConfidence, FlowOwnerProvider,
+    FlowOwnerResult, FlowProtocol, FlowTuple, ListenSocket, NeighborEntry, NeighProvider,
     NetInterface, Platform, PlatformError, PortsProvider, RootCert, RouteEntry, SysProvider,
 };
 use x509_parser::oid_registry::{
@@ -19,6 +21,7 @@ pub fn platform() -> Platform {
         ports: Arc::new(LinuxPortsProvider),
         cert: Arc::new(LinuxCertProvider),
         neigh: Arc::new(LinuxNeighProvider),
+        flow_owner: Arc::new(LinuxFlowOwnerProvider),
     }
 }
 
@@ -26,6 +29,7 @@ struct LinuxSysProvider;
 struct LinuxPortsProvider;
 struct LinuxCertProvider;
 struct LinuxNeighProvider;
+struct LinuxFlowOwnerProvider;
 
 #[async_trait]
 impl SysProvider for LinuxSysProvider {
@@ -375,6 +379,20 @@ fn parse_proc_socket_addr(value: &str, is_v6: bool) -> Option<String> {
     }
 }
 
+fn parse_proc_socket_addr_value(value: &str, is_v6: bool) -> Option<SocketAddr> {
+    let mut parts = value.split(':');
+    let addr_hex = parts.next()?;
+    let port_hex = parts.next()?;
+    let port = u16::from_str_radix(port_hex, 16).ok()?;
+    if is_v6 {
+        let addr = parse_ipv6_hex(addr_hex)?;
+        Some(SocketAddr::new(IpAddr::V6(addr), port))
+    } else {
+        let addr = parse_ipv4_hex(addr_hex)?;
+        Some(SocketAddr::new(IpAddr::V4(addr), port))
+    }
+}
+
 fn parse_linux_arp(contents: &str) -> Vec<NeighborEntry> {
     let mut neighbors = Vec::new();
     for (idx, line) in contents.lines().enumerate() {
@@ -482,6 +500,138 @@ fn read_ppid(pid: u32) -> Option<u32> {
     Some(ppid)
 }
 
+#[derive(Clone)]
+struct ProcSocketEntry {
+    local: SocketAddr,
+    remote: SocketAddr,
+    inode: String,
+}
+
+fn parse_proc_socket_entries(
+    path: &str,
+    is_v6: bool,
+) -> Result<Vec<ProcSocketEntry>, PlatformError> {
+    let contents = std::fs::read_to_string(path)
+        .map_err(|err| PlatformError::new(ErrorCode::IoError, err.to_string()))?;
+    let mut entries = Vec::new();
+    for (idx, line) in contents.lines().enumerate() {
+        if idx == 0 {
+            continue;
+        }
+        let parts: Vec<&str> = line.split_whitespace().collect();
+        if parts.len() < 10 {
+            continue;
+        }
+        let local = parts[1];
+        let remote = parts[2];
+        let inode = match parts.get(9) {
+            Some(value) => (*value).to_string(),
+            None => continue,
+        };
+        let local_addr = match parse_proc_socket_addr_value(local, is_v6) {
+            Some(addr) => addr,
+            None => continue,
+        };
+        let remote_addr = match parse_proc_socket_addr_value(remote, is_v6) {
+            Some(addr) => addr,
+            None => continue,
+        };
+        entries.push(ProcSocketEntry {
+            local: local_addr,
+            remote: remote_addr,
+            inode,
+        });
+    }
+    Ok(entries)
+}
+
+fn match_flow_entry<'a>(
+    flow: &FlowTuple,
+    entries: &'a [ProcSocketEntry],
+    match_remote: bool,
+) -> Option<(&'a ProcSocketEntry, FlowOwnerConfidence)> {
+    for entry in entries {
+        let local_match = entry.local.port() == flow.src_port
+            && (entry.local.ip() == flow.src_ip
+                || entry.local.ip().is_unspecified()
+                || entry.local.ip().is_loopback() && flow.src_ip.is_loopback());
+        if !local_match {
+            continue;
+        }
+        if match_remote {
+            let remote_match = entry.remote.port() == flow.dst_port
+                && (entry.remote.ip() == flow.dst_ip
+                    || entry.remote.ip().is_unspecified());
+            if remote_match {
+                return Some((entry, FlowOwnerConfidence::High));
+            }
+        } else {
+            return Some((entry, FlowOwnerConfidence::Medium));
+        }
+    }
+    None
+}
+
+fn resolve_flow_owner(
+    flow: &FlowTuple,
+) -> Result<FlowOwnerResult, PlatformError> {
+    let inode_map = build_inode_map();
+    let entries = match flow.proto {
+        FlowProtocol::Tcp => {
+            let mut out = parse_proc_socket_entries("/proc/net/tcp", false)?;
+            out.extend(parse_proc_socket_entries("/proc/net/tcp6", true)?);
+            out
+        }
+        FlowProtocol::Udp => {
+            let mut out = parse_proc_socket_entries("/proc/net/udp", false)?;
+            out.extend(parse_proc_socket_entries("/proc/net/udp6", true)?);
+            out
+        }
+    };
+
+    let match_remote = matches!(flow.proto, FlowProtocol::Tcp);
+    let matched = match_flow_entry(flow, &entries, match_remote)
+        .or_else(|| {
+            if matches!(flow.proto, FlowProtocol::Udp) {
+                match_flow_entry(flow, &entries, false)
+            } else {
+                None
+            }
+        });
+
+    let (entry, confidence) = match matched {
+        Some(value) => value,
+        None => {
+            return Ok(FlowOwnerResult {
+                owner: None,
+                confidence: FlowOwnerConfidence::None,
+                failure_reason: Some("no socket match".to_string()),
+            })
+        }
+    };
+
+    let owner = inode_map.get(&entry.inode).map(|info| FlowOwner {
+        pid: Some(info.pid),
+        ppid: info.ppid,
+        process_name: info.name.clone(),
+        process_path: info.path.clone(),
+    });
+
+    if owner.is_none() {
+        return Ok(FlowOwnerResult {
+            owner: None,
+            confidence: FlowOwnerConfidence::Low,
+            failure_reason: Some("socket owner not found".to_string()),
+        });
+    }
+
+    Ok(FlowOwnerResult {
+        owner,
+        confidence,
+        failure_reason: None,
+    })
+}
+
 fn load_native_roots(store: &str) -> Result<Vec<RootCert>, PlatformError> {
     let certs = rustls_native_certs::load_native_certs()
         .map_err(|err| PlatformError::new(ErrorCode::IoError, err.to_string()))?;
@@ -626,3 +776,10 @@ impl NeighProvider for LinuxNeighProvider {
         Ok(parse_linux_arp(&contents))
     }
 }
+
+#[async_trait]
+impl FlowOwnerProvider for LinuxFlowOwnerProvider {
+    async fn owner_of(&self, flow: FlowTuple) -> Result<FlowOwnerResult, PlatformError> {
+        resolve_flow_owner(&flow)
+    }
+}