Skip to main content

max / makenotwork

12.8 KB · 429 lines History Blame Raw
1 //! SFTP subsystem handler for file uploads.
2 //!
3 //! Presents a virtual filesystem with a single `/upload/` directory.
4 //! Files written here land in the staging directory on disk, from where
5 //! the TUI publish flow sends them to S3.
6
7 use std::collections::HashMap;
8 use std::path::PathBuf;
9
10 use russh_sftp::protocol::{
11 Attrs, Data, File, FileAttributes, Handle, Name, OpenFlags, Status, StatusCode, Version,
12 };
13
14 use crate::staging::{self, STAGING_QUOTA_BYTES, is_allowed_extension, sanitize_filename};
15
16 /// SFTP session handler for a single authenticated user.
17 pub(crate) struct SftpSession {
18 user_id: String,
19 creator_tier: Option<String>,
20 staging_dir: PathBuf,
21 open_files: HashMap<String, OpenFile>,
22 dir_handles: HashMap<String, bool>, // handle -> already_read
23 next_handle: u64,
24 }
25
26 struct OpenFile {
27 path: PathBuf,
28 file: tokio::fs::File,
29 }
30
31 impl SftpSession {
32 pub(crate) fn new(user_id: String, creator_tier: Option<String>, staging_dir: PathBuf) -> Self {
33 Self {
34 user_id,
35 creator_tier,
36 staging_dir,
37 open_files: HashMap::new(),
38 dir_handles: HashMap::new(),
39 next_handle: 1,
40 }
41 }
42
43 fn alloc_handle(&mut self) -> String {
44 let h = self.next_handle;
45 self.next_handle += 1;
46 format!("h{h}")
47 }
48
49 fn ok_status(id: u32) -> Status {
50 Status {
51 id,
52 status_code: StatusCode::Ok,
53 error_message: String::new(),
54 language_tag: String::new(),
55 }
56 }
57
58 fn is_basic_tier(&self) -> bool {
59 self.creator_tier.as_deref() == Some("basic")
60 }
61
62 fn is_upload_path(path: &str) -> bool {
63 let normalized = path.trim_matches('/');
64 normalized == "upload" || normalized.is_empty() || normalized == "."
65 }
66
67 fn extract_filename(path: &str) -> Option<&str> {
68 let normalized = path.trim_start_matches('/');
69 normalized.strip_prefix("upload/").or_else(|| {
70 // Direct filename without upload/ prefix
71 if !normalized.contains('/') && normalized != "upload" && !normalized.is_empty() {
72 Some(normalized)
73 } else {
74 None
75 }
76 })
77 }
78 }
79
80 impl russh_sftp::server::Handler for SftpSession {
81 type Error = StatusCode;
82
83 fn unimplemented(&self) -> Self::Error {
84 StatusCode::OpUnsupported
85 }
86
87 async fn init(
88 &mut self,
89 version: u32,
90 _extensions: HashMap<String, String>,
91 ) -> Result<Version, Self::Error> {
92 tracing::debug!(user = %self.user_id, sftp_version = version, "SFTP session initialized");
93
94 // Ensure staging dir exists
95 if let Err(e) = tokio::fs::create_dir_all(&self.staging_dir).await {
96 tracing::error!(error = ?e, "failed to create staging dir");
97 return Err(StatusCode::Failure);
98 }
99
100 Ok(Version::new())
101 }
102
103 async fn realpath(&mut self, id: u32, _path: String) -> Result<Name, Self::Error> {
104 Ok(Name {
105 id,
106 files: vec![File::dummy("/upload")],
107 })
108 }
109
110 async fn opendir(&mut self, id: u32, path: String) -> Result<Handle, Self::Error> {
111 if !Self::is_upload_path(&path) {
112 return Err(StatusCode::NoSuchFile);
113 }
114
115 let handle = self.alloc_handle();
116 self.dir_handles.insert(handle.clone(), false);
117
118 Ok(Handle { id, handle })
119 }
120
121 async fn readdir(&mut self, id: u32, handle: String) -> Result<Name, Self::Error> {
122 let already_read = self
123 .dir_handles
124 .get_mut(&handle)
125 .ok_or(StatusCode::Failure)?;
126
127 if *already_read {
128 return Err(StatusCode::Eof);
129 }
130 *already_read = true;
131
132 let staged = staging::list_staged_files(&self.staging_dir).await;
133
134 let files: Vec<File> = staged
135 .into_iter()
136 .map(|sf| {
137 let mut attrs = FileAttributes::empty();
138 attrs.set_regular(true);
139 attrs.size = Some(sf.size);
140 if let Ok(dur) = sf
141 .modified
142 .duration_since(std::time::SystemTime::UNIX_EPOCH)
143 {
144 attrs.mtime = Some(dur.as_secs() as u32);
145 }
146 File::new(sf.filename, attrs)
147 })
148 .collect();
149
150 Ok(Name { id, files })
151 }
152
153 async fn stat(&mut self, id: u32, path: String) -> Result<Attrs, Self::Error> {
154 if Self::is_upload_path(&path) {
155 let mut attrs = FileAttributes::empty();
156 attrs.set_dir(true);
157 attrs.permissions = Some(0o755);
158 return Ok(Attrs { id, attrs });
159 }
160
161 if let Some(filename) = Self::extract_filename(&path) {
162 let file_path = self.staging_dir.join(sanitize_filename(filename));
163 if let Ok(metadata) = tokio::fs::metadata(&file_path).await {
164 let attrs = FileAttributes::from(&metadata);
165 return Ok(Attrs { id, attrs });
166 }
167 }
168
169 Err(StatusCode::NoSuchFile)
170 }
171
172 async fn lstat(&mut self, id: u32, path: String) -> Result<Attrs, Self::Error> {
173 self.stat(id, path).await
174 }
175
176 async fn fstat(&mut self, id: u32, handle: String) -> Result<Attrs, Self::Error> {
177 if self.dir_handles.contains_key(&handle) {
178 let mut attrs = FileAttributes::empty();
179 attrs.set_dir(true);
180 attrs.permissions = Some(0o755);
181 return Ok(Attrs { id, attrs });
182 }
183
184 if let Some(of) = self.open_files.get(&handle)
185 && let Ok(metadata) = of.file.metadata().await
186 {
187 let attrs = FileAttributes::from(&metadata);
188 return Ok(Attrs { id, attrs });
189 }
190
191 Err(StatusCode::Failure)
192 }
193
194 async fn open(
195 &mut self,
196 id: u32,
197 filename: String,
198 pflags: OpenFlags,
199 _attrs: FileAttributes,
200 ) -> Result<Handle, Self::Error> {
201 // Only allow writing to /upload/<filename>
202 let raw_name = Self::extract_filename(&filename).ok_or(StatusCode::PermissionDenied)?;
203 let safe_name = sanitize_filename(raw_name);
204
205 if safe_name.is_empty() {
206 return Err(StatusCode::NoSuchFile);
207 }
208
209 // Check tier — Basic is text-only, no file uploads
210 if self.is_basic_tier() {
211 tracing::warn!(user = %self.user_id, "Basic tier user attempted SFTP upload");
212 return Err(StatusCode::PermissionDenied);
213 }
214
215 // Check extension
216 let ext = safe_name.rsplit('.').next().unwrap_or("").to_lowercase();
217 if !is_allowed_extension(&ext) {
218 tracing::warn!(user = %self.user_id, ext, "unsupported file extension");
219 return Err(StatusCode::PermissionDenied);
220 }
221
222 // Check staging quota before opening
223 let current_usage = staging::staging_usage(&self.staging_dir).await;
224 if current_usage >= STAGING_QUOTA_BYTES {
225 tracing::warn!(user = %self.user_id, usage = current_usage, "staging quota exceeded");
226 return Err(StatusCode::Failure);
227 }
228
229 let file_path = self.staging_dir.join(&safe_name);
230
231 if pflags.contains(OpenFlags::WRITE) || pflags.contains(OpenFlags::CREATE) {
232 // Ensure staging dir exists
233 if let Err(e) = tokio::fs::create_dir_all(&self.staging_dir).await {
234 tracing::error!(error = ?e, "failed to create staging dir");
235 return Err(StatusCode::Failure);
236 }
237
238 let file = tokio::fs::OpenOptions::new()
239 .write(true)
240 .create(true)
241 .truncate(pflags.contains(OpenFlags::TRUNCATE))
242 .open(&file_path)
243 .await
244 .map_err(|e| {
245 tracing::error!(error = ?e, "failed to open staging file for write");
246 StatusCode::Failure
247 })?;
248
249 let handle = self.alloc_handle();
250 self.open_files.insert(
251 handle.clone(),
252 OpenFile {
253 path: file_path,
254 file,
255 },
256 );
257
258 tracing::info!(user = %self.user_id, file = %safe_name, "staging file opened for write");
259 return Ok(Handle { id, handle });
260 }
261
262 if pflags.contains(OpenFlags::READ) {
263 let file = tokio::fs::File::open(&file_path)
264 .await
265 .map_err(|_| StatusCode::NoSuchFile)?;
266 let handle = self.alloc_handle();
267 self.open_files.insert(
268 handle.clone(),
269 OpenFile {
270 path: file_path,
271 file,
272 },
273 );
274 return Ok(Handle { id, handle });
275 }
276
277 Err(StatusCode::PermissionDenied)
278 }
279
280 async fn write(
281 &mut self,
282 id: u32,
283 handle: String,
284 offset: u64,
285 data: Vec<u8>,
286 ) -> Result<Status, Self::Error> {
287 use tokio::io::{AsyncSeekExt, AsyncWriteExt};
288
289 let of = self
290 .open_files
291 .get_mut(&handle)
292 .ok_or(StatusCode::Failure)?;
293
294 // Check staging quota (approximate — race-free enforcement at close time)
295 let current_usage = staging::staging_usage(&self.staging_dir).await;
296 if current_usage + data.len() as u64 > STAGING_QUOTA_BYTES {
297 return Err(StatusCode::Failure);
298 }
299
300 of.file
301 .seek(std::io::SeekFrom::Start(offset))
302 .await
303 .map_err(|_| StatusCode::Failure)?;
304
305 of.file
306 .write_all(&data)
307 .await
308 .map_err(|_| StatusCode::Failure)?;
309
310 Ok(Self::ok_status(id))
311 }
312
313 async fn read(
314 &mut self,
315 id: u32,
316 handle: String,
317 offset: u64,
318 len: u32,
319 ) -> Result<Data, Self::Error> {
320 use tokio::io::{AsyncReadExt, AsyncSeekExt};
321
322 let of = self
323 .open_files
324 .get_mut(&handle)
325 .ok_or(StatusCode::Failure)?;
326
327 of.file
328 .seek(std::io::SeekFrom::Start(offset))
329 .await
330 .map_err(|_| StatusCode::Failure)?;
331
332 let mut buf = vec![0u8; len as usize];
333 let n = of
334 .file
335 .read(&mut buf)
336 .await
337 .map_err(|_| StatusCode::Failure)?;
338
339 if n == 0 {
340 return Err(StatusCode::Eof);
341 }
342
343 buf.truncate(n);
344 Ok(Data { id, data: buf })
345 }
346
347 async fn close(&mut self, id: u32, handle: String) -> Result<Status, Self::Error> {
348 if self.dir_handles.remove(&handle).is_some() {
349 return Ok(Self::ok_status(id));
350 }
351
352 if let Some(of) = self.open_files.remove(&handle) {
353 drop(of.file);
354 tracing::debug!(user = %self.user_id, path = %of.path.display(), "file handle closed");
355 return Ok(Self::ok_status(id));
356 }
357
358 Err(StatusCode::Failure)
359 }
360
361 async fn remove(&mut self, id: u32, filename: String) -> Result<Status, Self::Error> {
362 let raw_name = Self::extract_filename(&filename).ok_or(StatusCode::NoSuchFile)?;
363 let safe_name = sanitize_filename(raw_name);
364 let file_path = self.staging_dir.join(&safe_name);
365
366 tokio::fs::remove_file(&file_path)
367 .await
368 .map_err(|_| StatusCode::NoSuchFile)?;
369
370 tracing::info!(user = %self.user_id, file = %safe_name, "staging file removed");
371 Ok(Self::ok_status(id))
372 }
373
374 async fn mkdir(
375 &mut self,
376 _id: u32,
377 _path: String,
378 _attrs: FileAttributes,
379 ) -> Result<Status, Self::Error> {
380 Err(StatusCode::PermissionDenied)
381 }
382
383 async fn rmdir(&mut self, _id: u32, _path: String) -> Result<Status, Self::Error> {
384 Err(StatusCode::PermissionDenied)
385 }
386
387 async fn rename(
388 &mut self,
389 _id: u32,
390 _oldpath: String,
391 _newpath: String,
392 ) -> Result<Status, Self::Error> {
393 Err(StatusCode::PermissionDenied)
394 }
395
396 async fn symlink(
397 &mut self,
398 _id: u32,
399 _linkpath: String,
400 _targetpath: String,
401 ) -> Result<Status, Self::Error> {
402 Err(StatusCode::OpUnsupported)
403 }
404
405 async fn readlink(&mut self, _id: u32, _path: String) -> Result<Name, Self::Error> {
406 Err(StatusCode::OpUnsupported)
407 }
408
409 async fn setstat(
410 &mut self,
411 id: u32,
412 _path: String,
413 _attrs: FileAttributes,
414 ) -> Result<Status, Self::Error> {
415 // Silently accept — some SFTP clients send setstat after upload
416 Ok(Self::ok_status(id))
417 }
418
419 async fn fsetstat(
420 &mut self,
421 id: u32,
422 _handle: String,
423 _attrs: FileAttributes,
424 ) -> Result<Status, Self::Error> {
425 // Silently accept
426 Ok(Self::ok_status(id))
427 }
428 }
429