Skip to main content

max / makenotwork

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