netget/client/ssh/
mod.rs

1//! SSH commands and bounded SFTP v3 operations on one authenticated owned session.
2pub mod actions;
3pub mod sftp;
4use crate::client::{command_support, llm_budget::call_llm_for_client};
5use crate::llm::actions::client_trait::{Client, ClientActionResult};
6use crate::logging::emit::Log;
7use crate::protocol::{ConnectContext, Event};
8use crate::state::{
9    client_handles::{ClientCommand, ClientSendOutcome},
10    AccessLogOwner, ClientStatus,
11};
12pub use actions::SshClientProtocol;
13use actions::{
14    SSH_CLIENT_CONNECTED_EVENT, SSH_CLIENT_OUTPUT_RECEIVED_EVENT, SSH_OPERATION_FAILED_EVENT,
15    SSH_SFTP_RESULT_EVENT,
16};
17use anyhow::{bail, ensure, Context, Result};
18use futures::{future::BoxFuture, stream::FuturesUnordered, FutureExt, StreamExt};
19use russh::{
20    client::{self, Handle},
21    ChannelMsg, Disconnect,
22};
23use russh_keys::key;
24use serde_json::{json, Value};
25use std::{
26    collections::HashMap,
27    net::{Shutdown, SocketAddr},
28    sync::Arc,
29    time::Duration,
30};
31
32pub const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
33pub const OPERATION_TIMEOUT: Duration = Duration::from_secs(30);
34pub const IDLE_TIMEOUT: Duration = Duration::from_secs(300);
35pub const MAX_OPERATIONS: usize = 16;
36pub const MAX_COMMAND_OUTPUT: usize = 1024 * 1024;
37pub const MAX_CHANNEL_BYTES: usize = 3 * 1024 * 1024;
38pub const MAX_CHANNEL_MESSAGES: usize = 4096;
39const MAX_FOLLOWUP_DEPTH: u8 = 4;
40const CONNECT_DEPTH: u8 = u8::MAX;
41type HandlerFuture = BoxFuture<'static, (u8, Result<Vec<Value>>)>;
42type OperationFuture = BoxFuture<'static, (u8, Event)>;
43struct SocketGuard(Arc<std::net::TcpStream>);
44impl Drop for SocketGuard {
45    fn drop(&mut self) {
46        let _ = self.0.shutdown(Shutdown::Both);
47    }
48}
49struct ClientHandler {
50    expected: Option<String>,
51    channels: HashMap<russh::ChannelId, (usize, usize)>,
52}
53impl ClientHandler {
54    fn receive(&mut self, channel: russh::ChannelId, bytes: usize) -> Result<()> {
55        let (received, messages) = self
56            .channels
57            .get_mut(&channel)
58            .context("SSH data for an unowned channel")?;
59        ensure!(
60            bytes <= MAX_CHANNEL_BYTES.saturating_sub(*received),
61            "SSH channel byte budget exceeded"
62        );
63        ensure!(
64            *messages < MAX_CHANNEL_MESSAGES,
65            "SSH channel message budget exceeded"
66        );
67        *received += bytes;
68        *messages += 1;
69        Ok(())
70    }
71}
72#[async_trait::async_trait]
73impl client::Handler for ClientHandler {
74    type Error = anyhow::Error;
75    async fn check_server_key(&mut self, key: &key::PublicKey) -> Result<bool, Self::Error> {
76        // Preserve legacy command-only connections; SFTP refuses an unpinned session.
77        Ok(self
78            .expected
79            .as_ref()
80            .is_none_or(|pin| key.fingerprint() == *pin))
81    }
82    async fn channel_open_confirmation(
83        &mut self,
84        channel: russh::ChannelId,
85        _: u32,
86        _: u32,
87        _: &mut client::Session,
88    ) -> Result<()> {
89        ensure!(
90            self.channels.len() < MAX_OPERATIONS,
91            "SSH channel count exceeds 16"
92        );
93        ensure!(
94            self.channels.insert(channel, (0, 0)).is_none(),
95            "Duplicate SSH channel"
96        );
97        Ok(())
98    }
99    async fn channel_close(
100        &mut self,
101        channel: russh::ChannelId,
102        _: &mut client::Session,
103    ) -> Result<()> {
104        self.channels.remove(&channel);
105        Ok(())
106    }
107    async fn data(
108        &mut self,
109        channel: russh::ChannelId,
110        data: &[u8],
111        _: &mut client::Session,
112    ) -> Result<()> {
113        self.receive(channel, data.len())
114    }
115    async fn extended_data(
116        &mut self,
117        channel: russh::ChannelId,
118        _: u32,
119        data: &[u8],
120        _: &mut client::Session,
121    ) -> Result<()> {
122        self.receive(channel, data.len())
123    }
124    async fn window_adjusted(
125        &mut self,
126        channel: russh::ChannelId,
127        _: u32,
128        _: &mut client::Session,
129    ) -> Result<()> {
130        self.receive(channel, 0)
131    }
132    async fn channel_success(
133        &mut self,
134        channel: russh::ChannelId,
135        _: &mut client::Session,
136    ) -> Result<()> {
137        self.receive(channel, 0)
138    }
139    async fn channel_failure(
140        &mut self,
141        channel: russh::ChannelId,
142        _: &mut client::Session,
143    ) -> Result<()> {
144        self.receive(channel, 0)
145    }
146    async fn channel_eof(
147        &mut self,
148        channel: russh::ChannelId,
149        _: &mut client::Session,
150    ) -> Result<()> {
151        self.receive(channel, 0)
152    }
153    async fn exit_status(
154        &mut self,
155        channel: russh::ChannelId,
156        _: u32,
157        _: &mut client::Session,
158    ) -> Result<()> {
159        self.receive(channel, 0)
160    }
161    async fn exit_signal(
162        &mut self,
163        channel: russh::ChannelId,
164        _: russh::Sig,
165        _: bool,
166        message: &str,
167        language: &str,
168        _: &mut client::Session,
169    ) -> Result<()> {
170        self.receive(channel, message.len() + language.len())
171    }
172    async fn xon_xoff(
173        &mut self,
174        channel: russh::ChannelId,
175        _: bool,
176        _: &mut client::Session,
177    ) -> Result<()> {
178        self.receive(channel, 0)
179    }
180    async fn server_channel_open_session(
181        &mut self,
182        _: russh::ChannelId,
183        _: &mut client::Session,
184    ) -> Result<()> {
185        bail!("Unsolicited SSH session channel")
186    }
187    async fn server_channel_open_direct_tcpip(
188        &mut self,
189        _: russh::ChannelId,
190        _: &str,
191        _: u32,
192        _: &str,
193        _: u32,
194        _: &mut client::Session,
195    ) -> Result<()> {
196        bail!("Unsolicited SSH TCP channel")
197    }
198    async fn server_channel_open_agent_forward(
199        &mut self,
200        _: russh::ChannelId,
201        _: &mut client::Session,
202    ) -> Result<()> {
203        bail!("Unsolicited SSH agent channel")
204    }
205    async fn server_channel_open_forwarded_tcpip(
206        &mut self,
207        _: russh::Channel<client::Msg>,
208        _: &str,
209        _: u32,
210        _: &str,
211        _: u32,
212        _: &mut client::Session,
213    ) -> Result<()> {
214        bail!("Unsolicited SSH forwarded channel")
215    }
216    async fn server_channel_open_x11(
217        &mut self,
218        _: russh::Channel<client::Msg>,
219        _: &str,
220        _: u32,
221        _: &mut client::Session,
222    ) -> Result<()> {
223        bail!("Unsolicited SSH X11 channel")
224    }
225}
226fn parameter(
227    p: &crate::protocol::StartupParams,
228    name: &str,
229    default: u64,
230    max: u64,
231) -> Result<u64> {
232    let n = p.get_optional_u64(name)?.unwrap_or(default);
233    ensure!(n > 0 && n <= max, "{name} must be 1..{max}");
234    Ok(n)
235}
236pub struct SshClient;
237impl SshClient {
238    /// Compatibility entry point for callers using the prior command-only API.
239    pub async fn connect_with_llm_actions(
240        remote_addr: String,
241        llm_client: crate::llm::OllamaClient,
242        app_state: Arc<crate::state::AppState>,
243        status_tx: tokio::sync::mpsc::UnboundedSender<String>,
244        client_id: crate::state::ClientId,
245        startup_params: Option<crate::protocol::StartupParams>,
246    ) -> Result<SocketAddr> {
247        Self::connect(ConnectContext {
248            remote_addr,
249            llm_client,
250            state: app_state,
251            status_tx,
252            client_id,
253            startup_params,
254        })
255        .await
256    }
257    pub async fn connect(ctx: ConnectContext) -> Result<SocketAddr> {
258        let p = ctx
259            .startup_params
260            .as_ref()
261            .context("Missing required startup parameters for SSH client")?;
262        let username = p.get_string("username")?;
263        let pin = p
264            .get_optional_string("host_key_sha256")?
265            .map(|pin| -> Result<String> {
266                use base64::Engine;
267                let hash = pin
268                    .strip_prefix("SHA256:")
269                    .context("host_key_sha256 must start with SHA256:")?;
270                ensure!(
271                    base64::engine::general_purpose::STANDARD_NO_PAD
272                        .decode(hash)?
273                        .len()
274                        == 32,
275                    "host_key_sha256 must contain a SHA256 fingerprint"
276                );
277                Ok(hash.to_owned())
278            })
279            .transpose()?;
280        let pinned = pin.is_some();
281        let handshake = Duration::from_secs(parameter(
282            p,
283            "handshake_timeout_secs",
284            HANDSHAKE_TIMEOUT.as_secs(),
285            60,
286        )?);
287        let deadline = Duration::from_secs(parameter(
288            p,
289            "operation_timeout_secs",
290            OPERATION_TIMEOUT.as_secs(),
291            300,
292        )?);
293        let idle = Duration::from_secs(parameter(
294            p,
295            "idle_timeout_secs",
296            IDLE_TIMEOUT.as_secs(),
297            3600,
298        )?);
299        let key_path = p.get_optional_string("private_key_path")?;
300        let method = p.get_optional_string("auth_method")?.unwrap_or_else(|| {
301            if key_path.is_some() {
302                "publickey"
303            } else {
304                "password"
305            }
306            .into()
307        });
308        enum Credential {
309            Password(String),
310            Key(Arc<key::KeyPair>),
311        }
312        let credential = match method.as_str() {
313            "password" => Credential::Password(
314                p.get_optional_string("password")?
315                    .context("auth_method 'password' needs the 'password' startup parameter")?,
316            ),
317            "publickey" => {
318                let path = key_path.context(
319                    "auth_method 'publickey' needs the 'private_key_path' startup parameter",
320                )?;
321                let passphrase = p.get_optional_string("private_key_passphrase")?;
322                Credential::Key(Arc::new(
323                    russh_keys::load_secret_key(&path, passphrase.as_deref())
324                        .with_context(|| format!("Failed to load SSH private key from {path}"))?,
325                ))
326            }
327            _ => bail!("Unknown SSH auth_method '{method}': expected 'password' or 'publickey'"),
328        };
329        let (session, socket, local, peer) = tokio::time::timeout(handshake, async {
330            let stream = tokio::net::TcpStream::connect(&ctx.remote_addr).await.context("Failed to connect to SSH server")?;
331            let local = stream.local_addr()?;
332            let peer = stream.peer_addr()?;
333            let std = stream.into_std()?;
334            let socket = SocketGuard(Arc::new(std.try_clone()?));
335            let stream = tokio::net::TcpStream::from_std(std)?;
336            let config = client::Config { inactivity_timeout: Some(idle), window_size: 256*1024, maximum_packet_size: 32768, ..Default::default() };
337            let mut session = client::connect_stream(Arc::new(config), stream, ClientHandler { expected: pin, channels: HashMap::new() }).await.context("Failed to connect to SSH server")?;
338            let authenticated = match credential {
339                Credential::Password(password) => session.authenticate_password(username.clone(),password).await,
340                Credential::Key(key) => session.authenticate_publickey(username.clone(),key).await,
341            }.context("SSH authentication failed")?;
342            ensure!(authenticated, "SSH authentication failed: the server refused the {method} credential for '{username}'");
343            Ok::<_,anyhow::Error>((session,socket,local,peer))
344        }).await.context("SSH handshake deadline exceeded")??;
345        let now = crate::utils::clock::Instant::now();
346        ctx.state
347            .with_client_mut(ctx.client_id, |client| {
348                client.connection = Some(crate::state::ClientConnectionState {
349                    id: ctx.client_id,
350                    remote_addr: ctx.remote_addr.clone(),
351                    connected_addr: Some(peer),
352                    local_addr: Some(local),
353                    bytes_sent: 0,
354                    bytes_received: 0,
355                    packets_sent: 0,
356                    packets_received: 0,
357                    last_activity: now,
358                    status: ClientStatus::Connected,
359                    status_changed_at: now,
360                    protocol_info: crate::state::server::ProtocolConnectionInfo::new(
361                        json!({"host_key_verified":pinned}),
362                    ),
363                });
364            })
365            .await;
366        ctx.state
367            .update_client_status(ctx.client_id, ClientStatus::Connected)
368            .await;
369        let commands = command_support::register_command_channel(&ctx.state, ctx.client_id).await;
370        let state = ctx.state.clone();
371        let id = ctx.client_id;
372        let event = Event::new(
373            &SSH_CLIENT_CONNECTED_EVENT,
374            json!({"remote_addr":ctx.remote_addr,"username":username,"host_key_verified":pinned}),
375        );
376        state
377            .spawn_client_task(
378                id,
379                Self::run(ctx, session, socket, commands, event, deadline, pinned),
380            )
381            .await;
382        Ok(local)
383    }
384    async fn run(
385        ctx: ConnectContext,
386        session: Handle<ClientHandler>,
387        socket: SocketGuard,
388        mut commands: tokio::sync::mpsc::Receiver<ClientCommand>,
389        event: Event,
390        deadline: Duration,
391        pinned: bool,
392    ) {
393        let session = Arc::new(session);
394        let mut operations: FuturesUnordered<OperationFuture> = FuturesUnordered::new();
395        let mut handlers: FuturesUnordered<HandlerFuture> = FuturesUnordered::new();
396        handlers.push(handle_event(ctx.clone(), event, CONNECT_DEPTH));
397        let mut check_closed = tokio::time::interval(Duration::from_millis(100));
398        loop {
399            tokio::select! {
400                command = commands.recv() => {
401                    let Some(command) = command else { break; };
402                    if enqueue(&ctx,&session,socket.0.clone(),deadline,pinned,&mut operations,handlers.len(),command.action.clone(),Some(command),0).await { break; }
403                }
404                Some((depth,event)) = operations.next(), if !operations.is_empty() => { handlers.push(handle_event(ctx.clone(),event,depth)); }
405                Some((depth,result)) = handlers.next(), if !handlers.is_empty() => {
406                    match result {
407                        Ok(actions) if actions.len() <= MAX_OPERATIONS => {
408                            let mut disconnect = false;
409                            for action in actions {
410                                if depth != CONNECT_DEPTH && depth >= MAX_FOLLOWUP_DEPTH && action["type"] != "disconnect" {
411                                    Log::new(Some(&ctx.status_tx)).warn("SSH follow-up depth bound reached");
412                                    continue;
413                                }
414                                let next_depth = if depth == CONNECT_DEPTH {0} else {depth+1};
415                                if enqueue(&ctx,&session,socket.0.clone(),deadline,pinned,&mut operations,handlers.len(),action,None,next_depth).await { disconnect = true; break; }
416                            }
417                            if disconnect { break; }
418                        }
419                        Ok(_) => Log::new(Some(&ctx.status_tx)).warn("SSH handler returned too many actions"),
420                        Err(e) => Log::new(Some(&ctx.status_tx)).warn(format!("SSH event handler failed: {e}")),
421                    }
422                }
423                _ = check_closed.tick() => { if session.is_closed() { break; } }
424            }
425        }
426        drop(operations);
427        drop(handlers);
428        if let Ok(mut session) = Arc::try_unwrap(session) {
429            let _ = session
430                .disconnect(Disconnect::ByApplication, "", "en")
431                .await;
432            // Let the transport flush the disconnect, then the socket guard forces
433            // closure even if the library's internally spawned driver is stalled.
434            let _ = tokio::time::timeout(Duration::from_secs(1), &mut session).await;
435        }
436        drop(socket);
437        ctx.state.remove_client_handle(ctx.client_id).await;
438        ctx.state
439            .with_client_mut(ctx.client_id, |c| {
440                if let Some(connection) = &mut c.connection {
441                    connection.status = ClientStatus::Disconnected;
442                    connection.status_changed_at = crate::utils::clock::Instant::now();
443                }
444            })
445            .await;
446        ctx.state
447            .update_client_status(ctx.client_id, ClientStatus::Disconnected)
448            .await;
449        let _ = ctx.status_tx.send("__UPDATE_UI__".into());
450    }
451}
452fn handle_event(ctx: ConnectContext, event: Event, depth: u8) -> HandlerFuture {
453    async move {
454        let result = async {
455            let instruction = ctx
456                .state
457                .get_instruction_for_client(ctx.client_id)
458                .await
459                .unwrap_or_default();
460            let memory = ctx
461                .state
462                .get_memory_for_client(ctx.client_id)
463                .await
464                .unwrap_or_default();
465            let result = call_llm_for_client(
466                &ctx.llm_client,
467                &ctx.state,
468                ctx.client_id.to_string(),
469                &instruction,
470                &memory,
471                Some(&event),
472                &SshClientProtocol,
473                &ctx.status_tx,
474            )
475            .await?;
476            if let Some(memory) = result.memory_updates {
477                ctx.state.set_memory_for_client(ctx.client_id, memory).await;
478            }
479            Ok(result.actions)
480        }
481        .await;
482        (depth, result)
483    }
484    .boxed()
485}
486#[allow(clippy::too_many_arguments)]
487async fn enqueue(
488    ctx: &ConnectContext,
489    session: &Arc<Handle<ClientHandler>>,
490    socket: Arc<std::net::TcpStream>,
491    deadline: Duration,
492    pinned: bool,
493    operations: &mut FuturesUnordered<OperationFuture>,
494    handlers: usize,
495    action: Value,
496    command: Option<ClientCommand>,
497    depth: u8,
498) -> bool {
499    let outcome = match SshClientProtocol.execute_action(action.clone()) {
500        Err(e) => Ok(ClientSendOutcome::Rejected {
501            error: e.to_string(),
502        }),
503        Ok(ClientActionResult::Disconnect) => Ok(ClientSendOutcome::Disconnected),
504        Ok(ClientActionResult::WaitForMore | ClientActionResult::NoAction) => {
505            Ok(ClientSendOutcome::Executed {
506                detail: "wait_for_more".into(),
507            })
508        }
509        Ok(ClientActionResult::Custom { name, .. })
510            if name == "execute_command" || name == "sftp_operation" =>
511        {
512            if name == "sftp_operation" && !pinned {
513                Ok(ClientSendOutcome::Rejected {
514                    error: "SFTP requires host_key_sha256 at startup".into(),
515                })
516            } else if operations.len() + handlers >= MAX_OPERATIONS {
517                Err(anyhow::anyhow!(
518                    "SSH client busy: 16 operations or handlers"
519                ))
520            } else {
521                operations.push(operation_future(
522                    ctx.clone(),
523                    session.clone(),
524                    socket,
525                    deadline,
526                    action,
527                    command,
528                    depth,
529                ));
530                return false;
531            }
532        }
533        Ok(_) => Ok(ClientSendOutcome::Rejected {
534            error: "Unsupported SSH action".into(),
535        }),
536    };
537    let disconnect = matches!(outcome, Ok(ClientSendOutcome::Disconnected));
538    record_action(ctx, &action, &outcome).await;
539    if let Some(command) = command {
540        command_support::reply(command, outcome);
541    }
542    disconnect
543}
544async fn record_action(ctx: &ConnectContext, action: &Value, outcome: &Result<ClientSendOutcome>) {
545    let result = match outcome {
546        Ok(v) => serde_json::to_value(v).unwrap_or(Value::Null),
547        Err(e) => json!({"error":e.to_string()}),
548    };
549    ctx.state
550        .record_access_log(
551            AccessLogOwner::Client(ctx.client_id.as_u32()),
552            "SSH",
553            None,
554            "injected_action",
555            action.clone(),
556            vec![result],
557        )
558        .await;
559}
560fn operation_future(
561    ctx: ConnectContext,
562    session: Arc<Handle<ClientHandler>>,
563    socket: Arc<std::net::TcpStream>,
564    deadline: Duration,
565    action: Value,
566    command: Option<ClientCommand>,
567    depth: u8,
568) -> OperationFuture {
569    async move {
570        let (outcome,event) = match execute(&session,&socket,&action,deadline).await {
571            Ok((detail,event)) => (Ok(ClientSendOutcome::Executed {detail}),event),
572            Err(error) => {
573                Log::new(Some(&ctx.status_tx)).warn(format!("SSH operation failed: {error:#}"));
574                let event = Event::new(&SSH_OPERATION_FAILED_EVENT,json!({"action_type":action["type"],"path":action["path"],"command":action["command"],"error":format!("{error:#}")}));
575                (Err(error),event)
576            }
577        };
578        ctx.state
579            .with_client_mut(ctx.client_id, |client| {
580                if let Some(connection) = &mut client.connection {
581                    connection.last_activity = crate::utils::clock::Instant::now();
582                }
583            })
584            .await;
585        record_action(&ctx,&action,&outcome).await;
586        if let Some(command) = command { command_support::reply(command,outcome); }
587        (depth,event)
588    }.boxed()
589}
590async fn execute(
591    session: &Handle<ClientHandler>,
592    socket: &std::net::TcpStream,
593    action: &Value,
594    deadline: Duration,
595) -> Result<(String, Event)> {
596    let end = tokio::time::Instant::now() + deadline;
597    let mut channel = tokio::time::timeout_at(end, session.channel_open_session())
598        .await
599        .context("SSH channel-open deadline exceeded")??;
600    let result = tokio::time::timeout_at(end, async {
601        if action["type"] == "execute_command" {
602            let command = action["command"].as_str().context("Missing command")?;
603            channel.exec(true, command).await?;
604            let mut output = Vec::new();
605            let mut stderr = Vec::new();
606            let mut exit_code = None;
607            let mut eof = false;
608            loop {
609                match channel.wait().await {
610                    Some(ChannelMsg::Data { data }) => {
611                        ensure!(
612                            data.len()
613                                <= MAX_COMMAND_OUTPUT.saturating_sub(output.len() + stderr.len()),
614                            "SSH command output exceeds 1 MiB"
615                        );
616                        output.extend_from_slice(&data);
617                    }
618                    Some(ChannelMsg::ExtendedData { data, ext: 1 }) => {
619                        ensure!(
620                            data.len()
621                                <= MAX_COMMAND_OUTPUT.saturating_sub(output.len() + stderr.len()),
622                            "SSH command output exceeds 1 MiB"
623                        );
624                        stderr.extend_from_slice(&data);
625                    }
626                    Some(ChannelMsg::ExitStatus { exit_status }) => {
627                        exit_code = Some(exit_status);
628                        if eof {
629                            break;
630                        }
631                    }
632                    Some(ChannelMsg::Eof) => {
633                        eof = true;
634                        if exit_code.is_some() {
635                            break;
636                        }
637                    }
638                    Some(ChannelMsg::Close) => break,
639                    None => {
640                        ensure!(!session.is_closed(), "SSH connection closed during command");
641                        break;
642                    }
643                    Some(ChannelMsg::Failure) => bail!("SSH server refused command"),
644                    _ => {}
645                }
646            }
647            let detail = format!(
648                "execute_command {command:?}: exit_code={}, {} bytes of output, {} of stderr",
649                exit_code
650                    .map(|n| n.to_string())
651                    .unwrap_or_else(|| "unknown".into()),
652                output.len(),
653                stderr.len()
654            );
655            let mut data = json!({"command":command,"output":String::from_utf8_lossy(&output)});
656            if !stderr.is_empty() {
657                data["stderr"] = json!(String::from_utf8_lossy(&stderr));
658            }
659            if let Some(code) = exit_code {
660                data["exit_code"] = json!(code);
661            }
662            Ok((detail, Event::new(&SSH_CLIENT_OUTPUT_RECEIVED_EVENT, data)))
663        } else {
664            channel.request_subsystem(true, "sftp").await?;
665            loop {
666                match channel.wait().await {
667                    Some(ChannelMsg::Success) => break,
668                    Some(ChannelMsg::WindowAdjusted { .. }) => {}
669                    _ => bail!("SSH server refused SFTP subsystem"),
670                }
671            }
672            let writer = channel.make_writer();
673            let reader = channel.make_reader();
674            let result = sftp::exchange(tokio::io::join(reader, writer), action).await?;
675            let detail = format!(
676                "{} {} completed",
677                action["type"].as_str().unwrap_or("sftp"),
678                action["path"].as_str().unwrap_or("")
679            );
680            Ok((detail, Event::new(&SSH_SFTP_RESULT_EVENT, result)))
681        }
682    })
683    .await
684    .map_err(|error| anyhow::anyhow!(error).context("SSH operation deadline exceeded"))
685    .and_then(|result| result);
686    // Closing a timed-out or malformed channel does not end unrelated operations.
687    // A blocked transport cannot retain an orphan channel past client removal.
688    if tokio::time::timeout(Duration::from_millis(250), async {
689        channel.eof().await?;
690        channel.close().await
691    })
692    .await
693    .is_err()
694    {
695        let _ = socket.shutdown(Shutdown::Both);
696    }
697    result
698}
netget/client/ssh/mod.rs

netget/client/ssh/
mod.rs
