From 2c50a98e63087566b73bbfafc0cbadfb5374867a Mon Sep 17 00:00:00 2001 From: Ishan Jain Date: Sun, 30 Jul 2023 20:18:17 +0530 Subject: [PATCH] restructured project --- Cargo.lock | 140 +----------------------------------------- Cargo.toml | 1 - config.toml | 21 ++++--- src/communications.rs | 47 -------------- src/main.rs | 30 ++------- src/mdns.rs | 128 ++++++++++++++++++++++++++++++++++++++ src/processor.rs | 112 --------------------------------- 7 files changed, 144 insertions(+), 335 deletions(-) delete mode 100644 src/communications.rs create mode 100644 src/mdns.rs delete mode 100644 src/processor.rs diff --git a/Cargo.lock b/Cargo.lock index bdf6547..d7134c2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,21 +2,6 @@ # It is not intended for manual editing. version = 3 -[[package]] -name = "addr2line" -version = "0.20.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f4fa78e18c64fce05e902adecd7a5eed15a5e0a3439f7b0e169f0252214865e3" -dependencies = [ - "gimli", -] - -[[package]] -name = "adler" -version = "1.0.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe" - [[package]] name = "aho-corasick" version = "1.0.2" @@ -26,27 +11,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "autocfg" -version = "1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa" - -[[package]] -name = "backtrace" -version = "0.3.68" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4319208da049c43661739c5fade2ba182f09d1dc2299b32298d3a31692b17e12" -dependencies = [ - "addr2line", - "cc", - "cfg-if", - "libc", - "miniz_oxide", - "object", - "rustc-demangle", -] - [[package]] name = "bitflags" version = "1.3.2" @@ -161,12 +125,6 @@ dependencies = [ "libc", ] -[[package]] -name = "gimli" -version = "0.27.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6c80984affa11d98d1b88b66ac8853f143217b399d3c74116778ff8fdb4ed2e" - [[package]] name = "hashbrown" version = "0.14.0" @@ -230,26 +188,6 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2dffe52ecf27772e601905b7522cb4ef790d2cc203488bbd0e2fe85fcb74566d" -[[package]] -name = "miniz_oxide" -version = "0.7.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e7810e0be55b428ada41041c41f32c9f1a42817901b4ccf45fa3d4b6561e74c7" -dependencies = [ - "adler", -] - -[[package]] -name = "mio" -version = "0.8.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "927a765cd3fc26206e66b296465fa9d3e5ab003e651c1b3c060e7956d96b19d2" -dependencies = [ - "libc", - "wasi", - "windows-sys", -] - [[package]] name = "multicast-socket" version = "0.2.2" @@ -259,7 +197,7 @@ dependencies = [ "get_if_addrs", "libc", "nix", - "socket2 0.3.19", + "socket2", "winapi 0.3.9", ] @@ -274,7 +212,6 @@ dependencies = [ "log", "multicast-socket", "serde", - "tokio", "toml", ] @@ -290,31 +227,6 @@ dependencies = [ "libc", ] -[[package]] -name = "num_cpus" -version = "1.16.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4161fcb6d602d4d2081af7c3a45852d875a03dd337a6bfdd6e06407b61342a43" -dependencies = [ - "hermit-abi", - "libc", -] - -[[package]] -name = "object" -version = "0.31.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8bda667d9f2b5051b8833f59f3bf748b28ef54f850f4fcb389a252aa383866d1" -dependencies = [ - "memchr", -] - -[[package]] -name = "pin-project-lite" -version = "0.2.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4c40d25201921e5ff0c862a505c6557ea88568a4e3ace775ab55e93f2f4f9d57" - [[package]] name = "proc-macro2" version = "1.0.66" @@ -368,12 +280,6 @@ version = "0.7.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e5ea92a5b6195c6ef2a0295ea818b312502c6fc94dde986c5553242e18fd4ce2" -[[package]] -name = "rustc-demangle" -version = "0.1.23" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d626bb9dae77e28219937af045c257c28bfd3f69333c512553507f5f9798cb76" - [[package]] name = "rustix" version = "0.38.4" @@ -427,16 +333,6 @@ dependencies = [ "winapi 0.3.9", ] -[[package]] -name = "socket2" -version = "0.4.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "64a4a911eed85daf18834cfaa86a79b7d266ff93ff5ba14005426219480ed662" -dependencies = [ - "libc", - "winapi 0.3.9", -] - [[package]] name = "syn" version = "2.0.27" @@ -457,34 +353,6 @@ dependencies = [ "winapi-util", ] -[[package]] -name = "tokio" -version = "1.29.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "532826ff75199d5833b9d2c5fe410f29235e25704ee5f0ef599fb51c21f4a4da" -dependencies = [ - "autocfg", - "backtrace", - "libc", - "mio", - "num_cpus", - "pin-project-lite", - "socket2 0.4.9", - "tokio-macros", - "windows-sys", -] - -[[package]] -name = "tokio-macros" -version = "2.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - [[package]] name = "toml" version = "0.7.6" @@ -525,12 +393,6 @@ version = "1.0.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "301abaae475aa91687eb82514b328ab47a211a533026cb25fc3e519b86adfc3c" -[[package]] -name = "wasi" -version = "0.11.0+wasi-snapshot-preview1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423" - [[package]] name = "winapi" version = "0.2.8" diff --git a/Cargo.toml b/Cargo.toml index 7e102c2..92131e9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,5 +11,4 @@ libc = "0.2.147" log = "0.4.18" multicast-socket = "0.2.2" serde = { version = "1.0.163", features = ["derive"] } -tokio = { version = "1.29.1", features = ["macros", "rt-multi-thread", "net"] } toml = "0.7.4" diff --git a/config.toml b/config.toml index b56d15a..08f11c6 100644 --- a/config.toml +++ b/config.toml @@ -2,14 +2,15 @@ sources = ["eth70"] destinations = ["eth10", "eth20", "eth30", "eth40"] filters = [ - "cups.local", - "_printer._tcp.local", - "_scanner._tcp.local", - "_ipp._tcp.local", - "_ipps._tcp.local", - "cups._ipps._tcp.local", - "lb._dns-sd._udp.local", - "EPSON1E715E.local", - "_pdl-datastream._tcp.local", - "EPSON M200 Series._pdl-datastream._tcp.local" + "cups.local", + "_printer._tcp.local", + "_scanner._tcp.local", + "_ipp._tcp.local", + "_ipps._tcp.local", + "cups._ipps._tcp.local", + "lb._dns-sd._udp.local", + "EPSON1E715E.local", + "_pdl-datastream._tcp.local", + "EPSON M200 Series._pdl-datastream._tcp.local", + "_ptp._tcp.local" ] diff --git a/src/communications.rs b/src/communications.rs deleted file mode 100644 index a0e38aa..0000000 --- a/src/communications.rs +++ /dev/null @@ -1,47 +0,0 @@ -use log::{info, trace, warn}; -use multicast_socket::{Message, MulticastOptions, MulticastSocket}; -use std::{net::SocketAddrV4, sync::mpsc::Sender}; - -pub struct Communications { - tx_chan: Sender, -} - -impl Communications { - pub fn new(tx_chan: Sender) -> Result { - Ok(Communications { tx_chan }) - } - - pub async fn start_listeners(&mut self) -> Result<(), String> { - info!("listener started"); - - let interfaces = get_if_addrs::get_if_addrs().unwrap(); - trace!("Interfaces list: {:?}", interfaces); - - // mdns - let mdns_address = SocketAddrV4::new([224, 0, 0, 251].into(), 5353); - let multicast_socket = MulticastSocket::with_options( - mdns_address, - multicast_socket::all_ipv4_interfaces().expect("could not fetch all interfaces"), - MulticastOptions { - loopback: false, - buffer_size: 4096, - ..Default::default() - }, - ) - .expect("error in creating multicast socket"); - - loop { - match multicast_socket.receive() { - Ok(msg) => { - self.tx_chan - .send(msg) - .expect("error in sending to mpsc channel"); - } - Err(e) if e.to_string().contains("EAGAIN") => continue, - Err(e) => { - warn!("error in reading from socket {:?} ", e); - } - }; - } - } -} diff --git a/src/main.rs b/src/main.rs index b3d746a..1b4f921 100644 --- a/src/main.rs +++ b/src/main.rs @@ -4,16 +4,11 @@ // Or a common listener/transmitter and then different modules to parse and transmit each type of // traffic use log::info; -use multicast_socket::Message; -use std::sync::mpsc::{self, Receiver, Sender}; -use tokio::runtime::Runtime; -pub mod communications; -pub use communications::*; pub mod config; pub use config::*; -pub mod processor; -pub use processor::*; +pub mod mdns; +pub use mdns::*; fn main() { env_logger::init(); @@ -24,24 +19,7 @@ fn main() { println!("{:?}", config); - let rt = Runtime::new().expect("error in creating runtime"); + let mdns_client = Mdns::new(config); - // This channel is used to send packets from this module to the processor - // TODO(ishan): Eventually, swap out std::mpsc for some thing faster - // or switch to a different model completely that doesn't use channels like this - let (tx, rx): (Sender, Receiver) = mpsc::channel(); - - // TODO(ishan): Start listeners and transmitters on v4 and v6 here - let mut comms = Communications::new(tx).expect("error in starting comms"); - - rt.spawn(async move { - comms - .start_listeners() - .await - .expect("error in comms listener"); - }); - - let processor = Processor::new(rx, config).expect("error in starting processor"); - - processor.start_read_loop(); + mdns_client.listener_loop(); } diff --git a/src/mdns.rs b/src/mdns.rs new file mode 100644 index 0000000..0b67088 --- /dev/null +++ b/src/mdns.rs @@ -0,0 +1,128 @@ +use crate::Config; +use dns_parser::Packet; +use log::{info, trace, warn}; +use multicast_socket::{Interface as MulticastInterface, MulticastOptions, MulticastSocket}; +use std::{ffi::CString, net::SocketAddrV4}; + +pub struct Mdns { + socket: MulticastSocket, + config: Config, +} + +impl Mdns { + pub fn new(config: Config) -> Self { + // mdns + let mdns_address = SocketAddrV4::new([224, 0, 0, 251].into(), 5353); + let multicast_socket = MulticastSocket::with_options( + mdns_address, + // TODO(ishan): Listen on ALL Interfaces, including ipv6 + multicast_socket::all_ipv4_interfaces().expect("could not fetch all interfaces"), + MulticastOptions { + loopback: false, + buffer_size: 4096, + ..Default::default() + }, + ) + .expect("error in creating multicast socket"); + + Self { + socket: multicast_socket, + config, + } + } + + pub fn listener_loop(&self) { + info!("listener started"); + + loop { + match self.socket.receive() { + Ok(msg) => self.process_packet(msg), + Err(e) if e.to_string().contains("EAGAIN") => continue, + Err(e) => { + warn!("error in reading from socket {:?} ", e); + } + }; + } + } + + pub fn process_packet(&self, msg: multicast_socket::Message) { + // TODO: Generalize this to parse any type of supported packet + let packet = Packet::parse(&msg.data).expect("failed to parse packet as a dns packet"); + + let src_ifname = if let MulticastInterface::Index(idx) = msg.interface { + ifidx_to_ifname(idx as u32) + } else { + "lo".to_string() + }; + + trace!( + "EVENT src-if = {} if-index {:?} address = {}, packet: {:?} answers = {:?}", + src_ifname, + msg.interface, + msg.origin_address, + packet.questions.iter().map(|q| q.qname).collect::>(), + packet.answers.iter().map(|q| q.name).collect::>() + ); + + let interfaces = get_if_addrs::get_if_addrs().unwrap(); + trace!("Interfaces: {:?}", interfaces); + for conf in &self.config.mdns { + let mut dst_ifs = vec![]; + + for query in &packet.questions { + if conf.destinations.contains(&src_ifname) + && (conf.filters.is_empty() || conf.filters.contains(&query.qname.to_string())) + { + dst_ifs.extend( + conf.sources + .iter() + .filter_map(|dst_if| interfaces.iter().find(|x| &x.name == dst_if)), + ); + } + } + + for answer in &packet.answers { + if conf.sources.contains(&src_ifname) + && (conf.filters.is_empty() || conf.filters.contains(&answer.name.to_string())) + { + dst_ifs.extend( + conf.destinations + .iter() + .filter_map(|dst_if| interfaces.iter().find(|x| &x.name == dst_if)), + ); + } + } + + for dst_if in dst_ifs { + let dst_ifid = ifname_to_ifidx(dst_if.name.to_string()); + + info!( + "forwarding packet questions {:?} answers = {:?} from {} to {}", + packet.questions, packet.answers, src_ifname, dst_if.name + ); + // TODO(ishan): Take a note of transaction id + // and avoid feedback loops + + self.socket + .send(&msg.data, &MulticastInterface::Index(dst_ifid as i32)) + .expect("error in sending mdns packet"); + } + } + } +} + +fn ifidx_to_ifname(idx: u32) -> String { + let out = CString::new("askdjhaskdjakdjadksa").unwrap(); + + unsafe { + let ptr = out.into_raw(); + let response = libc::if_indextoname(idx, ptr); + + CString::from_raw(response).into_string().unwrap() + } +} + +fn ifname_to_ifidx(name: String) -> u32 { + let out = name.as_ptr() as *const _; + unsafe { libc::if_nametoindex(out) } +} diff --git a/src/processor.rs b/src/processor.rs deleted file mode 100644 index c5e49d0..0000000 --- a/src/processor.rs +++ /dev/null @@ -1,112 +0,0 @@ -use crate::Config; -use dns_parser::Packet; -use log::{info, trace}; -use multicast_socket::{Interface, Message, MulticastOptions, MulticastSocket}; -use std::{ffi::CString, net::SocketAddrV4, sync::mpsc::Receiver}; - -pub struct Processor { - reader: Receiver, - config: Config, -} - -impl Processor { - pub fn new(reader: Receiver, config: Config) -> Result { - Ok(Self { reader, config }) - } - - pub fn start_read_loop(self) { - // mdns - let mdns_address = SocketAddrV4::new([224, 0, 0, 251].into(), 5353); - let socket = MulticastSocket::with_options( - mdns_address, - multicast_socket::all_ipv4_interfaces().expect("could not fetch all interfaces"), - MulticastOptions { - loopback: false, - buffer_size: 4096, - ..Default::default() - }, - ) - .expect("error in creating multicast socket"); - - for evt in self.reader { - // TODO: Generalize this to parse any type of supported packet - let packet = Packet::parse(&evt.data).expect("failed to parse packet as a dns packet"); - - let interfaces = get_if_addrs::get_if_addrs().unwrap(); - - let src_ifname = if let Interface::Index(idx) = evt.interface { - ifidx_to_ifname(idx as u32) - } else { - "lo".to_string() - }; - - trace!( - "EVENT src-if = {} if-index {:?} address = {}, packet: {:?} answers = {:?}", - src_ifname, - evt.interface, - evt.origin_address, - packet.questions.iter().map(|q| q.qname).collect::>(), - packet.answers.iter().map(|q| q.name).collect::>() - ); - for conf in &self.config.mdns { - let mut dst_ifs = vec![]; - - for query in &packet.questions { - if conf.destinations.contains(&src_ifname) - && (conf.filters.is_empty() - || conf.filters.contains(&query.qname.to_string())) - { - dst_ifs.extend( - conf.sources - .iter() - .filter_map(|dst_if| interfaces.iter().find(|x| &x.name == dst_if)), - ); - } - } - - for answer in &packet.answers { - if conf.sources.contains(&src_ifname) - && (conf.filters.is_empty() - || conf.filters.contains(&answer.name.to_string())) - { - dst_ifs.extend( - conf.destinations - .iter() - .filter_map(|dst_if| interfaces.iter().find(|x| &x.name == dst_if)), - ); - } - } - - for dst_if in dst_ifs { - let dst_ifid = ifname_to_ifidx(dst_if.name.to_string()); - - info!( - "forwarding packet questions {:?} answers = {:?} from {} to {}", - packet.questions, packet.answers, src_ifname, dst_if.name - ); - // TODO(ishan): Take a note of transaction id - // and avoid feedback loops - socket - .send(&evt.data, &Interface::Index(dst_ifid as i32)) - .expect("error in sending mdns packet"); - } - } - } - } -} - -fn ifidx_to_ifname(idx: u32) -> String { - let out = CString::new("askdjhaskdjakdjadksa").unwrap(); - - unsafe { - let ptr = out.into_raw(); - let response = libc::if_indextoname(idx, ptr); - - CString::from_raw(response).into_string().unwrap() - } -} - -fn ifname_to_ifidx(name: String) -> u32 { - let out = name.as_ptr() as *const _; - unsafe { libc::if_nametoindex(out) } -}