Scan IP neighbors on Windows

main/Cargo.toml

@@ -7,14 +7,16 @@ version.workspace = true
 
 [dependencies]
 anyhow = "1"
+csv = "1"
 futures-channel = "0.3"
 futures-util = "0.3"
 log = "0.4"
 mockall = "0.12"
 prost = "0.12"
+serde = { version = "1.0", features = ["derive"] }
 tansa-protocol = { path = "../protocol/rust" }
 thiserror = "1"
-tokio = { version = "1", features = ["fs"] }
+tokio = { version = "1", features = ["fs", "process"] }
 tokio-stream = { version = "0.1", features = ["net"] }
 tokio-util = { version = "0.7", features = ["codec", "net"]}
 tonic = "0.11"

main/src/lib.rs

@@ -31,6 +31,7 @@
 mod network;
 mod os;
 mod packet;
+mod process;
 mod response_collector;
 mod response_sender;
 mod scanner;

main/src/network/link_local/Print-IpNeighbors.ps1

@@ -0,0 +1 @@
+Get-NetNeighbor -AddressFamily IPv6 | Select-Object -Property InterfaceIndex, IPAddress, State | ConvertTo-Csv

main/src/network/link_local/mod.rs

@@ -1,4 +1,8 @@
+mod windows;
+
+use self::windows::PowerShellIpNeighborScanner;
 use crate::os::OperatingSystem;
+use crate::process::ProcessError;
 use futures_util::future::BoxFuture;
 use futures_util::FutureExt;
 use mockall::automock;
