Skip to main content

max / makenotwork

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