forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
19.2 KiB
1//! russh server: host key, pubkey auth, exec and env handling, piping, timeouts.
2
3use std::borrow::Cow;
4use std::collections::HashMap;
5use std::fs::OpenOptions;
6use std::io::{self, Write};
7use std::net::SocketAddr;
8use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
9use std::os::unix::process::ExitStatusExt;
10use std::path::{Path, PathBuf};
11use std::process::ExitStatus;
12use std::sync::Arc;
13use std::sync::atomic::{AtomicBool, Ordering};
14use std::time::Duration;
15
16use bytes::Bytes;
17use russh::keys::ssh_key::LineEnding;
18use russh::keys::ssh_key::private::{Ed25519Keypair, KeypairData};
19use russh::keys::{Algorithm, EcdsaCurve, HashAlg, PrivateKey, PublicKey};
20use russh::server::{Auth, Config, Handle, Handler, Msg, Session, run_stream};
21use russh::{Channel, ChannelId, ChannelOpenFailure, Disconnect, MethodKind, MethodSet, Preferred};
22use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
23use tokio::net::TcpListener;
24use tokio::process::{Child, ChildStdin};
25use tokio::sync::{Notify, mpsc};
26
27use crate::backend::GitBackend;
28use crate::error::{Error, Result};
29use crate::parse::{LFS_AUTHENTICATE_PREFIX, parse_exec, parse_lfs_authenticate};
30
31/// Stdout and stderr read size per message sent to the client.
32const PUMP_CHUNK: usize = 32 * 1024;
33/// Client input chunks buffered in front of the child's stdin. This bounds memory and
34/// propagates backpressure to the SSH read loop.
35const STDIN_QUEUE: usize = 32;
36/// Constant delay applied to failed authentication attempts.
37const AUTH_REJECTION_TIME: Duration = Duration::from_secs(1);
38/// Failed publickey attempts before the connection is dropped.
39const MAX_AUTH_ATTEMPTS: usize = 6;
40
41#[derive(Debug, Clone)]
42pub struct SshConfig {
43 pub listen: SocketAddr,
44 pub host_key_path: PathBuf,
45 pub auth_timeout: Duration,
46 pub inactivity_timeout: Duration,
47}
48
49/// Runs the SSH server until it fails. Creates the ed25519 host key on first boot.
50pub async fn run<B: GitBackend>(config: SshConfig, backend: Arc<B>) -> Result<()> {
51 let host_key = load_or_create_host_key(&config.host_key_path)?;
52 let russh_config = Arc::new(Config {
53 keys: vec![host_key],
54 methods: publickey_only(),
55 auth_rejection_time: AUTH_REJECTION_TIME,
56 max_auth_attempts: MAX_AUTH_ATTEMPTS,
57 inactivity_timeout: Some(config.inactivity_timeout),
58 preferred: preferred_algorithms(),
59 ..Config::default()
60 });
61
62 let listener = TcpListener::bind(config.listen).await?;
63 tracing::info!(addr = %config.listen, "ssh server listening");
64
65 loop {
66 let (stream, peer) = match listener.accept().await {
67 Ok(v) => v,
68 Err(e) => {
69 // Typically EMFILE. Back off briefly instead of tearing the server down.
70 tracing::warn!(error = %e, "ssh accept failed");
71 tokio::time::sleep(Duration::from_millis(100)).await;
72 continue;
73 }
74 };
75 let authed = Arc::new(AtomicBool::new(false));
76 let client = Client {
77 backend: Arc::clone(&backend),
78 peer: Some(peer),
79 authed: Arc::clone(&authed),
80 user_id: None,
81 git_protocol: None,
82 channels: HashMap::new(),
83 };
84 let russh_config = Arc::clone(&russh_config);
85 let auth_timeout = config.auth_timeout;
86 tokio::spawn(async move {
87 let running = match run_stream(russh_config, stream, client).await {
88 Ok(running) => running,
89 Err(e) => {
90 tracing::debug!(%peer, error = %e, "ssh handshake failed");
91 return;
92 }
93 };
94 // Drop unauthenticated connections after the auth timeout.
95 let handle = running.handle();
96 tokio::spawn(async move {
97 tokio::time::sleep(auth_timeout).await;
98 if !authed.load(Ordering::Acquire) {
99 let _ = handle
100 .disconnect(
101 Disconnect::ByApplication,
102 "authentication timeout".into(),
103 String::new(),
104 )
105 .await;
106 }
107 });
108 if let Err(e) = running.await {
109 tracing::debug!(%peer, error = %e, "ssh session ended with error");
110 }
111 });
112 }
113}
114
115/// Publickey is the only accepted method.
116fn publickey_only() -> MethodSet {
117 let mut methods = MethodSet::empty();
118 methods.push(MethodKind::PublicKey);
119 methods
120}
121
122/// Host and user key algorithms this server negotiates. `ssh-rsa` (SHA-1) is not listed.
123fn preferred_algorithms() -> Preferred {
124 Preferred {
125 key: Cow::Owned(vec![
126 Algorithm::Ed25519,
127 Algorithm::SkEd25519,
128 Algorithm::SkEcdsaSha2NistP256,
129 Algorithm::Ecdsa {
130 curve: EcdsaCurve::NistP256,
131 },
132 Algorithm::Ecdsa {
133 curve: EcdsaCurve::NistP384,
134 },
135 Algorithm::Ecdsa {
136 curve: EcdsaCurve::NistP521,
137 },
138 Algorithm::Rsa {
139 hash: Some(HashAlg::Sha512),
140 },
141 Algorithm::Rsa {
142 hash: Some(HashAlg::Sha256),
143 },
144 ]),
145 ..Preferred::default()
146 }
147}
148
149/// Key types accepted for authentication. RSA keys appear as `ssh-rsa` key blobs whatever the
150/// signature algorithm is, so this check cannot tell SHA-1 signatures apart. See `run`'s docs.
151fn key_type_allowed(algorithm: &Algorithm) -> bool {
152 matches!(
153 algorithm,
154 Algorithm::Ed25519
155 | Algorithm::SkEd25519
156 | Algorithm::SkEcdsaSha2NistP256
157 | Algorithm::Ecdsa { .. }
158 | Algorithm::Rsa { .. }
159 )
160}
161
162/// Accepts exactly `version=<digit>`, e.g. `version=2`.
163fn valid_git_protocol(value: &str) -> bool {
164 let bytes = value.as_bytes();
165 bytes.len() == 9 && value.starts_with("version=") && bytes[8].is_ascii_digit()
166}
167
168fn load_or_create_host_key(path: &Path) -> Result<PrivateKey> {
169 if path.exists() {
170 let key = PrivateKey::read_openssh_file(path)
171 .map_err(|e| Error::Ssh(format!("host key: {e}")))?;
172 if key.algorithm() != Algorithm::Ed25519 {
173 return Err(Error::Ssh("host key must be ed25519".into()));
174 }
175 return Ok(key);
176 }
177
178 let mut seed = [0u8; 32];
179 getrandom::fill(&mut seed).map_err(|e| Error::Ssh(format!("host key entropy: {e}")))?;
180 let key = PrivateKey::new(KeypairData::Ed25519(Ed25519Keypair::from_seed(&seed)), "")
181 .map_err(|e| Error::Ssh(format!("host key generation: {e}")))?;
182 let pem = key
183 .to_openssh(LineEnding::LF)
184 .map_err(|e| Error::Ssh(format!("host key encoding: {e}")))?;
185
186 if let Some(parent) = path.parent() {
187 std::fs::create_dir_all(parent)?;
188 }
189 // create_new so an existing key is never overwritten. The mode applies at creation.
190 let mut file = OpenOptions::new()
191 .write(true)
192 .create_new(true)
193 .mode(0o600)
194 .open(path)?;
195 file.write_all(pem.as_bytes())?;
196 file.sync_all()?;
197 std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?;
198 tracing::info!(path = %path.display(), "generated ssh host key");
199 Ok(key)
200}
201
202/// Per-connection SSH handler.
203struct Client<B> {
204 backend: Arc<B>,
205 peer: Option<SocketAddr>,
206 authed: Arc<AtomicBool>,
207 user_id: Option<String>,
208 git_protocol: Option<String>,
209 /// Open exec channels. Entries are removed on channel EOF or close, which drops the
210 /// sender and closes the child's stdin.
211 channels: HashMap<ChannelId, mpsc::Sender<Bytes>>,
212}
213
214impl<B: GitBackend> Client<B> {
215 async fn lookup(&self, key: &PublicKey) -> Option<String> {
216 if !key_type_allowed(&key.algorithm()) {
217 return None;
218 }
219 let blob = key.to_bytes().ok()?;
220 self.backend
221 .authenticate_key(key.algorithm().as_str(), &blob)
222 .await
223 }
224
225 /// Answers `git-lfs-authenticate` with the JSON reply on stdout and exit status 0. Nothing is
226 /// left open afterwards: the channel is closed once the reply is queued.
227 async fn exec_lfs_authenticate(
228 &self,
229 session: &mut Session,
230 channel: ChannelId,
231 user_id: &str,
232 command: &str,
233 ) -> std::result::Result<(), russh::Error> {
234 let (path, operation) = match parse_lfs_authenticate(command) {
235 Ok(parsed) => parsed,
236 Err(e) => return reject_exec(session, channel, &e.to_string()),
237 };
238 let reply = match self
239 .backend
240 .lfs_authenticate(user_id, &path, operation)
241 .await
242 {
243 Ok(reply) => reply,
244 Err(msg) => return reject_exec(session, channel, &msg),
245 };
246 session.channel_success(channel)?;
247 session.data(channel, reply)?;
248 session.exit_status_request(channel, 0)?;
249 session.eof(channel)?;
250 session.close(channel)?;
251 Ok(())
252 }
253}
254
255impl<B: GitBackend> Handler for Client<B> {
256 type Error = russh::Error;
257
258 async fn auth_publickey_offered(
259 &mut self,
260 _user: &str,
261 key: &PublicKey,
262 ) -> std::result::Result<Auth, Self::Error> {
263 // Probe before the client signs. Reject early so unknown keys never get a signature.
264 if self.lookup(key).await.is_some() {
265 Ok(Auth::Accept)
266 } else {
267 Ok(Auth::reject())
268 }
269 }
270
271 async fn auth_publickey(
272 &mut self,
273 _user: &str,
274 key: &PublicKey,
275 ) -> std::result::Result<Auth, Self::Error> {
276 match self.lookup(key).await {
277 Some(user_id) => {
278 tracing::info!(peer = ?self.peer, %user_id, "ssh publickey auth accepted");
279 self.user_id = Some(user_id);
280 self.authed.store(true, Ordering::Release);
281 Ok(Auth::Accept)
282 }
283 None => Ok(Auth::reject()),
284 }
285 }
286
287 async fn channel_open_session(
288 &mut self,
289 _channel: Channel<Msg>,
290 reply: russh::server::ChannelOpenHandle,
291 _session: &mut Session,
292 ) -> std::result::Result<(), Self::Error> {
293 if self.user_id.is_none() {
294 reply
295 .reject(ChannelOpenFailure::AdministrativelyProhibited)
296 .await;
297 } else {
298 reply.accept().await;
299 }
300 Ok(())
301 }
302
303 async fn env_request(
304 &mut self,
305 _channel: ChannelId,
306 variable_name: &str,
307 variable_value: &str,
308 _session: &mut Session,
309 ) -> std::result::Result<(), Self::Error> {
310 // Only GIT_PROTOCOL is kept. Everything else is ignored, never forwarded.
311 if variable_name == "GIT_PROTOCOL" && valid_git_protocol(variable_value) {
312 self.git_protocol = Some(variable_value.to_owned());
313 }
314 Ok(())
315 }
316
317 async fn exec_request(
318 &mut self,
319 channel: ChannelId,
320 data: &[u8],
321 session: &mut Session,
322 ) -> std::result::Result<(), Self::Error> {
323 if self.channels.contains_key(&channel) {
324 session.channel_failure(channel)?;
325 return Ok(());
326 }
327 let Some(user_id) = self.user_id.clone() else {
328 session.channel_failure(channel)?;
329 return Ok(());
330 };
331 let command = match std::str::from_utf8(data) {
332 Ok(command) => command,
333 Err(_) => return reject_exec(session, channel, "invalid command"),
334 };
335 if command.starts_with(LFS_AUTHENTICATE_PREFIX) {
336 return self
337 .exec_lfs_authenticate(session, channel, &user_id, command)
338 .await;
339 }
340 let (service, path) = match parse_exec(command) {
341 Ok(parsed) => parsed,
342 Err(e) => return reject_exec(session, channel, &e.to_string()),
343 };
344
345 let mut cmd = match self
346 .backend
347 .prepare(&user_id, service, &path, self.git_protocol.as_deref())
348 .await
349 {
350 Ok(cmd) => cmd,
351 Err(msg) => return reject_exec(session, channel, &msg),
352 };
353 cmd.stdin(std::process::Stdio::piped())
354 .stdout(std::process::Stdio::piped())
355 .stderr(std::process::Stdio::piped())
356 .kill_on_drop(true);
357
358 let child = match cmd.spawn() {
359 Ok(child) => child,
360 Err(e) => {
361 tracing::error!(error = %e, "failed to spawn git");
362 return reject_exec(session, channel, "failed to start git");
363 }
364 };
365
366 session.channel_success(channel)?;
367 let (tx, rx) = mpsc::channel(STDIN_QUEUE);
368 self.channels.insert(channel, tx);
369 tokio::spawn(run_child(session.handle(), channel, child, rx));
370 Ok(())
371 }
372
373 async fn data(
374 &mut self,
375 channel: ChannelId,
376 data: &[u8],
377 _session: &mut Session,
378 ) -> std::result::Result<(), Self::Error> {
379 if let Some(tx) = self.channels.get(&channel) {
380 // Err means the child has exited. Its output is still flushed, so this is not fatal.
381 let _ = tx.send(Bytes::copy_from_slice(data)).await;
382 }
383 Ok(())
384 }
385
386 async fn channel_eof(
387 &mut self,
388 channel: ChannelId,
389 _session: &mut Session,
390 ) -> std::result::Result<(), Self::Error> {
391 self.channels.remove(&channel);
392 Ok(())
393 }
394
395 async fn channel_close(
396 &mut self,
397 channel: ChannelId,
398 _session: &mut Session,
399 ) -> std::result::Result<(), Self::Error> {
400 self.channels.remove(&channel);
401 Ok(())
402 }
403
404 async fn pty_request(
405 &mut self,
406 channel: ChannelId,
407 _term: &str,
408 _col_width: u32,
409 _row_height: u32,
410 _pix_width: u32,
411 _pix_height: u32,
412 _modes: &[(russh::Pty, u32)],
413 session: &mut Session,
414 ) -> std::result::Result<(), Self::Error> {
415 session.channel_failure(channel)?;
416 Ok(())
417 }
418
419 async fn shell_request(
420 &mut self,
421 channel: ChannelId,
422 session: &mut Session,
423 ) -> std::result::Result<(), Self::Error> {
424 session.channel_failure(channel)?;
425 Ok(())
426 }
427
428 async fn subsystem_request(
429 &mut self,
430 channel: ChannelId,
431 _name: &str,
432 session: &mut Session,
433 ) -> std::result::Result<(), Self::Error> {
434 session.channel_failure(channel)?;
435 Ok(())
436 }
437
438 async fn x11_request(
439 &mut self,
440 channel: ChannelId,
441 _single_connection: bool,
442 _x11_auth_protocol: &str,
443 _x11_auth_cookie: &str,
444 _x11_screen_number: u32,
445 session: &mut Session,
446 ) -> std::result::Result<(), Self::Error> {
447 session.channel_failure(channel)?;
448 Ok(())
449 }
450
451 async fn tcpip_forward(
452 &mut self,
453 _address: &str,
454 _port: &mut u32,
455 _session: &mut Session,
456 ) -> std::result::Result<bool, Self::Error> {
457 Ok(false)
458 }
459
460 async fn agent_request(
461 &mut self,
462 _channel: ChannelId,
463 _session: &mut Session,
464 ) -> std::result::Result<bool, Self::Error> {
465 Ok(false)
466 }
467}
468
469/// Sends a rejection as the spec requires: the message on stderr, exit status 1, then EOF and close.
470/// The request itself is acknowledged so git reads the message instead of failing at the protocol level.
471fn reject_exec(
472 session: &mut Session,
473 channel: ChannelId,
474 msg: &str,
475) -> std::result::Result<(), russh::Error> {
476 session.channel_success(channel)?;
477 session.extended_data(channel, 1, format!("{msg}\n"))?;
478 session.exit_status_request(channel, 1)?;
479 session.eof(channel)?;
480 session.close(channel)?;
481 Ok(())
482}
483
484/// Runs one git process for an exec channel. Stdin is fed from `stdin_rx`, stdout goes to
485/// channel data and stderr to extended data 1. Exit status, EOF and close are sent last.
486async fn run_child(
487 handle: Handle,
488 channel: ChannelId,
489 mut child: Child,
490 stdin_rx: mpsc::Receiver<Bytes>,
491) {
492 // Raised by a pump when the client has gone away, so the child is killed.
493 let abort = Arc::new(Notify::new());
494
495 let feeder = tokio::spawn(feed_stdin(child.stdin.take(), stdin_rx));
496 let stdout_task = tokio::spawn(pump(
497 child.stdout.take(),
498 handle.clone(),
499 channel,
500 None,
501 Arc::clone(&abort),
502 ));
503 let stderr_task = tokio::spawn(pump(
504 child.stderr.take(),
505 handle.clone(),
506 channel,
507 Some(1),
508 Arc::clone(&abort),
509 ));
510
511 let status = tokio::select! {
512 status = child.wait() => status,
513 () = abort.notified() => {
514 let _ = child.start_kill();
515 child.wait().await
516 }
517 };
518
519 // Wait for the pumps so every byte of output is queued before exit status and EOF.
520 let _ = stdout_task.await;
521 let _ = stderr_task.await;
522 feeder.abort();
523
524 let _ = handle.exit_status_request(channel, exit_code(status)).await;
525 let _ = handle.eof(channel).await;
526 let _ = handle.close(channel).await;
527}
528
529async fn feed_stdin(stdin: Option<ChildStdin>, mut rx: mpsc::Receiver<Bytes>) {
530 let Some(mut stdin) = stdin else {
531 return;
532 };
533 while let Some(chunk) = rx.recv().await {
534 if stdin.write_all(&chunk).await.is_err() {
535 break;
536 }
537 }
538 // Dropping stdin here closes the child's input.
539 let _ = stdin.shutdown().await;
540}
541
542/// Forwards one child output stream to the client. `ext` = Some(n) sends extended data
543/// type n. Sets `abort` when the client has gone away.
544async fn pump<R>(
545 reader: Option<R>,
546 handle: Handle,
547 channel: ChannelId,
548 ext: Option<u32>,
549 abort: Arc<Notify>,
550) where
551 R: AsyncRead + Unpin + Send + 'static,
552{
553 let Some(mut reader) = reader else {
554 return;
555 };
556 let mut buf = vec![0u8; PUMP_CHUNK];
557 loop {
558 let n = match reader.read(&mut buf).await {
559 Ok(0) | Err(_) => break,
560 Ok(n) => n,
561 };
562 let chunk = Bytes::copy_from_slice(buf.get(..n).unwrap_or_default());
563 let sent = match ext {
564 None => handle.data(channel, chunk).await.is_ok(),
565 Some(kind) => handle.extended_data(channel, kind, chunk).await.is_ok(),
566 };
567 if !sent {
568 abort.notify_one();
569 break;
570 }
571 }
572}
573
574fn exit_code(status: io::Result<ExitStatus>) -> u32 {
575 match status {
576 Ok(status) => status
577 .code()
578 .and_then(|c| u32::try_from(c).ok())
579 .or_else(|| {
580 status
581 .signal()
582 .map(|sig| 128 + u32::try_from(sig).unwrap_or(0))
583 })
584 .unwrap_or(1),
585 Err(_) => 1,
586 }
587}
588
589#[cfg(test)]
590mod tests {
591 use super::*;
592
593 #[test]
594 fn git_protocol_is_strict() {
595 assert!(valid_git_protocol("version=2"));
596 assert!(valid_git_protocol("version=1"));
597 assert!(!valid_git_protocol("version=10"));
598 assert!(!valid_git_protocol("version=x"));
599 assert!(!valid_git_protocol("version=2\n"));
600 assert!(!valid_git_protocol("version=2;x"));
601 assert!(!valid_git_protocol(""));
602 }
603
604 #[test]
605 fn key_types_exclude_dsa() {
606 assert!(key_type_allowed(&Algorithm::Ed25519));
607 assert!(key_type_allowed(&Algorithm::Rsa { hash: None }));
608 assert!(!key_type_allowed(&Algorithm::Dsa));
609 }
610}