initial broken capture server

3 files changed, 206 insertions, 149 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 87b194f..afd29ef 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -62,12 +62,6 @@ 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"
@@ -75,7 +69,7 @@ checksum = "a66537f1bb974b254c98ed142ff995236e81b9d0fe4db0575f46612cb15eb0f9"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn",
+ "syn 2.0.39",
 ]
 
 [[package]]
@@ -175,6 +169,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
 
 [[package]]
+name = "byteorder_slice"
+version = "3.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0b294e30387378958e8bf8f4242131b930ea615ff81e8cac2440cea0a6013190"
+dependencies = [
+ "byteorder",
+]
+
+[[package]]
 name = "bytes"
 version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -315,7 +318,7 @@ checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn",
+ "syn 2.0.39",
 ]
 
 [[package]]
@@ -336,6 +339,17 @@ dependencies = [
 ]
 
 [[package]]
+name = "derive-into-owned"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2c9d94d81e3819a7b06a8638f448bc6339371ca9b6076a99d4a43eece3c4c923"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 1.0.109",
+]
+
+[[package]]
 name = "digest"
 version = "0.10.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -530,7 +544,7 @@ checksum = "53b153fd91e4b0147f4aced87be237c98248656bb01050b96bf3ee89220a8ddb"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn",
+ "syn 2.0.39",
 ]
 
 [[package]]
@@ -720,82 +734,6 @@ 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"
@@ -887,12 +825,6 @@ 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"
@@ -929,6 +861,32 @@ dependencies = [
 ]
 
 [[package]]
+name = "pcap-file"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1fc1f139757b058f9f37b76c48501799d12c9aa0aa4c0d4c980b062ee925d1b2"
+dependencies = [
+ "byteorder_slice",
+ "derive-into-owned",
+ "thiserror",
+]
+
+[[package]]
+name = "pcap-file-tokio"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3ee4f08e375f9aabbb17f4c031a2f0af1397835ce8d7909b167ada1dd8b572e6"
+dependencies = [
+ "async-trait",
+ "byteorder",
+ "derive-into-owned",
+ "pcap-file",
+ "thiserror",
+ "tokio",
+ "tokio-byteorder",
+]
+
+[[package]]
 name = "pem-rfc7468"
 version = "0.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1171,9 +1129,10 @@ name = "rsdsl_netdumpd"
 version = "0.1.0"
 dependencies = [
  "async-trait",
+ "byteorder",
  "pcap",
+ "pcap-file-tokio",
  "ringbuf",
- "rsdsl_netlinklib",
  "russh",
  "russh-keys",
  "thiserror",
@@ -1181,38 +1140,6 @@ 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"
@@ -1347,7 +1274,7 @@ checksum = "43576ca501357b9b071ac53cdc7da8ef0cbd9493d8df094cd821777ea6e894d3"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn",
+ "syn 2.0.39",
 ]
 
 [[package]]
@@ -1428,6 +1355,17 @@ checksum = "81cdd64d312baedb58e21336b31bc043b77e01cc99033ce76ef539f78e965ebc"
 
 [[package]]
 name = "syn"
+version = "1.0.109"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "unicode-ident",
+]
+
+[[package]]
+name = "syn"
 version = "2.0.39"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "23e78b90f2fcf45d3e842032ce32e3f2d1545ba6636271dcbf24fa306d87be7a"
@@ -1454,7 +1392,7 @@ checksum = "266b2e40bc00e5a6c09c3584011e08b06f123c00362c92b975ba9843aaaa14b8"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn",
+ "syn 2.0.39",
 ]
 
 [[package]]
@@ -1476,6 +1414,16 @@ dependencies = [
 ]
 
 [[package]]
+name = "tokio-byteorder"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5cf347e8ae1d1ffd16c8aed569172a71bd81098a001d0f4964d476c0097aba4a"
+dependencies = [
+ "byteorder",
+ "tokio",
+]
+
+[[package]]
 name = "tokio-macros"
 version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1483,7 +1431,7 @@ checksum = "5b8a1e28f2deaa14e508979454cb3a223b10b938b45af148bc0986de36f1923b"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn",
