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
134
135
136
137
138
139
140
141
142
143
|
pub mod types;
use parity_tokio_ipc::Endpoint as IpcEndpoint;
#[cfg(unix)]
use std::{env, fs, os::unix::fs::PermissionsExt};
use std::{
future::Future,
io,
pin::Pin,
task::{Context, Poll},
};
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use tonic::transport::{server::Connected, Endpoint, Server, Uri};
use tower::service_fn;
pub use tonic::{async_trait, transport::Channel, Code, Request, Response, Status};
pub type ManagementServiceClient =
types::management_service_client::ManagementServiceClient<Channel>;
pub use types::management_service_server::{ManagementService, ManagementServiceServer};
#[cfg(unix)]
lazy_static::lazy_static! {
static ref MULLVAD_MANAGEMENT_SOCKET_GROUP: Option<String> = env::var("MULLVAD_MANAGEMENT_SOCKET_GROUP")
.ok();
}
#[derive(err_derive::Error, Debug)]
#[error(no_from)]
pub enum Error {
#[error(display = "Management RPC server or client error")]
GrpcTransportError(#[error(source)] tonic::transport::Error),
#[error(display = "Failed to start IPC pipe/socket")]
StartServerError(#[error(source)] io::Error),
#[error(display = "Failed to initialize pipe/socket security attributes")]
SecurityAttributes(#[error(source)] io::Error),
#[error(display = "Unable to set permissions for IPC endpoint")]
PermissionsError(#[error(source)] io::Error),
#[cfg(unix)]
#[error(display = "Group not found")]
NoGidError,
#[cfg(unix)]
#[error(display = "Failed to obtain group ID")]
ObtainGidError(#[error(source)] nix::Error),
#[cfg(unix)]
#[error(display = "Failed to set group ID")]
SetGidError(#[error(source)] nix::Error),
}
pub async fn new_rpc_client() -> Result<ManagementServiceClient, Error> {
let ipc_path = mullvad_paths::get_rpc_socket_path();
// The URI will be ignored
let channel = Endpoint::from_static("lttp://[::]:50051")
.connect_with_connector(service_fn(move |_: Uri| {
IpcEndpoint::connect(ipc_path.clone())
}))
.await
.map_err(Error::GrpcTransportError)?;
Ok(ManagementServiceClient::new(channel))
}
pub type ServerJoinHandle = tokio::task::JoinHandle<Result<(), Error>>;
pub async fn spawn_rpc_server<T: ManagementService, F: Future<Output = ()> + Send + 'static>(
service: T,
abort_rx: F,
) -> std::result::Result<ServerJoinHandle, Error> {
use futures::stream::TryStreamExt;
use parity_tokio_ipc::SecurityAttributes;
let socket_path = mullvad_paths::get_rpc_socket_path();
let mut endpoint = IpcEndpoint::new(socket_path.to_string_lossy().to_string());
endpoint.set_security_attributes(
SecurityAttributes::allow_everyone_create()
.map_err(Error::SecurityAttributes)?
.set_mode(0o766)
.map_err(Error::SecurityAttributes)?,
);
let incoming = endpoint.incoming().map_err(Error::StartServerError)?;
#[cfg(unix)]
if let Some(group_name) = &*MULLVAD_MANAGEMENT_SOCKET_GROUP {
let group = nix::unistd::Group::from_name(group_name)
.map_err(Error::ObtainGidError)?
.ok_or(Error::NoGidError)?;
nix::unistd::chown(&socket_path, None, Some(group.gid)).map_err(Error::SetGidError)?;
fs::set_permissions(&socket_path, PermissionsExt::from_mode(0o760))
.map_err(Error::PermissionsError)?;
}
Ok(tokio::spawn(async move {
Server::builder()
.add_service(ManagementServiceServer::new(service))
.serve_with_incoming_shutdown(incoming.map_ok(StreamBox), abort_rx)
.await
.map_err(Error::GrpcTransportError)
}))
}
#[derive(Debug)]
struct StreamBox<T: AsyncRead + AsyncWrite>(pub T);
impl<T: AsyncRead + AsyncWrite> Connected for StreamBox<T> {
type ConnectInfo = Option<()>;
fn connect_info(&self) -> Self::ConnectInfo {
None
}
}
impl<T: AsyncRead + AsyncWrite + Unpin> AsyncRead for StreamBox<T> {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Pin::new(&mut self.0).poll_read(cx, buf)
}
}
impl<T: AsyncRead + AsyncWrite + Unpin> AsyncWrite for StreamBox<T> {
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
Pin::new(&mut self.0).poll_write(cx, buf)
}
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut self.0).poll_flush(cx)
}
fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut self.0).poll_shutdown(cx)
}
}
|
