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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
|
use crate::tls_stream::TlsStream;
use futures::Stream;
use hyper::client::connect::{Connected, Connection};
use serde::{Deserialize, Serialize};
use shadowsocks::relay::tcprelay::ProxyClientStream;
use std::{
fmt, io,
net::SocketAddr,
path::Path,
pin::Pin,
task::{self, Poll},
};
use talpid_types::{net::openvpn::ShadowsocksProxySettings, ErrorExt};
use tokio::{
fs,
io::{AsyncRead, AsyncWrite, AsyncWriteExt, ReadBuf},
net::TcpStream,
};
const CURRENT_CONFIG_FILENAME: &str = "api-endpoint.json";
#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
pub enum ApiConnectionMode {
/// Connect directly to the target.
Direct,
/// Connect to the destination via a proxy.
Proxied(ProxyConfig),
}
impl fmt::Display for ApiConnectionMode {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
match self {
ApiConnectionMode::Direct => write!(f, "unproxied"),
ApiConnectionMode::Proxied(settings) => settings.fmt(f),
}
}
}
#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
pub enum ProxyConfig {
Shadowsocks(ShadowsocksProxySettings),
}
impl fmt::Display for ProxyConfig {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
match self {
// TODO: Do not hardcode TCP
ProxyConfig::Shadowsocks(ss) => write!(f, "Shadowsocks {}/TCP", ss.peer),
}
}
}
impl ApiConnectionMode {
/// Reads the proxy config from `CURRENT_CONFIG_FILENAME`.
/// This returns `ApiConnectionMode::Direct` if reading from disk fails for any reason.
pub async fn try_from_cache(cache_dir: &Path) -> Self {
Self::from_cache(cache_dir).await.unwrap_or_else(|error| {
log::error!(
"{}",
error.display_chain_with_msg("Failed to read API endpoint cache")
);
ApiConnectionMode::Direct
})
}
/// Reads the proxy config from `CURRENT_CONFIG_FILENAME`.
/// If the file does not exist, this returns `Ok(ApiConnectionMode::Direct)`.
async fn from_cache(cache_dir: &Path) -> io::Result<Self> {
let path = cache_dir.join(CURRENT_CONFIG_FILENAME);
match fs::read_to_string(path).await {
Ok(s) => serde_json::from_str(&s).map_err(|error| {
log::error!(
"{}",
error.display_chain_with_msg(&format!(
"Failed to deserialize \"{}\"",
CURRENT_CONFIG_FILENAME
))
);
io::Error::new(io::ErrorKind::Other, "deserialization failed")
}),
Err(error) => {
if error.kind() == io::ErrorKind::NotFound {
Ok(ApiConnectionMode::Direct)
} else {
Err(error)
}
}
}
}
/// Stores this config to `CURRENT_CONFIG_FILENAME`.
pub async fn save(&self, cache_dir: &Path) -> io::Result<()> {
let mut file = crate::fs::AtomicFile::new(cache_dir.join(CURRENT_CONFIG_FILENAME)).await?;
let json = serde_json::to_string_pretty(self)
.map_err(|_| io::Error::new(io::ErrorKind::Other, "serialization failed"))?;
file.write_all(json.as_bytes()).await?;
file.write_all(b"\n").await?;
file.finalize().await
}
/// Attempts to remove `CURRENT_CONFIG_FILENAME`, if it exists.
pub async fn try_delete_cache(cache_dir: &Path) {
let path = cache_dir.join(CURRENT_CONFIG_FILENAME);
if let Err(err) = fs::remove_file(path).await {
if err.kind() != std::io::ErrorKind::NotFound {
log::error!(
"{}",
err.display_chain_with_msg("Failed to remove old API config")
);
}
}
}
/// Returns the remote address, or `None` for `ApiConnectionMode::Direct`.
pub fn get_endpoint(&self) -> Option<SocketAddr> {
match self {
ApiConnectionMode::Proxied(ProxyConfig::Shadowsocks(ss)) => Some(ss.peer),
ApiConnectionMode::Direct => None,
}
}
pub fn is_proxy(&self) -> bool {
*self != ApiConnectionMode::Direct
}
/// Convenience function that returns a stream that repeats
/// this config forever.
pub fn into_repeat(self) -> impl Stream<Item = ApiConnectionMode> {
futures::stream::repeat(self)
}
}
/// Stream that is either a regular TLS stream or TLS via shadowsocks
pub enum ApiConnection {
Direct(Box<TlsStream<TcpStream>>),
Proxied(Box<TlsStream<ProxyClientStream<TcpStream>>>),
}
impl AsyncRead for ApiConnection {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut task::Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
match Pin::get_mut(self) {
ApiConnection::Direct(s) => Pin::new(s).poll_read(cx, buf),
ApiConnection::Proxied(s) => Pin::new(s).poll_read(cx, buf),
}
}
}
impl AsyncWrite for ApiConnection {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut task::Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
match Pin::get_mut(self) {
ApiConnection::Direct(s) => Pin::new(s).poll_write(cx, buf),
ApiConnection::Proxied(s) => Pin::new(s).poll_write(cx, buf),
}
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<io::Result<()>> {
match Pin::get_mut(self) {
ApiConnection::Direct(s) => Pin::new(s).poll_flush(cx),
ApiConnection::Proxied(s) => Pin::new(s).poll_flush(cx),
}
}
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<io::Result<()>> {
match Pin::get_mut(self) {
ApiConnection::Direct(s) => Pin::new(s).poll_shutdown(cx),
ApiConnection::Proxied(s) => Pin::new(s).poll_shutdown(cx),
}
}
}
impl Connection for ApiConnection {
fn connected(&self) -> Connected {
match self {
ApiConnection::Direct(s) => s.connected(),
ApiConnection::Proxied(s) => s.connected(),
}
}
}
|