+ "syn 2.0.39",
 ]
 
 [[package]]
diff --git a/Cargo.toml b/Cargo.toml
index d893d39..5ba6621 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -7,9 +7,10 @@ edition = "2021"
 
 [dependencies]
 async-trait = "^0.1"
+byteorder = "^1.4.3"
 pcap = "1.1.0"
+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 = ["blocking", "status"] }
 russh = "0.40.0"
 russh-keys = "^0.40"
 thiserror = "1.0"
diff --git a/src/main.rs b/src/main.rs
index 1222b55..be11646 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,19 +1,19 @@
 use std::array;
 use std::collections::HashMap;
 use std::fs;
-use std::io::{self, Read};
+use std::io;
 use std::net::SocketAddr;
-use std::os::fd::{AsRawFd, RawFd};
-use std::sync::{mpsc, Arc};
-use std::thread;
-use std::time::{Duration, Instant};
+use std::sync::Arc;
 
-use tokio::sync::Mutex;
+use tokio::sync::{mpsc, Mutex};
+use tokio::time::Duration;
 
 use async_trait::async_trait;
-use pcap::Capture;
+use byteorder::LittleEndian;
+use pcap::{Capture, Device};
+use pcap_file_tokio::pcap::{PcapHeader, PcapPacket};
+use pcap_file_tokio::{Endianness, TsResolution};
 use ringbuf::{HeapRb, Rb};
-use rsdsl_netlinklib::blocking::Connection;
 use russh::server::{Auth, Handle, Msg, Session};
 use russh::{Channel, ChannelId, CryptoVec, MethodSet};
 use russh_keys::key::KeyPair;
