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

oxdock_ssh_plugin/
funcs.rs
