From c7d175b6d44ff7c4cc68f6eb208f1aa0587fde26 Mon Sep 17 00:00:00 2001 From: Mathias Peters Date: Thu, 5 Dec 2024 10:50:51 +0100 Subject: [PATCH] Add initial pcap analysis --- .github/workflows/tests.yml | 4 + Cargo.lock | 117 ++++++++++++++++++-- neptun/Cargo.toml | 1 + neptun/src/noise/mod.rs | 35 ++++++ neptun/src/noise/session.rs | 35 ++++++ xray/Cargo.toml | 4 +- xray/Pipfile | 1 - xray/analyze.py | 211 +++++++++++++++++------------------- xray/run.py | 38 ++++--- xray/src/client.rs | 2 +- xray/src/event_loop.rs | 45 ++++---- xray/src/key_pair.rs | 4 +- xray/src/main.rs | 83 +++++++++++--- xray/src/pcap.rs | 104 ++++++++++++++++++ xray/src/utils.rs | 38 ++++--- 15 files changed, 528 insertions(+), 194 deletions(-) create mode 100644 xray/src/pcap.rs diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 8d65a51..ee81e6e 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -17,6 +17,8 @@ jobs: steps: - uses: actions/checkout@a5ac7e51b41094c92402da3b24376905380afc29 # v4.1.6 - uses: actions-rs/toolchain@b2417cde72dcf67f306c0ae8e0828a81bf0b189f # v1.0.6 + - if: matrix.os == 'ubuntu-24.04' + run: sudo apt-get install -y libpcap-dev - name: Install hack run: cargo +stable install --git https://github.com/taiki-e/cargo-hack.git cargo-hack --rev c0b517b9eefa27cdaf27cca5f1b186c00ef1af47 --locked - run: cargo hack test --each-feature ${{ matrix.packages }} @@ -29,6 +31,8 @@ jobs: steps: - uses: actions/checkout@a5ac7e51b41094c92402da3b24376905380afc29 # v4.1.6 - uses: actions-rs/toolchain@b2417cde72dcf67f306c0ae8e0828a81bf0b189f # v1.0.6 + - if: matrix.os == 'ubuntu-24.04' + run: sudo apt-get install -y libpcap-dev - run: CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_RUNNER='sudo -E' cargo test -- --ignored crypto-bench: diff --git a/Cargo.lock b/Cargo.lock index 02cfca7..6bcf3c2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -130,6 +130,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "bitflags" +version = "1.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" + [[package]] name = "bitflags" version = "2.6.0" @@ -510,6 +516,27 @@ version = "1.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "60b1af1c220855b6ceac025d3f6ecdd2b7c4894bfe9cd9bda4fbb4bc7c0d4cf0" +[[package]] +name = "errno" +version = "0.2.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f639046355ee4f37944e44f60642c6f3a7efa3cf6b78c78a0d989a8ce6c396a1" +dependencies = [ + "errno-dragonfly", + "libc", + "winapi", +] + +[[package]] +name = "errno-dragonfly" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa68f1b12764fab894d2755d2518754e71b4fd80ecfb822714a1206c2aab39bf" +dependencies = [ + "cc", + "libc", +] + [[package]] name = "etherparse" version = "0.12.0" @@ -710,6 +737,16 @@ version = "0.2.159" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "561d97a539a36e26a9a5fad1ea11a3039a67714694aaa379433e580854bc3dc5" +[[package]] +name = "libloading" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc2f4eb4bc735547cfed7c0a4922cbd04a4655978c09b54f1f7b228750664c34" +dependencies = [ + "cfg-if", + "windows-targets", +] + [[package]] name = "lock_api" version = "0.4.12" @@ -808,7 +845,7 @@ version = "0.28.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ab2156c4fce2f8df6c499cc1c763e4394b7482525bf2a9701c9d79d215f519e4" dependencies = [ - "bitflags", + "bitflags 2.6.0", "cfg-if", "cfg_aliases", "libc", @@ -907,12 +944,33 @@ dependencies = [ "windows-targets", ] +[[package]] +name = "pcap" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "499125886165f62fbc0c095ead9189b253f48eb1c5fcab49f81a270f2f220652" +dependencies = [ + "bitflags 1.3.2", + "errno", + "libc", + "libloading", + "pkg-config", + "regex", + "windows-sys 0.36.1", +] + [[package]] name = "pin-project-lite" version = "0.2.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bda66fc9667c18cb2758a2ac84d1167245054bcf85d5d1aaa6923f45801bdd02" +[[package]] +name = "pkg-config" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "953ec861398dccce10c670dfeaf3ec4911ca479e9c02154b3a215178c5f566f2" + [[package]] name = "plotters" version = "0.3.7" @@ -1132,7 +1190,7 @@ version = "0.5.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b6dfecf2c74bce2466cabf93f6664d6998a69eb21e39f4207930065b27b771f" dependencies = [ - "bitflags", + "bitflags 2.6.0", ] [[package]] @@ -1647,6 +1705,19 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" +[[package]] +name = "windows-sys" +version = "0.36.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea04155a16a59f9eab786fe12a4a450e75cdb175f9e0d80da1e17db09f55b8d2" +dependencies = [ + "windows_aarch64_msvc 0.36.1", + "windows_i686_gnu 0.36.1", + "windows_i686_msvc 0.36.1", + "windows_x86_64_gnu 0.36.1", + "windows_x86_64_msvc 0.36.1", +] + [[package]] name = "windows-sys" version = "0.52.0" @@ -1672,13 +1743,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" dependencies = [ "windows_aarch64_gnullvm", - "windows_aarch64_msvc", - "windows_i686_gnu", + "windows_aarch64_msvc 0.52.6", + "windows_i686_gnu 0.52.6", "windows_i686_gnullvm", - "windows_i686_msvc", - "windows_x86_64_gnu", + "windows_i686_msvc 0.52.6", + "windows_x86_64_gnu 0.52.6", "windows_x86_64_gnullvm", - "windows_x86_64_msvc", + "windows_x86_64_msvc 0.52.6", ] [[package]] @@ -1687,12 +1758,24 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" +[[package]] +name = "windows_aarch64_msvc" +version = "0.36.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9bb8c3fd39ade2d67e9874ac4f3db21f0d710bee00fe7cab16949ec184eeaa47" + [[package]] name = "windows_aarch64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" +[[package]] +name = "windows_i686_gnu" +version = "0.36.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "180e6ccf01daf4c426b846dfc66db1fc518f074baa793aa7d9b9aaeffad6a3b6" + [[package]] name = "windows_i686_gnu" version = "0.52.6" @@ -1705,12 +1788,24 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" +[[package]] +name = "windows_i686_msvc" +version = "0.36.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e2e7917148b2812d1eeafaeb22a97e4813dfa60a3f8f78ebe204bcc88f12f024" + [[package]] name = "windows_i686_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" +[[package]] +name = "windows_x86_64_gnu" +version = "0.36.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4dcd171b8776c41b97521e5da127a2d86ad280114807d0b2ab1e462bc764d9e1" + [[package]] name = "windows_x86_64_gnu" version = "0.52.6" @@ -1723,6 +1818,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" +[[package]] +name = "windows_x86_64_msvc" +version = "0.36.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c811ca4a8c853ef420abd8592ba53ddbbac90410fab6903b3e79972a631f7680" + [[package]] name = "windows_x86_64_msvc" version = "0.52.6" @@ -1751,8 +1852,10 @@ dependencies = [ "csv", "curve25519-dalek", "neptun", + "pcap", "pnet", "rand", + "serde", "thiserror", "tokio", "x25519-dalek", diff --git a/neptun/Cargo.toml b/neptun/Cargo.toml index 6981736..f10922f 100644 --- a/neptun/Cargo.toml +++ b/neptun/Cargo.toml @@ -15,6 +15,7 @@ default = [] device = ["socket2", "thiserror"] # mocks std::time::Instant with mock_instant mock-instant = ["mock_instant"] +xray = [] [dependencies] base64 = "0.13" diff --git a/neptun/src/noise/mod.rs b/neptun/src/noise/mod.rs index 8517ecf..6efddab 100644 --- a/neptun/src/noise/mod.rs +++ b/neptun/src/noise/mod.rs @@ -331,6 +331,41 @@ impl Tunn { self.handle_verified_packet(packet, dst) } + #[cfg(feature = "xray")] + pub fn decrypt<'a>( + &mut self, + datagram: &[u8], + dst: &'a mut [u8], + ) -> Result<&'a [u8], WireGuardError> { + let packet = Tunn::parse_incoming_packet(datagram)?; + match packet { + Packet::PacketData(p) => { + let r_idx = p.receiver_idx as usize; + let idx = r_idx % N_SESSIONS; + + // Get the (probably) right session + let decapsulated_packet = { + let session = self.sessions[idx].as_ref(); + let session = session.ok_or_else(|| { + tracing::trace!( + message = "No current session available", + remote_idx = r_idx + ); + WireGuardError::NoCurrentSession + })?; + session.decrypt_data_packet(p, dst)? + }; + + match self.validate_decapsulated_packet(decapsulated_packet) { + TunnResult::WriteToTunnelV4(p, _) => Ok(p), + TunnResult::Err(err) => Err(err), + _ => Err(WireGuardError::UnexpectedPacket), + } + } + _ => Err(WireGuardError::WrongPacketType), + } + } + pub(crate) fn handle_verified_packet<'a>( &mut self, packet: Packet, diff --git a/neptun/src/noise/session.rs b/neptun/src/noise/session.rs index 3aa6408..1170b9a 100644 --- a/neptun/src/noise/session.rs +++ b/neptun/src/noise/session.rs @@ -273,6 +273,41 @@ impl Session { let counter_validator = self.receiving_key_counter.lock(); (counter_validator.next, counter_validator.receive_cnt) } + + #[cfg(feature = "xray")] + pub fn decrypt_data_packet<'a>( + &self, + packet: PacketData, + dst: &'a mut [u8], + ) -> Result<&'a mut [u8], WireGuardError> { + let ct_len = packet.encrypted_encapsulated_packet.len(); + if dst.len() < ct_len { + // This is a very incorrect use of the library, therefore panic and not error + panic!("The destination buffer is too small"); + } + let decrypt_key = if packet.receiver_idx == self.receiving_index { + &self.receiver + } else if packet.receiver_idx == self.sending_index { + &self.sender + } else { + return Err(WireGuardError::WrongIndex); + }; + + let ret = { + let mut nonce = [0u8; 12]; + nonce[4..12].copy_from_slice(&packet.counter.to_le_bytes()); + dst[..ct_len].copy_from_slice(packet.encrypted_encapsulated_packet); + decrypt_key + .open_in_place( + Nonce::assume_unique_for_key(nonce), + Aad::from(&[]), + &mut dst[..ct_len], + ) + .map_err(|_| WireGuardError::InvalidAeadTag)? + }; + + Ok(ret) + } } #[inline(always)] diff --git a/xray/Cargo.toml b/xray/Cargo.toml index dfdbd3d..24aa10c 100644 --- a/xray/Cargo.toml +++ b/xray/Cargo.toml @@ -11,8 +11,10 @@ clap = { version = "4.5", features = ["derive"] } color-eyre = "0.6" csv = "1.3.1" curve25519-dalek = "4.1" +pcap = "2.2" pnet = "0.35" rand = "0.8.5" +serde = { version = "1.0", features = ["derive"] } thiserror = "1.0" tokio = { version = "1.41", features = ["macros", "rt-multi-thread", "time", "net", "sync"] } x25519-dalek = "2.0" @@ -20,4 +22,4 @@ x25519-dalek = "2.0" [dependencies.neptun] version = "0.6.0" path = "../neptun" -features = ["device"] +features = ["device", "xray"] diff --git a/xray/Pipfile b/xray/Pipfile index 2d3defc..d3d17e1 100644 --- a/xray/Pipfile +++ b/xray/Pipfile @@ -6,7 +6,6 @@ name = "pypi" [packages] ruff = "==0.8" matplotlib = "==3.9" -scapy = "==2.6" [dev-packages] diff --git a/xray/analyze.py b/xray/analyze.py index 2dc413f..9188e1d 100644 --- a/xray/analyze.py +++ b/xray/analyze.py @@ -2,69 +2,57 @@ import math import matplotlib.pyplot as plt # type: ignore from functools import reduce -from scapy.all import PcapReader # type: ignore -from scapy.layers.inet import UDP # type: ignore def analyze(csv_path, pcap_path, count, test_type): Analyzer(csv_path, pcap_path, count, test_type) +def parse_int(val): + if val is None or val == "": + return None + else: + return int(val) + + +class PacketInfo: + def __init__(self, csv_row): + self.recv_index = parse_int(csv_row[0]) + self.send_ts = int(csv_row[1]) + self.pre_wg_ts = parse_int(csv_row[2]) + self.post_wg_ts = parse_int(csv_row[3]) + self.recv_ts = parse_int(csv_row[4]) + + def get_latencies(self): + pre_wg = self.get_latency(self.send_ts, self.pre_wg_ts) + post_wg = self.get_latency(self.pre_wg_ts, self.post_wg_ts) + recv = self.get_latency(self.post_wg_ts, self.recv_ts) + total = self.get_latency(self.send_ts, self.recv_ts) + return (pre_wg, post_wg, recv, total) + + def get_latency(self, left, right): + if left is not None and right is not None: + return right - left + else: + return None + + class CsvData: def __init__(self, csv_path): - self.indices = [] - self.timestamps = [] - self.latencies = [] - self.min_latency = -1 - self.max_latency = -1 + self.packets = [] with open(csv_path, newline="") as csvfile: - reader = csv.reader(csvfile, delimiter=",", quotechar="|") - next(reader) + reader = csv.reader(csvfile, delimiter=",") + next(reader) # skip header row for row in reader: - if len(row[0]) > 0: - self.indices.append(int(row[0])) - send_ts = int(row[1]) - if len(row[2]) > 0: - recv_ts = int(row[2]) - else: - recv_ts = -1 - self.timestamps.append((send_ts, recv_ts)) - if recv_ts >= 0: - latency = recv_ts - send_ts - self.latencies.append(latency) - if self.min_latency < 0 or latency < self.min_latency: - self.min_latency = latency - if self.max_latency < 0 or latency > self.max_latency: - self.max_latency = latency - - -class PcapData: - def __init__(self, pcap_path, test_type): - self.before_wg_packets = [] - self.after_wg_packets = [] - with PcapReader(pcap_path) as pcap_reader: - for pkt in pcap_reader: - if not pkt.haslayer(UDP): - continue - - if test_type == "crypto" and pkt.sport == 63636: - if pkt.dport == 41414: - self.before_wg_packets.append(pkt) - elif pkt.dport == 52525: - self.after_wg_packets.append(pkt) - elif test_type == "pt" and pkt.dport == 63636: - if pkt.sport == 52525: - self.before_wg_packets.append(pkt) - elif pkt.sport == 41414: - self.after_wg_packets.append(pkt) + packet_info = PacketInfo(row) + self.packets.append(packet_info) class Analyzer: - def __init__(self, csv_name, pcap_name, count, test_type): + def __init__(self, csv_path, pcap_path, count, test_type): self.count = count - self.csv_data = CsvData(csv_name) - self.pcap_data = PcapData(pcap_name, test_type) + self.csv_data = CsvData(csv_path) graphs = [ self.ordering_pie_chart, @@ -86,9 +74,11 @@ def __init__(self, csv_name, pcap_name, count, test_type): plt.show() def ordering_pie_chart(self, ax): - in_order = count_ordered(self.csv_data.indices, self.count) + in_order = count_ordered(self.csv_data.packets, self.count) dropped = reduce( - lambda count, e: count + (1 if e == 0 else 0), self.csv_data.indices, 0 + lambda count, packet: count + (1 if packet.recv_index is None else 0), + self.csv_data.packets, + 0, ) reordered = self.count - in_order - dropped data = [] @@ -106,46 +96,36 @@ def ordering_pie_chart(self, ax): ax.pie(data, labels=labels) def packet_ordering(self, ax): - y_axis = [None] - x_axis = [0] - for iter, index in enumerate(self.csv_data.indices): - if self.csv_data.timestamps[iter][1] >= 0: - y_axis.append(self.csv_data.indices[iter]) - else: - y_axis.append(None) - x_axis.append(iter + 1) + data = list(map(lambda p: p.recv_index, self.csv_data.packets)) ax.set_title("Packet order") ax.set_xlabel("Received order") ax.set_ylabel("Packet index") - ax.plot(x_axis, y_axis) + ax.plot(data) def packet_latency(self, ax): - millisec = 1000 - sec = 1000 * millisec - if self.csv_data.min_latency > sec: - divisor = sec - timeunit = "Seconds" - elif self.csv_data.min_latency > millisec: - divisor = millisec - timeunit = "Milliseconds" - else: - divisor = 1 - timeunit = "Microseconds" + data = list(map(lambda pi: pi.get_latencies(), self.csv_data.packets)) + pre_wg = [] + post_wg = [] + recv = [] + for l in data: # noqa: E741 (ambiguous name) + if l[0] is not None: + pre_wg.append(l[0]) + if l[1] is not None: + post_wg.append(l[1]) + if l[2] is not None: + recv.append(l[2]) - num_buckets = 15 - bucket_size = int( - (self.csv_data.max_latency - self.csv_data.min_latency) / (num_buckets - 1) - ) - buckets = [] - for latency in self.csv_data.latencies: - bucket_index = int((latency - self.csv_data.min_latency) / bucket_size) - buckets.append( - (self.csv_data.min_latency + (bucket_index * bucket_size)) / divisor - ) ax.set_title("Latency") - ax.set_xlabel(f"Latency ({timeunit})") + ax.set_xlabel("Latency (Microseconds)") ax.set_ylabel("Count") - ax.hist(buckets, color="blue", bins=num_buckets) + ax.hist( + [pre_wg, post_wg, recv], + label=["PreWG", "PostWG", "Recv"], + color=["orange", "green", "blue"], + stacked=True, + bins=15, + ) + ax.legend() def dropped_packets(self, ax): if self.count >= 100: @@ -154,36 +134,44 @@ def dropped_packets(self, ax): num_buckets = 10 else: num_buckets = self.count - bucket_size = int(self.count / (num_buckets - 1)) - buckets = [] - for iter, index in enumerate(self.csv_data.indices): - if self.csv_data.timestamps[iter][1] < 0: - bucket_index = int(iter / bucket_size) - buckets.append(bucket_index * bucket_size) + + pre_wg = [] + post_wg = [] + recv = [] + for i, packet in enumerate(self.csv_data.packets): + if packet.pre_wg_ts is None: + pre_wg.append(i) + if packet.post_wg_ts is None: + post_wg.append(i) + if packet.recv_ts is None: + recv.append(i) + ax.set_title("Dropped packets") ax.set_xlabel("Index") ax.set_ylabel("Count") - ax.hist(buckets, color="blue", bins=num_buckets) - - def dropped_packets2(self, ax): - data = [] - for iter, index in enumerate(self.csv_data.indices): - if self.csv_data.timestamps[iter][1] < 0: - data.append(index) - ax.set_title("Dropped packets 2") - ax.set_xlabel("Packet index") - ax.set_ylabel("Count") - ax.plot(data) + ax.hist( + [pre_wg, post_wg, recv], + label=["PreWG", "PostWG", "Recv"], + color=["orange", "green", "blue"], + stacked=True, + bins=num_buckets, + ) + ax.legend() def packet_funnel(self, ax): count = self.count - before_wg = len(self.pcap_data.before_wg_packets) - after_wg = len(self.pcap_data.after_wg_packets) - recv = len(list(filter(lambda x: x > 0, self.csv_data.indices))) + before_wg = 0 + after_wg = 0 + recv = 0 + for p in self.csv_data.packets: + latencies = p.get_latencies() + before_wg += 1 if latencies[0] is not None else 0 + after_wg += 1 if latencies[1] is not None else 0 + recv += 1 if latencies[2] is not None else 0 categories = [ f"Count ({count})", - f"before wg ({before_wg})", - f"after_wg ({after_wg})", + f"Before wg ({before_wg})", + f"After_wg ({after_wg})", f"Recv ({recv})", ] values = [self.count, before_wg, after_wg, recv] @@ -197,19 +185,20 @@ def packet_funnel(self, ax): def count_ordered(data, count): if len(data) == 0: return 0 + indices = list(map(lambda p: p.recv_index, data)) ordered = 0 - range_good_start = data[0] == 1 + range_good_start = indices[0] == 1 range_len = 1 - prev = data[0] - for i in range(1, len(data)): - if data[i] == 0: + prev = indices[0] + for i in range(1, len(indices)): + if indices[i] is None: continue - elif data[i] == prev + 1: + elif indices[i] == prev + 1: range_len += 1 else: ordered += range_len - (0 if range_good_start else 1) - range_good_start = data[i] = i + range_good_start = indices[i] == i range_len = 1 - prev = data[i] + prev = indices[i] ordered += range_len - (0 if range_good_start else 1) return ordered diff --git a/xray/run.py b/xray/run.py index 56c099a..716ba2c 100755 --- a/xray/run.py +++ b/xray/run.py @@ -13,15 +13,15 @@ def run_command(cmd, capture_output=False): args = shlex.split(cmd) - run = subprocess.run(args, capture_output=capture_output,check=True) + run = subprocess.run(args, capture_output=capture_output, check=True) return (run.stdout, run.stderr) -def get_csv_name(wg, test_type, count): +def get_csv_path(wg, test_type, count): return f"results/xray_metrics_{wg.lower()}_{test_type}_{count}.csv" -def get_pcap_name(wg, test_type, count): +def get_pcap_path(wg, test_type, count): return f"results/{WG_IFC_NAME}_{wg.lower()}_{test_type}_{count}.pcap" @@ -58,7 +58,7 @@ def setup_wireguard(wg, build_neptun): run_command(f"sudo ../target/release/boringtun-cli {WG_IFC_NAME}") else: if build_neptun: - run_command(f"cargo build --release -p neptun-cli") + run_command("cargo build --release -p neptun-cli") run_command(f"sudo ../target/release/neptun-cli {WG_IFC_NAME}") run_command(f"sudo ip link set dev {WG_IFC_NAME} mtu 1420") run_command(f"sudo ip link set dev {WG_IFC_NAME} up") @@ -67,9 +67,17 @@ def setup_wireguard(wg, build_neptun): ) # Not strictly necessary but keeps the pcaps a bit cleaner -def start_tcpdump(pcap_name): +def start_tcpdump(pcap_path): return subprocess.Popen( - ["sudo", "tcpdump", "-ni", "any", "-w", pcap_name], + [ + "sudo", + "tcpdump", + "-ni", + "any", + "-w", + pcap_path, + "udp and (port 41414 or port 52525 or port 63636)", + ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, @@ -78,10 +86,10 @@ def start_tcpdump(pcap_name): def run_xray(wg, test_type, count, build_xray): if build_xray: - run_command( - f"cargo build --release" - ) - run_command(f"sudo ../target/release/xray --wg {wg.lower()} --test-type {test_type} --packet-count {count} --csv-name {get_csv_name(wg, test_type, count)}") + run_command("cargo build --release") + run_command( + f"sudo ../target/release/xray --wg {wg.lower()} --test-type {test_type} --packet-count {count} --csv-path {get_csv_path(wg, test_type, count)} --pcap-path {get_pcap_path(wg, test_type, count)}" + ) def stop_tcpdump(tcpdump): @@ -119,13 +127,13 @@ def main(): Path("results/").mkdir(parents=True, exist_ok=True) try: - os.remove(get_csv_name(wg.name, test_type, count)) - os.remove(get_pcap_name(wg.name, test_type, count)) + os.remove(get_csv_path(wg.name, test_type, count)) + os.remove(get_pcap_path(wg.name, test_type, count)) except: # noqa: E722 pass setup_wireguard(wg, build_neptun) - tcpdump = start_tcpdump(get_pcap_name(wg.name, test_type, count)) + tcpdump = start_tcpdump(get_pcap_path(wg.name, test_type, count)) succeeded = True try: @@ -139,8 +147,8 @@ def main(): if succeeded: analyze( - get_csv_name(wg.name, test_type, count), - get_pcap_name(wg.name, test_type, count), + get_csv_path(wg.name, test_type, count), + get_pcap_path(wg.name, test_type, count), count, test_type, ) diff --git a/xray/src/client.rs b/xray/src/client.rs index a471843..9aa29b8 100644 --- a/xray/src/client.rs +++ b/xray/src/client.rs @@ -240,7 +240,7 @@ impl Client { Ok(udp_packet) } - fn parse_udp_packet(packet: &[u8]) -> XRayResult<(SocketAddrV4, usize, usize)> { + pub fn parse_udp_packet(packet: &[u8]) -> XRayResult<(SocketAddrV4, usize, usize)> { let ip_packet = Ipv4Packet::new(packet).ok_or(XRayError::PacketParse)?; let udp_packet = UdpPacket::new(ip_packet.payload()).ok_or(XRayError::PacketParse)?; let from = SocketAddrV4::new(ip_packet.get_source(), udp_packet.get_source()); diff --git a/xray/src/event_loop.rs b/xray/src/event_loop.rs index 24a14ea..c109332 100644 --- a/xray/src/event_loop.rs +++ b/xray/src/event_loop.rs @@ -1,10 +1,11 @@ use std::{net::SocketAddrV4, pin::Pin, time::Duration}; +use neptun::noise::Tunn; use tokio::{sync::mpsc, time::Instant}; use crate::{ client::Client, - utils::{write_to_csv, Packet, RecvType, SendType, TestCmd}, + utils::{Packet, RecvType, SendType, TestCmd}, CliArgs, XRayResult, }; @@ -14,7 +15,7 @@ pub struct EventLoop { crypto_client: Client, plaintext_client: Client, cmd_rx: mpsc::Receiver, - packets: Vec, + pub packets: Vec, can_send: bool, is_done: bool, crypto_buf: Vec, @@ -46,7 +47,7 @@ impl EventLoop { } } - pub async fn run(mut self) -> XRayResult<()> { + pub async fn run(mut self) -> XRayResult { let mut wg_tick_interval = tokio::time::interval(Duration::from_millis(250)); // This timeout is only actually used when the test is otherwise done // It is here set to one second just to initialize it, but is reset before it's actually used @@ -55,7 +56,7 @@ impl EventLoop { loop { tokio::select! { _ = &mut finish_timeout, if self.is_done => { - self.on_finished(self.recv_counter).await?; + println!("Test done, received {} packets", self.recv_counter); break; }, _ = wg_tick_interval.tick() => { @@ -72,7 +73,14 @@ impl EventLoop { } } } - Ok(()) + Ok(self) + } + + pub fn tunn(&mut self) -> Tunn { + self.crypto_client + .tunn + .take() + .expect("Crypto client is expected to have a Tunn object") } async fn on_recv_cmd( @@ -84,11 +92,7 @@ impl EventLoop { let send_ts = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH)? .as_micros(); - let packet = Packet { - send_ts, - recv_index: None, - recv_ts: None, - }; + let packet = Packet::new(send_ts); let mut payload = vec![0; Packet::send_size()]; payload[0..Packet::index_size()].copy_from_slice(&send_index.to_le_bytes()); @@ -110,7 +114,7 @@ impl EventLoop { packet_dst, send_index, } => { - if send_index % (self.cli_args.packet_count / 10) as u64 == 0 { + if send_index > 0 && send_index % (self.cli_args.packet_count / 10) as u64 == 0 { println!("[Crypto] Sending packet with index {send_index}"); } let (packet, payload) = prepare_packet(send_index)?; @@ -123,7 +127,7 @@ impl EventLoop { self.can_send = !matches!(sr, SendType::HandshakeInitiation); } TestCmd::SendPlaintext { dst, send_index } => { - if send_index % (self.cli_args.packet_count / 10) as u64 == 0 { + if send_index > 0 && send_index % (self.cli_args.packet_count / 10) as u64 == 0 { println!("[Plaintext] Sending packet with index {send_index}"); } let (packet, payload) = prepare_packet(send_index)?; @@ -149,8 +153,10 @@ impl EventLoop { } RecvType::Data { length: bytes_read } => { if bytes_read == Packet::send_size() { - if self.recv_counter % (self.cli_args.packet_count / 10) == 0 { - println!("[Crypto] Received {} packets", self.recv_counter + 1); + if self.recv_counter > 0 + && self.recv_counter % (self.cli_args.packet_count / 10) == 0 + { + println!("[Crypto] Received {} packets", self.recv_counter); } self.recv_counter += 1; let send_index = u64::from_le_bytes( @@ -177,8 +183,10 @@ impl EventLoop { ) -> XRayResult<()> { if let RecvType::Data { length: bytes_read } = rt { if bytes_read == Packet::send_size() { - if self.recv_counter % (self.cli_args.packet_count / 10) == 0 { - println!("[Plaintext] Received {} packets", self.recv_counter + 1); + if self.recv_counter > 0 + && self.recv_counter % (self.cli_args.packet_count / 10) == 0 + { + println!("[Plaintext] Received {} packets", self.recv_counter); } self.recv_counter += 1; let send_index = u64::from_le_bytes( @@ -205,9 +213,4 @@ impl EventLoop { .reset(Instant::now() + Duration::from_secs(1)); } } - - async fn on_finished(&mut self, recv_packet_count: usize) -> XRayResult<()> { - println!("Test done, received {recv_packet_count} packets"); - write_to_csv(&self.cli_args.csv_name(), &self.packets) - } } diff --git a/xray/src/key_pair.rs b/xray/src/key_pair.rs index 3b73983..8d315b7 100644 --- a/xray/src/key_pair.rs +++ b/xray/src/key_pair.rs @@ -30,8 +30,8 @@ pub trait NepTUNKey { BASE64_STANDARD.encode(self.bytes()) } - fn write_to_file(&self, file_name: &str) -> XRayResult<()> { - let mut f = File::create(file_name)?; + fn write_to_file(&self, path: &str) -> XRayResult<()> { + let mut f = File::create(path)?; f.write_all(self.as_b64().as_bytes())?; Ok(()) } diff --git a/xray/src/main.rs b/xray/src/main.rs index 5ca8b15..76ab0d3 100644 --- a/xray/src/main.rs +++ b/xray/src/main.rs @@ -1,6 +1,7 @@ mod client; mod event_loop; mod key_pair; +mod pcap; mod utils; use std::{ @@ -10,8 +11,10 @@ use std::{ use neptun::noise::{Tunn, TunnResult}; +use ::pcap::Error as PcapError; use clap::Parser; use color_eyre::eyre::Result as EyreResult; + use tokio::{ net::UdpSocket, sync::mpsc::{self, error::SendError}, @@ -21,15 +24,20 @@ use crate::{ client::Client, event_loop::EventLoop, key_pair::KeyPair, - utils::{configure_wg, TestCmd}, + pcap::process_pcap, + utils::{configure_wg, run_command, write_to_csv, TestCmd}, }; const WG_NAME: &str = "xraywg1"; -const WG_ADDR: SocketAddrV4 = SocketAddrV4::new(Ipv4Addr::new(100, 66, 0, 1), 41414); -const PLAINTEXT_ADDR: SocketAddrV4 = SocketAddrV4::new(*WG_ADDR.ip(), 52525); -const CRYPTO_ADDR: SocketAddrV4 = SocketAddrV4::new(Ipv4Addr::new(100, 66, 0, 2), 63636); -const CRYPTO_SOCK_ADDR: SocketAddrV4 = SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), 63636); +const WG_PORT: u16 = 41414; +const PLAINTEXT_PORT: u16 = 52525; +const CRYPTO_PORT: u16 = 63636; + +const WG_ADDR: SocketAddrV4 = SocketAddrV4::new(Ipv4Addr::new(100, 66, 0, 1), WG_PORT); +const PLAINTEXT_ADDR: SocketAddrV4 = SocketAddrV4::new(*WG_ADDR.ip(), PLAINTEXT_PORT); +const CRYPTO_ADDR: SocketAddrV4 = SocketAddrV4::new(Ipv4Addr::new(100, 66, 0, 2), CRYPTO_PORT); +const CRYPTO_SOCK_ADDR: SocketAddrV4 = SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), CRYPTO_PORT); type XRayResult = Result; @@ -57,6 +65,8 @@ enum XRayError { Time(#[from] SystemTimeError), #[error("Failed to send command over channel: {0:?}")] ChannelSend(#[from] SendError), + #[error("Pcap error: {0:?}")] + Pcap(#[from] PcapError), } impl From> for XRayError { @@ -74,20 +84,28 @@ struct CliArgs { #[arg(long, default_value_t = 10)] packet_count: usize, #[arg(long)] - csv_name: Option, + csv_path: Option, + #[arg(long)] + pcap_path: Option, } impl CliArgs { - fn csv_name(&self) -> String { - self.csv_name - .as_ref() - .map(|s| s.to_lowercase()) - .unwrap_or_else(|| { - format!( - "results/xray_metrics_{}_{}_{}.csv", - self.wg, self.test_type, self.packet_count - ) - }) + fn csv_path(&self) -> String { + self.csv_path.as_ref().cloned().unwrap_or_else(|| { + format!( + "results/xray_metrics_{}_{}_{}.csv", + self.wg, self.test_type, self.packet_count + ) + }) + } + + fn pcap_path(&self) -> String { + self.pcap_path.as_ref().cloned().unwrap_or_else(|| { + format!( + "results/{WG_NAME}_{}_{}_{}.csv", + self.wg, self.test_type, self.packet_count + ) + }) } } @@ -96,8 +114,11 @@ async fn main() -> EyreResult<()> { color_eyre::install()?; let cli_args = CliArgs::parse(); + let test_type = cli_args.test_type.clone(); let packet_count = cli_args.packet_count; + let csv_path = cli_args.csv_path(); + let pcap_path = cli_args.pcap_path(); let wg_keys = KeyPair::new(); let peer_keys = KeyPair::new(); @@ -147,8 +168,36 @@ async fn main() -> EyreResult<()> { } } cmd_tx.send(TestCmd::Done).await?; + let mut event_loop = task.await.expect("Awaiting task should be successful")?; + + run_command("killall -w tcpdump".to_owned()) + .map_err(|s| XRayError::ShellCommand(s.to_owned()))?; + let pcap_packets = process_pcap(&pcap_path, event_loop.tunn())?; + + let allowed_ports = [WG_PORT, PLAINTEXT_PORT, CRYPTO_PORT]; + let mut packets = event_loop.packets; + for p in pcap_packets { + if !allowed_ports.contains(&p.src.port()) || !allowed_ports.contains(&p.dst.port()) { + continue; + } + match (test_type.as_str(), p.src.port(), p.dst.port()) { + ("crypto", CRYPTO_PORT, WG_PORT) => { + packets[p.send_index as usize].pre_wg_ts = Some(p.ts) + } + ("crypto", CRYPTO_PORT, PLAINTEXT_PORT) => { + packets[p.send_index as usize].post_wg_ts = Some(p.ts) + } + ("plaintext", PLAINTEXT_PORT, CRYPTO_PORT) => { + packets[p.send_index as usize].pre_wg_ts = Some(p.ts) + } + ("plaintext", WG_PORT, CRYPTO_PORT) => { + packets[p.send_index as usize].post_wg_ts = Some(p.ts) + } + params => println!("Unexpected pcap packet found: {params:?}"), + } + } - task.await.expect("Awaiting task should be successful")?; + write_to_csv(&csv_path, &packets)?; Ok(()) } diff --git a/xray/src/pcap.rs b/xray/src/pcap.rs new file mode 100644 index 0000000..33d0629 --- /dev/null +++ b/xray/src/pcap.rs @@ -0,0 +1,104 @@ +use std::net::SocketAddrV4; + +use neptun::noise::Tunn; +use pcap::Capture; +use pnet::packet::{ + ethernet::EtherTypes, ip::IpNextHeaderProtocols, ipv4::Ipv4Packet, sll2::SLL2Packet, + udp::UdpPacket, Packet, +}; + +use crate::{client::Client, utils::Packet as XrayPacket, XRayResult}; + +#[derive(Debug)] +pub struct PcapPacket { + pub ts: u128, + pub src: SocketAddrV4, + pub dst: SocketAddrV4, + pub was_decrypted: bool, + pub send_index: u64, +} + +pub fn process_pcap(pcap_path: &str, mut tunn: Tunn) -> XRayResult> { + let mut packets = Vec::new(); + + let mut capture = Capture::from_file(pcap_path)?; + let mut decrypt_buf = vec![0; 1024]; + + while let Ok(packet) = capture.next_packet() { + let ts = packet.header.ts; + let ts = ts.tv_sec as u128 * 1_000_000 + ts.tv_usec as u128; + if let Some(packet) = process_packet(packet.data, ts, &mut tunn, &mut decrypt_buf) { + packets.push(packet); + } + } + + Ok(packets) +} + +fn process_packet( + packet_data: &[u8], + ts: u128, + tunn: &mut Tunn, + buf: &mut [u8], +) -> Option { + let sll2_packet = match SLL2Packet::new(packet_data) { + Some(packet) if matches!(packet.get_protocol_type(), EtherTypes::Ipv4) => packet, + _ => return None, + }; + + let ipv4_packet = match Ipv4Packet::new(sll2_packet.payload()) { + Some(packet) if matches!(packet.get_next_level_protocol(), IpNextHeaderProtocols::Udp) => { + packet + } + _ => return None, + }; + + let udp_packet = match UdpPacket::new(ipv4_packet.payload()) { + Some(packet) => packet, + _ => return None, + }; + + let src = SocketAddrV4::new(ipv4_packet.get_source(), udp_packet.get_source()); + let dst = SocketAddrV4::new(ipv4_packet.get_destination(), udp_packet.get_destination()); + + process_udp_packet(&udp_packet, tunn, buf).map(|(was_decrypted, send_index)| PcapPacket { + ts, + src, + dst, + was_decrypted, + send_index, + }) +} + +fn process_udp_packet( + udp_packet: &UdpPacket, + tunn: &mut Tunn, + buf: &mut [u8], +) -> Option<(bool, u64)> { + if udp_packet.payload().len() == XrayPacket::send_size() { + let send_index = extract_send_index(udp_packet.payload()); + Some((false, send_index)) + } else if udp_packet.payload().starts_with(&[4, 0, 0, 0]) { + let decrypted_packet = match tunn.decrypt(udp_packet.payload(), buf) { + Ok(packet) => packet, + Err(_) => return None, + }; + match Client::parse_udp_packet(decrypted_packet) { + Ok((_, start, end)) if end - start == XrayPacket::send_size() => { + let send_index = extract_send_index(&decrypted_packet[start..]); + Some((true, send_index)) + } + _ => None, + } + } else { + None + } +} + +fn extract_send_index(bytes: &[u8]) -> u64 { + u64::from_le_bytes( + bytes[0..8] + .try_into() + .expect("Slice should have exactly 8 bytes"), + ) +} diff --git a/xray/src/utils.rs b/xray/src/utils.rs index 61e5e76..e89c987 100644 --- a/xray/src/utils.rs +++ b/xray/src/utils.rs @@ -3,6 +3,7 @@ use std::{ process::Command, }; +use serde::Serialize; use tokio::net::UnixStream; use crate::{ @@ -28,14 +29,23 @@ pub enum RecvType { /// A `send_index` is not stored in the packet since they are added, in order, to a vector when they're sent /// so their index in that vector accurately represents the send index -#[derive(Copy, Clone)] +#[derive(Copy, Clone, Default, Serialize)] pub struct Packet { - pub send_ts: u128, pub recv_index: Option, + pub send_ts: u128, + pub pre_wg_ts: Option, + pub post_wg_ts: Option, pub recv_ts: Option, } impl Packet { + pub fn new(send_ts: u128) -> Self { + Self { + send_ts, + ..Default::default() + } + } + pub const fn send_size() -> usize { std::mem::size_of::() + std::mem::size_of::() } @@ -69,12 +79,12 @@ pub fn run_command(cmd: String) -> Result { Ok(output) => { if output.status.success() { Ok(format!( - "Command ran successfully with output: {}", + "Command ran successfully with output: \n{}", String::from_utf8(output.stdout).expect("Command output should be valid utf-8") )) } else { Err(format!( - "Command failed with output: {}", + "Command failed with output: \n{}", String::from_utf8(output.stderr).expect("Command output should be valid utf-8") )) } @@ -85,21 +95,13 @@ pub fn run_command(cmd: String) -> Result { pub fn write_to_csv(name: &str, packets: &[Packet]) -> XRayResult<()> { let file = std::fs::File::create(name)?; - let mut csv = csv::Writer::from_writer(file); - - csv.write_record(["Recv Index", "Send TS", "Recv TS"])?; - for info in packets { - csv.write_record([ - info.recv_index - .map(|i| i.to_string()) - .unwrap_or_else(String::new), - info.send_ts.to_string(), - info.recv_ts - .map(|ts| ts.to_string()) - .unwrap_or_else(String::new), - ])?; + let mut writer = csv::Writer::from_writer(file); + + for packet in packets { + writer.serialize(packet)?; } - csv.flush()?; + + writer.flush()?; Ok(()) }