error resistance: restart capturing and live pushing automatically on errors

3 files changed, 144 insertions, 8 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 3d1ede7..8a74243 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -62,6 +62,12 @@ dependencies = [
 ]
 
 [[package]]
+name = "anyhow"
+version = "1.0.75"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a4668cab20f66d8d020e1fbc0ebe47217433c1b6c8f2040faf858554e394ace6"
+
+[[package]]
 name = "async-trait"
 version = "0.1.74"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -734,6 +740,82 @@ dependencies = [
 ]
 
 [[package]]
+name = "netlink-packet-core"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "72724faf704479d67b388da142b186f916188505e7e0b26719019c525882eda4"
+dependencies = [
+ "anyhow",
+ "byteorder",
+ "netlink-packet-utils",
+]
+
+[[package]]
+name = "netlink-packet-route"
+version = "0.17.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "053998cea5a306971f88580d0829e90f270f940befd7cf928da179d4187a5a66"
+dependencies = [
+ "anyhow",
+ "bitflags 1.3.2",
+ "byteorder",
+ "libc",
+ "netlink-packet-core",
+ "netlink-packet-utils",
+]
+
+[[package]]
+name = "netlink-packet-utils"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0ede8a08c71ad5a95cdd0e4e52facd37190977039a4704eb82a283f713747d34"
+dependencies = [
+ "anyhow",
+ "byteorder",
+ "paste",
+ "thiserror",
+]
+
+[[package]]
+name = "netlink-proto"
+version = "0.11.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "842c6770fc4bb33dd902f41829c61ef872b8e38de1405aa0b938b27b8fba12c3"
+dependencies = [
+ "bytes",
+ "futures",
+ "log",
+ "netlink-packet-core",
+ "netlink-sys",
+ "thiserror",
+ "tokio",
+]
+
+[[package]]
+name = "netlink-sys"
+version = "0.8.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6471bf08e7ac0135876a9581bf3217ef0333c191c128d34878079f42ee150411"
+dependencies = [
+ "bytes",
+ "futures",
+ "libc",
+ "log",
+ "tokio",
+]
+
+[[package]]
+name = "nix"
+version = "0.26.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "598beaf3cc6fdd9a5dfb1630c2800c7acd31df7aaf0f565796fba2b53ca1af1b"
+dependencies = [
+ "bitflags 1.3.2",
+ "cfg-if",
+ "libc",
+]
+
+[[package]]
 name = "num-bigint"
 version = "0.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -825,6 +907,12 @@ dependencies = [
 ]
 
 [[package]]
+name = "paste"
+version = "1.0.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "de3145af08024dea9fa9914f381a17b8fc6034dfb00f3a84013f7ff43f29ed4c"
+
+[[package]]
 name = "pbkdf2"
 version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1136,6 +1224,7 @@ dependencies = [
  "pcap",
  "pcap-file-tokio",
  "ringbuf",
+ "rsdsl_netlinklib",
  "russh",
  "russh-keys",
  "thiserror",
@@ -1143,6 +1232,38 @@ dependencies = [
 ]
 
 [[package]]
+name = "rsdsl_netlinklib"
+version = "0.4.5"
+source = "git+https://github.com/rsdsl/netlinklib.git#f63d077934d1edd0c804d2f9d3ad736f4baead25"
+dependencies = [
+ "futures",
+ "libc",
+ "netlink-packet-route",
+ "netlink-proto",
+ "rtnetlink",
+ "thiserror",
+ "tokio",
+]
+
+[[package]]
+name = "rtnetlink"
+version = "0.13.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7a552eb82d19f38c3beed3f786bd23aa434ceb9ac43ab44419ca6d67a7e186c0"
+dependencies = [
+ "futures",
+ "log",
+ "netlink-packet-core",
+ "netlink-packet-route",
+ "netlink-packet-utils",
+ "netlink-proto",
+ "netlink-sys",
+ "nix",
+ "thiserror",
+ "tokio",
+]
+
+[[package]]
 name = "russh"
 version = "0.40.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
diff --git a/Cargo.toml b/Cargo.toml
index 059eaba..ea1154d 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -12,6 +12,7 @@ futures = { version = "^0.3", default-features = false, features = ["std"] }
 pcap = { version = "1.1.0", features = ["capture-stream"] }
 pcap-file-tokio = "0.1.0"
 ringbuf = "0.3.3"
+rsdsl_netlinklib = { git = "https://github.com/rsdsl/netlinklib.git", version = "0.4.5", default-features = false, features = ["status"] }
 russh = "0.40.0"
 russh-keys = "^0.40"
 thiserror = "1.0"
diff --git a/src/main.rs b/src/main.rs
index 351ea69..056b904 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -15,6 +15,7 @@ use pcap::{Capture, Device, Packet, PacketCodec};
 use pcap_file_tokio::pcap::{PcapHeader, PcapPacket};
 use pcap_file_tokio::{Endianness, TsResolution};
 use ringbuf::{HeapRb, Rb};
+use rsdsl_netlinklib::Connection;
 use russh::server::{Auth, Handle, Msg, Session};
 use russh::{Channel, ChannelId, CryptoVec, MethodSet};
 use russh_keys::key::KeyPair;
@@ -206,7 +207,7 @@ async fn capture(
     Ok(())
 }
 
-async fn live_push(server: Server, mut live_rx: mpsc::UnboundedReceiver<Vec<u8>>) -> Result<()> {
+async fn live_push(server: Server, live_rx: &mut mpsc::UnboundedReceiver<Vec<u8>>) -> Result<()> {
     while let Some(packet) = live_rx.recv().await {
         let clients = server.clients.lock().await;
         for ((_, channel), session) in clients.iter() {
@@ -232,7 +233,7 @@ async fn main() -> Result<()> {
         ..Default::default()
     });
 
-    let (live_tx, live_rx) = mpsc::unbounded_channel();
+    let (live_tx, mut live_rx) = mpsc::unbounded_channel();
 
     let server = Server {
         clients: Arc::new(Mutex::new(HashMap::new())),
@@ -253,18 +254,31 @@ async fn main() -> Result<()> {
         let server2 = server.clone();
         let live_tx2 = live_tx.clone();
         tokio::spawn(async move {
-            match capture(device.into(), server2, live_tx2).await {
-                Ok(_) => {}
-                Err(e) => println!("[fail] capture on {}: {}", device, e),
+            let conn = Connection::new().await.expect("netlinklib connection");
+
+            loop {
+                println!("[info] wait for {}", device);
+                conn.link_wait_up(device.to_string())
+                    .await
+                    .expect("link waiting");
+
+                match capture(device.into(), server2.clone(), live_tx2.clone()).await {
+                    Ok(_) => {}
+                    Err(e) => println!("[fail] capture on {}: {}", device, e),
+                }
             }
         });
     }
 
     let server2 = server.clone();
     tokio::spawn(async move {
-        match live_push(server2, live_rx).await {
-            Ok(_) => {}
-            Err(e) => println!("[fail] live push: {}", e),
+        loop {
+            match live_push(server2.clone(), &mut live_rx).await {
+                Ok(_) => {}
+                Err(e) => println!("[fail] live push: {}", e),
+            }
+
+            tokio::time::sleep(Duration::from_secs(8)).await;
         }
     });