1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
use crate::DaemonEventSender;
use futures::{
channel::{mpsc, oneshot},
stream, Stream, StreamExt,
};
use mullvad_rpc::{proxy::ApiConnectionMode, ApiEndpointUpdateCallback};
use std::{
net::SocketAddr,
sync::{Arc, Mutex, Weak},
};
use talpid_core::{mpsc::Sender, tunnel_state_machine::TunnelCommand};
use talpid_types::{
net::{AllowedEndpoint, Endpoint, TransportProtocol},
ErrorExt,
};
pub(crate) struct ApiConnectionModeRequest {
pub response_tx: oneshot::Sender<ApiConnectionMode>,
pub retry_attempt: u32,
}
/// Returns a stream that returns the next API bridge to try.
/// `initial_config` refers to the first config returned by the stream. The daemon is not notified
/// of this.
pub(crate) fn create_api_config_provider(
daemon_sender: DaemonEventSender<ApiConnectionModeRequest>,
initial_config: ApiConnectionMode,
) -> impl Stream<Item = ApiConnectionMode> + Unpin {
struct Context {
attempt: u32,
daemon_sender: DaemonEventSender<ApiConnectionModeRequest>,
}
let ctx = Context {
attempt: 1,
daemon_sender,
};
Box::pin(
stream::once(async move { initial_config }).chain(stream::unfold(
ctx,
|mut ctx| async move {
ctx.attempt = ctx.attempt.wrapping_add(1);
let (response_tx, response_rx) = oneshot::channel();
let _ = ctx.daemon_sender.send(ApiConnectionModeRequest {
response_tx,
retry_attempt: ctx.attempt,
});
let new_config = response_rx.await.unwrap_or_else(|error| {
log::error!(
"{}",
error.display_chain_with_msg("Failed to receive API proxy config")
);
// Fall back on unbridged connection
ApiConnectionMode::Direct
});
Some((new_config, ctx))
},
)),
)
}
/// Notifies the tunnel state machine that the API (real or proxied) endpoint has
/// changed. [ApiEndpointUpdaterHandle::callback()] creates a callback that may
/// be passed to the `mullvad-rpc` runtime.
pub(super) struct ApiEndpointUpdaterHandle {
tunnel_cmd_tx: Arc<Mutex<Option<Weak<mpsc::UnboundedSender<TunnelCommand>>>>>,
}
impl ApiEndpointUpdaterHandle {
pub fn new() -> Self {
Self {
tunnel_cmd_tx: Arc::new(Mutex::new(None)),
}
}
pub fn set_tunnel_command_tx(&self, tunnel_cmd_tx: Weak<mpsc::UnboundedSender<TunnelCommand>>) {
*self.tunnel_cmd_tx.lock().unwrap() = Some(tunnel_cmd_tx);
}
pub fn callback(&self) -> impl ApiEndpointUpdateCallback {
let tunnel_tx = self.tunnel_cmd_tx.clone();
move |address: SocketAddr| {
let inner_tx = tunnel_tx.clone();
async move {
let tunnel_tx = if let Some(Some(tunnel_tx)) = { inner_tx.lock().unwrap().as_ref() }
.map(|tx: &Weak<mpsc::UnboundedSender<TunnelCommand>>| tx.upgrade())
{
tunnel_tx
} else {
log::error!("Rejecting allowed endpoint: Tunnel state machine is not running");
return false;
};
let (result_tx, result_rx) = oneshot::channel();
let _ = tunnel_tx.unbounded_send(TunnelCommand::AllowEndpoint(
get_allowed_endpoint(address.clone()),
result_tx,
));
if result_rx.await.is_ok() {
log::debug!("API endpoint: {}", address);
true
} else {
log::error!("Failed to update allowed endpoint");
false
}
}
}
}
}
pub(super) fn get_allowed_endpoint(api_address: SocketAddr) -> AllowedEndpoint {
let endpoint = Endpoint::from_socket_address(api_address, TransportProtocol::Tcp);
#[cfg(windows)]
let daemon_exe = std::env::current_exe().expect("failed to obtain executable path");
#[cfg(windows)]
let clients = vec![
daemon_exe
.parent()
.expect("missing executable parent directory")
.join("mullvad-problem-report.exe"),
daemon_exe,
];
AllowedEndpoint {
#[cfg(windows)]
clients,
endpoint,
}
}
|
