oxdock_ssh_plugin/
funcs.rs

1//! The `SSH` host module: `SERVE`, `ACCEPT` / `DEQUEUE` + `PUMP_CHANNEL`,
2//! `CLOSE`, `CONNECT`, `PUMP`.
3//!
4//! Pipe direction convention (fixed for every func): `out_pipe` carries
5//! bytes produced by the wire side (the DSL reads them), `in_pipe`
6//! carries bytes consumed by the wire side (the DSL writes them).
7
8use std::collections::BTreeMap;
9use std::sync::Arc;
10use std::sync::atomic::AtomicBool;
11use std::time::Duration;
12
13use anyhow::{Context, Result, bail};
14use oxdock_core::{
15    FuncKind, FuncMeta, FuncParam, HostModule, HostRegistration, NativeFn, OxDockFn, OxDockType,
16    StepCtx, Value,
17};
18use oxdock_func_macro::oxdock_func;
19use oxdock_net_plugin::{AcquiredListener, EndpointRegistry, acquire_listener};
20use oxdock_process::ProcessManager;
21use russh::keys::{Algorithm, PrivateKey};
22
23use crate::bridge::{pump_pipe_to_pipe, pump_session};
24use crate::keys::load_or_create_host_key;
25use crate::runtime::{connect_runtime, connect_session};
26use crate::state::{
27    CLOSE_JOIN_TIMEOUT, Dequeue, PendingSession, ServerState, SessionQueue, ShutdownSignal,
28};
29use crate::types::{SshServerTag, SshSessionTag};
30use crate::validate::parse_serve_endpoint;
31
32/// Unique server ids per process.
33static SERVER_IDS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
34
35/// Dequeue tick: shutdown and cancellation surface within a few ticks.
36const DEQUEUE_TICK: Duration = Duration::from_millis(10);
37
38/// Read the `SSH_SERVER` payload out of a DSL value.
39fn server_state(value: &Value, func: &str) -> Result<Arc<ServerState>> {
40    let Some(tag) = value.read_heap::<SshServerTag>(SshServerTag::descriptor()) else {
41        bail!(
42            "{func} expects an SSH_SERVER value, got {}",
43            value.type_name()
44        );
45    };
46    Ok(Arc::clone(tag.state()))
47}
48
49/// Read the `SSH_SESSION` payload out of a DSL value.
50fn session_tag(value: &Value, func: &str) -> Result<SshSessionTag> {
51    let Some(tag) = value.read_heap::<SshSessionTag>(SshSessionTag::descriptor()) else {
52        bail!(
53            "{func} expects an SSH_SESSION value, got {}",
54            value.type_name()
55        );
56    };
57    Ok(tag.clone())
58}
59
60/// Block until the queue yields an authenticated session (or teardown).
61/// Shared by `SSH_DEQUEUE` and the `SSH_ACCEPT` wrapper.
62fn dequeue_session<P: ProcessManager>(
63    cx: &StepCtx<P>,
64    queue: &Arc<SessionQueue>,
65    func: &str,
66) -> Result<PendingSession> {
67    loop {
68        match queue.try_pop() {
69            Dequeue::Session(session) => return Ok(session),
70            Dequeue::Shutdown => bail!("{func}: server is closed"),
71            Dequeue::Empty => {}
72        }
73        match queue.wait_for_session(DEQUEUE_TICK) {
74            Dequeue::Session(session) => return Ok(session),
75            Dequeue::Shutdown => bail!("{func}: server is closed"),
76            Dequeue::Empty => {}
77        }
78        if cx.is_cancelled() {
79            bail!("{func}: task cancelled");
80        }
81    }
82}
83
84/// Dequeue one authenticated session and expose its metadata before any
85/// byte pumping starts, so scripts can route on the requested command or
86/// client identity. Must run inside `ASYNC`. Returns a MAP with
87/// `session` (SSH_SESSION), `command` (STRING, empty for shells),
88/// `username` and `addr` (STRINGs, empty when unknown).
89///
90/// Routing shape: compare the dequeued command against known commands,
91/// build a fresh pipe pair per session, and pump a synthetic reply with
92/// `SSH_PUMP_CHANNEL`. The server sends first: the client side never
93/// EOFs its input, so the reply cannot race teardown. This complete
94/// program runs end to end under the docs conformance suite.
95///
96/// ```oxdock
97/// IMPORT [STD, SSH]
98/// LET $m: MAP = SSH_SERVE("doc-ssh-demo", {username: "u", password: "p"})
99/// LET $in: PIPE
100/// LET $out: PIPE
101/// LET $w: HANDLE = ASYNC {
102///     LET $sess: MAP = SSH_DEQUEUE($m.server)
103///     ASSERT_CONTAINS $sess "session"
104///     ASSERT_CONTAINS $sess "command"
105///     ASSERT_CONTAINS $sess "username"
106///     ASSERT_CONTAINS $sess "addr"
107///     SSH_PUMP_CHANNEL($sess.session, $in, $out)
108/// }
109/// LET $cin: PIPE
110/// LET $cout: PIPE
111/// LET $c: HANDLE = ASYNC { SSH_CONNECT("doc-ssh-demo", "u", "p", $cin, $cout) }
112/// WITH_IO [stdout=$in] ECHO "server-greeting"
113/// LET $info: MAP = INSPECT($cout)
114/// LET $empty: BOOL = $info.buffer_bytes == 0
115/// WHILE $empty {
116///     SLEEP 100ms
117///     $info = INSPECT($cout)
118///     $empty = $info.buffer_bytes == 0
119/// }
120/// ASSERT_CONTAINS $cout "server-greeting"
121/// AWAIT $w
122/// CANCEL $c
123/// SSH_CLOSE($m.server)
124/// ```
125#[oxdock_func(
126    returns = "MAP",
127    summary = "Dequeue one SSH session with its metadata."
128)]
129fn ssh_dequeue<P: ProcessManager>(cx: &mut StepCtx<P>, server: Value) -> Result<Value> {
130    if !cx.is_async_task() {
131        bail!(
132            "SSH_DEQUEUE requires ASYNC: wrap it as LET $t: HANDLE = ASYNC {{ SSH_DEQUEUE($server.server) }}"
133        );
134    }
135    let state = server_state(&server, "SSH_DEQUEUE")?;
136    let session = dequeue_session(cx, state.queue(), "SSH_DEQUEUE")?;
137    let command = session.exec_command.clone().unwrap_or_default();
138    let username = session.username.clone().unwrap_or_default();
139    let addr = session
140        .peer_addr
141        .map(|addr| addr.to_string())
142        .unwrap_or_default();
143    let tag = SshSessionTag::new(
144        session.exec_command,
145        session.username,
146        session.peer_addr,
147        session.pty_size,
148        session.up_rx,
149        session.down_tx,
150    );
151    let mut map = BTreeMap::new();
152    map.insert(
153        "session".to_string(),
154        Value::mint_heap(SshSessionTag::descriptor(), tag),
155    );
156    map.insert("command".to_string(), Value::string(command));
157    map.insert("username".to_string(), Value::string(username));
158    map.insert("addr".to_string(), Value::string(addr));
159    Ok(Value::map(map))
160}
161
162/// Read a flat string list (an argv vector) out of a DSL value.
163fn argv_list(value: &Value, func: &str) -> Result<Vec<String>> {
164    let Some(items) = value.as_list() else {
165        bail!("{func} argv must be a LIST of strings");
166    };
167    if items.is_empty() {
168        bail!("{func} argv must not be empty");
169    }
170    items
171        .iter()
172        .map(|item| {
173            item.as_str().map(str::to_string).ok_or_else(|| {
174                anyhow::anyhow!("{func} argv must be strings, got {}", item.type_name())
175            })
176        })
177        .collect()
178}
179
180/// Read the options MAP for `SSH_SERVE`. The 2nd argument must be a MAP;
181/// unknown keys bail so script typos fail fast instead of silently ignored.
182fn serve_options(options: &Value) -> Result<&BTreeMap<String, Value>> {
183    options.as_map().ok_or_else(|| {
184        anyhow::anyhow!(
185            "SSH_SERVE options must be a MAP, got {}",
186            options.type_name()
187        )
188    })
189}
190
191/// Read a required STRING key from the options MAP. Missing, non-string,
192/// or blank binds bail naming the key: a server without credentials is
193/// meaningless, so there is no default.
194fn required_string(map: &BTreeMap<String, Value>, func: &str, key: &str) -> Result<String> {
195    let Some(value) = map.get(key) else {
196        bail!("{func} option '{key}' is required");
197    };
198    let Some(s) = value.as_str() else {
199        bail!(
200            "{func} option '{key}' must be a STRING, got {}",
201            value.type_name()
202        );
203    };
204    if s.trim().is_empty() {
205        bail!("{func} option '{key}' must not be empty");
206    }
207    Ok(s.to_string())
208}
209
210/// Read an optional STRING key from the options MAP. Missing, empty, or
211/// whitespace-only binds `None`; present non-strings bail.
212fn optional_string(map: &BTreeMap<String, Value>, func: &str, key: &str) -> Result<Option<String>> {
213    let Some(value) = map.get(key) else {
214        return Ok(None);
215    };
216    let Some(s) = value.as_str() else {
217        bail!(
218            "{func} option '{key}' must be a STRING, got {}",
219            value.type_name()
220        );
221    };
222    let trimmed = s.trim();
223    if trimmed.is_empty() {
224        return Ok(None);
225    }
226    Ok(Some(s.to_string()))
227}
228
229/// Spin up an SSH server on a virtual service endpoint with the given
230/// credentials. `bind` is a logical port (`"2251"`) or service name:
231/// physical binds in-script are rejected, `0` is reserved for the CLI
232/// outer mapping. `options` is a MAP with required `username`/`password`
233/// STRINGs (a server without credentials is meaningless, so blanks bail)
234/// and the optional `key_path` STRING (workspace-relative OpenSSH Ed25519
235/// file, load-or-create; a leading `/` anchors to the workspace root like
236/// WRITE, and escapes still bail; absent or blank keeps the ephemeral
237/// in-memory key).
238/// Non-blocking: returns a MAP with `server` (SSH_SERVER),
239/// `addr` (STRING: the physical bind, or the virtual endpoint echo when
240/// socketless), `username` and `password` (STRINGs), and `virtual`
241/// (STRING echo). Memory services bail: SSH needs a TCP socket, so map
242/// the name with `-p`/`--listen`.
243fn ssh_serve<P: ProcessManager>(
244    cx: &mut StepCtx<P>,
245    registry: &Arc<EndpointRegistry>,
246    bind: String,
247    options: Value,
248) -> Result<Value> {
249    let map = serve_options(&options)?;
250    for key in map.keys() {
251        if key != "username" && key != "password" && key != "key_path" {
252            bail!("SSH_SERVE() unknown option '{key}' (expected: username, password, key_path)");
253        }
254    }
255    let username = required_string(map, "SSH_SERVE", "username")?;
256    let password = required_string(map, "SSH_SERVE", "password")?;
257    let key_path = optional_string(map, "SSH_SERVE", "key_path")?;
258    let endpoint = parse_serve_endpoint(&bind)?;
259    let (acquired, registry) = acquire_listener(registry, &endpoint, "SSH_SERVE")?;
260    let host_key = match load_or_create_host_key(cx, "SSH_SERVE", key_path)? {
261        Some(key) => key,
262        None => PrivateKey::random(&mut rand::rng(), Algorithm::Ed25519)
263            .context("generate ephemeral Ed25519 host key")?,
264    };
265    let id = SERVER_IDS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
266    let id = format!("ssh-{pid}-{id}", pid = std::process::id());
267    let queue = Arc::new(SessionQueue::new());
268    let (shutdown_tx, shutdown_rx) = std::sync::mpsc::channel::<ShutdownSignal>();
269    let (local_addr, addr_text, thread) = match acquired {
270        AcquiredListener::Tcp { listener, addr } => {
271            // The slot keeps the shared backlog socket; the runtime owns
272            // its own clone.
273            let owned = listener
274                .try_clone()
275                .context("SSH_SERVE cannot clone its listener")?;
276            let thread_queue = Arc::clone(&queue);
277            let thread_user = username.clone();
278            let thread_pass = password.clone();
279            let thread = std::thread::Builder::new()
280                .name(id.clone())
281                .spawn(move || {
282                    crate::runtime::serve(
283                        owned,
284                        host_key,
285                        thread_user,
286                        thread_pass,
287                        thread_queue,
288                        shutdown_rx,
289                    )
290                })
291                .context("SSH_SERVE cannot spawn the server thread")?;
292            (addr, addr.to_string(), Some(thread))
293        }
294        AcquiredListener::Memory => {
295            drop(shutdown_rx);
296            bail!(
297                "SSH_SERVE: '{endpoint}' is a memory service (SSH needs a TCP socket; map it with -p/--listen)"
298            )
299        }
300        AcquiredListener::Offline => {
301            // Socketless servers spawn no thread: DEQUEUE waits on the
302            // queue until close (test drivers push sessions via the
303            // queue). Dropping the receiver makes later shutdown sends a
304            // silent no-op.
305            drop(shutdown_rx);
306            (
307                std::net::SocketAddr::from(([0, 0, 0, 0], 0)),
308                endpoint.to_string(),
309                None,
310            )
311        }
312    };
313    let state = Arc::new(ServerState::new(crate::state::ServerConfig {
314        id,
315        local_addr,
316        addr_text: addr_text.clone(),
317        queue,
318        shutdown_tx,
319        thread,
320        registry: Arc::clone(&registry),
321        endpoint: endpoint.clone(),
322    }));
323    let mut map = BTreeMap::new();
324    map.insert(
325        "server".to_string(),
326        Value::mint_heap(SshServerTag::descriptor(), SshServerTag::new(state)),
327    );
328    map.insert("addr".to_string(), Value::string(addr_text));
329    map.insert("username".to_string(), Value::string(username));
330    map.insert("password".to_string(), Value::string(password));
331    map.insert("virtual".to_string(), Value::string(endpoint.to_string()));
332    Ok(Value::map(map))
333}
334
335/// Accept the next authenticated session and pump it through explicit
336/// pipes until the channel closes. Must run inside `ASYNC`. Returns a
337/// MAP with `closed` (BOOL) and `command` (STRING, empty for shells).
338/// Thin wrapper over `SSH_DEQUEUE` + `SSH_PUMP_CHANNEL` for worker loops
339/// that need no pre-pump inspection; use those directly to route on
340/// session metadata first. Returns a MAP with `closed` (BOOL) and
341/// `command` (STRING, empty for shells); the example below asserts both
342/// keys on the awaited result. The server sends first: the client side
343/// never EOFs its input, so the reply cannot race teardown. This
344/// complete program runs end to end under the docs conformance suite.
345///
346/// ```oxdock
347/// IMPORT [STD, SSH]
348/// LET $m: MAP = SSH_SERVE("doc-ssh-demo", {username: "u", password: "p"})
349/// LET $in: PIPE
350/// LET $out: PIPE
351/// LET $acc: HANDLE = ASYNC { SSH_ACCEPT($m.server, $in, $out) }
352/// LET $cin: PIPE
353/// LET $cout: PIPE
354/// LET $c: HANDLE = ASYNC { SSH_CONNECT("doc-ssh-demo", "u", "p", $cin, $cout) }
355/// WITH_IO [stdout=$in] ECHO "server-greeting"
356/// LET $info: MAP = INSPECT($cout)
357/// LET $empty: BOOL = $info.buffer_bytes == 0
358/// WHILE $empty {
359///     SLEEP 100ms
360///     $info = INSPECT($cout)
361///     $empty = $info.buffer_bytes == 0
362/// }
363/// ASSERT_CONTAINS $cout "server-greeting"
364/// LET $done: MAP = AWAIT $acc
365/// ASSERT_CONTAINS $done "closed"
366/// ASSERT_CONTAINS $done "command"
367/// CANCEL $c
368/// SSH_CLOSE($m.server)
369/// ```
370#[oxdock_func(returns = "MAP", summary = "Accept one SSH session into pipes.")]
371fn ssh_accept<P: ProcessManager>(
372    cx: &mut StepCtx<P>,
373    server: Value,
374    in_pipe: Value,
375    out_pipe: Value,
376) -> Result<Value> {
377    if !cx.is_async_task() {
378        bail!(
379            "SSH_ACCEPT requires ASYNC: wrap it as LET $t: HANDLE = ASYNC {{ SSH_ACCEPT($server, $in, $out) }}"
380        );
381    }
382    let state = server_state(&server, "SSH_ACCEPT")?;
383    let cancel = AtomicBool::new(false);
384    let session = dequeue_session(cx, state.queue(), "SSH_ACCEPT")?;
385    pump_session(
386        cx,
387        &in_pipe,
388        &out_pipe,
389        session.up_rx,
390        session.down_tx,
391        &cancel,
392    )?;
393    let mut map = BTreeMap::new();
394    map.insert("closed".to_string(), Value::bool(true));
395    map.insert(
396        "command".to_string(),
397        Value::string(session.exec_command.unwrap_or_default()),
398    );
399    Ok(Value::map(map))
400}
401
402/// Pump a dequeued session between explicit DSL pipes until the channel
403/// closes. Must run inside `ASYNC`. The session ends are take-once: a
404/// second pump on the same session bails instead of splitting bytes.
405/// Returns a MAP with `closed` (BOOL).
406#[oxdock_func(
407    returns = "MAP",
408    summary = "Pump a dequeued SSH session through pipes."
409)]
410fn ssh_pump_channel<P: ProcessManager>(
411    cx: &mut StepCtx<P>,
412    session: Value,
413    in_pipe: Value,
414    out_pipe: Value,
415) -> Result<Value> {
416    if !cx.is_async_task() {
417        bail!("SSH_PUMP_CHANNEL requires ASYNC: pump it in its own task after SSH_DEQUEUE");
418    }
419    let tag = session_tag(&session, "SSH_PUMP_CHANNEL")?;
420    let (up_rx, down_tx) = tag.take_pump_ends()?;
421    let cancel = AtomicBool::new(false);
422    pump_session(cx, &in_pipe, &out_pipe, up_rx, down_tx, &cancel)?;
423    let mut map = BTreeMap::new();
424    map.insert("closed".to_string(), Value::bool(true));
425    Ok(Value::map(map))
426}
427
428/// Shut a server down and join its runtime thread (bounded). Idempotent:
429/// returns BOOL true when no thread remains.
430#[oxdock_func(returns = "BOOL", summary = "Shut down an SSH server.")]
431fn ssh_close<P: ProcessManager>(cx: &mut StepCtx<P>, server: Value) -> Result<Value> {
432    let _ = cx;
433    let state = server_state(&server, "SSH_CLOSE")?;
434    state.request_shutdown();
435    Ok(Value::bool(state.join_thread(CLOSE_JOIN_TIMEOUT)))
436}
437
438/// Connect to an SSH server with distinct inner credentials and pump the
439/// shell channel through explicit pipes until it closes. Must run inside
440/// `ASYNC`, concurrently with the `SSH_PUMP` tasks (never before them).
441/// `target` is a logical port (`"2251"`: CLI-mapped address or loopback
442/// default), a service name (CLI-mapped address only; unmapped names are
443/// memory services and SSH needs TCP), a served address (`$m.addr`), or
444/// a `host:port` dial. Under `--offline` the dial bails before any DNS
445/// or socket work. Returns a MAP with `closed` (BOOL).
446fn ssh_connect<P: ProcessManager>(
447    cx: &mut StepCtx<P>,
448    registry: &Arc<EndpointRegistry>,
449    target: String,
450    username: String,
451    password: String,
452    in_pipe: Value,
453    out_pipe: Value,
454) -> Result<Value> {
455    if !cx.is_async_task() {
456        bail!("SSH_CONNECT requires ASYNC: run it in its own task beside the SSH_PUMP tasks");
457    }
458    // Sandbox gate first: offline runs open no OS sockets.
459    if registry.is_offline() {
460        bail!("SSH_CONNECT failed: engine running in --offline mode");
461    }
462    if username.is_empty() {
463        bail!("SSH_CONNECT username must not be empty");
464    }
465    let addr = crate::validate::resolve_connect_addr(registry, &target)?;
466    let runtime = connect_runtime()?;
467    let session = runtime
468        .block_on(connect_session(&addr, &username, &password))
469        .context("SSH_CONNECT failed")?;
470    let crate::runtime::OutboundSession {
471        up_rx,
472        down_tx,
473        handle,
474        ..
475    } = session;
476    let cancel = AtomicBool::new(false);
477    let pump = pump_session(cx, &in_pipe, &out_pipe, up_rx, down_tx, &cancel);
478    let _ = runtime.block_on(handle.disconnect(russh::Disconnect::ByApplication, "", ""));
479    pump?;
480    let mut map = BTreeMap::new();
481    map.insert("closed".to_string(), Value::bool(true));
482    Ok(Value::map(map))
483}
484
485/// Copy one pipe into another until EOF, then close the target.
486/// Returns the INT byte count. Either task placement works, as long as
487/// the other end is live (usually an `ASYNC` task).
488#[oxdock_func(returns = "INT", summary = "Copy one pipe into another until EOF.")]
489fn ssh_pump<P: ProcessManager>(
490    cx: &mut StepCtx<P>,
491    from_pipe: Value,
492    to_pipe: Value,
493) -> Result<Value> {
494    let cancel = AtomicBool::new(false);
495    let total = pump_pipe_to_pipe(cx, &from_pipe, &to_pipe, &cancel)?;
496    Ok(Value::int(total))
497}
498
499/// Run `argv` under a local pseudo-terminal sized from the dequeued
500/// session and pump it through explicit pipes until the child exits.
501/// `rows`/`cols` seed the initial size when positive; non-positive falls
502/// back to the session's requested size (the outer pty request, 24x80
503/// default). Outer window-change requests resize this session's terminal
504/// live; every session owns its size cell, so concurrent guests never
505/// observe each other. Must run inside `ASYNC`. Returns the INT exit
506/// code. Environment is inherited from the host process and layered with
507/// the script environment like `RUN`: block-scoped `ENV` (such as the
508/// session's `SSH_USER` / `SSH_CLIENT` / `SSH_SERVER` / `SSH_COMMAND`
509/// relay) reaches the child; the working directory comes from the script.
510#[oxdock_func(
511    returns = "INT",
512    summary = "Run a command under a sized local terminal into pipes."
513)]
514fn ssh_pty_run<P: ProcessManager>(
515    cx: &mut StepCtx<P>,
516    session: Value,
517    argv: Value,
518    rows: i64,
519    cols: i64,
520    in_pipe: Value,
521    out_pipe: Value,
522) -> Result<Value> {
523    if !cx.is_async_task() {
524        bail!("SSH_PTY_RUN requires ASYNC: run it in its own task beside the session pump task");
525    }
526    let tag = session_tag(&session, "SSH_PTY_RUN")?;
527    let argv = argv_list(&argv, "SSH_PTY_RUN")?;
528    let initial = if rows > 0 && cols > 0 {
529        crate::state::PtySize::new(rows as u32, cols as u32)
530    } else {
531        tag.pty_size()
532    };
533    let cancel = AtomicBool::new(false);
534    let code = crate::pty::pump_pty_session(
535        cx,
536        &argv,
537        initial,
538        &tag.pty_size_handle(),
539        &in_pipe,
540        &out_pipe,
541        &cancel,
542    )?;
543    Ok(Value::int(code))
544}
545
546/// The `SSH` host module: virtual-endpoint server plus client, bridged
547/// to DSL pipes. Generic over the process manager like every host module.
548pub fn module_with<P: ProcessManager>() -> HostModule<P> {
549    module_with_endpoints(Arc::new(EndpointRegistry::new(false)))
550}
551
552/// The `SSH` host module resolving through `registry`: `SSH_SERVE` and
553/// `SSH_CONNECT` close over it (hand-built entries; the `#[oxdock_func]`
554/// macro only generates closers-over-nothing). Everything else reaches
555/// the same registry through its `SSH_SERVER` handle.
556pub fn module_with_endpoints<P: ProcessManager>(registry: Arc<EndpointRegistry>) -> HostModule<P> {
557    HostModule {
558        name: "SSH".to_string(),
559        funcs: vec![
560            ssh_serve_registration(Arc::clone(&registry)),
561            SshAccept::registration(),
562            SshDequeue::registration(),
563            SshPumpChannel::registration(),
564            SshClose::registration(),
565            ssh_connect_registration(registry),
566            SshPump::registration(),
567            SshPtyRun::registration(),
568        ],
569        types: vec![SshServerTag::descriptor(), SshSessionTag::descriptor()],
570    }
571}
572
573/// Hand-built `SSH_SERVE` entry: same shape the macro would emit (arity
574/// check, `STRING`/`Value` unpacking, metadata), plus the captured
575/// registry threaded into [`ssh_serve`].
576fn ssh_serve_registration<P: ProcessManager>(
577    registry: Arc<EndpointRegistry>,
578) -> HostRegistration<P> {
579    let func: NativeFn<P> = Arc::new(move |cx, values| {
580        if values.len() != 2 {
581            bail!("SSH_SERVE() expects 2 argument(s), got {}", values.len());
582        }
583        let mut values = values.into_iter();
584        let bind = match values.next().expect("arity checked above").as_str() {
585            Some(s) => s.to_string(),
586            None => bail!("SSH_SERVE() argument `$bind` must be a STRING"),
587        };
588        let options = values.next().expect("arity checked above");
589        ssh_serve(cx, &registry, bind, options)
590    });
591    HostRegistration::Stateful {
592        name: "SSH_SERVE".to_string(),
593        meta: FuncMeta {
594            name: "SSH_SERVE".to_string(),
595            // Assigned at registration, like the macro's markers.
596            module: String::new(),
597            kind: FuncKind::HostCtx,
598            params: Some(vec![
599                FuncParam {
600                    name: "bind".to_string(),
601                    param_type: Some("STRING".to_string()),
602                },
603                FuncParam {
604                    name: "options".to_string(),
605                    param_type: None,
606                },
607            ]),
608            returns: Some("MAP".to_string()),
609            rpn: false,
610            summary: "Serve SSH on a virtual service endpoint.",
611            docs: "Serve SSH on a virtual service endpoint.",
612        },
613        func,
614    }
615}
616
617/// Hand-built `SSH_CONNECT` entry: same shape the macro would emit, plus
618/// the captured registry threaded into [`ssh_connect`].
619fn ssh_connect_registration<P: ProcessManager>(
620    registry: Arc<EndpointRegistry>,
621) -> HostRegistration<P> {
622    let func: NativeFn<P> = Arc::new(move |cx, values| {
623        if values.len() != 5 {
624            bail!("SSH_CONNECT() expects 5 argument(s), got {}", values.len());
625        }
626        let mut values = values.into_iter();
627        let target = match values.next().expect("arity checked above").as_str() {
628            Some(s) => s.to_string(),
629            None => bail!("SSH_CONNECT() argument `$target` must be a STRING"),
630        };
631        let username = match values.next().expect("arity checked above").as_str() {
632            Some(s) => s.to_string(),
633            None => bail!("SSH_CONNECT() argument `$username` must be a STRING"),
634        };
635        let password = match values.next().expect("arity checked above").as_str() {
636            Some(s) => s.to_string(),
637            None => bail!("SSH_CONNECT() argument `$password` must be a STRING"),
638        };
639        let in_pipe = values.next().expect("arity checked above");
640        let out_pipe = values.next().expect("arity checked above");
641        ssh_connect(cx, &registry, target, username, password, in_pipe, out_pipe)
642    });
643    HostRegistration::Stateful {
644        name: "SSH_CONNECT".to_string(),
645        meta: FuncMeta {
646            name: "SSH_CONNECT".to_string(),
647            // Assigned at registration, like the macro's markers.
648            module: String::new(),
649            kind: FuncKind::HostCtx,
650            params: Some(vec![
651                FuncParam {
652                    name: "target".to_string(),
653                    param_type: Some("STRING".to_string()),
654                },
655                FuncParam {
656                    name: "username".to_string(),
657                    param_type: Some("STRING".to_string()),
658                },
659                FuncParam {
660                    name: "password".to_string(),
661                    param_type: Some("STRING".to_string()),
662                },
663                FuncParam {
664                    name: "in_pipe".to_string(),
665                    param_type: None,
666                },
667                FuncParam {
668                    name: "out_pipe".to_string(),
669                    param_type: None,
670                },
671            ]),
672            returns: Some("MAP".to_string()),
673            rpn: false,
674            summary: "Open an SSH client session into pipes.",
675            docs: "Open an SSH client session into pipes. Target shapes: a logical port (CLI-mapped address or loopback default), a service name (CLI-mapped address only), a served address, or a host:port dial.",
676        },
677        func,
678    }
679}
oxdock_ssh_plugin/funcs.rs

oxdock_ssh_plugin/
funcs.rs
