diff --git a/boringtun/benches/ffi_benches/.gitignore b/boringtun/benches/ffi_benches/.gitignore new file mode 100644 index 000000000..c23e1d0c7 --- /dev/null +++ b/boringtun/benches/ffi_benches/.gitignore @@ -0,0 +1,3 @@ +*.o +ffi-bench +release/ diff --git a/boringtun/benches/ffi_benches/Makefile b/boringtun/benches/ffi_benches/Makefile new file mode 100644 index 000000000..a29ecc15a --- /dev/null +++ b/boringtun/benches/ffi_benches/Makefile @@ -0,0 +1,38 @@ +SRCDIR := $(dir $(lastword ${MAKEFILE_LIST})) +PKGDIR := $(realpath ${SRCDIR}/../..) +OBJDIR := $(shell pwd) +HOST_OS := $(shell uname) + +ifeq (${HOST_OS},Darwin) +MACOSX_MAJOR_VERSION := $(shell sw_vers --productVersion | cut -d. -f1) +export MACOSX_DEPLOYMENT_TARGET=${MACOSX_MAJOR_VERSION}.0 +endif + +release/libboringtun.a: ${PKGDIR}/Cargo.toml + cd ${PKGDIR} && cargo build --lib --release --target-dir ${OBJDIR} --features ffi-bindings + @echo "$@: $$(cut -d: -f2- release/libboringtun.d)" > release/libboringtun-fixup.d + +-include release/libboringtun-fixup.d + +BENCH_CFLAGS := -I ${PKGDIR}/src +BENCH_SRCS := main.c wg_bench_client.c +BENCH_OBJS := $(patsubst %.c,%.o,${BENCH_SRCS}) + +# Build with profiling support +BENCH_CFLAGS += -g -fno-omit-frame-pointer +BENCH_LDFLAGS := -g -fno-omit-frame-pointer +release/libboringtun.a: export RUSTFLAGS=-C force-frame-pointers=y + +%.o : %.c + ${CC} ${CFLAGS} ${BENCH_CFLAGS} -c -o $@ $< + +ffi-bench: ${BENCH_OBJS} release/libboringtun.a + ${CC} ${LDLAGS} ${BENCH_LDFLAGS} -o $@ $^ + +clean: + rm -f ${OBJDIR}/*.o + rm -rf ${OBJDIR}/release + rm -f ${OBJDIR}/ffi-bench + +.PHONY: clean +.DEFAULT_GOAL := ffi-bench diff --git a/boringtun/benches/ffi_benches/main.c b/boringtun/benches/ffi_benches/main.c new file mode 100644 index 000000000..a00e9632f --- /dev/null +++ b/boringtun/benches/ffi_benches/main.c @@ -0,0 +1,295 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "wireguard_ffi.h" +#include "wg_bench_client.h" + +// Exiting because of a signal. +static int caught_sigint = 0; + +static void wg_printf(const char* format, ...) { + struct winsize ws; + if (ioctl(STDOUT_FILENO, TIOCGWINSZ, &ws) == 0) { + // If on a terminal - clear the line and reset before printing. + char fmtbuf[ws.ws_col + strlen(format) + 3]; + fmtbuf[0] = '\r'; + memset(&fmtbuf[1], ' ', ws.ws_col); + fmtbuf[ws.ws_col+1] = '\r'; + strcpy(&fmtbuf[2+ws.ws_col], format); + + va_list args; + va_start(args, format); + vprintf(fmtbuf, args); + va_end(args); + } else { + // Otherwise, just print it. + va_list args; + va_start(args, format); + vprintf(format, args); + va_end(args); + } +} + +static void wg_print_msg(const char* msg) { + wg_printf("%s", msg); +} + +static void handle_signal(int sig) { + switch (sig) { + case SIGINT: + wg_printf("benchmark interrupted\n"); + caught_sigint = 1; + break; + + case SIGTERM: + wg_printf("benchmark terminated\n"); + caught_sigint = 1; + break; + + case SIGHUP: + // Do nothing. + break; + } +} + +static long timespec_cmp(const struct timespec *a, const struct timespec* b) { + long i = (a->tv_sec - b->tv_sec); + return i ? i : (a->tv_nsec - b->tv_nsec); +} + +static double timespec_elapsed(const struct timespec *a, const struct timespec* b) { + double result = (a->tv_sec - b->tv_sec) * 1000000000.0; + return (double)(result + a->tv_nsec - b->tv_nsec) / 1000000000.0; +} + +static const char* print_bytes(uintmax_t value, char* buffer, size_t bufsize) { + const char* suffix = NULL; + double vfloat; + if (value > 1000000000) { + suffix = "GB"; + vfloat = value / 1000000000.0; + } else if (value > 1000000) { + suffix = "MB"; + vfloat = value / 1000000.0; + } else if (value > 1000.0) { + suffix = "kB"; + vfloat = value / 1000.0; + } else { + suffix = "B"; + vfloat = (double)value; + } + + int len = snprintf(buffer, bufsize, "%.3F %s", vfloat, suffix); + return buffer; +} + +static void print_stats(const struct wg_bench_statistics* st, double walltime, double cputime) { + uintmax_t total_errors = 0; + uintmax_t total_bytes = atomic_load(&st->tx_bytes) + atomic_load(&st->rx_bytes); + for (int i = 0; i < WG_BENCH_MAX_ERRORS; i++) { + total_errors += atomic_load(&st->errors[i]); + } + + char loadbuf[16]; + snprintf(loadbuf, sizeof(loadbuf), "%.1F%%", 100.0 * cputime / walltime); + + // Estimate the total throughput. + char xfer[32]; + char tpbuf[32]; + uintmax_t throughput = total_bytes / walltime; + + // Prepare the status to write. + char linebuf[120]; + int len = snprintf(linebuf, sizeof(linebuf), + " tx:%-12lu rx:%-12lu drops:%-8lu err:%-8lu load:%8s %s transferred (%s/s)", + atomic_load(&st->tx_packets), atomic_load(&st->rx_packets), atomic_load(&st->tx_drops), + total_errors, loadbuf, print_bytes(total_bytes, xfer, sizeof(xfer)), + print_bytes(total_bytes / walltime, tpbuf, sizeof(tpbuf))); + struct winsize ws; + if (ioctl(STDOUT_FILENO, TIOCGWINSZ, &ws) == 0) { + // If we are on an interactive terminal, refresh the status line + dprintf(STDOUT_FILENO, "\r%s%*s\r", linebuf, ws.ws_col - len - 1, ""); + } else { + // Some other file. + printf("%s\n", linebuf); + } +} + +static void print_errors(const struct wg_bench_statistics* st) { + // From errors.rs + const char* names[] = { + "DestinationBufferTooSmall", + "IncorrectPacketLength", + "UnexpectedPacket", + "WrongPacketType", + "WrongIndex", + "WrongKey", + "InvalidTai64nTimestamp", + "WrongTai64nTimestamp", + "InvalidMac", + "InvalidAeadTag", + "InvalidCounter", + "DuplicateCounter", + "InvalidPacket", + "NoCurrentSession", + "LockFailed", + "ConnectionExpired", + "UnderLoad", + }; + const int maxerr = sizeof(names)/sizeof(char*); + + int maxname = 0; + for (int i = 0; i < maxerr; i++) { + if (strlen(names[i]) > maxname) { + maxname = strlen(names[i]); + } + } + + wg_printf("\nError Report:\n"); + for (int i = 0; i < maxerr; i++) { + wg_printf(" %*s: %lu\n", maxname, names[i], atomic_load(&st->errors[i])); + } +} + +static void print_usage(FILE* fp, const char* name) { + fprintf(fp, "Usage: %s [OPTIONS]\n", name); + fprintf(fp, "Run FFI benchmarks for the boringtun library.\n"); + fprintf(fp, "\n"); + fprintf(fp, "Options:\n"); + fprintf(fp, "\t--time, -t DUR run benchmark for DUR seconds\n"); + fprintf(fp, "\t--jobs, -j NUM create NUM parallel threads\n"); + fprintf(fp, "\t--help, -h display this message and exit\n"); +} + +int main(int argc, char* argv[]) { + const char* shortopts = "t:j:h"; + const struct option longopts[] = { + {"time", required_argument, 0, 't'}, + {"jobs", required_argument, 0, 'j'}, + {"help", no_argument, 0, 'h'}, + {NULL, 0, 0, 0} + }; + unsigned int num_workers = 1; + unsigned int duration = 10; + + // Parse options + while (true) { + int index; + int opt = getopt_long(argc, argv, shortopts, longopts, &index); + if (opt < 0) { + break; + } + + char* endp; + switch (opt) { + case 't': + duration = strtoul(optarg, &endp, 10); + if (*endp != '\0' || (duration == 0)) { + fprintf(stderr, "Invalid duration: %s\n", optarg); + return 1; + } + break; + + case 'j': + num_workers = strtoul(optarg, &endp, 10); + if (*endp != '\0' || (num_workers == 0)) { + fprintf(stderr, "Invalid thread count: %s\n", optarg); + return 1; + } + break; + + case 'h': + print_usage(stdout, argv[0]); + return 0; + + default: + break; + } + } + + srand(time(0)); + set_logging_function(wg_print_msg); + + // The main thread should handle signals. + struct sigaction action = { + .sa_handler = handle_signal, + }; + sigaction(SIGINT, &action, NULL); + sigaction(SIGTERM, &action, NULL); + sigaction(SIGHUP, &action, NULL); + + // Create two benchmark clients. + struct wg_bench_client* a = wg_bench_create(); + struct wg_bench_client* b = wg_bench_create(); + + // Connect the two clients. + wg_bench_connect(a, b); + wg_bench_connect(b, a); + + struct timespec start; + struct timespec cpustart; + struct timespec end; + struct timespec renegotiate; + clock_gettime(CLOCK_MONOTONIC, &start); + clock_gettime(CLOCK_PROCESS_CPUTIME_ID, &cpustart); + end.tv_sec = start.tv_sec + duration; + end.tv_nsec = start.tv_nsec; + renegotiate.tv_sec = start.tv_sec + 1; + renegotiate.tv_nsec = start.tv_nsec; + + // Launch workers. + wg_bench_start_handshake(a); + for (int i = 0; i < num_workers; i++) { + wg_bench_start_worker(a); + wg_bench_start_worker(b); + } + + struct timespec now; + struct timespec cpu; + struct wg_bench_statistics stats; + while (!caught_sigint) { + clock_gettime(CLOCK_MONOTONIC, &now); + clock_gettime(CLOCK_PROCESS_CPUTIME_ID, &cpu); + + // Fetch and render the statistics. + memset(&stats, 0, sizeof(stats)); + wg_bench_fetch_stats(a, &stats); + wg_bench_fetch_stats(b, &stats); + print_stats(&stats, timespec_elapsed(&now, &start), timespec_elapsed(&cpu, &cpustart)); + + // Check for the end condition. + if (timespec_cmp(&end, &now) < 0) { + break; + } + if (timespec_cmp(&renegotiate, &now) < 0) { + renegotiate.tv_sec++; + wg_bench_start_handshake(a); + } + + // Sleep for more data. + usleep(100000); + } + + memset(&stats, 0, sizeof(stats)); + clock_gettime(CLOCK_MONOTONIC, &now); + clock_gettime(CLOCK_PROCESS_CPUTIME_ID, &cpu); + wg_bench_fetch_stats(a, &stats); + wg_bench_fetch_stats(b, &stats); + print_errors(&stats); + print_stats(&stats, timespec_elapsed(&now, &start), timespec_elapsed(&cpu, &cpustart)); + printf("\n"); + + wg_bench_close(a); + wg_bench_close(b); +} diff --git a/boringtun/benches/ffi_benches/wg_bench_client.c b/boringtun/benches/ffi_benches/wg_bench_client.c new file mode 100644 index 000000000..9a96529d0 --- /dev/null +++ b/boringtun/benches/ffi_benches/wg_bench_client.c @@ -0,0 +1,266 @@ +#include "wg_bench_client.h" + +#include +#include +#include +#include +#include +#include +#include + +#define WG_BENCH_MTU 2048 + +static void wg_worker_sigmask() { + sigset_t sigset; + sigemptyset(&sigset); + sigaddset(&sigset, SIGINT); + sigaddset(&sigset, SIGTERM); + pthread_sigmask(SIG_BLOCK, &sigset, NULL); + + sigemptyset(&sigset); + sigaddset(&sigset, SIGHUP); + pthread_sigmask(SIG_UNBLOCK, &sigset, NULL); + + pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL); +} + +struct wg_bench_iphdr { + uint8_t vlen; + uint8_t qos; + uint16_t tot_len; + uint16_t id; + uint16_t frag; + uint8_t ttl; + uint8_t protocol; + uint16_t cksum; + uint32_t saddr; + uint32_t daddr; +}; + +struct wg_bench_udphdr { + uint16_t sport; + uint16_t dport; + uint16_t length; + uint16_t cksum; + uint8_t dgram[]; +}; + +static void wg_bench_input_packet(struct wg_bench_client *client, const void *data, size_t len) { + uint8_t plaintext[WG_BENCH_MTU]; + struct wireguard_result result; + result = wireguard_read(client->tunnel, data, len, plaintext, sizeof(plaintext)); + switch (result.op) { + case WIREGUARD_DONE: + break; + + case WIREGUARD_ERROR: + if (result.size < WG_BENCH_MAX_ERRORS) { + atomic_fetch_add(&client->stats.errors[result.size], 1); + } + break; + + case WRITE_TO_NETWORK: + return wg_bench_input_packet(client->peer, plaintext, result.size); + + case WRITE_TO_TUNNEL_IPV4: + case WRITE_TO_TUNNEL_IPV6: + atomic_fetch_add(&client->stats.rx_packets, 1); + atomic_fetch_add(&client->stats.rx_bytes, result.size); + break; + + default: + break; + } +} + +static void* wg_bench_worker(void *arg) { + struct wg_bench_client *client = (struct wg_bench_client *)arg; + struct wireguard_result result; + uint8_t ciphertext[WG_BENCH_MTU + 32]; + uint8_t plaintext[WG_BENCH_MTU]; + + struct wg_bench_iphdr *ip = (struct wg_bench_iphdr*)plaintext; + struct wg_bench_udphdr *udp = (struct wg_bench_udphdr*)(ip+1); + const in_addr_t src = inet_addr("172.16.0.123"); + const in_addr_t dst = inet_addr("172.16.0.1"); + + // Generate a sample IPv4 packet header. + memset(ip, 0, sizeof(struct wg_bench_iphdr)); + ip->vlen = 0x40 + (sizeof(struct wg_bench_iphdr) / 4); + ip->ttl = 64; + ip->protocol = IPPROTO_UDP; + ip->saddr = htonl(src); + ip->daddr = htonl(dst); + udp->sport = 0x1234; + udp->dport = 0x5678; + udp->cksum = 0; + + wg_worker_sigmask(); + while (!atomic_load(&client->worker_shutdown)) { + // Pick a random datagram size between 512-1024 and fill with a random byte. + int dsize = 512 + (rand() % 512); + int pktlen = sizeof(struct wg_bench_iphdr) + sizeof(struct wg_bench_udphdr) + dsize; + memset(udp->dgram, rand() & 0xff, dsize); + udp->length = htons(sizeof(struct wg_bench_udphdr) + dsize); + udp->cksum = 0; + ip->tot_len = htons(pktlen); + ip->cksum = 0; + + // Encrypt the packet. + result = wireguard_write(client->tunnel, plaintext, pktlen, + ciphertext, sizeof(ciphertext)); + switch (result.op) { + case WIREGUARD_DONE: + break; + + case WIREGUARD_ERROR: + if (result.size < WG_BENCH_MAX_ERRORS) { + atomic_fetch_add(&client->stats.errors[result.size], 1); + } + fprintf(stderr, "worker encrypt error: %zu\n", result.size); + break; + + case WRITE_TO_NETWORK: + atomic_fetch_add(&client->stats.tx_packets, 1); + atomic_fetch_add(&client->stats.tx_bytes, pktlen); + wg_bench_input_packet(client->peer, ciphertext, result.size); + break; + + case WRITE_TO_TUNNEL_IPV4: + case WRITE_TO_TUNNEL_IPV6: + // Not expected in this case. + fprintf(stderr, "worker encrypt tunnel\n"); + break; + + default: + fprintf(stderr, "worker encrypt unknown: %d\n", result.op); + break; + } + } + + return NULL; +} + +static void *wg_bench_background(void* arg) { + struct wg_bench_client *client = (struct wg_bench_client *)arg; + uint8_t ciphertext[WG_BENCH_MTU + 32]; + + wg_worker_sigmask(); + while (!atomic_load(&client->worker_shutdown)) { + struct wireguard_result result; + result = wireguard_tick(client->tunnel, ciphertext, sizeof(ciphertext)); + + // Process timeouts and state updates. + switch (result.op) { + case WIREGUARD_DONE: + usleep(100000); + break; + + case WIREGUARD_ERROR: + fprintf(stderr, "worker tick error: %zu\n", result.size); + break; + + case WRITE_TO_NETWORK: + wg_bench_input_packet(client->peer, ciphertext, result.size); + break; + + case WRITE_TO_TUNNEL_IPV4: + case WRITE_TO_TUNNEL_IPV6: + // not expected + fprintf(stderr, "worker tick tunnel"); + break; + + default: + fprintf(stderr, "worker tick unknown: %d\n", result.op); + return NULL; + } + } + + return NULL; +} + +struct wg_bench_client* wg_bench_create() { + struct wg_bench_client* client = calloc(sizeof(struct wg_bench_client), 1); + if (!client) { + return NULL; + } + client->secret = x25519_secret_key(); + client->pubkey = x25519_key_to_base64(x25519_public_key(client->secret)); + return client; +} + +void wg_bench_connect(struct wg_bench_client* client, struct wg_bench_client* peer) { + const char* statickey = x25519_key_to_base64(client->secret); + client->peer = peer; + client->tunnel = new_tunnel(statickey, peer->pubkey, NULL, 5, rand() & 0xffffff); + x25519_key_to_str_free(statickey); + + pthread_create(&client->background, NULL, wg_bench_background, client); +} + +void wg_bench_start_handshake(struct wg_bench_client* client) { + uint8_t ciphertext[WG_BENCH_MTU + 32]; + struct wireguard_result result; + result = wireguard_force_handshake(client->tunnel, ciphertext, sizeof(ciphertext)); + + const struct sockaddr* peer = (const struct sockaddr*)&client->peer; + socklen_t peerlen = sizeof(client->peer); + + // Process timeouts and state updates. + switch (result.op) { + case WIREGUARD_DONE: + break; + + case WIREGUARD_ERROR: + fprintf(stderr, "worker handshake error: %zu\n", result.size); + break; + + case WRITE_TO_NETWORK: + wg_bench_input_packet(client->peer, ciphertext, result.size); + break; + + case WRITE_TO_TUNNEL_IPV4: + case WRITE_TO_TUNNEL_IPV6: + // not expected + fprintf(stderr, "worker handshake tunnel"); + break; + + default: + fprintf(stderr, "worker handshake unknown: %d\n", result.op); + return; + } +} + +void wg_bench_start_worker(struct wg_bench_client* client) { + if (client->worker_count >= WG_BENCH_MAX_THREADS) { + return; + } + if (pthread_create(&client->workers[client->worker_count], NULL, wg_bench_worker, client) != 0) { + return; + } + client->worker_count++; +} + +void wg_bench_fetch_stats(const struct wg_bench_client* client, struct wg_bench_statistics* st) { + atomic_fetch_add(&st->tx_packets, atomic_load(&client->stats.tx_packets)); + atomic_fetch_add(&st->tx_drops, atomic_load(&client->stats.tx_drops)); + atomic_fetch_add(&st->tx_bytes, atomic_load(&client->stats.tx_bytes)); + atomic_fetch_add(&st->rx_packets, atomic_load(&client->stats.rx_packets)); + atomic_fetch_add(&st->rx_bytes, atomic_load(&client->stats.rx_bytes)); + for (int i = 0; i < WG_BENCH_MAX_ERRORS; i++) { + atomic_fetch_add(&st->errors[i], atomic_load(&client->stats.errors[i])); + } +} + +void wg_bench_close(struct wg_bench_client* client) { + // Signal that we are shutting down and terminate workers. + atomic_store(&client->worker_shutdown, true); + pthread_join(client->background, NULL); + for (int i = 0; i < client->worker_count; i++) { + pthread_cancel(client->workers[i]); + pthread_join(client->workers[i], NULL); + } + + x25519_key_to_str_free(client->pubkey); + tunnel_free(client->tunnel); +} diff --git a/boringtun/benches/ffi_benches/wg_bench_client.h b/boringtun/benches/ffi_benches/wg_bench_client.h new file mode 100644 index 000000000..4f1d2feb2 --- /dev/null +++ b/boringtun/benches/ffi_benches/wg_bench_client.h @@ -0,0 +1,43 @@ +#pragma once + +#include +#include +#include "wireguard_ffi.h" + +#define WG_BENCH_MAX_THREADS 256 +#define WG_BENCH_MAX_ERRORS 32 + +struct wg_bench_statistics { + atomic_uintmax_t tx_packets; + atomic_uintmax_t tx_drops; + atomic_uintmax_t tx_bytes; + atomic_uintmax_t rx_packets; + atomic_uintmax_t rx_bytes; + atomic_uintmax_t errors[WG_BENCH_MAX_ERRORS]; +}; + +struct wg_bench_client { + struct wireguard_tunnel *tunnel; + struct x25519_key secret; + const char *pubkey; + + struct wg_bench_client* peer; + + // The packet send and receive worker pool. + int worker_handshake; + atomic_bool worker_shutdown; + int worker_count; + pthread_t background; + pthread_t workers[WG_BENCH_MAX_THREADS]; + + // Statistics. + struct wg_bench_statistics stats; +}; + +struct wg_bench_client* wg_bench_create(); +void wg_bench_connect(struct wg_bench_client* client, struct wg_bench_client* peer); +void wg_bench_start_handshake(struct wg_bench_client* client); +void wg_bench_start_send(struct wg_bench_client* client); +void wg_bench_start_worker(struct wg_bench_client* client); +void wg_bench_fetch_stats(const struct wg_bench_client* client, struct wg_bench_statistics* stats); +void wg_bench_close(struct wg_bench_client* client); diff --git a/boringtun/src/ffi/mod.rs b/boringtun/src/ffi/mod.rs index 1e5a2a9f3..2a2c26a16 100644 --- a/boringtun/src/ffi/mod.rs +++ b/boringtun/src/ffi/mod.rs @@ -11,7 +11,7 @@ use crate::x25519::{PublicKey, StaticSecret}; use base64::{decode, encode}; use hex::encode as encode_hex; use libc::{raise, SIGSEGV}; -use parking_lot::Mutex; +use parking_lot::RwLock; use rand_core::OsRng; use tracing; use tracing_subscriber::fmt; @@ -247,7 +247,7 @@ pub unsafe extern "C" fn new_tunnel( preshared_key: *const c_char, keep_alive: u16, index: u32, -) -> *mut Mutex { +) -> *mut RwLock { let c_str = CStr::from_ptr(static_private); let static_private = match c_str.to_str() { Err(_) => return ptr::null_mut(), @@ -292,7 +292,7 @@ pub unsafe extern "C" fn new_tunnel( Some(keep_alive) }; - let tunnel = Box::new(Mutex::new(Tunn::new( + let tunnel = Box::new(RwLock::new(Tunn::new( private_key, public_key, preshared_key, @@ -313,7 +313,7 @@ pub unsafe extern "C" fn new_tunnel( /// Drops the Tunn object #[no_mangle] -pub unsafe extern "C" fn tunnel_free(tunnel: *mut Mutex) { +pub unsafe extern "C" fn tunnel_free(tunnel: *mut RwLock) { drop(Box::from_raw(tunnel)); } @@ -321,33 +321,59 @@ pub unsafe extern "C" fn tunnel_free(tunnel: *mut Mutex) { /// For more details check noise::tunnel_to_network functions. #[no_mangle] pub unsafe extern "C" fn wireguard_write( - tunnel: *const Mutex, + tunnel: *const RwLock, src: *const u8, src_size: u32, dst: *mut u8, dst_size: u32, ) -> wireguard_result { - let mut tunnel = tunnel.as_ref().unwrap().lock(); // Slices are not owned, and therefore will not be freed by Rust let src = slice::from_raw_parts(src, src_size as usize); let dst = slice::from_raw_parts_mut(dst, dst_size as usize); - wireguard_result::from(tunnel.encapsulate(src, dst)) + + // Try handling the packet with only a read lock, this covers the common + // case where we are encrypting data packets and a valid session exists. + { + let rotunnel = tunnel.as_ref().unwrap().read(); + let result = rotunnel.try_encapsulate(src, dst); + if !matches!(result, TunnResult::Done) { + return wireguard_result::from(result); + } + } + + // Otherwise, acquire a write lock to queue the packet and start a new + // handshake if there isn't already one in progress. + let mut tunnel = tunnel.as_ref().unwrap().write(); + tunnel.queue_packet(src); + wireguard_result::from(tunnel.format_handshake_initiation(dst, false)) } /// Read a UDP packet from the server. /// For more details check noise::network_to_tunnel functions. #[no_mangle] pub unsafe extern "C" fn wireguard_read( - tunnel: *const Mutex, + tunnel: *const RwLock, src: *const u8, src_size: u32, dst: *mut u8, dst_size: u32, ) -> wireguard_result { - let mut tunnel = tunnel.as_ref().unwrap().lock(); // Slices are not owned, and therefore will not be freed by Rust let src = slice::from_raw_parts(src, src_size as usize); let dst = slice::from_raw_parts_mut(dst, dst_size as usize); + + // Try handling the packet with only a read lock, this covers the common + // case where we are processing data packets and doing validity checks. + { + let rotunnel = tunnel.as_ref().unwrap().read(); + if let Some(result) = rotunnel.try_decapsulate(None, src, dst) { + return wireguard_result::from(result); + } + } + + // Otherwise, we must acquire a write lock to continue processing this + // packet. This is likely a verified handshake packet of some sort. + let mut tunnel = tunnel.as_ref().unwrap().write(); wireguard_result::from(tunnel.decapsulate(None, src, dst)) } @@ -355,11 +381,11 @@ pub unsafe extern "C" fn wireguard_read( /// Recommended interval: 100ms. #[no_mangle] pub unsafe extern "C" fn wireguard_tick( - tunnel: *const Mutex, + tunnel: *const RwLock, dst: *mut u8, dst_size: u32, ) -> wireguard_result { - let mut tunnel = tunnel.as_ref().unwrap().lock(); + let mut tunnel = tunnel.as_ref().unwrap().write(); // Slices are not owned, and therefore will not be freed by Rust let dst = slice::from_raw_parts_mut(dst, dst_size as usize); wireguard_result::from(tunnel.update_timers(dst)) @@ -368,11 +394,11 @@ pub unsafe extern "C" fn wireguard_tick( /// Force the tunnel to initiate a new handshake, dst buffer must be at least 148 byte long. #[no_mangle] pub unsafe extern "C" fn wireguard_force_handshake( - tunnel: *const Mutex, + tunnel: *const RwLock, dst: *mut u8, dst_size: u32, ) -> wireguard_result { - let mut tunnel = tunnel.as_ref().unwrap().lock(); + let mut tunnel = tunnel.as_ref().unwrap().write(); // Slices are not owned, and therefore will not be freed by Rust let dst = slice::from_raw_parts_mut(dst, dst_size as usize); wireguard_result::from(tunnel.format_handshake_initiation(dst, true)) @@ -383,8 +409,8 @@ pub unsafe extern "C" fn wireguard_force_handshake( /// Number of data bytes encapsulated /// Number of data bytes decapsulated #[no_mangle] -pub unsafe extern "C" fn wireguard_stats(tunnel: *const Mutex) -> stats { - let tunnel = tunnel.as_ref().unwrap().lock(); +pub unsafe extern "C" fn wireguard_stats(tunnel: *const RwLock) -> stats { + let tunnel = tunnel.as_ref().unwrap().read(); let (time, tx_bytes, rx_bytes, estimated_loss, estimated_rtt) = tunnel.stats(); stats { time_since_last_handshake: time.map(|t| t.as_secs() as i64).unwrap_or(-1), diff --git a/boringtun/src/jni.rs b/boringtun/src/jni.rs index 7bc2bdcd7..5ed9d57b5 100644 --- a/boringtun/src/jni.rs +++ b/boringtun/src/jni.rs @@ -12,7 +12,7 @@ use jni::objects::{JByteBuffer, JClass, JString}; use jni::strings::JNIStr; use jni::sys::{jbyteArray, jint, jlong, jshort, jstring}; use jni::JNIEnv; -use parking_lot::Mutex; +use parking_lot::RwLock; use crate::ffi::new_tunnel; use crate::ffi::wireguard_read; @@ -194,7 +194,7 @@ pub unsafe extern "C" fn encrypt_raw_packet( }; let output: wireguard_result = wireguard_write( - tunnel as *const Mutex, + tunnel as *const RwLock, env.convert_byte_array(src).unwrap().as_mut_ptr(), src_size as u32, dst_ptr, @@ -229,7 +229,7 @@ pub unsafe extern "C" fn decrypt_to_raw_packet( }; let output: wireguard_result = wireguard_read( - tunnel as *const Mutex, + tunnel as *const RwLock, env.convert_byte_array(src).unwrap().as_mut_ptr(), src_size as u32, dst_ptr, @@ -263,7 +263,7 @@ pub unsafe extern "C" fn run_periodic_task( }; let output: wireguard_result = - wireguard_tick(tunnel as *const Mutex, dst_ptr, dst_size as u32); + wireguard_tick(tunnel as *const RwLock, dst_ptr, dst_size as u32); *op_ptr = output.op as u8; diff --git a/boringtun/src/noise/mod.rs b/boringtun/src/noise/mod.rs index 76e377b63..1c0767c66 100644 --- a/boringtun/src/noise/mod.rs +++ b/boringtun/src/noise/mod.rs @@ -14,6 +14,7 @@ use crate::noise::rate_limiter::RateLimiter; use crate::noise::timers::{TimerName, Timers}; use crate::x25519; +use portable_atomic::{AtomicUsize, Ordering}; use std::collections::VecDeque; use std::convert::{TryFrom, TryInto}; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; @@ -63,13 +64,13 @@ pub struct Tunn { /// The N_SESSIONS most recent sessions, index is session id modulo N_SESSIONS sessions: [Option; N_SESSIONS], /// Index of most recently used session - current: usize, + current: AtomicUsize, /// Queue to store blocked packets packet_queue: VecDeque>, /// Keeps tabs on the expiring timers timers: timers::Timers, - tx_bytes: usize, - rx_bytes: usize, + tx_bytes: AtomicUsize, + rx_bytes: AtomicUsize, rate_limiter: Arc, } @@ -241,14 +242,14 @@ impl Tunn { } } - /// Encapsulate a single packet from the tunnel interface. + /// Encapsulate a single packet from the tunnel interface or queue it. /// Returns TunnResult. /// /// # Panics /// Panics if dst buffer is too small. /// Size of dst should be at least src.len() + 32, and no less than 148 bytes. pub fn encapsulate<'a>(&mut self, src: &[u8], dst: &'a mut [u8]) -> TunnResult<'a> { - let current = self.current; + let current = self.current.load(Ordering::Relaxed); if let Some(ref session) = self.sessions[current % N_SESSIONS] { // Send the packet using an established session let packet = session.format_packet_data(src, dst); @@ -257,7 +258,7 @@ impl Tunn { if !src.is_empty() { self.timer_tick(TimerName::TimeLastDataPacketSent); } - self.tx_bytes += src.len(); + self.tx_bytes.fetch_add(src.len(), Ordering::Relaxed); return TunnResult::WriteToNetwork(packet); } @@ -267,6 +268,30 @@ impl Tunn { self.format_handshake_initiation(dst, false) } + /// Encapsulate a single packet from the tunnel interface. + /// Returns TunnResult::WriteToNetwork if there is a valid session, or + /// TunnResult::Done if the packet could not be handled yet. + /// + /// # Panics + /// Panics if dst buffer is too small. + /// Size of dst should be at least src.len() + 32, and no less than 148 bytes. + pub fn try_encapsulate<'a>(&self, src: &[u8], dst: &'a mut [u8]) -> TunnResult<'a> { + let current = self.current.load(Ordering::Relaxed); + if let Some(ref session) = self.sessions[current % N_SESSIONS] { + // Send the packet using an established session + let packet = session.format_packet_data(src, dst); + self.timer_tick(TimerName::TimeLastPacketSent); + // Exclude Keepalive packets from timer update. + if !src.is_empty() { + self.timer_tick(TimerName::TimeLastDataPacketSent); + } + self.tx_bytes.fetch_add(src.len(), Ordering::Relaxed); + return TunnResult::WriteToNetwork(packet); + } + + TunnResult::Done + } + /// Receives a UDP datagram from the network and parses it. /// Returns TunnResult. /// @@ -301,6 +326,47 @@ impl Tunn { self.handle_verified_packet(packet, dst) } + /// Receives a UDP datagram from the network and parses it. + /// Returns TunnResult. + /// + /// This is a subset of decapsulate that operates on a read only tunnel. + /// Will return Some(TunnResult) if the packet was handled successfully, or + /// None if processing requires a mutable reference to the tunnel. + /// + /// This method can handle the common case of data packet decryption and + /// cookie verification without needing to hold a write lock on the tunnel. + pub fn try_decapsulate<'a>( + &self, + src_addr: Option, + datagram: &[u8], + dst: &'a mut [u8], + ) -> Option> { + // Packet dequeue operations require a write lock. + if datagram.is_empty() { + return if self.packet_queue.is_empty() { + Some(TunnResult::Done) + } else { + None + }; + } + + let mut cookie = [0u8; COOKIE_REPLY_SZ]; + match self + .rate_limiter + .verify_packet(src_addr, datagram, &mut cookie) + { + Ok(Packet::PacketData(p)) => { + Some(self.handle_data(p, dst).unwrap_or_else(TunnResult::from)) + } + Err(TunnResult::WriteToNetwork(cookie)) => { + dst[..cookie.len()].copy_from_slice(cookie); + return Some(TunnResult::WriteToNetwork(&mut dst[..cookie.len()])); + } + Err(TunnResult::Err(e)) => return Some(TunnResult::Err(e)), + _ => None, + } + } + pub(crate) fn handle_verified_packet<'a>( &mut self, packet: Packet, @@ -387,24 +453,31 @@ impl Tunn { } /// Update the index of the currently used session, if needed - fn set_current_session(&mut self, new_idx: usize) { - let cur_idx = self.current; + fn set_current_session(&self, new_idx: usize) { + let cur_idx = self.current.load(Ordering::Relaxed); if cur_idx == new_idx { // There is nothing to do, already using this session, this is the common case return; } + if self.sessions[cur_idx % N_SESSIONS].is_none() || self.timers.session_timers[new_idx % N_SESSIONS] >= self.timers.session_timers[cur_idx % N_SESSIONS] { - self.current = new_idx; - tracing::debug!(message = "New session", session = new_idx); + if let Ok(idx) = self.current.compare_exchange( + cur_idx, + new_idx, + Ordering::Acquire, + Ordering::Relaxed, + ) { + tracing::debug!(message = "New session", session = idx); + } } } /// Decrypts a data packet, and stores the decapsulated packet in dst. fn handle_data<'a>( - &mut self, + &self, packet: PacketData, dst: &'a mut [u8], ) -> Result, WireGuardError> { @@ -461,7 +534,7 @@ impl Tunn { /// Check if an IP packet is v4 or v6, truncate to the length indicated by the length field /// Returns the truncated packet and the source IP as TunnResult - fn validate_decapsulated_packet<'a>(&mut self, packet: &'a mut [u8]) -> TunnResult<'a> { + fn validate_decapsulated_packet<'a>(&self, packet: &'a mut [u8]) -> TunnResult<'a> { let (computed_len, src_ip_address) = match packet.len() { 0 => return TunnResult::Done, // This is keepalive, and not an error _ if packet[0] >> 4 == 4 && packet.len() >= IPV4_MIN_HEADER_SIZE => { @@ -498,7 +571,7 @@ impl Tunn { } self.timer_tick(TimerName::TimeLastDataPacketReceived); - self.rx_bytes += computed_len; + self.rx_bytes.fetch_add(computed_len, Ordering::Relaxed); match src_ip_address { IpAddr::V4(addr) => TunnResult::WriteToTunnelV4(&mut packet[..computed_len], addr), @@ -521,7 +594,7 @@ impl Tunn { } /// Push packet to the back of the queue - fn queue_packet(&mut self, packet: &[u8]) { + pub fn queue_packet(&mut self, packet: &[u8]) { if self.packet_queue.len() < MAX_QUEUE_DEPTH { // Drop if too many are already in queue self.packet_queue.push_back(packet.to_vec()); @@ -541,7 +614,7 @@ impl Tunn { } fn estimate_loss(&self) -> f32 { - let session_idx = self.current; + let session_idx = self.current.load(Ordering::Relaxed); let mut weight = 9.0; let mut cur_avg = 0.0; @@ -576,8 +649,8 @@ impl Tunn { /// * Data bytes received pub fn stats(&self) -> (Option, usize, usize, f32, Option) { let time = self.time_since_last_handshake(); - let tx_bytes = self.tx_bytes; - let rx_bytes = self.rx_bytes; + let tx_bytes = self.tx_bytes.load(Ordering::Relaxed); + let rx_bytes = self.rx_bytes.load(Ordering::Relaxed); let loss = self.estimate_loss(); let rtt = self.handshake.last_rtt; diff --git a/boringtun/src/noise/rate_limiter.rs b/boringtun/src/noise/rate_limiter.rs index 421d5680f..fcecc7a9c 100644 --- a/boringtun/src/noise/rate_limiter.rs +++ b/boringtun/src/noise/rate_limiter.rs @@ -2,6 +2,8 @@ use super::handshake::{b2s_hash, b2s_keyed_mac_16, b2s_keyed_mac_16_2, b2s_mac_2 use crate::noise::handshake::{LABEL_COOKIE, LABEL_MAC1}; use crate::noise::{HandshakeInit, HandshakeResponse, Packet, Tunn, TunnResult, WireGuardError}; +use std::time::Duration; + #[cfg(feature = "mock-instant")] use mock_instant::Instant; use portable_atomic::{AtomicU64, Ordering}; @@ -13,7 +15,6 @@ use crate::sleepyinstant::Instant; use aead::generic_array::GenericArray; use aead::{AeadInPlace, KeyInit}; use chacha20poly1305::{Key, XChaCha20Poly1305}; -use parking_lot::Mutex; use rand_core::{OsRng, RngCore}; use ring::constant_time::verify_slices_are_equal; @@ -22,7 +23,7 @@ const COOKIE_SIZE: usize = 16; const COOKIE_NONCE_SIZE: usize = 24; /// How often should reset count in seconds -const RESET_PERIOD: u64 = 1; +const RESET_PERIOD: Duration = Duration::from_secs(1); type Cookie = [u8; COOKIE_SIZE]; @@ -47,8 +48,8 @@ pub struct RateLimiter { limit: u64, /// The counter since last reset count: AtomicU64, - /// The time last reset was performed on this rate limiter - last_reset: Mutex, + /// The time last reset was performed on this rate limiter, in milliseconds from start_time + last_reset: AtomicU64, } impl RateLimiter { @@ -64,7 +65,7 @@ impl RateLimiter { cookie_key: b2s_hash(LABEL_COOKIE, public_key.as_bytes()).into(), limit, count: AtomicU64::new(0), - last_reset: Mutex::new(Instant::now()), + last_reset: AtomicU64::new(0), } } @@ -77,11 +78,22 @@ impl RateLimiter { /// Reset packet count (ideally should be called with a period of 1 second) pub fn reset_count(&self) { // The rate limiter is not very accurate, but at the scale we care about it doesn't matter much - let current_time = Instant::now(); - let mut last_reset_time = self.last_reset.lock(); - if current_time.duration_since(*last_reset_time).as_secs() >= RESET_PERIOD { - self.count.store(0, Ordering::SeqCst); - *last_reset_time = current_time; + let now = Instant::now().duration_since(self.start_time); + let last_msec = self.last_reset.load(Ordering::Acquire); + let last_reset = Duration::from_millis(last_msec); + if now - last_reset >= RESET_PERIOD { + if self + .last_reset + .compare_exchange( + last_msec, + now.as_millis() as u64, + Ordering::SeqCst, + Ordering::Relaxed, + ) + .is_ok() + { + self.count.store(0, Ordering::SeqCst); + } } } diff --git a/boringtun/src/noise/session.rs b/boringtun/src/noise/session.rs index d547b9819..18299811f 100644 --- a/boringtun/src/noise/session.rs +++ b/boringtun/src/noise/session.rs @@ -3,9 +3,9 @@ use super::PacketData; use crate::noise::errors::WireGuardError; -use parking_lot::Mutex; use portable_atomic::{AtomicU64, Ordering}; use ring::aead::{Aad, LessSafeKey, Nonce, UnboundKey, CHACHA20_POLY1305}; +use std::hint; pub struct Session { pub(crate) receiving_index: u32, @@ -13,7 +13,7 @@ pub struct Session { receiver: LessSafeKey, sender: LessSafeKey, sending_key_counter: AtomicU64, - receiving_key_counter: Mutex, + receiving_key_counter: ReceivingKeyCounterValidator, } impl std::fmt::Debug for Session { @@ -31,122 +31,202 @@ const DATA_OFFSET: usize = 16; /// The overhead of the AEAD const AEAD_SIZE: usize = 16; -// Receiving buffer constants -const WORD_SIZE: u64 = 64; -const N_WORDS: u64 = 16; // Suffice to reorder 64*16 = 1024 packets; can be increased at will -const N_BITS: u64 = WORD_SIZE * N_WORDS; +#[cfg(target_has_atomic = "64")] +type CounterBitmap = AtomicU64; +#[cfg(not(target_has_atomic = "64"))] +use portable_atomic::AtomicU32 as CounterBitmap; -#[derive(Debug, Clone, Default)] +// Receiving buffer constants +const WORD_SIZE: u64 = if cfg!(target_has_atomic = "64") { + 64 +} else { + 32 +}; +const N_BITS: u64 = 1024; +const N_WORDS: usize = (N_BITS / WORD_SIZE) as usize; + +// The most significant bit used as a flag to indicate that the bitmap is being updated, +// and acts effectively as a spinlock. +const COUNTER_LOCK: u64 = 1u64 << 63; +const COUNTER_MASK: u64 = !COUNTER_LOCK; + +#[derive(Debug, Default)] struct ReceivingKeyCounterValidator { /// In order to avoid replays while allowing for some reordering of the packets, we keep a /// bitmap of received packets, and the value of the highest counter - next: u64, + next: AtomicU64, /// Used to estimate packet loss - receive_cnt: u64, - bitmap: [u64; N_WORDS as usize], + receive_cnt: AtomicU64, + bitmap: [CounterBitmap; N_WORDS], } impl ReceivingKeyCounterValidator { - #[inline(always)] - fn set_bit(&mut self, idx: u64) { - let bit_idx = idx % N_BITS; - let word = (bit_idx / WORD_SIZE) as usize; - let bit = (bit_idx % WORD_SIZE) as usize; - self.bitmap[word] |= 1 << bit; + pub const fn new() -> ReceivingKeyCounterValidator { + ReceivingKeyCounterValidator { + next: AtomicU64::new(0), + receive_cnt: AtomicU64::new(0), + bitmap: [const { CounterBitmap::new(0) }; N_WORDS], + } } #[inline(always)] - fn clear_bit(&mut self, idx: u64) { - let bit_idx = idx % N_BITS; - let word = (bit_idx / WORD_SIZE) as usize; - let bit = (bit_idx % WORD_SIZE) as usize; - self.bitmap[word] &= !(1u64 << bit); + fn get_next(&self) -> u64 { + self.next.load(Ordering::SeqCst) & COUNTER_MASK } - /// Clear the word that contains idx + /// Returns true if bit is set, false otherwise #[inline(always)] - fn clear_word(&mut self, idx: u64) { + fn check_bit(&self, idx: u64) -> bool { let bit_idx = idx % N_BITS; let word = (bit_idx / WORD_SIZE) as usize; - self.bitmap[word] = 0; + let bit = bit_idx % WORD_SIZE; + ((self.bitmap[word].load(Ordering::SeqCst) >> bit) & 1) == 1 } - /// Returns true if bit is set, false otherwise + /// Mark the packet as received, release the spinlock and return the verdict. #[inline(always)] - fn check_bit(&self, idx: u64) -> bool { + fn mark_and_unlock(&self, idx: u64) -> Result<(), WireGuardError> { let bit_idx = idx % N_BITS; let word = (bit_idx / WORD_SIZE) as usize; - let bit = (bit_idx % WORD_SIZE) as usize; - ((self.bitmap[word] >> bit) & 1) == 1 + let bit = bit_idx % WORD_SIZE; + let previous = self.bitmap[word].fetch_or(1 << bit, Ordering::SeqCst); + self.next.fetch_and(COUNTER_MASK, Ordering::SeqCst); + if (previous >> bit) & 1 == 1 { + Err(WireGuardError::DuplicateCounter) + } else { + Ok(()) + } } - /// Returns true if the counter was not yet received, and is not too far back + /// Clear all the packets between prev and next. #[inline(always)] - fn will_accept(&self, counter: u64) -> Result<(), WireGuardError> { - if counter >= self.next { - // As long as the counter is growing no replay took place for sure - return Ok(()); + fn clear_range(&self, prev: u64, next: u64) { + if next < prev { + return; } - if counter + N_BITS < self.next { - // Drop if too far back - return Err(WireGuardError::InvalidCounter); + + if next - prev >= N_BITS { + // Too far ahead, clear all the bits + for i in 0..N_WORDS { + self.bitmap[i].store(0, Ordering::SeqCst); + } + return; } - if !self.check_bit(counter) { - Ok(()) + + #[cfg(target_has_atomic = "64")] + const ONE: u64 = 1; + #[cfg(not(target_has_atomic = "64"))] + const ONE: u32 = 1; + + let prev_idx = (prev / WORD_SIZE) as usize; + let prev_mask = (ONE << (prev % WORD_SIZE)) - 1; + let next_idx = (next / WORD_SIZE) as usize; + let next_mask = (!ONE) << (next % WORD_SIZE); + + if next_idx == prev_idx { + // The bits to clear all fit within a single word. + self.bitmap[prev_idx % N_WORDS].fetch_and(next_mask | prev_mask, Ordering::SeqCst); } else { - Err(WireGuardError::DuplicateCounter) + // The bits to clear span multiple words. + self.bitmap[prev_idx % N_WORDS].fetch_and(prev_mask, Ordering::SeqCst); + for i in prev_idx + 1..next_idx { + self.bitmap[i % N_WORDS].store(0, Ordering::SeqCst); + } + self.bitmap[next_idx % N_WORDS].fetch_and(next_mask, Ordering::SeqCst); } } - /// Marks the counter as received, and returns true if it is still good (in case during - /// decryption something changed) + /// Returns true if the counter was not yet received, and is not too far back. #[inline(always)] - fn mark_did_receive(&mut self, counter: u64) -> Result<(), WireGuardError> { - if counter + N_BITS < self.next { - // Drop if too far back + fn will_accept(&self, counter: u64) -> Result<(), WireGuardError> { + if counter >= COUNTER_MASK { + // Too many packets, counter would overflow. return Err(WireGuardError::InvalidCounter); } - if counter == self.next { - // Usually the packets arrive in order, in that case we simply mark the bit and - // increment the counter - self.set_bit(counter); - self.next += 1; - return Ok(()); - } - if counter < self.next { - // A packet arrived out of order, check if it is valid, and mark - if self.check_bit(counter) { + + // Spin while checking the counter until the bitmap is updated, as + // indicated by the COUNTER_LOCK bit being cleared. + let mut next = self.next.load(Ordering::SeqCst); + loop { + if counter >= (next & COUNTER_MASK) { + // As long as the counter is growing no replay took place for sure + return Ok(()); + } + if counter + N_BITS < (next & COUNTER_MASK) { + // Drop if too far back return Err(WireGuardError::InvalidCounter); } - self.set_bit(counter); - return Ok(()); - } - // Packets where dropped, or maybe reordered, skip them and mark unused - if counter - self.next >= N_BITS { - // Too far ahead, clear all the bits - for c in self.bitmap.iter_mut() { - *c = 0; + if (next & COUNTER_LOCK) == 0 { + break; } + next = self.next.load(Ordering::SeqCst); + hint::spin_loop(); + } + + // Check the bitmap for duplicates, then re-check the counter + // one last time in case a race conditioned occurred. + let duplicate = self.check_bit(counter); + let next = self.next.load(Ordering::SeqCst) & COUNTER_MASK; + return if counter + N_BITS < next { + Err(WireGuardError::InvalidCounter) + } else if duplicate { + Err(WireGuardError::DuplicateCounter) } else { - let mut i = self.next; - while i % WORD_SIZE != 0 && i < counter { - // Clear until i aligned to word size - self.clear_bit(i); - i += 1; - } - while i + WORD_SIZE < counter { - // Clear whole word at a time - self.clear_word(i); - i = (i + WORD_SIZE) & 0u64.wrapping_sub(WORD_SIZE); + Ok(()) + }; + } + + /// Marks the counter as received, and returns true if it is still good (in case during + /// decryption something changed) + #[inline(always)] + fn mark_did_receive(&self, counter: u64) -> Result<(), WireGuardError> { + if counter >= COUNTER_MASK { + // Too many packets, counter would overflow. + return Err(WireGuardError::InvalidCounter); + } + + let mut prev = self.next.load(Ordering::Acquire); + loop { + if (counter + N_BITS) < (prev & COUNTER_MASK) { + // Drop if too far back that the packet would fall outside the bitmap. + return Err(WireGuardError::InvalidCounter); } - while i < counter { - // Clear any remaining bits - self.clear_bit(i); - i += 1; + if prev & COUNTER_LOCK == COUNTER_LOCK { + // Someone else has the spinlock. Try again. + prev = self.next.load(Ordering::SeqCst); + } else if counter < prev { + // This packet arrived out of order, acquire the spinlock to mark the packet. + match self.next.compare_exchange_weak( + prev, + prev | COUNTER_LOCK, + Ordering::SeqCst, + Ordering::Acquire, + ) { + Ok(_) => break, + Err(x) => prev = x, + } + } else { + // This packet arrived in-order. + // Acquire the spinlock, update the next packet counter, and clear bits that will wrap over. + match self.next.compare_exchange_weak( + prev, + (counter + 1) | COUNTER_LOCK, + Ordering::SeqCst, + Ordering::Acquire, + ) { + Ok(_) => { + self.clear_range(prev, counter); + break; + } + Err(x) => prev = x, + } } + hint::spin_loop(); } - self.set_bit(counter); - self.next = counter + 1; + + // And finally try to mark the packet, release the spinlock, and return the verdict. + self.mark_and_unlock(counter)?; + self.receive_cnt.fetch_add(1, Ordering::Relaxed); Ok(()) } } @@ -166,7 +246,7 @@ impl Session { ), sender: LessSafeKey::new(UnboundKey::new(&CHACHA20_POLY1305, &sending_key).unwrap()), sending_key_counter: AtomicU64::new(0), - receiving_key_counter: Mutex::new(Default::default()), + receiving_key_counter: ReceivingKeyCounterValidator::new(), } } @@ -174,22 +254,6 @@ impl Session { self.receiving_index as usize } - /// Returns true if receiving counter is good to use - fn receiving_counter_quick_check(&self, counter: u64) -> Result<(), WireGuardError> { - let counter_validator = self.receiving_key_counter.lock(); - counter_validator.will_accept(counter) - } - - /// Returns true if receiving counter is good to use, and marks it as used { - fn receiving_counter_mark(&self, counter: u64) -> Result<(), WireGuardError> { - let mut counter_validator = self.receiving_key_counter.lock(); - let ret = counter_validator.mark_did_receive(counter); - if ret.is_ok() { - counter_validator.receive_cnt += 1; - } - ret - } - /// src - an IP packet from the interface /// dst - pre-allocated space to hold the encapsulating UDP packet to send over the network /// returns the size of the formatted packet @@ -247,7 +311,7 @@ impl Session { return Err(WireGuardError::WrongIndex); } // Don't reuse counters, in case this is a replay attack we want to quickly check the counter without running expensive decryption - self.receiving_counter_quick_check(packet.counter)?; + self.receiving_key_counter.will_accept(packet.counter)?; let ret = { let mut nonce = [0u8; 12]; @@ -263,23 +327,81 @@ impl Session { }; // After decryption is done, check counter again, and mark as received - self.receiving_counter_mark(packet.counter)?; + self.receiving_key_counter + .mark_did_receive(packet.counter)?; Ok(ret) } /// Returns the estimated downstream packet loss for this session pub(super) fn current_packet_cnt(&self) -> (u64, u64) { - let counter_validator = self.receiving_key_counter.lock(); - (counter_validator.next, counter_validator.receive_cnt) + let rx = self + .receiving_key_counter + .receive_cnt + .load(Ordering::Relaxed); + (self.receiving_key_counter.get_next(), rx) } } +#[cfg(test)] +use std::thread; + #[cfg(test)] mod tests { use super::*; + + #[cfg(test)] + fn check_replay_clear_range(start: u64, end: u64) { + // Setup the replay bitmap with all bits set. + let c: ReceivingKeyCounterValidator = Default::default(); + for i in 0..N_WORDS { + c.bitmap[i].store(!0, Ordering::Release); + } + c.next.store(N_BITS, Ordering::Release); + for i in 0..N_BITS { + assert!(c.check_bit(i)); + } + + // Clear a range, and recheck the bitmap. + c.clear_range(start, end); + let check_start = (start / WORD_SIZE) * WORD_SIZE; + for i in check_start..check_start + N_BITS { + if i < start || i > end { + assert!(c.check_bit(i), "expected bit {} to be set", i); + } else { + assert!(!c.check_bit(i), "expected bit {} to be clear", i); + } + } + } + + #[test] + fn test_replay_clear_range() { + // Clear a single bit and check edge cases. + check_replay_clear_range(0, 0); + check_replay_clear_range(42, 42); + check_replay_clear_range(WORD_SIZE - 1, WORD_SIZE - 1); + check_replay_clear_range(WORD_SIZE, WORD_SIZE); + check_replay_clear_range(N_BITS - 1, N_BITS - 1); + check_replay_clear_range(N_BITS, N_BITS); + + // Clear some bits in the middle of a single word. + check_replay_clear_range(13, 19); + + // Clear precisely one word. + check_replay_clear_range(WORD_SIZE, WORD_SIZE * 2 - 1); + + // Clear some bits spanning two words. + check_replay_clear_range(WORD_SIZE + 7, WORD_SIZE * 2 + 13); + + // Clear some bits that span many words. + check_replay_clear_range(WORD_SIZE + 7, N_BITS - 7); + + // Clear some bits that wrap around to the start of the index. + check_replay_clear_range(N_BITS - 7, N_BITS + 7); + } + #[test] fn test_replay_counter() { - let mut c: ReceivingKeyCounterValidator = Default::default(); + let c: ReceivingKeyCounterValidator = Default::default(); assert!(c.mark_did_receive(0).is_ok()); assert!(c.mark_did_receive(0).is_err()); @@ -326,4 +448,92 @@ mod tests { assert!(c.mark_did_receive(N_BITS * 3 + 71).is_err()); assert!(c.mark_did_receive(N_BITS * 3 + 72).is_err()); } + + const RACE_MAX_PACKETS: u64 = 1024 * 1024; + + // Race check worker to try and send valid packets, they should be accepted. + #[cfg(test)] + fn racecheck_counter_worker(counter: &AtomicU64, validator: &ReceivingKeyCounterValidator) { + loop { + let mut value = counter.fetch_add(1, Ordering::Relaxed); + if value > RACE_MAX_PACKETS { + break; + } + + // To drive the out-of-order packet handling a little harder, simulate some packet + // reordering. If the packet is a multiple of 29, increment the value being sent by + // 29. This should still never generate a duplicate but forces the validator to + // interact with the bitmap to figure it out. + if value % 29 == 0 { + value += 29; + } + + match validator.will_accept(value) { + Ok(_) => {} + Err(WireGuardError::InvalidCounter) => { + // This error is allowed if, and only if, the thread hit an + // unlucky interrupt and the counter is too old now. + assert!(validator.get_next() >= value + N_BITS); + continue; + } + Err(WireGuardError::DuplicateCounter) => { + panic!("duplicate while checking {}", value) + } + _ => panic!("error while checking packet {}", value), + }; + + match validator.mark_did_receive(value) { + Ok(_) => {} + Err(WireGuardError::InvalidCounter) => { + // This error is allowed if, and only if, the thread hit an + // unlucky interrupt and the counter is too old now. + assert!(validator.get_next() >= value + N_BITS); + continue; + } + Err(WireGuardError::DuplicateCounter) => { + panic!("duplicate while marking {}", value) + } + _ => panic!("error while marking {}", value), + }; + + // Resend it as a duplicate, it must be rejected. + assert!( + validator.mark_did_receive(value).is_err(), + "race encountered while checking duplicate {}", + value + ); + + thread::yield_now(); + } + } + + // Race check worker to try and send duplicate packets, they must all be rejected. + #[cfg(test)] + fn racecheck_dup_worker(counter: &AtomicU64, validator: &ReceivingKeyCounterValidator) { + while counter.load(Ordering::Relaxed) < RACE_MAX_PACKETS { + let value = validator.get_next(); + if value > 0 { + assert!(validator.mark_did_receive(value - 1).is_err()); + } + } + } + + #[test] + fn test_replay_racecheck() { + static COUNTER: AtomicU64 = AtomicU64::new(0); + static VALIDATOR: ReceivingKeyCounterValidator = ReceivingKeyCounterValidator::new(); + let mut threads = Vec::new(); + let num_threads = thread::available_parallelism().map_or(8, |x| x.get()); + + for _ in 0..num_threads - 1 { + threads.push(thread::spawn(|| { + racecheck_counter_worker(&COUNTER, &VALIDATOR) + })); + } + threads.push(thread::spawn(|| racecheck_dup_worker(&COUNTER, &VALIDATOR))); + + for handle in threads { + handle.join().unwrap(); + } + } } diff --git a/boringtun/src/noise/timers.rs b/boringtun/src/noise/timers.rs index 6b91d5767..a54504e9e 100644 --- a/boringtun/src/noise/timers.rs +++ b/boringtun/src/noise/timers.rs @@ -3,8 +3,7 @@ use super::errors::WireGuardError; use crate::noise::{Tunn, TunnResult}; -use std::mem; -use std::ops::{Index, IndexMut}; +use portable_atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::Duration; @@ -25,8 +24,6 @@ const COOKIE_EXPIRATION_TIME: Duration = Duration::from_secs(120); #[derive(Debug)] pub enum TimerName { - /// Current time, updated each call to `update_timers` - TimeCurrent, /// Time when last handshake was completed TimeSessionEstablished, /// Time the last attempt for a new handshake began @@ -54,12 +51,14 @@ pub struct Timers { is_initiator: bool, /// Start time of the tunnel time_started: Instant, - timers: [Duration; TimerName::Top as usize], + /// Current time, updated each call to `update_timers` + current: Duration, + timers: [AtomicU64; TimerName::Top as usize], pub(super) session_timers: [Duration; super::N_SESSIONS], /// Did we receive data without sending anything back? - want_keepalive: bool, + want_keepalive: AtomicBool, /// Did we send data without hearing back? - want_handshake: bool, + want_handshake: AtomicBool, persistent_keepalive: usize, /// Should this timer call reset rr function (if not a shared rr instance) pub(super) should_reset_rr: bool, @@ -70,10 +69,11 @@ impl Timers { Timers { is_initiator: false, time_started: Instant::now(), + current: Default::default(), timers: Default::default(), session_timers: Default::default(), - want_keepalive: Default::default(), - want_handshake: Default::default(), + want_keepalive: AtomicBool::new(false), + want_handshake: AtomicBool::new(false), persistent_keepalive: usize::from(persistent_keepalive.unwrap_or(0)), should_reset_rr: reset_rr, } @@ -87,43 +87,41 @@ impl Timers { // so the reference time frame is the same pub(super) fn clear(&mut self) { let now = Instant::now().duration_since(self.time_started); + let msec = now.as_millis() as u64; for t in &mut self.timers[..] { - *t = now; + t.store(msec, Ordering::Release); } - self.want_handshake = false; - self.want_keepalive = false; - } -} - -impl Index for Timers { - type Output = Duration; - fn index(&self, index: TimerName) -> &Duration { - &self.timers[index as usize] - } -} - -impl IndexMut for Timers { - fn index_mut(&mut self, index: TimerName) -> &mut Duration { - &mut self.timers[index as usize] + self.want_handshake.store(false, Ordering::Release); + self.want_keepalive.store(false, Ordering::Release); } } impl Tunn { - pub(super) fn timer_tick(&mut self, timer_name: TimerName) { + pub(super) fn timer_tick(&self, timer_name: TimerName) { match timer_name { TimeLastPacketReceived => { - self.timers.want_keepalive = true; - self.timers.want_handshake = false; + self.timers.want_keepalive.store(true, Ordering::Release); + self.timers.want_handshake.store(false, Ordering::Release); } TimeLastPacketSent => { - self.timers.want_handshake = true; - self.timers.want_keepalive = false; + self.timers.want_handshake.store(true, Ordering::Release); + self.timers.want_keepalive.store(false, Ordering::Release); } _ => {} } - let time = self.timers[TimeCurrent]; - self.timers[timer_name] = time; + let index = timer_name as usize; + let current = self.timers.current.as_millis() as u64; + let mut prev = self.timers.timers[index].load(Ordering::Acquire); + while prev < current { + prev = self.timers.timers[index] + .compare_exchange_weak(prev, current, Ordering::SeqCst, Ordering::Acquire) + .unwrap_or_else(|x| x); + } + } + + pub(super) fn timer_fetch(&self, timer_name: TimerName) -> Duration { + Duration::from_millis(self.timers.timers[timer_name as usize].load(Ordering::Acquire)) } pub(super) fn timer_tick_session_established( @@ -132,8 +130,7 @@ impl Tunn { session_idx: usize, ) { self.timer_tick(TimeSessionEstablished); - self.timers.session_timers[session_idx % crate::noise::N_SESSIONS] = - self.timers[TimeCurrent]; + self.timers.session_timers[session_idx % crate::noise::N_SESSIONS] = self.timers.current; self.timers.is_initiator = is_initiator; } @@ -178,17 +175,17 @@ impl Tunn { // All the times are counted from tunnel initiation, for efficiency our timers are rounded // to a second, as there is no real benefit to having highly accurate timers. let now = time.duration_since(self.timers.time_started); - self.timers[TimeCurrent] = now; + self.timers.current = now; self.update_session_timers(now); // Load timers only once: - let session_established = self.timers[TimeSessionEstablished]; - let handshake_started = self.timers[TimeLastHandshakeStarted]; - let aut_packet_received = self.timers[TimeLastPacketReceived]; - let aut_packet_sent = self.timers[TimeLastPacketSent]; - let data_packet_received = self.timers[TimeLastDataPacketReceived]; - let data_packet_sent = self.timers[TimeLastDataPacketSent]; + let session_established = self.timer_fetch(TimeSessionEstablished); + let handshake_started = self.timer_fetch(TimeLastHandshakeStarted); + let aut_packet_received = self.timer_fetch(TimeLastPacketReceived); + let aut_packet_sent = self.timer_fetch(TimeLastPacketSent); + let data_packet_received = self.timer_fetch(TimeLastDataPacketReceived); + let data_packet_sent = self.timer_fetch(TimeLastDataPacketSent); let persistent_keepalive = self.timers.persistent_keepalive; { @@ -198,7 +195,7 @@ impl Tunn { // Clear cookie after COOKIE_EXPIRATION_TIME if self.handshake.has_cookie() - && now - self.timers[TimeCookieReceived] >= COOKIE_EXPIRATION_TIME + && now - self.timer_fetch(TimeCookieReceived) >= COOKIE_EXPIRATION_TIME { self.handshake.clear_cookie(); } @@ -270,7 +267,7 @@ impl Tunn { // we initiate a new handshake. if data_packet_sent > aut_packet_received && now - aut_packet_received >= KEEPALIVE_TIMEOUT + REKEY_TIMEOUT - && mem::replace(&mut self.timers.want_handshake, false) + && self.timers.want_handshake.swap(false, Ordering::AcqRel) { tracing::warn!("HANDSHAKE(KEEPALIVE + REKEY_TIMEOUT)"); handshake_initiation_required = true; @@ -281,7 +278,7 @@ impl Tunn { // to the given peer in KEEPALIVE ms, we send an empty packet. if data_packet_received > aut_packet_sent && now - aut_packet_sent >= KEEPALIVE_TIMEOUT - && mem::replace(&mut self.timers.want_keepalive, false) + && self.timers.want_keepalive.swap(false, Ordering::AcqRel) { tracing::debug!("KEEPALIVE(KEEPALIVE_TIMEOUT)"); keepalive_required = true; @@ -289,7 +286,7 @@ impl Tunn { // Persistent KEEPALIVE if persistent_keepalive > 0 - && (now - self.timers[TimePersistentKeepalive] + && (now - self.timer_fetch(TimePersistentKeepalive) >= Duration::from_secs(persistent_keepalive as _)) { tracing::debug!("KEEPALIVE(PERSISTENT_KEEPALIVE)"); @@ -312,10 +309,10 @@ impl Tunn { } pub fn time_since_last_handshake(&self) -> Option { - let current_session = self.current; + let current_session = self.current.load(Ordering::Relaxed); if self.sessions[current_session % super::N_SESSIONS].is_some() { let duration_since_tun_start = Instant::now().duration_since(self.timers.time_started); - let duration_since_session_established = self.timers[TimeSessionEstablished]; + let duration_since_session_established = self.timer_fetch(TimeSessionEstablished); Some(duration_since_tun_start - duration_since_session_established) } else {