hyper_connector_mullvad/
lib.rs

1//! A Hyper connector pooling Mullvad WireGuard tunnels.
2//!
3//! Hyper owns HTTP connection pooling and request dispatch. New connections
4//! are assigned to healthy tunnels; failed tunnels are repaired for future
5//! connections without replaying requests.
6//!
7//! ```no_run
8//! use hyper_connector_mullvad::{LocationFilter, MullvadConnector};
9//! use http_body_util::Empty;
10//! use hyper::body::Bytes;
11//! use hyper_util::{client::legacy::Client, rt::TokioExecutor};
12//! # async fn run() -> Result<(), Box<dyn std::error::Error>> {
13//! let mullvad = MullvadConnector::builder()
14//!     .max_connections(3)
15//!     .location_filter(LocationFilter::parse("Sweden")?)
16//!     .build().await?;
17//! let _client = Client::builder(TokioExecutor::new())
18//!     .build::<_, Empty<Bytes>>(mullvad.clone());
19//! mullvad.shutdown().await;
20//! # Ok(()) }
21//! ```
22
23mod discovery;
24mod location;
25
26use std::{
27    path::PathBuf,
28    sync::{
29        Arc,
30        atomic::{AtomicBool, AtomicU64, Ordering},
31    },
32    time::Duration,
33};
34
35use discovery::Candidate;
36use http::Uri;
37use rand::{Rng, seq::SliceRandom};
38use tokio::sync::{Mutex, Notify, RwLock};
39use tower_service::Service;
40use wireguard_hyper_connector::{
41    Error as WireGuardError, ManagedTunnel, WgConnector, WgTlsStream, WireGuardConfig,
42};
43
44pub use location::{Continent, LocationFilter};
45
46/// The default number of Mullvad tunnels maintained by the connector.
47pub const DEFAULT_MAX_CONNECTIONS: usize = 4;
48/// The maximum pool size supported by this crate.
49pub const MULLVAD_MAX_CONNECTIONS: usize = 5;
50
51/// Errors produced during discovery, connection, or a single HTTP send.
52#[derive(Debug, thiserror::Error)]
53#[non_exhaustive]
54pub enum Error {
55    #[error("max connections must be between 1 and {MULLVAD_MAX_CONNECTIONS}, got {0}")]
56    InvalidMaxConnections(usize),
57    #[error("could not read device.json at {path}")]
58    DeviceJsonRead {
59        path: PathBuf,
60        #[source]
61        source: std::io::Error,
62    },
63    #[error(
64        "Mullvad device.json could not be read from {system} or {user}; specify it with MullvadConnector::builder().device_json(path)"
65    )]
66    DeviceJsonNotFound { system: PathBuf, user: PathBuf },
67    #[error("could not parse Mullvad device.json at {path}")]
68    DeviceJsonParse {
69        path: PathBuf,
70        #[source]
71        source: serde_json::Error,
72    },
73    #[error("invalid Mullvad device.json: {0}")]
74    InvalidDeviceJson(String),
75    #[error("could not read Mullvad relay cache at {path}")]
76    RelayCache {
77        path: PathBuf,
78        #[source]
79        source: std::io::Error,
80    },
81    #[error("could not parse Mullvad relay cache at {path}")]
82    RelayCacheParse {
83        path: PathBuf,
84        #[source]
85        source: serde_json::Error,
86    },
87    #[error("invalid Mullvad relay cache: {0}")]
88    InvalidRelayCache(String),
89    #[error("unknown location filter: {0}")]
90    InvalidLocationFilter(String),
91    #[error("no Mullvad configurations match the location filter")]
92    NoConfigsMatchingFilter,
93    #[error("all {count} matching Mullvad configurations were invalid")]
94    AllCandidatesInvalid { count: usize },
95    #[error("none of the {attempted} Mullvad relays could establish a tunnel")]
96    InitialConnectionExhausted { attempted: usize },
97    #[error("connection failed through relay {relay}")]
98    Transport {
99        relay: String,
100        #[source]
101        source: WireGuardError,
102    },
103    #[error("no healthy Mullvad connection became available")]
104    NoHealthyConnections,
105}
106
107/// Non-sensitive state for an active pool slot.
108#[derive(Clone, Debug, Eq, PartialEq)]
109pub struct ConnectionSnapshot {
110    pub relay_id: String,
111    pub country_code: String,
112    pub healthy: bool,
113    pub generation: u64,
114}
115
116/// Summary of discovery and initial connection attempts.
117#[derive(Clone, Debug, Eq, PartialEq)]
118pub struct StartupReport {
119    pub matching_configs: usize,
120    pub invalid_configs: usize,
121    pub active_connections: usize,
122}
123
124/// Builder for [`MullvadConnector`].
125#[derive(Clone, Debug)]
126pub struct MullvadConnectorBuilder {
127    max: usize,
128    config_dir: PathBuf,
129    device_json: Option<PathBuf>,
130    filter: Option<LocationFilter>,
131    connection_timeout: Duration,
132    reconnect_base: Duration,
133    reconnect_max: Duration,
134}
135
136impl Default for MullvadConnectorBuilder {
137    fn default() -> Self {
138        Self {
139            max: DEFAULT_MAX_CONNECTIONS,
140            config_dir: PathBuf::from("/etc/wireguard"),
141            device_json: None,
142            filter: None,
143            connection_timeout: Duration::from_secs(30),
144            reconnect_base: Duration::from_millis(250),
145            reconnect_max: Duration::from_secs(10),
146        }
147    }
148}
149
150impl MullvadConnectorBuilder {
151    #[must_use]
152    pub fn max_connections(mut self, max: usize) -> Self {
153        self.max = max;
154        self
155    }
156    #[must_use]
157    pub fn location_filter(mut self, filter: LocationFilter) -> Self {
158        self.filter = Some(filter);
159        self
160    }
161    #[must_use]
162    pub fn config_dir(mut self, path: impl Into<PathBuf>) -> Self {
163        self.config_dir = path.into();
164        self
165    }
166    /// Override the native Mullvad device credential file.
167    #[must_use]
168    pub fn device_json(mut self, path: impl Into<PathBuf>) -> Self {
169        self.device_json = Some(path.into());
170        self
171    }
172    #[must_use]
173    pub fn connection_timeout(mut self, value: Duration) -> Self {
174        self.connection_timeout = value;
175        self
176    }
177    #[must_use]
178    pub fn reconnect_backoff(mut self, base: Duration, max: Duration) -> Self {
179        self.reconnect_base = base;
180        self.reconnect_max = max;
181        self
182    }
183    pub async fn build(self) -> Result<MullvadConnector, Error> {
184        if !(1..=MULLVAD_MAX_CONNECTIONS).contains(&self.max) {
185            return Err(Error::InvalidMaxConnections(self.max));
186        }
187        let (mut candidates, invalid) = discovery::discover(
188            &self.config_dir,
189            self.device_json.as_deref(),
190            self.filter.as_ref(),
191        )
192        .await?;
193        let matching = candidates.len();
194        candidates.shuffle(&mut rand::rng());
195        let wanted = self.max.min(candidates.len());
196        let mut jobs = tokio::task::JoinSet::new();
197        let mut pending = candidates.iter().cloned();
198        for candidate in pending.by_ref().take(wanted) {
199            let timeout = self.connection_timeout;
200            jobs.spawn(async move {
201                let result =
202                    tokio::time::timeout(timeout, connect_tunnel(candidate.config.clone())).await;
203                (candidate, result)
204            });
205        }
206        let mut slots = Vec::new();
207        while let Some(result) = jobs.join_next().await {
208            if let Ok((candidate, Ok(Ok(tunnel)))) = result {
209                slots.push(Arc::new(Slot {
210                    state: Mutex::new(SlotState {
211                        candidate,
212                        tunnel: Some(tunnel),
213                        healthy: true,
214                        generation: 0,
215                        failures: 0,
216                    }),
217                    repairing: AtomicBool::new(false),
218                }));
219                if slots.len() == wanted {
220                    jobs.abort_all();
221                    break;
222                }
223            } else if let Some(candidate) = pending.next() {
224                let timeout = self.connection_timeout;
225                jobs.spawn(async move {
226                    let result =
227                        tokio::time::timeout(timeout, connect_tunnel(candidate.config.clone()))
228                            .await;
229                    (candidate, result)
230                });
231            }
232        }
233        if slots.is_empty() {
234            return Err(Error::InitialConnectionExhausted {
235                attempted: matching,
236            });
237        }
238        let active = slots.len();
239        let inner = Inner {
240            max: self.max,
241            candidates,
242            slots: RwLock::new(slots),
243            cursor: AtomicU64::new(rand::rng().random()),
244            notify: Notify::new(),
245            shutting_down: AtomicBool::new(false),
246            connection_timeout: self.connection_timeout,
247            reconnect_base: self.reconnect_base,
248            reconnect_max: self.reconnect_max,
249        };
250        Ok(MullvadConnector {
251            inner: Arc::new(inner),
252            report: StartupReport {
253                matching_configs: matching,
254                invalid_configs: invalid,
255                active_connections: active,
256            },
257        })
258    }
259}
260
261struct Slot {
262    state: Mutex<SlotState>,
263    repairing: AtomicBool,
264}
265struct SlotState {
266    candidate: Candidate,
267    tunnel: Option<Tunnel>,
268    healthy: bool,
269    generation: u64,
270    failures: u32,
271}
272
273struct Tunnel {
274    owner: ManagedTunnel,
275    connector: WgConnector,
276}
277struct Inner {
278    max: usize,
279    candidates: Vec<Candidate>,
280    slots: RwLock<Vec<Arc<Slot>>>,
281    cursor: AtomicU64,
282    notify: Notify,
283    shutting_down: AtomicBool,
284    connection_timeout: Duration,
285    reconnect_base: Duration,
286    reconnect_max: Duration,
287}
288
289/// A cloneable Hyper connector routed through a pool of Mullvad tunnels.
290#[derive(Clone)]
291pub struct MullvadConnector {
292    inner: Arc<Inner>,
293    report: StartupReport,
294}
295
296impl MullvadConnector {
297    pub async fn new() -> Result<Self, Error> {
298        Self::builder().build().await
299    }
300    pub async fn with_max_connections(max: usize) -> Result<Self, Error> {
301        Self::builder().max_connections(max).build().await
302    }
303    #[must_use]
304    pub fn builder() -> MullvadConnectorBuilder {
305        MullvadConnectorBuilder::default()
306    }
307    #[must_use]
308    pub fn configured_max_connections(&self) -> usize {
309        self.inner.max
310    }
311    #[must_use]
312    pub fn startup_report(&self) -> &StartupReport {
313        &self.report
314    }
315    pub async fn active_connections(&self) -> usize {
316        let slots = self.inner.slots.read().await;
317        let mut n = 0;
318        for s in slots.iter() {
319            if s.state.lock().await.healthy {
320                n += 1;
321            }
322        }
323        n
324    }
325    pub async fn connections(&self) -> Vec<ConnectionSnapshot> {
326        let slots = self.inner.slots.read().await;
327        let mut out = Vec::new();
328        for s in slots.iter() {
329            let x = s.state.lock().await;
330            out.push(ConnectionSnapshot {
331                relay_id: x.candidate.id.clone(),
332                country_code: x.candidate.country.clone(),
333                healthy: x.healthy,
334                generation: x.generation,
335            });
336        }
337        out
338    }
339
340    /// Open a connection through a healthy tunnel.
341    pub async fn connect(&self, uri: Uri) -> Result<WgTlsStream, Error> {
342        let (slot, mut connector, generation, relay) = self.select().await?;
343        match connector.call(uri).await {
344            Ok(stream) => {
345                let mut state = slot.state.lock().await;
346                if state.generation == generation {
347                    state.failures = 0;
348                }
349                Ok(stream)
350            }
351            Err(source) => {
352                self.fail(slot, generation);
353                Err(Error::Transport { relay, source })
354            }
355        }
356    }
357
358    async fn select(&self) -> Result<(Arc<Slot>, WgConnector, u64, String), Error> {
359        let deadline = tokio::time::Instant::now() + self.inner.connection_timeout;
360        loop {
361            let slots = self.inner.slots.read().await;
362            let len = slots.len();
363            if len > 0 {
364                let start = self.inner.cursor.fetch_add(1, Ordering::Relaxed) as usize % len;
365                for offset in 0..len {
366                    let slot = slots[(start + offset) % len].clone();
367                    let s = slot.state.lock().await;
368                    if s.healthy {
369                        if let Some(tunnel) = &s.tunnel {
370                            return Ok((
371                                slot.clone(),
372                                tunnel.connector.clone(),
373                                s.generation,
374                                s.candidate.id.clone(),
375                            ));
376                        }
377                    }
378                }
379            }
380            drop(slots);
381            if tokio::time::timeout_at(deadline, self.inner.notify.notified())
382                .await
383                .is_err()
384            {
385                return Err(Error::NoHealthyConnections);
386            }
387        }
388    }
389
390    fn fail(&self, slot: Arc<Slot>, generation: u64) {
391        let inner = self.inner.clone();
392        tokio::spawn(async move {
393            repair(inner, slot, generation).await;
394        });
395    }
396
397    /// Stop all currently owned tunnels. Other clones must no longer be used.
398    pub async fn shutdown(self) {
399        self.inner.shutting_down.store(true, Ordering::Release);
400        let slots = self.inner.slots.read().await.clone();
401        for slot in slots {
402            let old = slot.state.lock().await.tunnel.take();
403            if let Some(tunnel) = old {
404                tunnel.owner.shutdown().await;
405            }
406        }
407    }
408}
409
410impl Service<Uri> for MullvadConnector {
411    type Response = WgTlsStream;
412    type Error = Error;
413    type Future = std::pin::Pin<
414        Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>> + Send>,
415    >;
416
417    fn poll_ready(
418        &mut self,
419        _cx: &mut std::task::Context<'_>,
420    ) -> std::task::Poll<Result<(), Self::Error>> {
421        std::task::Poll::Ready(Ok(()))
422    }
423
424    fn call(&mut self, uri: Uri) -> Self::Future {
425        let connector = self.clone();
426        Box::pin(async move { connector.connect(uri).await })
427    }
428}
429
430async fn connect_tunnel(config: WireGuardConfig) -> Result<Tunnel, wireguard_netstack::Error> {
431    let owner = ManagedTunnel::connect(config).await?;
432    let connector = WgConnector::new(owner.netstack());
433    Ok(Tunnel { owner, connector })
434}
435
436async fn repair(inner: Arc<Inner>, slot: Arc<Slot>, generation: u64) {
437    if slot.repairing.swap(true, Ordering::AcqRel) {
438        return;
439    }
440    {
441        let mut s = slot.state.lock().await;
442        if s.generation != generation {
443            slot.repairing.store(false, Ordering::Release);
444            return;
445        }
446        s.healthy = false;
447        s.failures += 1;
448    }
449    if inner.shutting_down.load(Ordering::Acquire) {
450        slot.repairing.store(false, Ordering::Release);
451        return;
452    }
453    let (same, failures) = {
454        let s = slot.state.lock().await;
455        (s.candidate.clone(), s.failures)
456    };
457    let used: Vec<String> = {
458        let slots = inner.slots.read().await;
459        let mut ids = Vec::new();
460        for x in slots.iter() {
461            if !Arc::ptr_eq(x, &slot) {
462                ids.push(x.state.lock().await.candidate.id.clone());
463            }
464        }
465        ids
466    };
467    let alternates: Vec<_> = inner
468        .candidates
469        .iter()
470        .filter(|c| c.id != same.id && !used.contains(&c.id))
471        .cloned()
472        .collect();
473    let mut cycle = failures;
474    'repair: loop {
475        if inner.shutting_down.load(Ordering::Acquire) {
476            break;
477        }
478        let exp = cycle.saturating_sub(1).min(8);
479        let delay = inner
480            .reconnect_base
481            .saturating_mul(1 << exp)
482            .min(inner.reconnect_max);
483        let jitter = if delay.is_zero() {
484            Duration::ZERO
485        } else {
486            Duration::from_millis(
487                rand::rng().random_range(0..=delay.as_millis().min(u128::from(u64::MAX)) as u64),
488            )
489        };
490        tokio::time::sleep(jitter).await;
491        let mut choices = vec![same.clone()];
492        let mut shuffled = alternates.clone();
493        shuffled.shuffle(&mut rand::rng());
494        choices.extend(shuffled);
495        for candidate in choices {
496            if inner.shutting_down.load(Ordering::Acquire) {
497                break 'repair;
498            }
499            let connected = tokio::time::timeout(
500                inner.connection_timeout,
501                connect_tunnel(candidate.config.clone()),
502            )
503            .await;
504            if let Ok(Ok(new_tunnel)) = connected {
505                if inner.shutting_down.load(Ordering::Acquire) {
506                    new_tunnel.owner.shutdown().await;
507                    break 'repair;
508                }
509                let old = {
510                    let mut s = slot.state.lock().await;
511                    if s.generation != generation {
512                        Some(new_tunnel)
513                    } else {
514                        let old = s.tunnel.replace(new_tunnel);
515                        s.candidate = candidate;
516                        s.generation += 1;
517                        s.healthy = true;
518                        old
519                    }
520                };
521                slot.repairing.store(false, Ordering::Release);
522                inner.notify.notify_waiters();
523                if let Some(tunnel) = old {
524                    tunnel.owner.shutdown().await;
525                }
526                return;
527            }
528        }
529        cycle = cycle.saturating_add(1);
530    }
531    slot.repairing.store(false, Ordering::Release);
532}
533
534#[cfg(test)]
535mod tests {
536    use super::*;
537
538    #[test]
539    fn defaults_to_four_connections() {
540        assert_eq!(MullvadConnectorBuilder::default().max, 4);
541    }
542
543    #[tokio::test]
544    async fn bounds_are_validated_without_io() {
545        assert!(matches!(
546            MullvadConnectorBuilder::default()
547                .max_connections(0)
548                .build()
549                .await,
550            Err(Error::InvalidMaxConnections(0))
551        ));
552    }
553}
hyper_connector_mullvad/lib.rs

hyper_connector_mullvad/
lib.rs