@@ -8,7 +12,7 @@ use thiserror::Error;
 
 pub async fn ip_neighbor_scanner() -> Box<dyn IpNeighborScanner> {
     match crate::os::detect_operating_system().await {
-        Ok(OperatingSystem::Windows) => Box::new(DummyIpNeighborScanner), // TODO
+        Ok(OperatingSystem::Windows) => Box::new(PowerShellIpNeighborScanner),
         Err(e) => {
             log::warn!("Failed to detect operating system: {}", e);
             log::info!("Unknown operating system, disabling IP neighbor discovery.");
@@ -18,7 +22,13 @@ pub async fn ip_neighbor_scanner() -> Box<dyn IpNeighborScanner> {
 }
 
 #[derive(Error, Debug)]
-pub enum IpNeighborScanError {}
+pub enum IpNeighborScanError {
+    #[error("Failed in running an external command")]
+    ChildProcess(#[from] ProcessError),
+
+    #[error("Failed to parse the CSV output of a child process")]
+    ChildProcessCsvOutput(#[from] csv::Error),
+}
 
 #[automock]
 pub trait IpNeighborScanner {
@@ -33,6 +43,7 @@ impl IpNeighborScanner for DummyIpNeighborScanner {
     }
 }
 
+#[derive(Debug, PartialEq, Eq)]
 pub struct IpNeighbor {
     pub address: Ipv6Addr,
     pub network_interface_index: u32,

main/src/network/link_local/windows.rs

@@ -0,0 +1,114 @@
+use super::IpNeighbor;
+use super::IpNeighborScanError;
+use super::IpNeighborScanner;
+use csv::Reader;
+use futures_util::future::BoxFuture;
+use futures_util::FutureExt;
+use serde::Deserialize;
+use std::net::Ipv6Addr;
+
+pub struct PowerShellIpNeighborScanner;
+
+impl PowerShellIpNeighborScanner {
+    async fn scan() -> Result<Vec<IpNeighbor>, IpNeighborScanError> {
+        let stdout = crate::process::eval(
+            "pwsh",
+            &["-NonInteractive", "-Command", "-"],
+            include_bytes!("./Print-IpNeighbors.ps1"),
+        )
+        .await?;
+        Self::parse_output(&stdout).map_err(Into::into)
+    }
+
+    fn parse_output(output: &[u8]) -> Result<Vec<IpNeighbor>, csv::Error> {
+        let neighbors = Reader::from_reader(output)
+            .deserialize()
+            .collect::<Result<Vec<NetNeighbor>, _>>()?;
+        neighbors
+            .iter()
+            .for_each(|n| log::debug!("Scanned IP neighbor: {:?}", n));
+
+        let neighbors: Vec<_> = neighbors
+            .into_iter()
+            .filter(|n| n.State != "Unreachable")
+            .filter(|n| n.IPAddress.segments().starts_with(&[0xFE80, 0, 0, 0]))
+            .map(Into::into)
+            .collect();
+        neighbors
+            .iter()
+            .for_each(|n| log::info!("Valid IP neighbor: {:?}", n));
+
+        Ok(neighbors)
+    }
+}
+
+impl IpNeighborScanner for PowerShellIpNeighborScanner {
+    fn scan(&self) -> BoxFuture<'static, Result<Vec<IpNeighbor>, IpNeighborScanError>> {
+        Self::scan().boxed()
+    }
+}
+
+#[derive(Deserialize, Debug)]
+#[allow(non_snake_case)]
+struct NetNeighbor {
+    InterfaceIndex: u32,
+    IPAddress: Ipv6Addr,
+    State: String,
+}
+
+impl From<NetNeighbor> for IpNeighbor {
+    fn from(value: NetNeighbor) -> Self {
+        Self {
+            address: value.IPAddress,
+            network_interface_index: value.InterfaceIndex,
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use crate::os::OperatingSystem;
+
+    #[tokio::test]
+    async fn scan() {
+        crate::test::init();
+
+        if crate::os::detect_operating_system().await.unwrap() != OperatingSystem::Windows {
+            println!("Operating system is not Windows, skipping.");
+            return;
+        }
+
+        let neighbors = PowerShellIpNeighborScanner.scan().await.unwrap();
+        assert!(!neighbors.is_empty());
+    }
+
+    #[test]
+    fn parse_output() {
+        let output = r#"
+"InterfaceIndex","IPAddress","State"
+"1","fe80::1:abcd","Reachable"
+"2","fe80::2:abcd","Permanent"
+"3","fe80::3:abcd","Unreachable"
+"4","ff02::4:abcd","Reachable"
+        "#
+        .trim();
+        let expected_neighbors = vec![
+            IpNeighbor {
+                network_interface_index: 1,
+                address: "fe80::1:abcd".parse().unwrap(),
+            },
+            IpNeighbor {
+                network_interface_index: 2,
+                address: "fe80::2:abcd".parse().unwrap(),
+            },
+        ];
+
+        // When
+        let actual_neighbors =
+            PowerShellIpNeighborScanner::parse_output(output.as_bytes()).unwrap();
+
+        // Then
+        assert_eq!(actual_neighbors, expected_neighbors);
+    }
+}

main/src/process.rs

@@ -0,0 +1,46 @@
+use std::process::Stdio;
+use thiserror::Error;
+use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::process::Command;
+
+pub async fn eval(command: &str, args: &[&str], stdin: &[u8]) -> Result<Vec<u8>, ProcessError> {
+    let mut process = Command::new(command)
+        .args(args)
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::inherit())
+        .spawn()?;
+
+    let mut stdin_pipe = process
+        .stdin
+        .take()
+        .ok_or_else(|| ProcessError::StdioRedirection)?;
+    stdin_pipe.write_all(stdin).await?;
+    drop(stdin_pipe);
+
+    if !process.wait().await?.success() {
+        return Err(ProcessError::ExternalCommand);
+    }
+
+    let mut stdout = process
+        .stdout
+        .take()
+        .ok_or_else(|| ProcessError::StdioRedirection)?;
+
+    let mut buffer = Default::default();
+    stdout.read_to_end(&mut buffer).await?;
+    Ok(buffer)
+}
+
+#[derive(Error, Debug)]
+pub enum ProcessError {
+    #[error("Failed in create a child process")]
+    ChildProcessCreation(#[from] std::io::Error),
+
+    #[error("Failed to redirect standard I/O")]
+    StdioRedirection,
+
+    #[error("External command failed")]
+    ExternalCommand,
+}