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
#![allow(dead_code)]
use libc::{close, dup2, getpid};
use std::env;
use std::io::{Read, Write};
use std::net::{Shutdown, TcpStream};
use std::os::unix::io::{FromRawFd, IntoRawFd};
use std::os::unix::net::UnixStream;
use std::os::unix::process::CommandExt;
use std::process::{Child, Command};
use tempfile::tempdir;
use tempfile::TempDir;
use unix_socket::UnixStream as AbstractStream;
use {ErrorKind, Result};
pub enum VarlinkStream {
TCP(TcpStream),
UNIX(UnixStream),
}
pub fn varlink_exec<S: ?Sized + AsRef<str>>(
address: &S,
) -> Result<(Child, String, Option<TempDir>)> {
let executable = String::from("exec ") + address.as_ref();
use unix_socket::UnixListener;
let dir = tempdir()?;
let file_path = dir.path().join("varlink-socket");
let listener = UnixListener::bind(file_path.clone())?;
let fd = listener.into_raw_fd();
let child = Command::new("sh")
.arg("-c")
.arg(executable)
.before_exec({
let file_path = file_path.clone();
move || {
unsafe {
if fd != 3 {
close(3);
dup2(fd, 3);
}
env::set_var("VARLINK_ADDRESS", format!("unix:{}", file_path.display()));
env::set_var("LISTEN_FDS", "1");
env::set_var("LISTEN_FDNAMES", "varlink");
env::set_var("LISTEN_PID", format!("{}", getpid()));
}
Ok(())
}
})
.spawn()?;
Ok((child, format!("unix:{}", file_path.display()), Some(dir)))
}
pub fn varlink_bridge<S: ?Sized + AsRef<str>>(address: &S) -> Result<(Child, VarlinkStream)> {
let executable = address.as_ref();
let (stream0, stream1) = UnixStream::pair()?;
let fd = stream1.into_raw_fd();
let childin = unsafe { ::std::fs::File::from_raw_fd(fd) };
let childout = unsafe { ::std::fs::File::from_raw_fd(fd) };
let child = Command::new("sh")
.arg("-c")
.arg(executable)
.stdin(childin)
.stdout(childout)
.spawn()?;
Ok((child, VarlinkStream::UNIX(stream0)))
}
impl<'a> VarlinkStream {
pub fn connect<S: ?Sized + AsRef<str>>(address: &S) -> Result<(Self, String)> {
let address = address.as_ref();
let new_address: String = address.into();
if new_address.starts_with("tcp:") {
Ok((
VarlinkStream::TCP(TcpStream::connect(&new_address[4..])?),
new_address,
))
} else if new_address.starts_with("unix:") {
let mut addr = String::from(new_address[5..].split(';').next().unwrap());
if addr.starts_with('@') {
addr = addr.replacen('@', "\0", 1);
let l = AbstractStream::connect(addr)?;
unsafe {
return Ok((
VarlinkStream::UNIX(UnixStream::from_raw_fd(l.into_raw_fd())),
new_address,
));
}
}
Ok((VarlinkStream::UNIX(UnixStream::connect(addr)?), new_address))
} else {
Err(ErrorKind::InvalidAddress)?
}
}
pub fn split(&mut self) -> Result<(Box<Read + Send + Sync>, Box<Write + Send + Sync>)> {
match *self {
VarlinkStream::TCP(ref mut s) => {
Ok((Box::new(s.try_clone()?), Box::new(s.try_clone()?)))
}
VarlinkStream::UNIX(ref mut s) => {
Ok((Box::new(s.try_clone()?), Box::new(s.try_clone()?)))
}
}
}
pub fn shutdown(&mut self) -> Result<()> {
match *self {
VarlinkStream::TCP(ref mut s) => s.shutdown(Shutdown::Both)?,
VarlinkStream::UNIX(ref mut s) => s.shutdown(Shutdown::Both)?,
}
Ok(())
}
pub fn set_nonblocking(&self, b: bool) -> Result<()> {
match *self {
VarlinkStream::TCP(ref l) => l.set_nonblocking(b)?,
VarlinkStream::UNIX(ref l) => l.set_nonblocking(b)?,
}
Ok(())
}
}
impl Drop for VarlinkStream {
fn drop(&mut self) {
let _r = self.shutdown();
}
}