CH-J Toolsbainisteoireacht freastalaí SSH
Roghchlár
Foinse foilsithe

CH-J Tools

Brabhsáil comhadlanna agus comhaid chun feidhmchlár áirithe a scaoileadh.

Íosluchtaigh foinse ZIP
CH-J / Server Manager 1704 Comhaid
0.1.2-beta Suimigh: · Tógtha:
bridge.rs Suimigh: Dáta gan taifeadadh 10.8 KB · 261 líne
Cód Íosluchtaigh comhad
1//! Carries rsync's protocol over the SSH sessions of the application, so a transfer logs in the way the session did
2//! (password or key, host key checked, jump hosts) and no secret reaches a process argument.
3//!
4//! A local rsync starts this same executable as its remote shell. That child connects back to Core on the loopback
5//! interface and proves itself with a one-time token from the environment.
7use crate::core::error::{CoreError, CoreResult};
8use crate::core::ssh::privileged::{PrivilegedExecutor, RemoteStream, StreamEvent};
9use std::io::{Read, Write};
10use std::sync::Arc;
11use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
12use tokio::net::TcpListener;
14pub const BRIDGE_FLAG: &str = "--chj-rsync-bridge";
15pub const BRIDGE_ENV: &str = "CHJ_RSYNC_BRIDGE";
17/// Dropping a task handle leaves the task running; this one takes the task along.
18struct AbortOnDrop<T>(tokio::task::JoinHandle<T>);
20impl<T> Drop for AbortOnDrop<T> {
21 fn drop(&mut self) {
22 self.0.abort();
23 }
26/// The arguments rsync gives its remote shell: a host, then `rsync --server ...`. Only that role may run.
27pub fn server_arguments(argv: &[String]) -> CoreResult<Vec<String>> {
28 let start = argv.iter().position(|arg| arg == "rsync").ok_or_else(|| CoreError::new("RSYNC_BRIDGE_REFUSED", "The transfer asked for something other than rsync."))?;
29 let rest = &argv[start + 1..];
30 if rest.first().map(String::as_str) != Some("--server") {
31 return Err(CoreError::new("RSYNC_BRIDGE_REFUSED", "The transfer asked for something other than the rsync server role."));
32 }
33 Ok(rest.to_vec())
36/// The child side, in the process rsync started. Returns the exit code of the process.
37pub fn client_main(args: &[String]) -> i32 {
38 let Ok(contact) = std::env::var(BRIDGE_ENV) else { return 12 };
39 let Some((port, token)) = contact.split_once(':') else { return 12 };
40 let Ok(mut socket) = std::net::TcpStream::connect(("127.0.0.1", port.parse::<u16>().unwrap_or(0))) else { return 12 };
41 let header = serde_json::to_string(&serde_json::json!({ "token": token, "argv": args })).unwrap_or_default();
42 if writeln!(socket, "{header}").is_err() {
43 return 12;
44 }
45 let Ok(mut writer) = socket.try_clone() else { return 12 };
46 std::thread::spawn(move || {
47 let mut stdin = std::io::stdin();
48 let mut buffer = [0u8; 65536];
49 loop {
50 match read_patiently(&mut stdin, &mut buffer) {
51 Some(count) if count > 0 => {
52 if writer.write_all(&buffer[..count]).is_err() {
53 break;
54 }
55 }
56 _ => break,
57 }
58 }
59 let _ = writer.shutdown(std::net::Shutdown::Write);
60 });
61 let mut buffer = [0u8; 65536];
62 let mut stdout = std::io::stdout();
63 loop {
64 match socket.read(&mut buffer) {
65 Ok(0) | Err(_) => return 0,
66 Ok(count) => {
67 if !write_patiently(&mut stdout, &buffer[..count]) {
68 return 12;
69 }
70 }
71 }
72 }
75/// openrsync makes its end of the pipe non-blocking, and that reaches this process too.
76fn read_patiently(source: &mut impl Read, buffer: &mut [u8]) -> Option<usize> {
77 loop {
78 match source.read(buffer) {
79 Ok(count) => return Some(count),
80 Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => std::thread::sleep(std::time::Duration::from_millis(1)),
81 Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
82 Err(_) => return None,
83 }
84 }
87fn write_patiently(target: &mut impl Write, mut data: &[u8]) -> bool {
88 while !data.is_empty() {
89 match target.write(data) {
90 Ok(0) => return false,
91 Ok(count) => data = &data[count..],
92 Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => std::thread::sleep(std::time::Duration::from_millis(1)),
93 Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
94 Err(_) => return false,
95 }
96 }
97 loop {
98 match target.flush() {
99 Ok(()) => return true,
100 Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => std::thread::sleep(std::time::Duration::from_millis(1)),
101 Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
102 Err(_) => return false,
103 }
104 }
107/// Waits for the local rsync's remote shell and connects it to `rsync --server` on the session.
108pub async fn serve_local(listener: TcpListener, token: String, executor: Arc<PrivilegedExecutor>, session_id: String, privileged: bool, password: Option<zeroize::Zeroizing<String>>, on_error: impl Fn(String) + Send + 'static) -> CoreResult<()> {
109 let (socket, peer) = listener.accept().await.map_err(|e| CoreError::new("RSYNC_BRIDGE_FAILED", e.to_string()))?;
110 if !peer.ip().is_loopback() {
111 return Err(CoreError::new("RSYNC_BRIDGE_REFUSED", "The transfer helper connected from elsewhere."));
112 }
113 drop(listener);
114 let (read, mut write) = socket.into_split();
115 let mut reader = BufReader::new(read);
116 let mut header = String::new();
117 reader.read_line(&mut header).await.map_err(|e| CoreError::new("RSYNC_BRIDGE_FAILED", e.to_string()))?;
118 let header: serde_json::Value = serde_json::from_str(header.trim()).map_err(|_| CoreError::new("RSYNC_BRIDGE_REFUSED", "The transfer helper did not identify itself."))?;
119 if header["token"].as_str() != Some(token.as_str()) {
120 return Err(CoreError::new("RSYNC_BRIDGE_REFUSED", "The transfer helper did not identify itself."));
121 }
122 let argv: Vec<String> = header["argv"].as_array().into_iter().flatten().filter_map(|arg| arg.as_str().map(str::to_string)).collect();
123 let args = server_arguments(&argv)?;
124 let mut stream = executor.stream(&session_id, "rsync.server", &args, privileged, password.as_ref().map(|p| p.as_str())).await?;
125 let writer = stream.write.clone();
126 let _upload = AbortOnDrop(tokio::spawn(async move {
127 let mut buffer = vec![0u8; 65536];
128 loop {
129 match reader.read(&mut buffer).await {
130 Ok(0) | Err(_) => break,
131 Ok(count) => {
132 if writer.data_bytes(buffer[..count].to_vec()).await.is_err() {
133 break;
134 }
135 }
136 }
137 }
138 let _ = writer.eof().await;
139 }));
140 loop {
141 match stream.next().await {
142 StreamEvent::Out(data) => {
143 if write.write_all(&data).await.is_err() {
144 break;
145 }
146 }
147 StreamEvent::Err(data) => on_error(String::from_utf8_lossy(&data).into_owned()),
148 StreamEvent::BadPassword => {
149 on_error("The sudo password was rejected.".into());
150 break;
151 }
152 StreamEvent::Exit(_) => {}
153 StreamEvent::Closed => break,
154 }
155 }
156 let _ = write.shutdown().await;
157 Ok(())
160/// Connects the rsync client on one server with `rsync --server` on another, through this application.
161pub async fn serve_relay(mut near: RemoteStream, mut far: RemoteStream, on_error: impl Fn(String) + Send + Sync + 'static, leftover: Vec<u8>) -> CoreResult<()> {
162 let on_error = Arc::new(on_error);
163 if !leftover.is_empty() {
164 far.write.data_bytes(leftover).await.map_err(|e| CoreError::new("RSYNC_BRIDGE_FAILED", e.to_string()))?;
165 }
166 let far_write = far.write.clone();
167 let near_write = near.write.clone();
168 let errors = on_error.clone();
169 let mut forward = AbortOnDrop(tokio::spawn(async move {
170 loop {
171 match near.next().await {
172 StreamEvent::Out(data) => {
173 if far_write.data_bytes(data).await.is_err() {
174 break;
175 }
176 }
177 StreamEvent::Err(data) => errors(String::from_utf8_lossy(&data).into_owned()),
178 StreamEvent::BadPassword => errors("The sudo password was rejected.".into()),
179 StreamEvent::Exit(_) => {}
180 StreamEvent::Closed => break,
181 }
182 }
183 let _ = far_write.eof().await;
184 }));
185 loop {
186 match far.next().await {
187 StreamEvent::Out(data) => {
188 if near_write.data_bytes(data).await.is_err() {
189 break;
190 }
191 }
192 StreamEvent::Err(data) => on_error(String::from_utf8_lossy(&data).into_owned()),
193 StreamEvent::BadPassword => {
194 on_error("The sudo password of the other server was rejected.".into());
195 break;
196 }
197 StreamEvent::Exit(_) => {}
198 StreamEvent::Closed => break,
199 }
200 }
201 // The server is done: the end of this stream lets the helper on the near side finish.
202 let _ = near_write.eof().await;
203 let _ = (&mut forward.0).await;
204 Ok(())
207#[cfg(test)]
208mod tests {
209 use super::*;
211 fn strings(list: &[&str]) -> Vec<String> {
212 list.iter().map(|text| text.to_string()).collect()
213 }
215 /// Reports "not yet" twice before every piece of work, as a non-blocking pipe does.
216 struct Flaky { calls: u32, written: Vec<u8> }
218 impl Read for Flaky {
219 fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
220 self.calls += 1;
221 if self.calls <= 2 {
222 return Err(std::io::ErrorKind::WouldBlock.into());
223 }
224 buffer[0] = b'x';
225 Ok(1)
226 }
227 }
229 impl Write for Flaky {
230 fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
231 self.calls += 1;
232 if self.calls % 3 != 0 {
233 return Err(std::io::ErrorKind::WouldBlock.into());
234 }
235 self.written.push(data[0]);
236 Ok(1)
237 }
238 fn flush(&mut self) -> std::io::Result<()> {
239 Ok(())
240 }
241 }
243 #[test]
244 fn a_non_blocking_pipe_does_not_end_the_transfer() {
245 let mut flaky = Flaky { calls: 0, written: Vec::new() };
246 let mut buffer = [0u8; 4];
247 assert_eq!(read_patiently(&mut flaky, &mut buffer), Some(1));
248 let mut flaky = Flaky { calls: 0, written: Vec::new() };
249 assert!(write_patiently(&mut flaky, b"abc"));
250 assert_eq!(flaky.written, b"abc");
251 }
253 #[test]
254 fn only_the_rsync_server_role_passes() {
255 assert_eq!(server_arguments(&strings(&["x", "rsync", "--server", "-logDtpre.iLsfxCIvu", ".", "/dst/"])).unwrap(), strings(&["--server", "-logDtpre.iLsfxCIvu", ".", "/dst/"]));
256 assert_eq!(server_arguments(&strings(&["-l", "user", "x", "rsync", "--server", "--sender", ".", "/src/"])).unwrap()[1], "--sender");
257 for argv in [strings(&["x", "sh", "-c", "id"]), strings(&["x", "rsync", "--daemon"]), strings(&["x", "rsync"]), strings(&[])] {
258 assert_eq!(server_arguments(&argv).unwrap_err().code, "RSYNC_BRIDGE_REFUSED", "{argv:?}");
259 }
260 }
SHA-2565de1975cda987a01cea5af8dcf9273dbe35135448c235c3d07d1479750f985d0

CH-J Proprietary Software License 1.14 · Tá Foinse ar fáil faoin CH-J Proprietary Software License 1.14. Ní chuireann a infhaighteacht a athrú ar na téarmaí ceadúnais nó cearta breise a dheonú.

Cartlann SHA-256ac8fe790f4ba60fc098f8c985fc2656cfcc6cdc21fb34b30e308a2b54e62258f
Comhaid