irosh 0.1.0

SSH sessions over Iroh peer-to-peer transport
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110

use crate::error::{Result, ServerError, TransportError};
use crate::transport::stream::IrohDuplex;
use crate::transport::transfer::{
    MAX_CHUNK_BYTES, TransferComplete, TransferFailure, TransferFailureCode, TransferReady,
    write_get_chunk, write_get_complete, write_get_ready, write_transfer_error,
};
use tokio::io::AsyncReadExt;

use crate::server::transfer::helpers::{probe_download_size, spawn_download_helper};
use crate::server::transfer::{ConnectionShellState, LiveShellContext, resolve_remote_path};

pub(crate) async fn handle_get_request(
    stream: &mut IrohDuplex,
    request: crate::transport::transfer::GetRequest,
    shell_state: ConnectionShellState,
) -> Result<()> {
    let Some(shell) = LiveShellContext::from_state(&shell_state) else {
        write_transfer_error(
            stream,
            &TransferFailure::new(
                TransferFailureCode::RemoteShellUnavailable,
                "no live shell process",
            ),
        )
        .await
        .map_err(TransportError::from)?;
        return Ok(());
    };

    let source_path = resolve_remote_path(&request.path)?;
    let expected_size = match probe_download_size(shell, &source_path).await? {
        Ok(size) => size,
        Err(failure) => {
            write_transfer_error(stream, &failure)
                .await
                .map_err(TransportError::from)?;
            return Ok(());
        }
    };

    let (mut child, helper_source) = spawn_download_helper(shell, &source_path).await?;

    let mut stdout = child
        .stdout
        .take()
        .ok_or_else(|| ServerError::TransferFailed {
            details: "stdout pipe unavailable".to_string(),
        })?;

    write_get_ready(
        stream,
        &TransferReady {
            size: expected_size,
            mode: None,
        },
    )
    .await
    .map_err(TransportError::from)?;

    let mut buffer = vec![0u8; MAX_CHUNK_BYTES];
    loop {
        let count = stdout
            .read(&mut buffer)
            .await
            .map_err(|e| ServerError::TransferFailed {
                details: format!("reading download helper stdout failed: {e}"),
            })?;
        if count == 0 {
            break;
        }
        write_get_chunk(stream, &buffer[..count])
            .await
            .map_err(TransportError::from)?;
    }

    let output = child
        .wait_with_output()
        .await
        .map_err(|e| ServerError::TransferFailed {
            details: format!("waiting for download helper failed: {e}"),
        })?;
    if !output.status.success() {
        write_transfer_error(
            stream,
            &TransferFailure::new(
                TransferFailureCode::HelperFailed,
                format!(
                    "{}; shell_pid={}; requested={}; helper_arg={}",
                    String::from_utf8_lossy(&output.stderr).trim(),
                    shell.pid(),
                    source_path.display(),
                    helper_source
                ),
            ),
        )
        .await
        .map_err(TransportError::from)?;
        return Ok(());
    }

    write_get_complete(
        stream,
        &TransferComplete {
            size: expected_size,
        },
    )
    .await
    .map_err(TransportError::from)?;
    Ok(())
}