ssh_mcp/transfer/
mod.rs

1//! File and directory transfer tool support.
2//!
3//! Transports:
4//! - `sftp`: OpenSSH `sftp` client (batch mode)
5//! - `scp`: OpenSSH `scp` client
6//! - `exec-raw`: stdin/stdout streaming over the existing SSH session
7//! - `auto`: fallback chain `sftp -> scp -> exec-raw`
8
9mod exec_raw;
10mod local_root;
11mod openssh;
12mod process;
13mod rsync;
14mod skeleton;
15mod staging;
16mod tar;
17mod types;
18mod walk;
19
20pub use types::{
21    CompactTransferResponse, ResolvedPaths, RsyncOptions, StagingLocal, StagingRemote,
22    TransferCounts, TransferKind, TransferOperation, TransferParams, TransferResponse,
23    TransferStaging, TransferTransport,
24};
25pub(crate) use types::{TransferEvent, TransferEventSink, TransferProgressTarget};
26
27use std::collections::HashSet;
28use std::fmt::Write;
29use std::path::{Path, PathBuf};
30use std::sync::Arc;
31use std::sync::Mutex;
32use std::time::Duration;
33
34use tokio::time::Instant;
35use tokio_util::sync::CancellationToken;
36
37use crate::error::{Result, SshMcpError};
38use crate::ssh::{HostKeyCheckMode, SshConnectionManager};
39
40fn io_to_transport_attempt(err: std::io::Error) -> TransportAttemptError {
41    TransportAttemptError::Other(SshMcpError::Io(err))
42}
43
44struct StepCtx<'a> {
45    conn: &'a SshConnectionManager,
46    remote_home: &'a str,
47    id: &'a str,
48    kind: TransferKind,
49    resolved: &'a ResolvedPaths,
50    timeout: Duration,
51    cancellation: &'a CancellationToken,
52    progress: Option<&'a TransferEventSink>,
53    response: &'a mut TransferResponse,
54}
55
56struct OpenSshContext<'a> {
57    conn: &'a SshConnectionManager,
58    remote_home: &'a str,
59    key_path: Option<&'a Path>,
60    ssh: &'a TransferSshOptions,
61    id: &'a str,
62    timeout: Duration,
63    cancellation: &'a CancellationToken,
64    progress: Option<&'a TransferEventSink>,
65}
66
67struct OpenSshOperation<'a> {
68    transport: openssh::OpenSshTransport,
69    kind: TransferKind,
70    response: &'a mut TransferResponse,
71}
72
73struct ExecRawOperation<'a> {
74    conn: &'a SshConnectionManager,
75    remote_home: &'a str,
76    id: &'a str,
77    kind: TransferKind,
78    timeout: Duration,
79    cancellation: &'a CancellationToken,
80    progress: Option<&'a TransferEventSink>,
81    response: &'a mut TransferResponse,
82}
83
84/// Core transfer engine.
85///
86/// For now this selects the EXEC-RAW transport, but it is structured so that
87/// SFTP/SCP can be added as additional implementations.
88#[derive(Clone, Debug)]
89pub struct TransferEngine {
90    local_root: Arc<PathBuf>,
91    active_destinations: Arc<Mutex<HashSet<String>>>,
92}
93
94#[derive(Clone, Debug)]
95pub struct TransferRunContext {
96    pub timeout: Duration,
97    pub ssh: TransferSshOptions,
98}
99
100struct TransferExecutionContext {
101    timeout: Duration,
102    ssh: TransferSshOptions,
103    cancellation: CancellationToken,
104    progress: Option<TransferEventSink>,
105}
106
107#[derive(Clone, Debug)]
108pub struct TransferSshOptions {
109    pub host: String,
110    pub port: u16,
111    pub user: String,
112    pub key_path: Option<PathBuf>,
113    pub host_key_checking: HostKeyCheckMode,
114    pub known_hosts: Option<PathBuf>,
115}
116
117struct DestinationGuard {
118    key: String,
119    active: Arc<Mutex<HashSet<String>>>,
120}
121
122impl Drop for DestinationGuard {
123    fn drop(&mut self) {
124        if let Ok(mut active) = self.active.lock() {
125            active.remove(&self.key);
126        }
127    }
128}
129
130impl TransferEngine {
131    pub fn new(local_root: PathBuf) -> Self {
132        Self {
133            local_root: Arc::new(local_root),
134            active_destinations: Arc::new(Mutex::new(HashSet::new())),
135        }
136    }
137
138    pub fn local_root(&self) -> &Path {
139        self.local_root.as_path()
140    }
141
142    fn next_attempt_token(&self) -> Result<String> {
143        let mut bytes = [0u8; 16];
144        getrandom::fill(&mut bytes).map_err(|error| {
145            SshMcpError::connection(format!("failed to create staging token: {error}"))
146        })?;
147        let mut token = String::with_capacity(32);
148        for byte in bytes {
149            write!(&mut token, "{byte:02x}")
150                .map_err(|_| SshMcpError::connection("failed to format staging token"))?;
151        }
152        Ok(token)
153    }
154
155    fn reserve_destination(
156        &self,
157        params: &TransferParams,
158        kind: TransferKind,
159    ) -> Result<DestinationGuard> {
160        let key = match params.operation {
161            TransferOperation::Put => {
162                exec_raw::validate_remote_user_path(&params.remote_path, "remote_path")?;
163                format!("put:{}", normalize_remote_path(&params.remote_path))
164            }
165            TransferOperation::Get => {
166                let resolved = local_root::resolve_paths(self.local_root(), params, kind)
167                    .map_err(SshMcpError::invalid_params)?;
168                format!("get:{}", resolved.local_path.display())
169            }
170        };
171
172        let mut active = self
173            .active_destinations
174            .lock()
175            .map_err(|_| SshMcpError::connection("destination guard poisoned"))?;
176        if !active.insert(key.clone()) {
177            return Err(SshMcpError::invalid_params(format!(
178                "destination busy: {}",
179                match params.operation {
180                    TransferOperation::Put => &params.remote_path,
181                    TransferOperation::Get => &params.local_path,
182                }
183            )));
184        }
185        drop(active);
186
187        Ok(DestinationGuard {
188            key,
189            active: Arc::clone(&self.active_destinations),
190        })
191    }
192
193    pub async fn run(
194        &self,
195        conn: &SshConnectionManager,
196        params: TransferParams,
197        ctx: TransferRunContext,
198    ) -> TransferResponse {
199        self.run_controlled(conn, params, ctx, CancellationToken::new(), None)
200            .await
201    }
202
203    pub(crate) async fn run_controlled(
204        &self,
205        conn: &SshConnectionManager,
206        params: TransferParams,
207        ctx: TransferRunContext,
208        external_cancellation: CancellationToken,
209        progress: Option<TransferEventSink>,
210    ) -> TransferResponse {
211        const TEARDOWN_GRACE: Duration = Duration::from_secs(5);
212
213        let started_at = Instant::now();
214        let transfer_timeout = ctx.timeout;
215        let response_params = params.clone();
216        let work_cancellation = CancellationToken::new();
217        let execution_ctx = TransferExecutionContext {
218            timeout: ctx.timeout,
219            ssh: ctx.ssh,
220            cancellation: work_cancellation.clone(),
221            progress,
222        };
223
224        let work = self.run_inner(conn, params, execution_ctx, started_at);
225        tokio::pin!(work);
226
227        enum StopReason {
228            Cancelled,
229            TimedOut,
230        }
231
232        let stop_reason = tokio::select! {
233            response = &mut work => return response,
234            _ = external_cancellation.cancelled() => StopReason::Cancelled,
235            _ = tokio::time::sleep_until(started_at + transfer_timeout) => StopReason::TimedOut,
236        };
237
238        work_cancellation.cancel();
239        let _ = tokio::time::timeout(TEARDOWN_GRACE, &mut work).await;
240
241        let message = match stop_reason {
242            StopReason::Cancelled => "transfer cancelled".to_string(),
243            StopReason::TimedOut => {
244                format!("transfer timeout after {}ms", transfer_timeout.as_millis())
245            }
246        };
247        let mut response = TransferResponse::error(response_params, self.local_root(), &message);
248        response.elapsed_ms = Some(started_at.elapsed().as_millis() as u64);
249        response
250    }
251
252    async fn run_inner(
253        &self,
254        conn: &SshConnectionManager,
255        params: TransferParams,
256        ctx: TransferExecutionContext,
257        started_at: Instant,
258    ) -> TransferResponse {
259        let key_path_opt = ctx.ssh.key_path.clone();
260
261        if let Some(progress) = &ctx.progress {
262            progress.emit(types::TransferEvent::Preparing);
263        }
264
265        let remote_home = match exec_raw::resolve_remote_home(conn, ctx.timeout).await {
266            Ok(home) => home,
267            Err(e) => {
268                return TransferResponse::error(
269                    params,
270                    self.local_root(),
271                    &format!("failed to resolve remote HOME: {e}"),
272                );
273            }
274        };
275
276        let mut response = TransferResponse::ok_stub(
277            params,
278            TransferTransport::ExecRaw,
279            &remote_home,
280            self.local_root(),
281        );
282
283        let kind = match resolve_kind(
284            conn,
285            self.local_root(),
286            &response.params,
287            ctx.timeout,
288            &ctx.cancellation,
289        )
290        .await
291        {
292            Ok(kind) => kind,
293            Err(e) => {
294                response.set_error(&format!("failed to resolve transfer kind: {e}"));
295                response.elapsed_ms = Some(started_at.elapsed().as_millis() as u64);
296                return response;
297            }
298        };
299        response.kind = Some(kind);
300
301        let _destination_guard = match self.reserve_destination(&response.params, kind) {
302            Ok(guard) => guard,
303            Err(error) => {
304                response.set_error(&error.to_string());
305                response.elapsed_ms = Some(started_at.elapsed().as_millis() as u64);
306                return response;
307            }
308        };
309
310        let transports = match response.params.transport {
311            TransferTransport::Auto => {
312                vec![
313                    TransferTransport::Rsync,   // Try rsync first (most efficient)
314                    TransferTransport::Sftp,    // Fallback to sftp
315                    TransferTransport::Scp,     // Fallback to scp
316                    TransferTransport::ExecRaw, // Last resort
317                ]
318            }
319            other => vec![other],
320        };
321
322        let mut attempted_transports: Vec<TransferTransport> = Vec::new();
323        let mut unsupported_reasons: Vec<String> = Vec::new();
324
325        for transport in transports {
326            let id = match self.next_attempt_token() {
327                Ok(id) => id,
328                Err(error) => {
329                    response.set_error(&error.to_string());
330                    break;
331                }
332            };
333            attempted_transports.push(transport);
334            response.transport_used = transport;
335            if let Some(progress) = &ctx.progress {
336                progress.emit(types::TransferEvent::Transferring(transport));
337            }
338            let attempt = match transport {
339                TransferTransport::ExecRaw => self
340                    .run_exec_raw(ExecRawOperation {
341                        conn,
342                        remote_home: &remote_home,
343                        id: &id,
344                        kind,
345                        timeout: ctx.timeout,
346                        cancellation: &ctx.cancellation,
347                        progress: ctx.progress.as_ref(),
348                        response: &mut response,
349                    })
350                    .await
351                    .map_err(TransportAttemptError::Other),
352                TransferTransport::Sftp => {
353                    self.run_openssh(
354                        OpenSshContext {
355                            conn,
356                            remote_home: &remote_home,
357                            key_path: key_path_opt.as_deref(),
358                            ssh: &ctx.ssh,
359                            id: &id,
360                            timeout: ctx.timeout,
361                            cancellation: &ctx.cancellation,
362                            progress: ctx.progress.as_ref(),
363                        },
364                        OpenSshOperation {
365                            transport: openssh::OpenSshTransport::Sftp,
366                            kind,
367                            response: &mut response,
368                        },
369                    )
370                    .await
371                }
372                TransferTransport::Scp => {
373                    self.run_openssh(
374                        OpenSshContext {
375                            conn,
376                            remote_home: &remote_home,
377                            key_path: key_path_opt.as_deref(),
378                            ssh: &ctx.ssh,
379                            id: &id,
380                            timeout: ctx.timeout,
381                            cancellation: &ctx.cancellation,
382                            progress: ctx.progress.as_ref(),
383                        },
384                        OpenSshOperation {
385                            transport: openssh::OpenSshTransport::Scp,
386                            kind,
387                            response: &mut response,
388                        },
389                    )
390                    .await
391                }
392                TransferTransport::Auto => {
393                    Err(TransportAttemptError::Other(SshMcpError::connection(
394                        "internal error: transport=auto should have been expanded",
395                    )))
396                }
397                TransferTransport::Rsync => {
398                    self.run_rsync(
399                        OpenSshContext {
400                            conn,
401                            remote_home: &remote_home,
402                            key_path: key_path_opt.as_deref(),
403                            ssh: &ctx.ssh,
404                            id: &id,
405                            timeout: ctx.timeout,
406                            cancellation: &ctx.cancellation,
407                            progress: ctx.progress.as_ref(),
408                        },
409                        kind,
410                        &mut response,
411                    )
412                    .await
413                }
414            };
415
416            match attempt {
417                Ok(()) => {
418                    response.ok = true;
419                    break;
420                }
421                Err(TransportAttemptError::FallbackSafe { transport, reason }) => {
422                    unsupported_reasons.push(format!("{transport:?}: {reason}"));
423                    continue;
424                }
425                Err(e) => {
426                    response.set_error(&e.to_string());
427                    break;
428                }
429            }
430        }
431
432        // Only populate fallback_chain when the original transport was Auto
433        if response.params.transport == TransferTransport::Auto {
434            response.fallback_chain = attempted_transports;
435        }
436
437        if !response.ok
438            && response.error.is_none()
439            && response.params.transport == TransferTransport::Auto
440        {
441            let all_reasons = unsupported_reasons;
442
443            if all_reasons.is_empty() {
444                response.set_error("all transfer transports failed");
445            } else {
446                response.set_error(&format!(
447                    "all auto transports failed: {}",
448                    all_reasons.join("; ")
449                ));
450            }
451        }
452
453        response.elapsed_ms = Some(started_at.elapsed().as_millis() as u64);
454        response
455    }
456
457    async fn run_exec_raw(&self, operation: ExecRawOperation<'_>) -> Result<()> {
458        let ExecRawOperation {
459            conn,
460            remote_home,
461            id,
462            kind,
463            timeout,
464            cancellation,
465            progress,
466            response,
467        } = operation;
468        let resolved = self
469            .resolve_and_validate_local_paths(&response.params, kind)
470            .await?;
471        response.resolved_paths = Some(resolved.clone());
472
473        let mut ctx = StepCtx {
474            conn,
475            remote_home,
476            id,
477            kind,
478            resolved: &resolved,
479            timeout,
480            cancellation,
481            progress,
482            response,
483        };
484
485        match ctx.response.params.operation {
486            TransferOperation::Put => self.put(&mut ctx).await,
487            TransferOperation::Get => self.get(&mut ctx).await,
488        }
489    }
490
491    async fn put(&self, ctx: &mut StepCtx<'_>) -> Result<()> {
492        let raw_ctx = exec_raw::ExecRawCtx {
493            conn: ctx.conn,
494            id: ctx.id,
495            timeout: ctx.timeout,
496            cancellation: ctx.cancellation,
497            progress: ctx.progress,
498        };
499
500        match ctx.kind {
501            TransferKind::File => {
502                let (staging, counts) = exec_raw::put_file_exec_raw(exec_raw::PutFileExecRawArgs {
503                    ctx: raw_ctx,
504                    remote_home: ctx.remote_home,
505                    local_src: &ctx.resolved.local_path,
506                    remote_dst: &ctx.response.params.remote_path,
507                    overwrite: ctx.response.params.overwrite,
508                })
509                .await?;
510                ctx.response.staging = Some(staging);
511                ctx.response.counts = Some(counts);
512                Ok(())
513            }
514            TransferKind::Directory => {
515                let (staging, counts) = exec_raw::put_dir_exec_raw(exec_raw::PutDirExecRawArgs {
516                    ctx: raw_ctx,
517                    remote_home: ctx.remote_home,
518                    local_src_dir: &ctx.resolved.local_path,
519                    remote_dst_dir: &ctx.response.params.remote_path,
520                    overwrite: ctx.response.params.overwrite,
521                })
522                .await?;
523                ctx.response.staging = Some(staging);
524                ctx.response.counts = Some(counts);
525                ctx.response.semantics = Some(
526                    "directory transfer behavior depends on overwrite: if overwrite=true, it uses an exclusively created sibling staging directory and rollback-protected rename; if overwrite=false, it creates the destination directory and writes directly into it (no atomic swap); on upload error it attempts to remove the stage directory (best-effort; for overwrite=false this is the created destination directory, and partial contents may remain)"
527                        .to_string(),
528                );
529                Ok(())
530            }
531        }
532    }
533
534    async fn get(&self, ctx: &mut StepCtx<'_>) -> Result<()> {
535        let raw_ctx = exec_raw::ExecRawCtx {
536            conn: ctx.conn,
537            id: ctx.id,
538            timeout: ctx.timeout,
539            cancellation: ctx.cancellation,
540            progress: ctx.progress,
541        };
542
543        // If the client explicitly provided a kind, validate the remote path kind
544        // before starting any streaming transfer.
545        if ctx.response.params.kind.is_some() {
546            let remote_kind = exec_raw::probe_remote_kind(exec_raw::ProbeRemoteKindArgs {
547                ctx: raw_ctx,
548                remote_path: &ctx.response.params.remote_path,
549            })
550            .await?;
551
552            if remote_kind != ctx.kind {
553                let msg = match ctx.kind {
554                    TransferKind::File => "remote_path is not a file",
555                    TransferKind::Directory => "remote_path is not a directory",
556                };
557                return Err(SshMcpError::invalid_params(msg));
558            }
559        }
560
561        match ctx.kind {
562            TransferKind::File => {
563                let (staging, counts) = exec_raw::get_file_exec_raw(exec_raw::GetFileExecRawArgs {
564                    ctx: raw_ctx,
565                    remote_src: &ctx.response.params.remote_path,
566                    local_dst: &ctx.resolved.local_path,
567                    local_root: self.local_root(),
568                    overwrite: ctx.response.params.overwrite,
569                })
570                .await?;
571                ctx.response.staging = Some(staging);
572                ctx.response.counts = Some(counts);
573                Ok(())
574            }
575            TransferKind::Directory => {
576                let (staging, counts) = exec_raw::get_dir_exec_raw(exec_raw::GetDirExecRawArgs {
577                    ctx: raw_ctx,
578                    remote_src_dir: &ctx.response.params.remote_path,
579                    local_dst_dir: &ctx.resolved.local_path,
580                    local_root: self.local_root(),
581                    overwrite: ctx.response.params.overwrite,
582                })
583                .await?;
584                ctx.response.staging = Some(staging);
585                ctx.response.counts = Some(counts);
586                ctx.response.semantics = Some(
587                    "directory transfer writes into an exclusively created sibling staging directory under local_root, then installs it with rollback-protected rename; local_path must not normalize to '.'"
588                        .to_string(),
589                );
590                Ok(())
591            }
592        }
593    }
594
595    async fn run_openssh(
596        &self,
597        ctx: OpenSshContext<'_>,
598        op: OpenSshOperation<'_>,
599    ) -> std::result::Result<(), TransportAttemptError> {
600        let key_path = match ctx.key_path {
601            Some(p) => p,
602            None => {
603                return Err(TransportAttemptError::FallbackSafe {
604                    transport: match op.transport {
605                        openssh::OpenSshTransport::Sftp => TransferTransport::Sftp,
606                        openssh::OpenSshTransport::Scp => TransferTransport::Scp,
607                    },
608                    reason: "SSH key required for OpenSSH transports (sftp/scp)".to_string(),
609                });
610            }
611        };
612
613        let kind = op.kind;
614        let response = op.response;
615
616        let resolved = self
617            .resolve_and_validate_local_paths(&response.params, kind)
618            .await
619            .map_err(TransportAttemptError::Other)?;
620        response.resolved_paths = Some(resolved.clone());
621
622        // If the client explicitly provided a kind for get, validate the remote path kind
623        // before invoking OpenSSH tooling.
624        let (operation, remote_path, kind_override) = {
625            let params = &response.params;
626            (params.operation, params.remote_path.clone(), params.kind)
627        };
628
629        if matches!(operation, TransferOperation::Get) && kind_override.is_some() {
630            let remote_kind = exec_raw::probe_remote_kind(exec_raw::ProbeRemoteKindArgs {
631                ctx: exec_raw::ExecRawCtx {
632                    conn: ctx.conn,
633                    id: ctx.id,
634                    timeout: ctx.timeout,
635                    cancellation: ctx.cancellation,
636                    progress: ctx.progress,
637                },
638                remote_path: &remote_path,
639            })
640            .await
641            .map_err(TransportAttemptError::Other)?;
642
643            if remote_kind != kind {
644                let msg = match kind {
645                    TransferKind::File => "remote_path is not a file",
646                    TransferKind::Directory => "remote_path is not a directory",
647                };
648                return Err(TransportAttemptError::Other(SshMcpError::invalid_params(
649                    msg,
650                )));
651            }
652        }
653
654        let endpoint = openssh::OpenSshEndpoint {
655            host: ctx.ssh.host.clone(),
656            port: ctx.ssh.port,
657            user: ctx.ssh.user.clone(),
658            key_path: key_path.to_path_buf(),
659            host_key_checking: ctx.ssh.host_key_checking,
660            known_hosts: ctx.ssh.known_hosts.clone(),
661        };
662
663        let overwrite = response.params.overwrite;
664
665        let openssh_args = openssh::OpenSshTransferArgs {
666            transport: op.transport,
667            conn: ctx.conn,
668            remote_home: ctx.remote_home,
669            local_root: self.local_root(),
670            id: ctx.id.to_string(),
671            timeout: ctx.timeout,
672            cancellation: ctx.cancellation.clone(),
673            progress: ctx.progress.cloned(),
674            operation,
675            kind,
676            local_path: resolved.local_path,
677            remote_path,
678            overwrite,
679        };
680
681        let (staging, counts) = openssh::run_transfer(endpoint, openssh_args).await?;
682        response.staging = Some(staging);
683        response.counts = Some(counts);
684        if kind == TransferKind::Directory {
685            response.semantics = Some(match operation {
686                TransferOperation::Put => "directory transfer behavior depends on overwrite: if overwrite=true, it uses an exclusively created sibling staging directory and rollback-protected rename; if overwrite=false, it creates the destination directory and writes directly into it (no atomic swap)".to_string(),
687                TransferOperation::Get => "directory transfer writes into an exclusively created sibling staging directory under local_root, then installs it with rollback-protected rename; local_path must not normalize to '.'".to_string(),
688            });
689        }
690        Ok(())
691    }
692
693    async fn run_rsync(
694        &self,
695        ctx: OpenSshContext<'_>,
696        kind: TransferKind,
697        response: &mut TransferResponse,
698    ) -> std::result::Result<(), TransportAttemptError> {
699        if ctx.key_path.is_none() {
700            return Err(TransportAttemptError::FallbackSafe {
701                transport: TransferTransport::Rsync,
702                reason: "SSH key required for rsync transport".to_string(),
703            });
704        }
705
706        let resolved = self
707            .resolve_and_validate_local_paths(&response.params, kind)
708            .await
709            .map_err(TransportAttemptError::Other)?;
710        response.resolved_paths = Some(resolved.clone());
711
712        // If the client explicitly provided a kind for get, validate the remote path kind
713        // before invoking rsync.
714        let (operation, remote_path, kind_override) = {
715            let params = &response.params;
716            (params.operation, params.remote_path.clone(), params.kind)
717        };
718
719        if matches!(operation, TransferOperation::Get) && kind_override.is_some() {
720            let remote_kind = exec_raw::probe_remote_kind(exec_raw::ProbeRemoteKindArgs {
721                ctx: exec_raw::ExecRawCtx {
722                    conn: ctx.conn,
723                    id: ctx.id,
724                    timeout: ctx.timeout,
725                    cancellation: ctx.cancellation,
726                    progress: ctx.progress,
727                },
728                remote_path: &remote_path,
729            })
730            .await
731            .map_err(TransportAttemptError::Other)?;
732
733            if remote_kind != kind {
734                let msg = match kind {
735                    TransferKind::File => "remote_path is not a file",
736                    TransferKind::Directory => "remote_path is not a directory",
737                };
738                return Err(TransportAttemptError::Other(SshMcpError::invalid_params(
739                    msg,
740                )));
741            }
742        }
743
744        let endpoint = rsync::RsyncEndpoint {
745            host: ctx.ssh.host.clone(),
746            port: ctx.ssh.port,
747            user: ctx.ssh.user.clone(),
748            key_path: ctx.key_path.map(|p| p.to_path_buf()),
749            host_key_checking: ctx.ssh.host_key_checking,
750            known_hosts: ctx.ssh.known_hosts.clone(),
751        };
752
753        let overwrite = response.params.overwrite;
754        let rsync_options = response.params.rsync_options.clone();
755
756        let rsync_args = rsync::RsyncTransferArgs {
757            conn: ctx.conn,
758            remote_home: ctx.remote_home,
759            local_root: self.local_root(),
760            id: ctx.id.to_string(),
761            timeout: ctx.timeout,
762            cancellation: ctx.cancellation.clone(),
763            progress: ctx.progress.cloned(),
764            operation,
765            kind,
766            local_path: &resolved.local_path,
767            remote_path: &remote_path,
768            overwrite,
769            rsync_options,
770        };
771
772        let (staging, counts) = rsync::run_transfer(endpoint, rsync_args).await?;
773        response.staging = Some(staging);
774        response.counts = Some(counts);
775        if kind == TransferKind::Directory {
776            response.semantics = Some(match operation {
777                TransferOperation::Put => "directory transfer behavior depends on overwrite: if overwrite=true, it uses an exclusively created sibling staging directory and rollback-protected rename; if overwrite=false, it creates the destination directory and writes directly into it (no atomic swap)".to_string(),
778                TransferOperation::Get => "directory transfer writes into an exclusively created sibling staging directory under local_root, then installs it with rollback-protected rename; local_path must not normalize to '.'".to_string(),
779            });
780        }
781        Ok(())
782    }
783
784    async fn resolve_and_validate_local_paths(
785        &self,
786        params: &TransferParams,
787        kind: TransferKind,
788    ) -> Result<ResolvedPaths> {
789        let resolved = local_root::resolve_paths(self.local_root(), params, kind)
790            .map_err(SshMcpError::invalid_params)?;
791
792        if matches!(params.operation, TransferOperation::Get) {
793            local_root::validate_get_target_no_symlinks(self.local_root(), &resolved.local_path)
794                .await
795                .map_err(SshMcpError::invalid_params)?;
796
797            // Create missing parent directories without following symlinks (best-effort).
798            local_root::ensure_parent_dirs_no_symlinks(self.local_root(), &resolved.local_path)
799                .await?;
800        } else {
801            // Best-effort: reject symlink components for put sources to prevent escaping local_root.
802            local_root::validate_put_source_no_symlinks(self.local_root(), &resolved.local_path)
803                .await
804                .map_err(SshMcpError::invalid_params)?;
805        }
806
807        Ok(resolved)
808    }
809}
810
811fn normalize_remote_path(path: &str) -> String {
812    let absolute = path.starts_with('/');
813    let mut parts: Vec<&str> = Vec::new();
814    for part in path.split('/') {
815        match part {
816            "" | "." => {}
817            ".." if parts.last().is_some_and(|last| *last != "..") => {
818                parts.pop();
819            }
820            ".." if !absolute => parts.push(part),
821            ".." => {}
822            _ => parts.push(part),
823        }
824    }
825
826    let joined = parts.join("/");
827    if absolute {
828        format!("/{joined}")
829    } else if joined.is_empty() {
830        ".".to_string()
831    } else {
832        joined
833    }
834}
835
836#[derive(Debug)]
837enum TransportAttemptError {
838    FallbackSafe {
839        transport: TransferTransport,
840        reason: String,
841    },
842    Other(SshMcpError),
843}
844
845impl std::fmt::Display for TransportAttemptError {
846    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
847        match self {
848            Self::FallbackSafe { transport, reason } => {
849                write!(f, "transport {transport:?} unsupported: {reason}")
850            }
851            Self::Other(e) => write!(f, "{e}"),
852        }
853    }
854}
855
856impl std::error::Error for TransportAttemptError {}
857
858async fn resolve_kind(
859    conn: &SshConnectionManager,
860    local_root: &Path,
861    params: &TransferParams,
862    timeout: Duration,
863    cancellation: &CancellationToken,
864) -> Result<TransferKind> {
865    match params.kind {
866        Some(kind) => Ok(kind),
867        None => match params.operation {
868            TransferOperation::Put => {
869                let local_src = local_root::safe_join_local_root(local_root, &params.local_path)
870                    .map_err(SshMcpError::invalid_params)?;
871                let meta = tokio::fs::symlink_metadata(&local_src).await?;
872                if meta.is_dir() {
873                    Ok(TransferKind::Directory)
874                } else {
875                    Ok(TransferKind::File)
876                }
877            }
878            TransferOperation::Get => {
879                exec_raw::probe_remote_kind(exec_raw::ProbeRemoteKindArgs {
880                    ctx: exec_raw::ExecRawCtx {
881                        conn,
882                        id: "",
883                        timeout,
884                        cancellation,
885                        progress: None,
886                    },
887                    remote_path: &params.remote_path,
888                })
889                .await
890            }
891        },
892    }
893}
894
895#[cfg(test)]
896mod tests {
897    use super::*;
898
899    #[test]
900    fn destination_guard_rejects_normalized_alias_until_release() {
901        let engine = TransferEngine::new(PathBuf::from("/tmp/local-root"));
902        let first = TransferParams {
903            remote_path: "/tmp/a/../target".to_string(),
904            ..TransferParams::default()
905        };
906        let second = TransferParams {
907            remote_path: "/tmp/target".to_string(),
908            ..TransferParams::default()
909        };
910
911        let guard = engine
912            .reserve_destination(&first, TransferKind::File)
913            .expect("first destination reservation");
914        let error = engine
915            .reserve_destination(&second, TransferKind::File)
916            .err()
917            .expect("normalized alias must be busy");
918        assert!(error.to_string().contains("destination busy"));
919
920        drop(guard);
921        assert!(
922            engine
923                .reserve_destination(&second, TransferKind::File)
924                .is_ok()
925        );
926    }
927
928    #[test]
929    fn attempt_tokens_are_128_bit_hex() {
930        let engine = TransferEngine::new(PathBuf::from("/tmp/local-root"));
931        let first = engine.next_attempt_token().expect("first token");
932        let second = engine.next_attempt_token().expect("second token");
933
934        assert_eq!(first.len(), 32);
935        assert!(first.bytes().all(|byte| byte.is_ascii_hexdigit()));
936        assert_ne!(first, second);
937    }
938}
ssh_mcp/transfer/mod.rs

ssh_mcp/transfer/
mod.rs