@@ -23,33 +23,25 @@ use thiserror::Error;
 enum Error {
     #[error("io error: {0}")]
     Io(#[from] io::Error),
-    #[error("can't receive from mpsc channel: {0}")]
-    MpscRecv(#[from] mpsc::RecvError),
-    #[error("can't send Vec<u8> to mpsc channel")]
-    MpscSendU8Vec,
     #[error("can't convert slice to array: {0}")]
     ArrayTryFromSlice(#[from] array::TryFromSliceError),
 
-    #[error("netlinklib error: {0}")]
-    Netlinklib(#[from] rsdsl_netlinklib::Error),
     #[error("pcap error: {0}")]
     Pcap(#[from] pcap::Error),
+    #[error("pcap_file_tokio error: {0}")]
+    PcapFileTokio(#[from] pcap_file_tokio::PcapError),
     #[error("russh error: {0}")]
     Russh(#[from] russh::Error),
 }
 
-impl From<mpsc::SendError<Vec<u8>>> for Error {
-    fn from(_: mpsc::SendError<Vec<u8>>) -> Self {
-        Self::MpscSendU8Vec
-    }
-}
-
 type Result<T> = std::result::Result<T, Error>;
 
 #[derive(Clone)]
 struct Server {
     clients: Arc<Mutex<HashMap<(usize, ChannelId), Handle>>>,
     id: usize,
+
+    packets: Arc<Mutex<HeapRb<Vec<u8>>>>,
 }
 
 impl russh::server::Server for Server {
@@ -71,6 +63,8 @@ impl russh::server::Handler for Server {
         channel: Channel<Msg>,
         session: Session,
     ) -> Result<(Self, bool, Session)> {
+        println!("[info] [{}] open session", channel.id());
+
         {
             let mut clients = self.clients.lock().await;
             clients.insert((self.id, channel.id()), session.handle());
@@ -83,8 +77,10 @@ impl russh::server::Handler for Server {
         let correct_password = fs::read("/data/admind.passwd")?;
 
         if user == "rustkrazy" && password.as_bytes() == correct_password {
+            println!("[info] auth ok");
             Ok((self, Auth::Accept))
         } else {
+            println!("[warn] auth err");
             Ok((
                 self,
                 Auth::Reject {
@@ -93,10 +89,91 @@ impl russh::server::Handler for Server {
             ))
         }
     }
+
+    async fn exec_request(
+        self,
+        channel: ChannelId,
+        _: &[u8],
+        mut session: Session,
+    ) -> Result<(Self, Session)> {
+        println!("[info] [{}] exec", channel);
+
+        let header = PcapHeader {
+            endianness: Endianness::Little,
+            ..Default::default()
+        };
+
+        let mut buf = Vec::new();
+        header.write_to(&mut buf).await?;
+
+        let s = session.handle();
+        let _ = s.data(channel, CryptoVec::from(buf)).await;
+
+        {
+            let packets = self.packets.lock().await;
+            for packet in packets.iter() {
+                let _ = s.data(channel, CryptoVec::from(packet.clone())).await;
+            }
+        }
+
+        Ok((self, session))
+    }
+
+    async fn channel_close(self, channel: ChannelId, session: Session) -> Result<(Self, Session)> {
+        println!("[info] [{}] close session", channel);
+        Ok((self, session))
+    }
+}
+
+async fn capture(
+    dev: Device,
+    server: Server,
+    live_tx: mpsc::UnboundedSender<Vec<u8>>,
+) -> Result<()> {
+    let mut cap = Capture::from_device(dev)?.immediate_mode(true).open()?;
+    loop {
+        let packet = cap.next_packet()?;
+
+        {
+            let pcap_packet = PcapPacket::new(
+                Duration::new(
+                    packet.header.ts.tv_sec as u64,
+                    (packet.header.ts.tv_usec * 1000) as u32,
+                ),
+                packet.header.len,
+                packet.data,
+            );
+
+            let mut buf = Vec::new();
+            pcap_packet
+                .write_to::<_, LittleEndian>(&mut buf, TsResolution::MicroSecond, 65535)
+                .await?;
+
+            let mut packets = server.packets.lock().await;
+            packets.push_overwrite(buf.clone());
+
+            let _ = live_tx.send(buf);
+        }
+    }
+}
+
+async fn live_push(server: Server, mut live_rx: 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() {
+            let _ = session
+                .data(*channel, CryptoVec::from(packet.clone()))
+                .await;
+        }
+    }
+
+    Ok(())
 }
 
 #[tokio::main]
 async fn main() -> Result<()> {
+    println!("[info] init");
+
     let config = Arc::new(russh::server::Config {
         methods: MethodSet::PASSWORD,
         inactivity_timeout: Some(Duration::from_secs(3600)),
@@ -106,11 +183,42 @@ async fn main() -> Result<()> {
         ..Default::default()
     });
 
+    let (live_tx, live_rx) = mpsc::unbounded_channel();
+
     let server = Server {
         clients: Arc::new(Mutex::new(HashMap::new())),
         id: 0,
+
+        packets: Arc::new(Mutex::new(HeapRb::new(64000))),
     };
 
-    russh::server::run(config, "[::]:2222", server).await?;
+    let devs = ["wlan0"];
+
+    for dev in devs {
+        if dev == "any" {
+            continue;
+        }
+
+        println!("[info] capture on {}", dev);
+
+        let server2 = server.clone();
+        let live_tx2 = live_tx.clone();
+        tokio::spawn(async move {
+            match capture(dev.into(), server2, live_tx2).await {
+                Ok(_) => {}
+                Err(e) => println!("[fail] capture on {}: {}", dev, e),
+            }
+        });
+    }
+
+    let server2 = server.clone();
+    tokio::spawn(async move {
+        match live_push(server2, live_rx).await {
+            Ok(_) => {}
+            Err(e) => println!("[fail] live push: {}", e),
+        }
+    });
+
+    russh::server::run(config, "[::]:22", server).await?;
     Ok(())
 }