Skip to main content

max / makenotwork

27.3 KB · 690 lines History Blame Raw
1 //! SSH session handler: authentication, PTY, shell, SFTP, and input dispatch.
2
3 use std::collections::HashMap;
4 use std::net::SocketAddr;
5 use std::path::PathBuf;
6 use std::sync::Arc;
7
8 use russh::keys::{HashAlg, PublicKey};
9 use russh::server::{Auth, ChannelOpenHandle, Msg, Session};
10 use russh::{Channel, ChannelId};
11 use tokio::sync::Mutex;
12
13 use crate::api::{MnwApiClient, UserInfo};
14 use crate::rate_limit::AuthRateLimiter;
15 use crate::ssh::git;
16 use crate::ssh::sftp::SftpSession;
17 use crate::ssh::terminal::TerminalHandle;
18 use crate::staging;
19 use crate::tui;
20
21 /// Per-connection handler. Created by `MnwServer::new_client()`.
22 pub(crate) struct MnwHandler {
23 api: MnwApiClient,
24 peer_addr: Option<SocketAddr>,
25 staging_dir: Arc<PathBuf>,
26 git_user: Arc<str>,
27 rate_limiter: Arc<AuthRateLimiter>,
28 /// Populated after successful auth.
29 user: Option<UserInfo>,
30 /// Terminal dimensions (cols, rows).
31 term_size: (u16, u16),
32 /// Whether the client requested a PTY. The TUI needs one; without it a
33 /// shell session gets the help text instead of hanging on a dead terminal.
34 pty_requested: bool,
35 /// What the client's terminal said about itself, for the theme.
36 ///
37 /// Collected here rather than read from this process's environment because
38 /// the terminal being drawn on is at the other end of the connection; see
39 /// `tui::theme`. `TERM` arrives on the PTY request, the rest on `env`
40 /// requests if the client was configured to forward them.
41 client_terminal: tui::theme::ClientTerminal,
42 /// TUI application handle for forwarding keypresses.
43 app: Option<tui::AppHandle>,
44 /// Channels stored between open and shell/subsystem request. Only SFTP
45 /// consumes one; see [`MnwHandler::release_channel`] for why every other
46 /// path has to drop its entry rather than leave it parked here.
47 channels: Arc<Mutex<HashMap<ChannelId, Channel<Msg>>>>,
48 /// Active git subprocess stdin handles, keyed by channel.
49 git_processes: HashMap<ChannelId, tokio::process::ChildStdin>,
50 /// Pending pipe upload: buffered stdin data + parsed args.
51 pipe_uploads: HashMap<ChannelId, PipeUpload>,
52 }
53
54 /// State for a pipe-mode upload (`cat file | ssh ... upload ...`).
55 pub(crate) struct PipeUpload {
56 pub user: UserInfo,
57 pub filename: String,
58 pub project_slug: String,
59 pub title: String,
60 pub price_cents: i32,
61 pub data: Vec<u8>,
62 }
63
64 impl MnwHandler {
65 pub(crate) fn new(
66 api: MnwApiClient,
67 peer_addr: Option<SocketAddr>,
68 staging_dir: Arc<PathBuf>,
69 git_user: Arc<str>,
70 rate_limiter: Arc<AuthRateLimiter>,
71 ) -> Self {
72 Self {
73 api,
74 peer_addr,
75 staging_dir,
76 git_user,
77 rate_limiter,
78 user: None,
79 term_size: (80, 24),
80 pty_requested: false,
81 client_terminal: tui::theme::ClientTerminal::default(),
82 app: None,
83 channels: Arc::new(Mutex::new(HashMap::new())),
84 git_processes: HashMap::new(),
85 pipe_uploads: HashMap::new(),
86 }
87 }
88
89 /// Drop the parked [`Channel`] for a session we are going to drive through
90 /// the [`Handle`](russh::server::Handle) and the handler callbacks instead.
91 ///
92 /// Not bookkeeping: it is what keeps a push from hanging. russh delivers
93 /// every inbound `CHANNEL_DATA` twice, first by `send().await` onto the
94 /// `Channel`'s bounded queue (`channel_buffer_size`, 100 messages) and only
95 /// then to `Handler::data`. Holding a `Channel` nobody reads means that
96 /// queue fills, the `send().await` blocks the session loop forever, and the
97 /// handler stops being called at all. The client sits in `Writing objects`
98 /// with a window the server already granted and never spends.
99 ///
100 /// It only bites above ~100 packets, which is why a small repo pushed fine
101 /// and `shop` (4.5 MiB) stalled around 12%. Dropping the receiver makes
102 /// russh's `send(...).unwrap_or(())` a no-op, and the channel keeps working
103 /// through the handle.
104 async fn release_channel(&self, channel: ChannelId) {
105 self.channels.lock().await.remove(&channel);
106 }
107 }
108
109 impl russh::server::Handler for MnwHandler {
110 type Error = anyhow::Error;
111
112 async fn auth_publickey_offered(
113 &mut self,
114 _user: &str,
115 key: &PublicKey,
116 ) -> Result<Auth, Self::Error> {
117 // Per-IP rate limiting: reject early if threshold exceeded
118 if let Some(addr) = self.peer_addr
119 && !self.rate_limiter.check(addr.ip())
120 {
121 tracing::warn!(peer = %addr, "auth rate limit exceeded");
122 return Ok(Auth::Reject {
123 proceed_with_methods: None,
124 partial_success: false,
125 });
126 }
127
128 let fingerprint = key.fingerprint(HashAlg::Sha256).to_string();
129 tracing::debug!(%fingerprint, peer = ?self.peer_addr, "key offered");
130
131 match self.api.lookup_ssh_key(&fingerprint).await {
132 Ok(Some(info)) => {
133 if info.suspended {
134 tracing::warn!(user = %info.username, "suspended user attempted SSH login");
135 if let Some(addr) = self.peer_addr {
136 self.rate_limiter.record_failure(addr.ip());
137 }
138 return Ok(Auth::Reject {
139 proceed_with_methods: None,
140 partial_success: false,
141 });
142 }
143 // Forward the SSH-authenticated actor assertion on this session's
144 // internal calls so the server verifies identity from the token,
145 // not a caller-supplied user_id.
146 self.api.set_actor_token(info.actor_token.clone());
147 self.user = Some(info);
148 Ok(Auth::Accept)
149 }
150 Ok(None) => {
151 tracing::debug!(%fingerprint, "key not found");
152 if let Some(addr) = self.peer_addr {
153 self.rate_limiter.record_failure(addr.ip());
154 }
155 Ok(Auth::Reject {
156 proceed_with_methods: None,
157 partial_success: false,
158 })
159 }
160 Err(e) => {
161 tracing::error!(error = ?e, "SSH key lookup failed");
162 Ok(Auth::Reject {
163 proceed_with_methods: None,
164 partial_success: false,
165 })
166 }
167 }
168 }
169
170 async fn auth_publickey(&mut self, _user: &str, _key: &PublicKey) -> Result<Auth, Self::Error> {
171 // If auth_publickey_offered accepted, the user is already stored.
172 if self.user.is_some() {
173 Ok(Auth::Accept)
174 } else {
175 Ok(Auth::Reject {
176 proceed_with_methods: None,
177 partial_success: false,
178 })
179 }
180 }
181
182 async fn channel_open_session(
183 &mut self,
184 channel: Channel<Msg>,
185 reply: ChannelOpenHandle,
186 _session: &mut Session,
187 ) -> Result<(), Self::Error> {
188 let channel_id = channel.id();
189 tracing::debug!(channel = %channel_id, "session channel opened");
190 // Store channel for later consumption by shell_request or subsystem_request
191 self.channels.lock().await.insert(channel_id, channel);
192 // russh 0.62 replaced the `Ok(true)` accept with an explicit handle.
193 // Registering the channel before accepting keeps the old ordering: the
194 // peer never proceeds against a channel this handler cannot find. An
195 // undropped handle rejects with AdministrativelyProhibited, so failing
196 // to accept closes the channel rather than hanging it.
197 reply.accept().await;
198 Ok(())
199 }
200
201 async fn pty_request(
202 &mut self,
203 _channel: ChannelId,
204 term: &str,
205 col_width: u32,
206 row_height: u32,
207 _pix_width: u32,
208 _pix_height: u32,
209 _modes: &[(russh::Pty, u32)],
210 _session: &mut Session,
211 ) -> Result<(), Self::Error> {
212 self.term_size = (col_width as u16, row_height as u16);
213 self.pty_requested = true;
214 // The one capability hint every client sends. `COLORTERM` would be the
215 // better answer and openssh strips it, so on most sessions this is the
216 // whole basis for the fidelity decision.
217 self.client_terminal.term = term.to_string();
218 tracing::debug!(cols = col_width, rows = row_height, term, "PTY requested");
219 Ok(())
220 }
221
222 /// Record the environment variables that decide how the TUI is painted.
223 ///
224 /// Only the two that do, and only when the client sends them at all — most
225 /// do not, since forwarding either takes a `SendEnv` line the creator wrote.
226 /// Everything else is ignored rather than accumulated: this is not a shell,
227 /// nothing here spawns a process, and an environment nobody reads is a place
228 /// for a client to park unbounded data.
229 ///
230 /// Answered with `channel_success` either way. A client that asked politely
231 /// and was refused has no way to tell whether the request was unsupported or
232 /// the session is broken, and openssh's `-o SendEnv` warns on failure.
233 async fn env_request(
234 &mut self,
235 channel: ChannelId,
236 variable_name: &str,
237 variable_value: &str,
238 session: &mut Session,
239 ) -> Result<(), Self::Error> {
240 // Bounded before it is stored. Both values are short by their own
241 // specifications (`truecolor`, `15;0`), and neither is worth a
242 // per-session allocation a client chooses the size of.
243 const MAX: usize = 64;
244 match variable_name {
245 "COLORTERM" if variable_value.len() <= MAX => {
246 self.client_terminal.colorterm = variable_value.to_string();
247 }
248 "COLORFGBG" if variable_value.len() <= MAX => {
249 self.client_terminal.colorfgbg = Some(variable_value.to_string());
250 }
251 _ => {}
252 }
253 session.channel_success(channel)?;
254 Ok(())
255 }
256
257 async fn shell_request(
258 &mut self,
259 channel: ChannelId,
260 session: &mut Session,
261 ) -> Result<(), Self::Error> {
262 // The TUI writes through the session handle and reads keypresses from
263 // `Handler::data`, so the parked channel would only queue up.
264 self.release_channel(channel).await;
265
266 let Some(ref user) = self.user else {
267 tracing::warn!("shell_request without authenticated user");
268 session.close(channel)?;
269 return Ok(());
270 };
271
272 // No PTY means no terminal to drive the TUI. Print the help text and
273 // exit rather than launching a TUI that renders into nothing and
274 // blocks forever waiting on input that can never arrive.
275 if !self.pty_requested {
276 tracing::info!(user = %user.username, "shell request without PTY, returning help");
277 let handle = session.handle();
278 let bytes = bytes::Bytes::from(crate::commands::help_text());
279 let _ = handle.data(channel, bytes).await;
280 let _ = handle.exit_status_request(channel, 0).await;
281 let _ = handle.eof(channel).await;
282 let _ = handle.close(channel).await;
283 return Ok(());
284 }
285
286 // Resolved before the terminal is put into its alternate screen, so a
287 // theme that will not load reports itself on the creator's own prompt
288 // rather than on a screen that is then torn down.
289 let selection = tui::theme::selection(user.theme_id.as_deref());
290 let theme = match tui::theme::load(&selection, &self.client_terminal) {
291 Ok(theme) => theme,
292 Err(e) => {
293 // Every id this binary can resolve to is embedded in it and
294 // covered by a test, so this is a packaging fault rather than
295 // anything the creator chose wrongly. Say so instead of leaving
296 // them looking for a setting to change.
297 tracing::error!(user = %user.username, error = ?e, "theme failed to resolve");
298 let handle = session.handle();
299 let message = format!("mnw: cannot resolve a theme to render in ({e}).\r\n");
300 let _ = handle.data(channel, bytes::Bytes::from(message)).await;
301 let _ = handle.exit_status_request(channel, 1).await;
302 let _ = handle.eof(channel).await;
303 let _ = handle.close(channel).await;
304 return Ok(());
305 }
306 };
307
308 tracing::info!(user = %user.username, "launching TUI");
309
310 let handle = session.handle();
311 let terminal_handle = TerminalHandle::new(handle.clone(), channel);
312 let (cols, rows) = self.term_size;
313
314 let user_clone = user.clone();
315 let staging_dir = staging::user_staging_dir(&self.staging_dir, &user.user_id);
316
317 match tui::launch(
318 terminal_handle,
319 user_clone,
320 theme,
321 cols,
322 rows,
323 handle,
324 channel,
325 self.api.clone(),
326 staging_dir,
327 ) {
328 Ok(app_handle) => {
329 self.app = Some(app_handle);
330 session.channel_success(channel)?;
331 }
332 Err(e) => {
333 tracing::error!(error = ?e, "TUI launch failed");
334 session.close(channel)?;
335 }
336 }
337
338 Ok(())
339 }
340
341 async fn subsystem_request(
342 &mut self,
343 channel_id: ChannelId,
344 name: &str,
345 session: &mut Session,
346 ) -> Result<(), Self::Error> {
347 if name != "sftp" {
348 tracing::debug!(subsystem = name, "unsupported subsystem requested");
349 session.close(channel_id)?;
350 return Ok(());
351 }
352
353 let Some(ref user) = self.user else {
354 tracing::warn!("subsystem_request without authenticated user");
355 session.close(channel_id)?;
356 return Ok(());
357 };
358
359 // Take the stored channel for this ID
360 let channel = self.channels.lock().await.remove(&channel_id);
361
362 let Some(channel) = channel else {
363 tracing::error!("no stored channel for SFTP subsystem");
364 session.close(channel_id)?;
365 return Ok(());
366 };
367
368 let user_staging = staging::user_staging_dir(&self.staging_dir, &user.user_id);
369 let sftp_session = SftpSession::new(
370 user.user_id.clone(),
371 user.creator_tier.clone(),
372 user_staging,
373 );
374
375 tracing::info!(user = %user.username, "starting SFTP session");
376
377 let stream = channel.into_stream();
378 tokio::spawn(async move {
379 russh_sftp::server::run(stream, sftp_session).await;
380 });
381
382 Ok(())
383 }
384
385 async fn exec_request(
386 &mut self,
387 channel: ChannelId,
388 data: &[u8],
389 session: &mut Session,
390 ) -> Result<(), Self::Error> {
391 let command_line = String::from_utf8_lossy(data);
392 let handle = session.handle();
393
394 // Every exec path below (git, pipe upload, scp notice, plain command)
395 // streams through `handle` and takes its stdin from `Handler::data`.
396 // A push is the one that moves enough bytes to deadlock on the parked
397 // channel's queue, but none of them read it.
398 self.release_channel(channel).await;
399
400 let Some(ref user) = self.user else {
401 let _ = handle.close(channel).await;
402 return Ok(());
403 };
404
405 // Git operations: bidirectional streaming via subprocess proxy
406 if let Some(request) = git::parse_command(&command_line) {
407 let operation = request.operation.command();
408 let (owner, repo_name) = (request.owner, request.repo);
409
410 tracing::info!(
411 user = %user.username,
412 %operation,
413 %owner,
414 %repo_name,
415 "git operation"
416 );
417
418 match self
419 .api
420 .git_authorize(&user.user_id, operation, owner, repo_name)
421 .await
422 {
423 Ok(auth) => {
424 // Auto-create bare repo on disk if it doesn't exist yet.
425 // The server only registers the repo in the DB — we create
426 // it here as the git user so ownership is correct.
427 if !std::path::Path::new(&auth.repo_path).exists() {
428 // Run git init directly (not via sudo) — mnw-cli is in the
429 // git group and the parent dir has setgid, so the repo
430 // gets git group ownership. Avoids systemd security
431 // restrictions that block sudo child processes.
432 match tokio::process::Command::new("git")
433 .args([
434 "init",
435 "--bare",
436 "--shared=group",
437 "-b",
438 "main",
439 &auth.repo_path,
440 ])
441 .stdout(std::process::Stdio::null())
442 .stderr(std::process::Stdio::null())
443 .status()
444 .await
445 {
446 Ok(s) if s.success() => {
447 tracing::info!(path = %auth.repo_path, "auto-created bare repository");
448 // Install post-receive hook if build trigger token is configured
449 if let Ok(token) = std::env::var("BUILD_TRIGGER_TOKEN") {
450 let _ = git::install_post_receive_hook(
451 &self.git_user,
452 &auth.repo_path,
453 &token,
454 )
455 .await;
456 }
457 }
458 Ok(s) => {
459 tracing::error!(path = %auth.repo_path, code = ?s.code(), "git init --bare failed");
460 let msg =
461 bytes::Bytes::from("fatal: failed to create repository\r\n");
462 let _ = handle.extended_data(channel, 1, msg).await;
463 let _ = handle.exit_status_request(channel, 1).await;
464 let _ = handle.eof(channel).await;
465 let _ = handle.close(channel).await;
466 return Ok(());
467 }
468 Err(e) => {
469 tracing::error!(error = ?e, "failed to spawn git init");
470 let msg = bytes::Bytes::from("fatal: internal error\r\n");
471 let _ = handle.extended_data(channel, 1, msg).await;
472 let _ = handle.exit_status_request(channel, 1).await;
473 let _ = handle.eof(channel).await;
474 let _ = handle.close(channel).await;
475 return Ok(());
476 }
477 }
478 }
479
480 match git::spawn_git_process(
481 &self.git_user,
482 operation,
483 &auth.repo_path,
484 channel,
485 handle.clone(),
486 ) {
487 Ok(stdin) => {
488 self.git_processes.insert(channel, stdin);
489 session.channel_success(channel)?;
490 }
491 Err(e) => {
492 tracing::error!(error = ?e, "failed to spawn git process");
493 let msg = bytes::Bytes::from("fatal: internal error\r\n".to_string());
494 let _ = handle.data(channel, msg).await;
495 let _ = handle.exit_status_request(channel, 1).await;
496 let _ = handle.eof(channel).await;
497 let _ = handle.close(channel).await;
498 }
499 }
500 }
501 Err(e) => {
502 let msg = bytes::Bytes::from(format!(
503 "fatal: {}\r\n",
504 crate::commands::sanitize_api_error(&e)
505 ));
506 let _ = handle.extended_data(channel, 1, msg).await;
507 let _ = handle.exit_status_request(channel, 1).await;
508 let _ = handle.eof(channel).await;
509 let _ = handle.close(channel).await;
510 }
511 }
512 return Ok(());
513 }
514
515 // Pipe upload: `cat file | ssh ... upload --filename X --project SLUG ...`
516 if command_line.starts_with("upload ") || command_line.as_ref() == "upload" {
517 let parts: Vec<&str> = command_line.split_whitespace().collect();
518 let mut filename = String::new();
519 let mut project_slug = String::new();
520 let mut title = String::new();
521 let mut price_cents: i32 = 0;
522 let mut i = 1;
523 while i < parts.len() {
524 match parts[i] {
525 "--filename" | "-f" if i + 1 < parts.len() => {
526 filename = parts[i + 1].to_string();
527 i += 1;
528 }
529 "--project" | "-p" if i + 1 < parts.len() => {
530 project_slug = parts[i + 1].to_string();
531 i += 1;
532 }
533 "--title" | "-t" if i + 1 < parts.len() => {
534 title = parts[i + 1].to_string();
535 i += 1;
536 }
537 "--price" if i + 1 < parts.len() => {
538 price_cents = parts[i + 1].parse().unwrap_or(0);
539 i += 1;
540 }
541 _ => {}
542 }
543 i += 1;
544 }
545
546 if filename.is_empty() || project_slug.is_empty() {
547 let msg = b"Usage: upload --filename NAME.ext --project SLUG [--title TITLE] [--price CENTS]\r\nPipe file data via stdin: cat file.wav | ssh cli.makenot.work upload ...\r\n";
548 let bytes = bytes::Bytes::copy_from_slice(msg);
549 tokio::spawn(async move {
550 let _ = handle.data(channel, bytes).await;
551 let _ = handle.exit_status_request(channel, 1).await;
552 let _ = handle.eof(channel).await;
553 let _ = handle.close(channel).await;
554 });
555 return Ok(());
556 }
557
558 if title.is_empty() {
559 title = staging::derive_title(&filename);
560 }
561
562 self.pipe_uploads.insert(
563 channel,
564 PipeUpload {
565 user: user.clone(),
566 filename,
567 project_slug,
568 title,
569 price_cents,
570 data: Vec::new(),
571 },
572 );
573 session.channel_success(channel)?;
574 return Ok(());
575 }
576
577 // Check if this looks like a legacy SCP transfer
578 if command_line.starts_with("scp ") {
579 let msg: &[u8] =
580 b"Use scp (not scp -O) or sftp to upload files to cli.makenot.work\r\n";
581 let bytes = bytes::Bytes::copy_from_slice(msg);
582 tokio::spawn(async move {
583 let _ = handle.data(channel, bytes).await;
584 let _ = handle.close(channel).await;
585 });
586 return Ok(());
587 }
588
589 // Execute the command
590 let user = user.clone();
591 let api = self.api.clone();
592 let cmd = command_line.to_string();
593 tracing::info!(user = %user.username, command = %cmd, "exec command");
594
595 tokio::spawn(async move {
596 let output = crate::commands::execute(&cmd, &user, &api).await;
597 let bytes = bytes::Bytes::from(output);
598 let _ = handle.data(channel, bytes).await;
599 let _ = handle.exit_status_request(channel, 0).await;
600 let _ = handle.eof(channel).await;
601 let _ = handle.close(channel).await;
602 });
603
604 Ok(())
605 }
606
607 async fn data(
608 &mut self,
609 channel: ChannelId,
610 data: &[u8],
611 _session: &mut Session,
612 ) -> Result<(), Self::Error> {
613 if let Some(stdin) = self.git_processes.get_mut(&channel) {
614 use tokio::io::AsyncWriteExt;
615 let _ = stdin.write_all(data).await;
616 } else if let Some(upload) = self.pipe_uploads.get_mut(&channel) {
617 upload.data.extend_from_slice(data);
618 } else if let Some(ref app) = self.app {
619 app.send_input(data).await;
620 }
621 Ok(())
622 }
623
624 async fn channel_eof(
625 &mut self,
626 channel: ChannelId,
627 session: &mut Session,
628 ) -> Result<(), Self::Error> {
629 if let Some(stdin) = self.git_processes.remove(&channel) {
630 drop(stdin); // Closes pipe → subprocess sees EOF
631 } else if let Some(upload) = self.pipe_uploads.remove(&channel) {
632 let handle = session.handle();
633 let api = self.api.clone();
634 tokio::spawn(async move {
635 let result = crate::commands::execute_pipe_upload(&api, upload).await;
636 let (msg, exit_code) = match result {
637 Ok(msg) => (msg, 0),
638 Err(e) => (
639 format!("Error: {}\r\n", crate::commands::sanitize_api_error(&e)),
640 1,
641 ),
642 };
643 let _ = handle.data(channel, bytes::Bytes::from(msg)).await;
644 let _ = handle.exit_status_request(channel, exit_code).await;
645 let _ = handle.eof(channel).await;
646 let _ = handle.close(channel).await;
647 });
648 }
649 Ok(())
650 }
651
652 async fn channel_close(
653 &mut self,
654 channel: ChannelId,
655 _session: &mut Session,
656 ) -> Result<(), Self::Error> {
657 // A channel that was opened and then closed without a shell, exec or
658 // subsystem request still has an entry parked in `channels`; a client
659 // that closes without sending EOF still has a git child holding a pipe.
660 // Both live for the length of the connection otherwise.
661 self.release_channel(channel).await;
662 self.git_processes.remove(&channel);
663 self.pipe_uploads.remove(&channel);
664 Ok(())
665 }
666
667 async fn window_change_request(
668 &mut self,
669 _channel: ChannelId,
670 col_width: u32,
671 row_height: u32,
672 _pix_width: u32,
673 _pix_height: u32,
674 _session: &mut Session,
675 ) -> Result<(), Self::Error> {
676 self.term_size = (col_width as u16, row_height as u16);
677 if let Some(ref app) = self.app {
678 app.send_resize(col_width as u16, row_height as u16).await;
679 }
680 Ok(())
681 }
682
683 async fn auth_succeeded(&mut self, _session: &mut Session) -> Result<(), Self::Error> {
684 if let Some(ref user) = self.user {
685 tracing::info!(user = %user.username, peer = ?self.peer_addr, "authenticated");
686 }
687 Ok(())
688 }
689 }
690