From 522343628273ebc9e48c91bf2f2e3727fe369b07 Mon Sep 17 00:00:00 2001 From: John Vandenberg Date: Mon, 20 Jul 2026 11:01:54 +0800 Subject: [PATCH 1/2] Add R lang --- .dockerignore | 2 + .github/workflows/check.yaml | 2 +- .github/workflows/codeql.yaml | 2 +- .github/workflows/coverage.yaml | 4 +- .github/workflows/dependencies.yaml | 2 +- .github/workflows/docker-linux.yaml | 2 +- .github/workflows/docker-windows.yaml | 2 +- .github/workflows/test.yaml | 4 +- .github/workflows/upstream-cache.yaml | 2 +- .gitignore | 2 + .mise/config.r.toml | 49 +++ .mise/config.toml | 8 +- CLAUDE.md | 2 + Cargo.lock | 14 + Cargo.toml | 2 + Dockerfile | 2 +- Dockerfile.windows | 2 +- README.md | 2 + config/clippy.toml | 1 + config/conftest/policy/gha/gha.rego | 2 +- libs/edge-toolkit/src/config.rs | 1 + services/websockify/Cargo.toml | 26 ++ services/websockify/src/lib.rs | 365 ++++++++++++++++++ services/websockify/tests/relay.rs | 315 +++++++++++++++ .../ws-modules/rcomm1/pkg/et_ws_rcomm1.js | 66 ++++ services/ws-modules/rcomm1/pkg/module.R | 109 ++++++ services/ws-modules/rcomm1/pkg/package.json | 8 + .../ws-modules/rdata1/pkg/et_ws_rdata1.js | 50 +++ services/ws-modules/rdata1/pkg/module.R | 115 ++++++ services/ws-modules/rdata1/pkg/package.json | 8 + services/ws-server/Cargo.toml | 1 + services/ws-server/src/lib.rs | 10 + services/ws-web-runner/tests/modules.rs | 59 +++ 33 files changed, 1227 insertions(+), 14 deletions(-) create mode 100644 .mise/config.r.toml create mode 100644 services/websockify/Cargo.toml create mode 100644 services/websockify/src/lib.rs create mode 100644 services/websockify/tests/relay.rs create mode 100644 services/ws-modules/rcomm1/pkg/et_ws_rcomm1.js create mode 100644 services/ws-modules/rcomm1/pkg/module.R create mode 100644 services/ws-modules/rcomm1/pkg/package.json create mode 100644 services/ws-modules/rdata1/pkg/et_ws_rdata1.js create mode 100644 services/ws-modules/rdata1/pkg/module.R create mode 100644 services/ws-modules/rdata1/pkg/package.json diff --git a/.dockerignore b/.dockerignore index 3723f547..d4e6650b 100644 --- a/.dockerignore +++ b/.dockerignore @@ -11,6 +11,8 @@ **/.DS_Store services/ws-wasm-agent/pkg/ services/ws-modules/pywasm1/pkg/ +services/ws-modules/rdata1/pkg/webr/ +services/ws-modules/rcomm1/pkg/webr/ services/ws-server/static/models/ **/.zig-cache/ **/zig-out/ diff --git a/.github/workflows/check.yaml b/.github/workflows/check.yaml index 09afbf2d..89a3b70d 100644 --- a/.github/workflows/check.yaml +++ b/.github/workflows/check.yaml @@ -19,7 +19,7 @@ defaults: shell: bash --noprofile --norc -euo pipefail {0} env: - MISE_ENV: dart,dotnet,java,js,python,rust,zig + MISE_ENV: dart,dotnet,java,js,python,r,rust,zig jobs: check: diff --git a/.github/workflows/codeql.yaml b/.github/workflows/codeql.yaml index a9b34f6a..14cc0d06 100644 --- a/.github/workflows/codeql.yaml +++ b/.github/workflows/codeql.yaml @@ -33,7 +33,7 @@ defaults: shell: bash --noprofile --norc -euo pipefail {0} env: - MISE_ENV: dart,dotnet,java,js,python,rust,zig + MISE_ENV: dart,dotnet,java,js,python,r,rust,zig jobs: analyze: diff --git a/.github/workflows/coverage.yaml b/.github/workflows/coverage.yaml index 5ce35436..38c21130 100644 --- a/.github/workflows/coverage.yaml +++ b/.github/workflows/coverage.yaml @@ -21,14 +21,14 @@ defaults: shell: bash --noprofile --norc -euo pipefail {0} env: - MISE_ENV: dart,dotnet,java,js,python,rust,zig + MISE_ENV: dart,dotnet,java,js,python,r,rust,zig jobs: coverage: runs-on: ubuntu-latest timeout-minutes: 150 env: - MISE_ENV: dart,dotnet,java,js,python,rust,zig,coverage + MISE_ENV: dart,dotnet,java,js,python,r,rust,zig,coverage ET_TEST_COVERAGE: "true" steps: - name: Checkout diff --git a/.github/workflows/dependencies.yaml b/.github/workflows/dependencies.yaml index cab0ccfc..c6201fe6 100644 --- a/.github/workflows/dependencies.yaml +++ b/.github/workflows/dependencies.yaml @@ -25,7 +25,7 @@ defaults: shell: bash --noprofile --norc -euo pipefail {0} env: - MISE_ENV: dart,dotnet,java,js,python,rust,zig + MISE_ENV: dart,dotnet,java,js,python,r,rust,zig jobs: dependencies: diff --git a/.github/workflows/docker-linux.yaml b/.github/workflows/docker-linux.yaml index 0635cc62..1b386fcc 100644 --- a/.github/workflows/docker-linux.yaml +++ b/.github/workflows/docker-linux.yaml @@ -20,7 +20,7 @@ defaults: shell: bash --noprofile --norc -euo pipefail {0} env: - MISE_ENV: dart,dotnet,java,js,python,rust,zig + MISE_ENV: dart,dotnet,java,js,python,r,rust,zig jobs: build: diff --git a/.github/workflows/docker-windows.yaml b/.github/workflows/docker-windows.yaml index f1bfc750..f6b738ed 100644 --- a/.github/workflows/docker-windows.yaml +++ b/.github/workflows/docker-windows.yaml @@ -22,7 +22,7 @@ defaults: shell: bash --noprofile --norc -euo pipefail {0} env: - MISE_ENV: dart,dotnet,java,js,python,rust,zig + MISE_ENV: dart,dotnet,java,js,python,r,rust,zig jobs: build: diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index 1d2c25ac..33c377ef 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -31,7 +31,7 @@ defaults: shell: bash --noprofile --norc -euo pipefail {0} env: - MISE_ENV: "dart,dotnet,java,js,python,rust,zig" + MISE_ENV: "dart,dotnet,java,js,python,r,rust,zig" jobs: default: @@ -164,7 +164,7 @@ jobs: # the rust toolchains' extra target std, and mise does not reconcile a changed `targets` list on an # already-installed toolchain. env: - MISE_ENV: "dart,dotnet,java,js,python,rust,zig,${{ matrix.compiler }}" + MISE_ENV: "dart,dotnet,java,js,python,r,rust,zig,${{ matrix.compiler }}" # Static busybox-w32 ash default (this job is always windows-latest, so no per-OS ternary). defaults: run: diff --git a/.github/workflows/upstream-cache.yaml b/.github/workflows/upstream-cache.yaml index d7331660..bccc6ded 100644 --- a/.github/workflows/upstream-cache.yaml +++ b/.github/workflows/upstream-cache.yaml @@ -44,7 +44,7 @@ defaults: shell: bash --noprofile --norc -euo pipefail {0} env: - MISE_ENV: dart,dotnet,java,js,python,rust,zig + MISE_ENV: dart,dotnet,java,js,python,r,rust,zig jobs: rustpython: diff --git a/.gitignore b/.gitignore index ba086bcc..9275c687 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,8 @@ target/ .DS_Store services/ws-wasm-agent/pkg/ services/ws-modules/pywasm1/pkg/ +services/ws-modules/rdata1/pkg/webr/ +services/ws-modules/rcomm1/pkg/webr/ services/ws-server/static/models/ .zig-cache/ zig-out/ diff --git a/.mise/config.r.toml b/.mise/config.r.toml new file mode 100644 index 00000000..0f2d4178 --- /dev/null +++ b/.mise/config.r.toml @@ -0,0 +1,49 @@ +# R toolchain + tasks. Loaded when MISE_ENV includes `r`. +# +# The R-language modules (rdata1, rcomm1) run R via webR -- the R interpreter compiled to WebAssembly. There is +# no host R toolchain, compiler, or linter to install: the R workload is an inline R program the module's JS shim +# feeds to webR at runtime, the shim itself is linted by the `js` env (oxlint/oxfmt/dprint), and the webR +# distribution is a prebuilt wasm bundle. So this env supplies just the webR tool plus the build tasks that +# vendor it into each module's pkg/ -- there is no check:r / fmt:r / test:r because nothing R-specific is +# statically analysable here. The build-ws-*-module tasks join the globbed `build-modules` aggregate when this +# env is loaded, exactly like the Zig module's build task. + +[tools] +# webR: the R interpreter compiled to WebAssembly, vendored into the R modules by their build tasks below. +# Platform-independent (it is wasm), so one url covers every OS. Direct upstream release, no checksum -- same +# posture as http:openobserve. +[tools."http:webr"] +url = "https://github.com/r-wasm/webr/releases/download/v{{version}}/webr-{{version}}.tar.gz" +version = "0.6.0" + +# The two R modules share an identical vendoring step. +# webR is a prebuilt wasm distribution, so "building" a module is just staging that distribution into the +# module's pkg/webr/. `http:webr` downloads and unpacks it flat; copy that install dir into the module's +# pkg/webr/ (dropping mise's own metadata.json). The module's package.json makes pkg/ a served package, so the +# copied webr/ subtree rides along as static assets under /modules//webr/. + +[tasks.build-ws-rdata1-module] +description = "Vendor the webR distribution into rdata1's pkg/webr/ (served at /modules/et-ws-rdata1/webr/)" +dir = "services/ws-modules/rdata1" +run = """ +src="$(mise where "http:webr")" +[ -n "$src" ] || { echo "http:webr not installed; run 'MISE_ENV=r mise install' first" >&2; exit 1; } +coreutils rm -rf pkg/webr +coreutils mkdir -p pkg/webr +coreutils cp -R "$src/." pkg/webr/ +coreutils rm -f pkg/webr/metadata.json +""" +shell = "bash -euo pipefail -c" + +[tasks.build-ws-rcomm1-module] +description = "Vendor the webR distribution into rcomm1's pkg/webr/ (served at /modules/et-ws-rcomm1/webr/)" +dir = "services/ws-modules/rcomm1" +run = """ +src="$(mise where "http:webr")" +[ -n "$src" ] || { echo "http:webr not installed; run 'MISE_ENV=r mise install' first" >&2; exit 1; } +coreutils rm -rf pkg/webr +coreutils mkdir -p pkg/webr +coreutils cp -R "$src/." pkg/webr/ +coreutils rm -f pkg/webr/metadata.json +""" +shell = "bash -euo pipefail -c" diff --git a/.mise/config.toml b/.mise/config.toml index d59a13e9..cf92df0d 100644 --- a/.mise/config.toml +++ b/.mise/config.toml @@ -468,7 +468,7 @@ rp_wasm_dir = '''{% if os() == "windows" -%} # Used as MISE_ENV by the *-all tasks to act on every language at once (mise has # no MISE_ENV=all). Hardcoded rather than shell-discovered so it works on Windows # too -- keep in sync when adding/removing a config..toml. -ALL_LANGS = "dart,dotnet,java,js,python,rust,zig" +ALL_LANGS = "dart,dotnet,java,js,python,r,rust,zig" CLIPPY_CONF_DIR = "{{ config_root }}/config" TAPLO_CONFIG = "{{ config_root }}/config/taplo.toml" # augtool's lens search path is compile-time pinned to the build-time --prefix. @@ -1209,6 +1209,12 @@ run = "watchexec --restart --clear --exts rs,toml -i '**/tests/**' --shell=none OTLP_AUTH_PASSWORD = "1234" OTLP_AUTH_USERNAME = "root@example.com" +[tasks.ws-server-no-otel] +description = "Run the WebSocket server without OTLP/o2 observability (OpenTelemetry init is skipped)" +# Same as ws-server but sets no OTLP_* env, so Config.otlp deserialises to None and et_otlp::init is skipped. +# Use this when you don't want the o2/OpenObserve stack (`mise run o2`) running alongside the server. +run = "watchexec --restart --clear --exts rs,toml -i '**/tests/**' --shell=none -- cargo run -p et-ws-server" + [tasks.ws-wasi-runner] description = "Run the WebSocket WASI runner (set RUNNER_MODULE, e.g. wasi-graphics-info; RUNNER_FEATURES=cuda on CUDA)" dir = "services/ws-wasi-runner" diff --git a/CLAUDE.md b/CLAUDE.md index fc1cf3e6..0af8f829 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -310,6 +310,8 @@ Languages: - **Python (Pyodide)**: pydata1, pyeye1, pyface1 - **C# (.NET WASM)**: dotnet-data1 - **Java (TeaVM -> JS)**: java-data1 +- **R (webR -> WASM)**: rdata1, rcomm1 -- browser-only (webR spawns a classic Worker, unsupported by Deno's + ws-web-runner). Their JS shims are linted by the `js` env; `MISE_ENV=r` supplies webR + the vendoring build tasks. - **Zig -> WASM**: zig-data1 - **Python (componentize-py -> WASI Preview 2 component)**: wasi-graphics-info -- runs in `et-ws-wasi-runner` rather than the browser. The WIT world the component implements is at diff --git a/Cargo.lock b/Cargo.lock index 108b9b1d..eb376e3e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4362,6 +4362,19 @@ dependencies = [ "web-sys", ] +[[package]] +name = "et-websockify-service" +version = "0.1.0" +dependencies = [ + "actix-web", + "actix-ws", + "bytes", + "futures-util", + "log", + "tokio", + "tokio-tungstenite", +] + [[package]] name = "et-ws-audio1" version = "0.1.0" @@ -4606,6 +4619,7 @@ dependencies = [ "et-modules-service", "et-otlp", "et-storage-service", + "et-websockify-service", "et-ws-service", "fs-err", "futures-util", diff --git a/Cargo.toml b/Cargo.toml index 383f144d..a8bbf4c3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,6 +29,7 @@ members = [ "services/ws-modules/wasi-data1", "services/modules", "services/storage", + "services/websockify", "services/ws", "services/ws-pyo3-runner", "services/ws-server", @@ -79,6 +80,7 @@ et-rest-client = { path = "generated/rust-rest", version = "0.1.0", default-feat et-storage-service = { path = "services/storage", version = "0.1.0" } et-test-helpers = { path = "libs/test-helpers", version = "0.1.0" } et-web = { path = "libs/web", version = "0.1.0" } +et-websockify-service = { path = "services/websockify", version = "0.1.0" } et-ws-runner-common = { path = "libs/ws-runner-common", version = "0.1.0" } et-ws-server = { path = "services/ws-server", version = "0.1.0" } et-ws-service = { path = "services/ws", version = "0.1.0" } diff --git a/Dockerfile b/Dockerfile index 631ad06c..d08ec773 100644 --- a/Dockerfile +++ b/Dockerfile @@ -219,7 +219,7 @@ EOF FROM build-minimal AS build COPY .mise/ .mise/ RUN mise trust -ARG MISE_ENV=dart,dotnet,java,js,python,rust,zig +ARG MISE_ENV=dart,dotnet,java,js,python,r,rust,zig ENV MISE_ENV=${MISE_ENV} RUN --mount=type=secret,id=gh_token,required=false bash <<'EOF' set -euo pipefail diff --git a/Dockerfile.windows b/Dockerfile.windows index 05e4da0d..b55ed560 100644 --- a/Dockerfile.windows +++ b/Dockerfile.windows @@ -144,7 +144,7 @@ RUN (if exist C:\token\gh_token set /p GITHUB_TOKEN=). This service is that relay. +//! +//! It recognises two client shapes on the same `/websockify` route, by the first byte: +//! +//! - **SOCKS5** (first byte `0x05`): webR's curl is configured to reach a SOCKS5 proxy, so the relay speaks just +//! enough SOCKS5 (no-auth, CONNECT) to be that proxy. The requested CONNECT target is honoured only when it is +//! loopback; every connection is then bridged to the single server-configured target (its own plain-HTTP +//! port). A non-loopback CONNECT (e.g. curl's probe to the public r-universe proxy) is refused with a SOCKS5 +//! error, so it never reaches the app server. +//! - **Direct byte stream** (anything else, e.g. a raw HTTP request): bridged straight to the same target. +//! +//! Either way the target is fixed by the server (see [`configure`]), never taken from the client, so a browser +//! cannot point it at an arbitrary host (no SSRF / open proxy). It is a separate route from the agent hub's +//! `/ws`: Emscripten frames carry raw TCP bytes with no marker of their own, indistinguishable from the hub's +//! binary-broadcast fallback, so the two must never share one socket -- the separate path is the separation. + +use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr}; + +use actix_web::http::header::{HeaderValue, SEC_WEBSOCKET_PROTOCOL}; +use actix_web::{Error, HttpRequest, HttpResponse, web}; +use actix_ws::{AggregatedMessage, AggregatedMessageStream, MessageStream, Session}; +use bytes::Bytes; +use futures_util::StreamExt as _; +use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; +use tokio::net::TcpStream; + +/// Loopback TCP endpoint every `/websockify` connection is bridged to. +/// +/// Set by the server in [`configure`], never by the client -- this is what keeps the relay from being an open +/// proxy: a browser chooses neither the host nor the port. +#[derive(Clone, Copy)] +struct RelayTarget(SocketAddr); + +/// Register the `/websockify` relay route, bridging every connection to `target`. +/// +/// `target` is expected to be a loopback address -- the ws-server passes its own plain-HTTP port. Register this +/// before any catch-all `Files::new("/")` mount so the explicit route wins. +#[expect( + clippy::literal_string_with_formatting_args, + reason = "`{tail:.*}` is an actix-web route path pattern, not a format string" +)] +pub fn configure(cfg: &mut web::ServiceConfig, target: SocketAddr) { + // Match a trailing path too: Emscripten may append the connect target to the configured WebSocket URL + // (e.g. `/websockify/127.0.0.1:8080`). The suffix is ignored -- the relay always bridges to the fixed + // server-configured target -- so it stays SSRF-safe regardless of what the client puts there. + let _routed = cfg + .app_data(web::Data::new(RelayTarget(target))) + .route("/websockify{tail:.*}", web::get().to(relay_handler)); +} + +/// Bytes copied per TCP->WebSocket read. 16 KiB is a typical socket read; large reads just loop. +const RELAY_BUF: usize = 16 * 1024; + +/// Upper bound on a single inbound WebSocket frame / aggregated message the relay will accept. +/// +/// Emscripten sends the tunnelled TCP stream as many small binary messages, so this is only a safety ceiling; +/// it is set well above any realistic single send (and above actix-ws's smaller defaults, which would otherwise +/// reject a large upload frame and tear the relay down). +const RELAY_MAX_MESSAGE: usize = 8 * 1024 * 1024; + +/// How many bytes of a relayed chunk to show in the debug preview log line. +const PREVIEW_BYTES: usize = 256; + +// SOCKS5 (RFC 1928) constants for the minimal proxy the relay speaks. +const SOCKS5_VERSION: u8 = 0x05; +const SOCKS5_CMD_CONNECT: u8 = 0x01; +const SOCKS5_ATYP_IPV4: u8 = 0x01; +const SOCKS5_ATYP_DOMAIN: u8 = 0x03; +const SOCKS5_ATYP_IPV6: u8 = 0x04; +/// Method-selection reply: version 5, "no authentication required". +const SOCKS5_NO_AUTH: [u8; 2] = [SOCKS5_VERSION, 0x00]; +/// CONNECT reply, succeeded: VER, REP=0, RSV, ATYP=IPv4, BND.ADDR=0.0.0.0, BND.PORT=0. +const SOCKS5_REPLY_OK: [u8; 10] = [SOCKS5_VERSION, 0x00, 0x00, SOCKS5_ATYP_IPV4, 0, 0, 0, 0, 0, 0]; +/// CONNECT reply, connection not allowed by ruleset (REP=2). +const SOCKS5_REPLY_REFUSED: [u8; 10] = [SOCKS5_VERSION, 0x02, 0x00, SOCKS5_ATYP_IPV4, 0, 0, 0, 0, 0, 0]; + +/// Render the first [`PREVIEW_BYTES`] of a relayed chunk as an escaped string for logging. +/// +/// Escaping makes CR/LF and any non-printable bytes visible, so a malformed HTTP request line or header (or an +/// unexpectedly base64-encoded frame) is obvious in the log rather than mangling the terminal. +fn preview(bytes: &[u8]) -> String { + let end = bytes.len().min(PREVIEW_BYTES); + String::from_utf8_lossy(bytes.get(..end).unwrap_or_default()) + .escape_default() + .to_string() +} + +#[expect( + clippy::future_not_send, + clippy::single_call_fn, + reason = "actix handler: HttpRequest/Payload are Rc-backed and !Send; registered once in configure" +)] +async fn relay_handler( + req: HttpRequest, + body: web::Payload, + target: web::Data, +) -> Result { + // Emscripten offers the `binary` subprotocol; echo it so the browser WebSocket sees a negotiated protocol + // and delivers unencoded binary frames rather than falling back to base64-encoded text. + let offered = req + .headers() + .get(SEC_WEBSOCKET_PROTOCOL) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let offers_binary = offered + .as_deref() + .is_some_and(|protocols| protocols.split(',').any(|protocol| protocol.trim() == "binary")); + log::info!( + "websockify: upgrade on {} -> target {} (offered subprotocols: {:?}, echoing binary: {offers_binary})", + req.path(), + target.0, + offered.as_deref().unwrap_or(""), + ); + + let (mut response, session, stream) = actix_ws::handle(&req, body)?; + if offers_binary { + let _prev = response + .headers_mut() + .insert(SEC_WEBSOCKET_PROTOCOL, HeaderValue::from_static("binary")); + } + + let target = target.0; + let _join = actix_web::rt::spawn(async move { + relay(session, stream, target).await; + }); + Ok(response) +} + +/// Read from the WebSocket into `buf` until it holds at least `need` bytes; false if the socket closed/errored. +#[expect( + clippy::future_not_send, + reason = "actix-ws Session/stream are Rc-backed and !Send; runs on the current-thread actix runtime" +)] +async fn fill(stream: &mut AggregatedMessageStream, session: &mut Session, buf: &mut Vec, need: usize) -> bool { + while buf.len() < need { + match stream.next().await { + Some(Ok(AggregatedMessage::Binary(bytes))) => buf.extend_from_slice(bytes.as_ref()), + Some(Ok(AggregatedMessage::Text(text))) => buf.extend_from_slice(text.as_bytes()), + Some(Ok(AggregatedMessage::Ping(ping))) => { + let _pong = session.pong(&ping).await; + } + Some(Ok(AggregatedMessage::Pong(_))) => {} + _ => return false, + } + } + true +} + +/// Perform the minimal SOCKS5 server handshake over the WebSocket. +/// +/// Returns `Some(true)` if the client issued a CONNECT to a loopback target (success reply sent; `buf` left +/// holding any leftover tunnelled bytes), `Some(false)` if the request was refused (error reply sent), or `None` +/// if the socket closed mid-handshake. Only no-auth CONNECT to a loopback address is accepted; the relay then +/// bridges to its own fixed target, so the requested port is not otherwise used. +#[expect( + clippy::future_not_send, + clippy::single_call_fn, + reason = "!Send actix-ws types; called once from relay" +)] +async fn socks5_handshake( + stream: &mut AggregatedMessageStream, + session: &mut Session, + buf: &mut Vec, +) -> Option { + // Method selection: VER, NMETHODS, METHODS... + if !fill(stream, session, buf, 2).await { + return None; + } + let nmethods = usize::from(*buf.get(1)?); + let select_len = 2_usize.saturating_add(nmethods); + if !fill(stream, session, buf, select_len).await { + return None; + } + // Discard the consumed method-selection bytes, keeping any tail already buffered. + *buf = buf.split_off(select_len); + session.binary(Bytes::from_static(&SOCKS5_NO_AUTH)).await.ok()?; + + // Request: VER, CMD, RSV, ATYP, DST.ADDR, DST.PORT. + if !fill(stream, session, buf, 4).await { + return None; + } + let command = *buf.get(1)?; + let atyp = *buf.get(3)?; + let (loopback, request_len) = match atyp { + SOCKS5_ATYP_IPV4 => { + if !fill(stream, session, buf, 10).await { + return None; + } + let octets: [u8; 4] = buf.get(4..8)?.try_into().ok()?; + (Ipv4Addr::from(octets).is_loopback(), 10) + } + SOCKS5_ATYP_IPV6 => { + if !fill(stream, session, buf, 22).await { + return None; + } + let octets: [u8; 16] = buf.get(4..20)?.try_into().ok()?; + (Ipv6Addr::from(octets).is_loopback(), 22) + } + SOCKS5_ATYP_DOMAIN => { + if !fill(stream, session, buf, 5).await { + return None; + } + let name_len = usize::from(*buf.get(4)?); + let request_len = 5_usize.saturating_add(name_len).saturating_add(2); + if !fill(stream, session, buf, request_len).await { + return None; + } + let name_end = 5_usize.saturating_add(name_len); + let name = String::from_utf8_lossy(buf.get(5..name_end)?).into_owned(); + (matches!(name.as_str(), "localhost" | "127.0.0.1" | "::1"), request_len) + } + _ => { + session.binary(Bytes::from_static(&SOCKS5_REPLY_REFUSED)).await.ok()?; + return Some(false); + } + }; + + if command == SOCKS5_CMD_CONNECT && loopback { + session.binary(Bytes::from_static(&SOCKS5_REPLY_OK)).await.ok()?; + // Discard the consumed request bytes, keeping any tunnelled data the client already sent after it. + *buf = buf.split_off(request_len); + Some(true) + } else { + session.binary(Bytes::from_static(&SOCKS5_REPLY_REFUSED)).await.ok()?; + Some(false) + } +} + +#[expect( + clippy::future_not_send, + clippy::single_call_fn, + reason = "actix-ws Session/stream are Rc-backed and !Send; called once from relay_handler" +)] +async fn relay(mut session: Session, stream: MessageStream, target: SocketAddr) { + let mut stream = stream + .max_frame_size(RELAY_MAX_MESSAGE) + .aggregate_continuations() + .max_continuation_size(RELAY_MAX_MESSAGE); + let mut head: Vec = Vec::new(); + + // Recognise the client from its first byte: SOCKS5 (webR's curl proxy) vs a direct byte stream. + if !fill(&mut stream, &mut session, &mut head, 1).await { + let _closed = session.close(None).await; + return; + } + if head.first() == Some(&SOCKS5_VERSION) { + match socks5_handshake(&mut stream, &mut session, &mut head).await { + Some(true) => log::info!("websockify: SOCKS5 CONNECT to loopback accepted; bridging to {target}"), + Some(false) => { + log::info!("websockify: refused SOCKS5 CONNECT (non-loopback target or unsupported command)"); + let _closed = session.close(None).await; + return; + } + None => { + let _closed = session.close(None).await; + return; + } + } + } else { + log::info!("websockify: direct (non-SOCKS5) client; bridging to {target}"); + } + + let tcp = match TcpStream::connect(target).await { + Ok(tcp) => tcp, + Err(err) => { + log::warn!("websockify: cannot reach relay target {target}: {err}"); + let _closed = session.close(None).await; + return; + } + }; + log::info!("websockify: bridged to target {target}"); + bridge(session, stream, tcp, head).await; +} + +#[expect( + clippy::cognitive_complexity, + clippy::future_not_send, + clippy::integer_division_remainder_used, + clippy::single_call_fn, + reason = "actix-ws Session/stream are Rc-backed and !Send; select! uses % internally; one call site" +)] +async fn bridge(mut session: Session, mut stream: AggregatedMessageStream, tcp: TcpStream, head: Vec) { + let (mut tcp_read, mut tcp_write) = tcp.into_split(); + let mut to_target = 0_usize; + let mut to_client = 0_usize; + + // Forward any bytes already read during protocol detection (the HTTP request start, or SOCKS5 tunnel data). + if !head.is_empty() { + log::info!( + "websockify: first client->target bytes ({} bytes): {}", + head.len(), + preview(&head) + ); + to_target = head.len(); + if tcp_write.write_all(&head).await.is_err() { + let _closed = session.close(None).await; + return; + } + } + + let mut buf = vec![0_u8; RELAY_BUF]; + loop { + tokio::select! { + inbound = stream.next() => match inbound { + // Browser -> TCP: forward the raw bytes of each message to the target socket. + Some(Ok(AggregatedMessage::Binary(bytes))) => { + to_target = to_target.saturating_add(bytes.len()); + if tcp_write.write_all(&bytes).await.is_err() { + break; + } + } + Some(Ok(AggregatedMessage::Text(text))) => { + to_target = to_target.saturating_add(text.len()); + if tcp_write.write_all(text.as_bytes()).await.is_err() { + break; + } + } + Some(Ok(AggregatedMessage::Ping(ping))) => { + let _pong = session.pong(&ping).await; + } + Some(Ok(AggregatedMessage::Pong(_))) => {} + Some(Ok(AggregatedMessage::Close(reason))) => { + log::info!("websockify: client closed the websocket ({reason:?})"); + break; + } + None => break, + Some(Err(err)) => { + log::debug!("websockify: websocket receive error: {err}"); + break; + } + }, + outbound = tcp_read.read(&mut buf) => match outbound { + // TCP -> browser: 0 bytes means the target closed; anything else is a binary frame. + Ok(0) => { + log::debug!("websockify: target closed the connection"); + break; + } + Err(err) => { + log::debug!("websockify: target read failed: {err}"); + break; + } + Ok(read) => { + // `read` is always <= buf.len() per AsyncReadExt::read, so get(..read) is always Some. + let Some(chunk) = buf.get(..read) else { break }; + if to_client == 0 { + log::info!("websockify: first target->client chunk ({read} bytes): {}", preview(chunk)); + } + to_client = to_client.saturating_add(read); + if session.binary(Bytes::copy_from_slice(chunk)).await.is_err() { + break; + } + } + }, + } + } + + let _shutdown = tcp_write.shutdown().await; + let _closed = session.close(None).await; + log::info!("websockify: closed (client->target {to_target} bytes, target->client {to_client} bytes)"); +} diff --git a/services/websockify/tests/relay.rs b/services/websockify/tests/relay.rs new file mode 100644 index 00000000..1a797076 --- /dev/null +++ b/services/websockify/tests/relay.rs @@ -0,0 +1,315 @@ +//! End-to-end tests for the `/websockify` WebSocket-to-TCP relay. +//! +//! These stand in for the browser: a real WebSocket client (tokio-tungstenite) connects to the relay, and the +//! relay bridges to a plain TCP target we control. That is exactly the shape of the browser path -- webR's +//! Emscripten libcurl opens a WebSocket and writes/reads raw TCP bytes -- minus the Emscripten framing, which is +//! ordinary WebSocket binary. Each target below (echo, a minimal HTTP responder, a closing socket, an +//! unreachable port) pins one behaviour the browser depends on. + +#![cfg(test)] + +use std::net::SocketAddr; +use std::time::Duration; + +use actix_web::{App, HttpServer}; +use futures_util::{SinkExt as _, StreamExt as _}; +use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; +use tokio::net::{TcpListener, TcpStream}; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::tungstenite::client::IntoClientRequest as _; +use tokio_tungstenite::tungstenite::http::HeaderValue; +use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, connect_async}; + +type WsStream = WebSocketStream>; + +/// Start the relay app bridging to `target`, on an ephemeral port; return that port once it accepts. +#[expect( + clippy::future_not_send, + reason = "test helper; the actix App factory is !Send and runs on the current-thread test runtime" +)] +async fn start_relay(target: SocketAddr) -> u16 { + let server = HttpServer::new(move || App::new().configure(|cfg| et_websockify_service::configure(cfg, target))) + .workers(1) + .bind(("127.0.0.1", 0)) + .unwrap(); + let port = server.addrs()[0].port(); + let _server = actix_web::rt::spawn(server.run()); + for _ in 0_u32..100 { + if TcpStream::connect(("127.0.0.1", port)).await.is_ok() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + port +} + +/// A TCP target that echoes every byte back -- exercises both relay directions. +async fn start_echo_target() -> SocketAddr { + let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap(); + let addr = listener.local_addr().unwrap(); + let _task = actix_web::rt::spawn(async move { + while let Ok((mut sock, _)) = listener.accept().await { + let _conn = actix_web::rt::spawn(async move { + let mut buf = [0_u8; 4096]; + loop { + match sock.read(&mut buf).await { + Ok(0) | Err(_) => break, + Ok(read) => { + if sock.write_all(&buf[..read]).await.is_err() { + break; + } + } + } + } + }); + } + }); + addr +} + +/// A TCP target that answers any request with a fixed HTTP/1.1 200 -- what libcurl over the tunnel expects. +#[expect(clippy::single_call_fn, reason = "test target used by one case")] +async fn start_http_target() -> SocketAddr { + let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap(); + let addr = listener.local_addr().unwrap(); + let _task = actix_web::rt::spawn(async move { + while let Ok((mut sock, _)) = listener.accept().await { + let _conn = actix_web::rt::spawn(async move { + let mut buf = [0_u8; 4096]; + let _read = sock.read(&mut buf).await; + let response = "HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: close\r\n\r\npong"; + let _written = sock.write_all(response.as_bytes()).await; + let _shutdown = sock.shutdown().await; + }); + } + }); + addr +} + +/// A TCP target that accepts then immediately closes -- the relay must propagate that as a WebSocket close. +#[expect(clippy::single_call_fn, reason = "test target used by one case")] +async fn start_closing_target() -> SocketAddr { + let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap(); + let addr = listener.local_addr().unwrap(); + let _task = actix_web::rt::spawn(async move { + while let Ok((_sock, _)) = listener.accept().await { + // `_sock` drops at the end of the loop body, closing the connection immediately. + } + }); + addr +} + +/// Reserve then release a port so nothing is listening on it -- an unreachable relay target. +#[expect(clippy::single_call_fn, reason = "test target used by one case")] +async fn unreachable_target() -> SocketAddr { + let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap(); + listener.local_addr().unwrap() +} + +async fn open_ws(port: u16) -> WsStream { + let (ws, _response) = connect_async(format!("ws://127.0.0.1:{port}/websockify")) + .await + .unwrap(); + ws +} + +/// Accumulate relayed bytes until at least `want` have arrived (or the socket closes), with a timeout. +async fn recv_bytes(ws: &mut WsStream, want: usize) -> Vec { + let mut acc = Vec::new(); + let collected = tokio::time::timeout(Duration::from_secs(10), async { + while acc.len() < want { + match ws.next().await { + Some(Ok(Message::Binary(bytes))) => acc.extend_from_slice(&bytes), + Some(Ok(Message::Text(text))) => acc.extend_from_slice(text.as_bytes()), + Some(Ok(Message::Close(_))) | None => break, + Some(Ok(_)) => {} + Some(Err(err)) => panic!("websocket error: {err}"), + } + } + }) + .await; + collected.unwrap(); + acc +} + +/// Return true if the relay closes the WebSocket within the timeout (a `Close` frame, end of stream, or error). +async fn closed_within(ws: &mut WsStream, secs: u64) -> bool { + tokio::time::timeout(Duration::from_secs(secs), async { + loop { + match ws.next().await { + Some(Ok(Message::Close(_)) | Err(_)) | None => return true, + Some(Ok(_)) => {} + } + } + }) + .await + .unwrap_or(false) +} + +#[actix_web::test] +async fn relay_echoes_bytes_both_directions() { + let port = start_relay(start_echo_target().await).await; + let mut ws = open_ws(port).await; + + ws.send(Message::Binary(b"hello relay".to_vec())).await.unwrap(); + let echoed = recv_bytes(&mut ws, b"hello relay".len()).await; + + assert_eq!(echoed, b"hello relay"); +} + +#[actix_web::test] +async fn relay_carries_a_full_http_get() { + let port = start_relay(start_http_target().await).await; + let mut ws = open_ws(port).await; + + // Exactly what libcurl/httr2 would write over the tunnel. + let request = "GET /storage/agent/data.txt HTTP/1.1\r\nHost: relay\r\nConnection: close\r\n\r\n"; + ws.send(Message::Binary(request.as_bytes().to_vec())).await.unwrap(); + + let response = String::from_utf8(recv_bytes(&mut ws, 48).await).unwrap(); + assert!( + response.starts_with("HTTP/1.1 200 OK"), + "unexpected response: {response}" + ); + assert!(response.contains("pong"), "response missing body: {response}"); +} + +#[actix_web::test] +async fn relay_handles_a_large_chunked_payload() { + let port = start_relay(start_echo_target().await).await; + let mut ws = open_ws(port).await; + + // Larger than the relay's 16 KiB read buffer, so both directions must span multiple reads/frames. + let payload = vec![0xAB_u8; 200 * 1024]; + ws.send(Message::Binary(payload.clone())).await.unwrap(); + let echoed = recv_bytes(&mut ws, payload.len()).await; + + assert_eq!(echoed.len(), payload.len()); + assert_eq!(echoed, payload); +} + +#[actix_web::test] +async fn relay_closes_when_target_closes() { + let port = start_relay(start_closing_target().await).await; + let mut ws = open_ws(port).await; + // Send a (non-SOCKS5) byte so the relay sniffs the protocol and connects to the target, which closed. + ws.send(Message::Binary(b"GET / HTTP/1.1\r\n\r\n".to_vec())) + .await + .unwrap(); + + assert!( + closed_within(&mut ws, 10).await, + "relay did not close after target closed" + ); +} + +#[actix_web::test] +async fn relay_closes_when_target_unreachable() { + let port = start_relay(unreachable_target().await).await; + let mut ws = open_ws(port).await; + // The relay only connects to the target after the client's first byte; send one so it tries (and fails). + ws.send(Message::Binary(b"GET / HTTP/1.1\r\n\r\n".to_vec())) + .await + .unwrap(); + + assert!( + closed_within(&mut ws, 10).await, + "relay did not close when target was unreachable" + ); +} + +#[actix_web::test] +async fn relay_accepts_an_appended_target_path() { + // Emscripten may append the connect target to the configured URL: ws://host/websockify/:. + // The relay must still match and bridge to its own fixed target, ignoring the suffix. + let port = start_relay(start_echo_target().await).await; + let (mut ws, _response) = connect_async(format!("ws://127.0.0.1:{port}/websockify/127.0.0.1:8080")) + .await + .unwrap(); + + ws.send(Message::Binary(b"suffixed".to_vec())).await.unwrap(); + let echoed = recv_bytes(&mut ws, b"suffixed".len()).await; + + assert_eq!(echoed, b"suffixed"); +} + +/// Do the SOCKS5 no-auth method negotiation, send `request`, and return the open socket + the CONNECT reply. +async fn socks5_connect(port: u16, request: &[u8]) -> (WsStream, Vec) { + let mut ws = open_ws(port).await; + // Method selection: version 5, one method, "no authentication". + ws.send(Message::Binary(vec![0x05, 0x01, 0x00])).await.unwrap(); + assert_eq!( + recv_bytes(&mut ws, 2).await, + vec![0x05, 0x00], + "SOCKS5 method-selection reply" + ); + ws.send(Message::Binary(request.to_vec())).await.unwrap(); + let reply = recv_bytes(&mut ws, 10).await; + (ws, reply) +} + +#[actix_web::test] +async fn relay_socks5_connect_to_loopback_bridges() { + let port = start_relay(start_echo_target().await).await; + // CONNECT 127.0.0.1:80 -- loopback, so accepted; the relay bridges to its own (echo) target. + let request = [0x05, 0x01, 0x00, 0x01, 127, 0, 0, 1, 0x00, 0x50]; + let (mut ws, reply) = socks5_connect(port, &request).await; + assert_eq!( + reply.get(0..2), + Some([0x05, 0x00].as_slice()), + "expected SOCKS5 success, got {reply:?}" + ); + + // Tunnel is live -> bytes echo back through the bridged target. + ws.send(Message::Binary(b"socks-hi".to_vec())).await.unwrap(); + assert_eq!(recv_bytes(&mut ws, b"socks-hi".len()).await, b"socks-hi"); +} + +#[actix_web::test] +async fn relay_socks5_refuses_non_loopback_ipv4() { + let port = start_relay(start_echo_target().await).await; + // CONNECT 8.8.8.8:443 -- not loopback, must be refused (REP=2) and never reach the target. + let request = [0x05, 0x01, 0x00, 0x01, 8, 8, 8, 8, 0x01, 0xBB]; + let (_ws, reply) = socks5_connect(port, &request).await; + assert_eq!( + reply.get(1), + Some(&0x02), + "expected SOCKS5 refusal (REP=2), got {reply:?}" + ); +} + +#[actix_web::test] +async fn relay_socks5_refuses_external_domain() { + let port = start_relay(start_echo_target().await).await; + // CONNECT get-ws-proxy.r-universe.dev:443 -- the exact stray probe webR's curl makes; must be refused. + let host = b"get-ws-proxy.r-universe.dev"; + let mut request = vec![0x05, 0x01, 0x00, 0x03, u8::try_from(host.len()).unwrap()]; + request.extend_from_slice(host); + request.extend_from_slice(&[0x01, 0xBB]); + let (_ws, reply) = socks5_connect(port, &request).await; + assert_eq!( + reply.get(1), + Some(&0x02), + "expected SOCKS5 refusal for external domain, got {reply:?}" + ); +} + +#[actix_web::test] +async fn relay_echoes_the_binary_subprotocol() { + let port = start_relay(start_echo_target().await).await; + + // Emscripten offers `binary`; the server must echo it so the browser gets unencoded binary frames. + let mut request = format!("ws://127.0.0.1:{port}/websockify") + .into_client_request() + .unwrap(); + let _prev = request + .headers_mut() + .insert("Sec-WebSocket-Protocol", HeaderValue::from_static("binary")); + let (_ws, response) = connect_async(request).await.unwrap(); + + let negotiated = response + .headers() + .get("sec-websocket-protocol") + .and_then(|value| value.to_str().ok()); + assert_eq!(negotiated, Some("binary")); +} diff --git a/services/ws-modules/rcomm1/pkg/et_ws_rcomm1.js b/services/ws-modules/rcomm1/pkg/et_ws_rcomm1.js new file mode 100644 index 00000000..821c1baf --- /dev/null +++ b/services/ws-modules/rcomm1/pkg/et_ws_rcomm1.js @@ -0,0 +1,66 @@ +// et_ws_rcomm1.js -- bootstrap only. The control loop is module.R's run(); this shim boots webR, sets up the +// agent WebSocket transport that the R code drives via webr::eval_js, and hands control to run(). +// +// webR cannot open the agent WebSocket itself, so the transport (the shared et-ws-wasm-agent WsClient) lives +// here and is exposed on globalThis.__etAgent for R to drive. The on_message callback stashes the latest agent +// list (R selects the peer from it) and logs inbound messages. All sequencing and message composition happen in +// module.R. webR is vendored under pkg/webr/ (see build-ws-rcomm1-module). + +const WEBR_BASE_URL = "/modules/et-ws-rcomm1/webr/"; +const R_SOURCE_URL = "/modules/et-ws-rcomm1/module.R"; + +let webR = null; + +export default async function init() { + const { WebR } = await import(`${WEBR_BASE_URL}webr.mjs`); + webR = new WebR({ baseUrl: WEBR_BASE_URL }); + await webR.init(); + // Cache-bust so edits to module.R are picked up on reload. + const rSource = await (await fetch(`${R_SOURCE_URL}?v=${Date.now()}`)).text(); + await webR.evalRVoid(rSource); +} + +export async function run() { + if (!webR) throw new Error("rcomm1: not initialized"); + await setupAgent(); + // Hand control to R -- run() is the control loop. + await webR.evalRVoid("run()"); +} + +// Expose the agent WebSocket to R on globalThis.__etAgent. R drives it (state, agent_id, send, disconnect) via +// webr::eval_js; the shim creates it, connects, and keeps the latest agent list for R's peer selection. +async function setupAgent() { + const wasmAgent = await import("/modules/et-ws-wasm-agent/et_ws_wasm_agent.js"); + await wasmAgent.default(); + const { WsClient, WsClientConfig } = wasmAgent; + const loc = typeof location !== "undefined" ? location : null; + const wsProto = loc?.protocol === "https:" ? "wss:" : "ws:"; + const wsHost = loc?.host ?? "localhost:8080"; + const wsUrl = globalThis.__ET_WS_URL || `${wsProto}//${wsHost}/ws`; + const client = new WsClient(new WsClientConfig(wsUrl)); + const agent = { + client, + lastAgents: "", + log(msg) { + console.log(`[rcomm1] ${msg}`); + const el = typeof document !== "undefined" ? document.getElementById("module-output") : null; + if (el) el.value = (el.value ? `${el.value}\n` : "") + msg; + }, + }; + client.set_on_message((data) => { + if (typeof data !== "string") return; + let type; + try { + type = JSON.parse(data).type; + } catch { + return; + } + if (type === "et-list-agents-response") { + agent.lastAgents = data; + } else if (type === "et-agent-message" || type === "et-message-status") { + agent.log(`rcomm1: received: ${data}`); + } + }); + client.connect(); + globalThis.__etAgent = agent; +} diff --git a/services/ws-modules/rcomm1/pkg/module.R b/services/ws-modules/rcomm1/pkg/module.R new file mode 100644 index 00000000..f0267951 --- /dev/null +++ b/services/ws-modules/rcomm1/pkg/module.R @@ -0,0 +1,109 @@ +# module.R -- the rcomm1 communication module. run() IS the control loop; the JS shim only boots webR, sets up +# the agent WebSocket transport, and hands control here. +# +# Mirrors comm1 / dart-comm1: connect, discover a peer via et-list-agents, broadcast a message to every other +# agent, then send a direct message to the peer. R drives the JS-side WsClient through webr::eval_js (webR +# cannot open the agent WebSocket itself) and composes every ClientMessage with jsonlite (installed at runtime; +# base R has no JSON tooling). The message payloads match comm1's exactly (module / step / from_agent_id / +# message), and the timing matches comm1's (100 ms connect poll, 1 s list-agents poll, 3 s message pauses). + +# --- transport: drive the shim's WsClient via webr::eval_js (await = TRUE -> main thread) --- +js <- function(code) { + webr::eval_js(code, await = TRUE) +} +agent_state <- function() as.character(js("globalThis.__etAgent.client.get_state()")) +agent_id <- function() as.character(js("globalThis.__etAgent.client.get_agent_id()")) +agent_last_agents <- function() as.character(js("globalThis.__etAgent.lastAgents")) +agent_disconnect <- function() js("globalThis.__etAgent.client.disconnect(); true") +agent_send <- function(message) { + js(sprintf("globalThis.__etAgent.client.send(%s); true", encodeString(message, quote = "\""))) +} +agent_log <- function(msg) { + js(sprintf("globalThis.__etAgent.log(%s); true", encodeString(msg, quote = "\""))) +} +sleep_ms <- function(ms) { + js(sprintf("new Promise(function(resolve){setTimeout(resolve, %d);})", as.integer(ms))) +} + +# --- messages (matching comm1's payload shapes) --- + +# The et-list-agents request carries no fields. +rcomm1_list_agents <- function() { + '{"type":"et-list-agents"}' +} + +# Compose the broadcast ClientMessage (et-broadcast-message). +rcomm1_broadcast <- function(self_id) { + message <- list( + module = "rcomm1", + step = "broadcast", + from_agent_id = self_id, + message = "rcomm1 broadcast to all other connected agents" + ) + as.character(jsonlite::toJSON(list(type = "et-broadcast-message", message = message), auto_unbox = TRUE)) +} + +# Compose the direct ClientMessage (et-send-agent-message) to `peer_id`. +rcomm1_direct <- function(self_id, peer_id) { + message <- list( + module = "rcomm1", + step = "direct", + from_agent_id = self_id, + message = "rcomm1 direct message" + ) + envelope <- list(type = "et-send-agent-message", to_agent_id = peer_id, message = message) + as.character(jsonlite::toJSON(envelope, auto_unbox = TRUE)) +} + +# Pick the target peer from an et-list-agents-response: the first connected agent that is not this agent. +# Returns "" when no peer is present yet, so the poll loop keeps trying (as comm1 / dart-comm1 do). +rcomm1_pick_peer <- function(response_json, self_id) { + parsed <- tryCatch(jsonlite::fromJSON(response_json, simplifyVector = FALSE), error = function(e) NULL) + if (is.null(parsed) || !identical(parsed$type, "et-list-agents-response")) { + return("") + } + for (agent in parsed$agents) { + connected <- identical(agent$state, "connected") + if (connected && !is.null(agent$agent_id) && !identical(agent$agent_id, self_id)) { + return(agent$agent_id) + } + } + "" +} + +# The control loop the JS shim hands control to. +run <- function() { + webr::install("jsonlite") + agent_log("rcomm1: entered run()") + + # Wait for the agent WebSocket to connect and hand us our agent_id. + repeat { + if (identical(agent_state(), "connected")) break + sleep_ms(100) + } + self_id <- "" + repeat { + self_id <- agent_id() + if (nzchar(self_id)) break + sleep_ms(100) + } + agent_log(sprintf("rcomm1: websocket connected with agent_id=%s", self_id)) + + # Poll for a peer, exactly like comm1 -- ask the server for the agent list until R finds a connected peer. + peer <- "" + repeat { + agent_send(rcomm1_list_agents()) + sleep_ms(1000) + peer <- rcomm1_pick_peer(agent_last_agents(), self_id) + if (nzchar(peer)) break + } + agent_log(sprintf("rcomm1: found connected peer agent %s; sending broadcast", peer)) + + agent_send(rcomm1_broadcast(self_id)) + sleep_ms(3000) + agent_log(sprintf("rcomm1: sending direct message to %s", peer)) + agent_send(rcomm1_direct(self_id, peer)) + sleep_ms(3000) + agent_disconnect() + agent_log("rcomm1: workflow complete") +} diff --git a/services/ws-modules/rcomm1/pkg/package.json b/services/ws-modules/rcomm1/pkg/package.json new file mode 100644 index 00000000..96a53bd9 --- /dev/null +++ b/services/ws-modules/rcomm1/pkg/package.json @@ -0,0 +1,8 @@ +{ + "description": "R comm 1", + "license": "Apache-2.0 OR MIT", + "main": "et_ws_rcomm1.js", + "name": "et-ws-rcomm1", + "type": "module", + "version": "0.1.0" +} diff --git a/services/ws-modules/rdata1/pkg/et_ws_rdata1.js b/services/ws-modules/rdata1/pkg/et_ws_rdata1.js new file mode 100644 index 00000000..e9156535 --- /dev/null +++ b/services/ws-modules/rdata1/pkg/et_ws_rdata1.js @@ -0,0 +1,50 @@ +// et_ws_rdata1.js -- bootstrap only. The control loop is module.R's run(); this shim boots webR, sets up the +// agent WebSocket transport that the R code drives via webr::eval_js, and hands control to run(). +// +// webR cannot open the agent WebSocket itself, so the transport (the shared et-ws-wasm-agent WsClient) lives +// here and is exposed on globalThis.__etAgent for R to drive. Everything else -- sequencing, the storage +// round-trip (httr2 over the /websockify relay), verification -- happens in module.R. webR is vendored under +// pkg/webr/ (see build-ws-rdata1-module) and served at the path below. + +const WEBR_BASE_URL = "/modules/et-ws-rdata1/webr/"; +const R_SOURCE_URL = "/modules/et-ws-rdata1/module.R"; + +let webR = null; + +export default async function init() { + const { WebR } = await import(`${WEBR_BASE_URL}webr.mjs`); + webR = new WebR({ baseUrl: WEBR_BASE_URL }); + await webR.init(); + // Cache-bust so edits to module.R are picked up on reload. + const rSource = await (await fetch(`${R_SOURCE_URL}?v=${Date.now()}`)).text(); + await webR.evalRVoid(rSource); +} + +export async function run() { + if (!webR) throw new Error("rdata1: not initialized"); + await setupAgent(); + // Hand control to R -- run() is the control loop. + await webR.evalRVoid("run()"); +} + +// Expose the agent WebSocket to R on globalThis.__etAgent. R drives it (connect state, agent_id, disconnect) +// via webr::eval_js; the shim only creates and connects it. +async function setupAgent() { + const wasmAgent = await import("/modules/et-ws-wasm-agent/et_ws_wasm_agent.js"); + await wasmAgent.default(); + const { WsClient, WsClientConfig } = wasmAgent; + const loc = typeof location !== "undefined" ? location : null; + const wsProto = loc?.protocol === "https:" ? "wss:" : "ws:"; + const wsHost = loc?.host ?? "localhost:8080"; + const wsUrl = globalThis.__ET_WS_URL || `${wsProto}//${wsHost}/ws`; + const client = new WsClient(new WsClientConfig(wsUrl)); + client.connect(); + globalThis.__etAgent = { + client, + log(msg) { + console.log(`[rdata1] ${msg}`); + const el = typeof document !== "undefined" ? document.getElementById("module-output") : null; + if (el) el.value = (el.value ? `${el.value}\n` : "") + msg; + }, + }; +} diff --git a/services/ws-modules/rdata1/pkg/module.R b/services/ws-modules/rdata1/pkg/module.R new file mode 100644 index 00000000..a0fb1ec4 --- /dev/null +++ b/services/ws-modules/rdata1/pkg/module.R @@ -0,0 +1,115 @@ +# module.R -- the rdata1 data module. run() IS the control loop; the JS shim only boots webR, sets up the agent +# WebSocket transport, and hands control here. +# +# run() computes a payload in R and round-trips it through the ws-server storage API entirely in R (httr2), +# tunnelled via the /websockify SOCKS5 relay. The agent WebSocket -- used only to obtain the storage bucket's +# agent_id -- is driven through the JS-side WsClient via webr::eval_js, since webR cannot open it directly. +# httr2 is installed at runtime from the webR package repo; base R (stats/datasets) provides the computation. + +# --- transport: drive the shim's WsClient + browser globals via webr::eval_js (await = TRUE -> main thread) --- +js <- function(code) { + webr::eval_js(code, await = TRUE) +} +agent_state <- function() as.character(js("globalThis.__etAgent.client.get_state()")) +agent_id <- function() as.character(js("globalThis.__etAgent.client.get_agent_id()")) +agent_disconnect <- function() js("globalThis.__etAgent.client.disconnect(); true") +agent_log <- function(msg) { + js(sprintf("globalThis.__etAgent.log(%s); true", encodeString(msg, quote = "\""))) +} +sleep_ms <- function(ms) { + js(sprintf("new Promise(function(resolve){setTimeout(resolve, %d);})", as.integer(ms))) +} + +# Point webR's Emscripten sockets at the /websockify relay so httr2/curl reach the server through it. This runs +# in the webR worker (no await = TRUE), where the SOCKFS filesystem lives. +rdata1_configure_socket <- function(relay_url) { + code <- paste0( + "SOCKFS.websocketArgs = SOCKFS.websocketArgs || {};", + " SOCKFS.websocketArgs.url = ", encodeString(relay_url, quote = "\""), ";", + " SOCKFS.websocketArgs.subprotocol = 'binary'; 0" + ) + webr::eval_js(code) +} + +# Storage object name the payload is written to (application data, so R owns it). +rdata1_filename <- function() { + "rdata1.txt" +} + +# Compute the payload: a linear regression of fuel efficiency on weight + horsepower over the built-in +# `mtcars` dataset, summarised as deterministic CSV so the round-trip comparison is byte-stable. +rdata1_payload <- function() { + model <- lm(mpg ~ wt + hp, data = mtcars) + fit <- summary(model) + rows <- data.frame( + metric = c("observations", "mean_mpg", "sd_mpg", "intercept", "beta_wt", "beta_hp", "r_squared"), + value = round( + c( + nrow(mtcars), + mean(mtcars$mpg), + sd(mtcars$mpg), + coef(model)[["(Intercept)"]], + coef(model)[["wt"]], + coef(model)[["hp"]], + fit$r.squared + ), + 6 + ) + ) + paste(capture.output(write.csv(rows, row.names = FALSE)), collapse = "\n") +} + +# Verify the round-trip. Returns a status line prefixed OK/FAIL so run() can decide success. +rdata1_verify <- function(sent, received) { + if (identical(sent, received)) { + "OK rdata1: VERIFICATION SUCCESS -- storage round-trip preserved the R payload" + } else { + paste0("FAIL rdata1: VERIFICATION FAILURE -- data mismatch\nSent:\n", sent, "\nGot:\n", received) + } +} + +# The control loop the JS shim hands control to. +run <- function() { + webr::install("httr2") + # Route curl through the /websockify relay's SOCKS5 front end (see the relay service). The proxy host is + # nominal -- SOCKFS sends every socket to relay_url -- but the scheme must be socks5h so curl speaks SOCKS5. + Sys.setenv(ALL_PROXY = "socks5h://127.0.0.1:8080") + agent_log("rdata1: entered run()") + + # Point webR's sockets at this origin's /websockify relay (http(s) origin -> ws(s) relay URL). + origin <- as.character(js("location.protocol + '//' + location.host")) + relay_url <- sub("^http", "ws", paste0(origin, "/websockify")) + rdata1_configure_socket(relay_url) + + # Wait for the agent WebSocket to connect and hand us the agent_id (the storage bucket). + repeat { + if (identical(agent_state(), "connected")) break + sleep_ms(100) + } + bucket <- "" + repeat { + bucket <- agent_id() + if (nzchar(bucket)) break + sleep_ms(100) + } + agent_log(sprintf("rdata1: registered as %s", bucket)) + + # Compute in R, then PUT and GET the payload with httr2 (tunnelled to the server via the relay), and verify. + content <- rdata1_payload() + url <- sprintf("http://127.0.0.1:8080/storage/%s/%s", bucket, rdata1_filename()) + httr2::request(url) |> + httr2::req_method("PUT") |> + httr2::req_body_raw(content) |> + httr2::req_perform() + retrieved <- httr2::request(url) |> + httr2::req_perform() |> + httr2::resp_body_string() + + status <- rdata1_verify(content, retrieved) + agent_disconnect() + agent_log(sub("^(OK|FAIL) ", "", status)) + if (startsWith(status, "FAIL")) { + stop("rdata1: verification failed") + } + agent_log("rdata1: workflow complete") +} diff --git a/services/ws-modules/rdata1/pkg/package.json b/services/ws-modules/rdata1/pkg/package.json new file mode 100644 index 00000000..08d47fab --- /dev/null +++ b/services/ws-modules/rdata1/pkg/package.json @@ -0,0 +1,8 @@ +{ + "description": "R data 1", + "license": "Apache-2.0 OR MIT", + "main": "et_ws_rdata1.js", + "name": "et-ws-rdata1", + "type": "module", + "version": "0.1.0" +} diff --git a/services/ws-server/Cargo.toml b/services/ws-server/Cargo.toml index 98a29a41..0a811737 100644 --- a/services/ws-server/Cargo.toml +++ b/services/ws-server/Cargo.toml @@ -27,6 +27,7 @@ edge-toolkit.workspace = true et-modules-service.workspace = true et-otlp.workspace = true et-storage-service.workspace = true +et-websockify-service.workspace = true et-ws-service.workspace = true fs-err.workspace = true futures-util.workspace = true diff --git a/services/ws-server/src/lib.rs b/services/ws-server/src/lib.rs index 43d42be1..eca279ab 100644 --- a/services/ws-server/src/lib.rs +++ b/services/ws-server/src/lib.rs @@ -1,3 +1,5 @@ +use std::net::{Ipv4Addr, SocketAddr}; + use actix_web::{HttpResponse, web}; pub use et_ws_service::{AgentSession, WsAgentRegistry}; @@ -23,6 +25,14 @@ pub fn configure_app(cfg: &mut web::ServiceConfig, agent_registry: web::Data(cfg, &config.storage); + // Relay for browser WASM runtimes (webR) that can only open a WebSocket: bridge /websockify to this + // server's own plain-HTTP loopback port so their libcurl/httr2 can reach the storage API. Loopback-only + // and server-fixed, so it is not an open proxy. Registered before the modules catch-all below. + let relay_target = SocketAddr::from(( + Ipv4Addr::LOCALHOST, + edge_toolkit::ports::Services::InsecureWebSocketServer.port(), + )); + et_websockify_service::configure(cfg, relay_target); // Must be last: registers a catch-all Files::new("/", ...) for the root module. et_modules_service::configure(cfg, &config.modules); } diff --git a/services/ws-web-runner/tests/modules.rs b/services/ws-web-runner/tests/modules.rs index 0caf1046..03fb07fd 100644 --- a/services/ws-web-runner/tests/modules.rs +++ b/services/ws-web-runner/tests/modules.rs @@ -33,6 +33,24 @@ //! workers can spawn their own workers, sharing a //! `CrossIsolateStore>` across isolates for //! the `SharedArrayBuffer` transfer. +//! +//! ### R modules (webR spawns a *classic* Worker) +//! +//! - **et-ws-rdata1, et-ws-rcomm1** -- both run R via webR, whose +//! `WebR.init()` spawns its interpreter on a **classic** Web Worker +//! (`new Worker(url)`, no `{ type: "module" }`). Deno implements only +//! module workers, so that call throws `NotSupportedError: Classic +//! workers are not supported` -- the module aborts in its `init()` +//! default export before any of its own logic (rdata1's httr2 +//! storage round-trip, rcomm1's WebSocket) runs. Unlike the +//! zig-data1 case above (a *module* worker the runner's recursive +//! `CreateWebWorkerCb` supports), there is no runner-side fix: the +//! limitation is Deno's, and webR's classic worker is upstream. These +//! modules work in a real browser (which supports classic workers + +//! the `SharedArrayBuffer` channel the ws-server's COOP/COEP headers +//! enable); `r_module_load_fails` pins the Deno failure so a future +//! runner that gains classic-worker support is caught and the module +//! is promoted to `module_runs_successfully`. #![cfg(test)] @@ -179,6 +197,47 @@ fn hardware_module_load_fails(#[case] module: &str, #[case] language: Language) ); } +/// Load + run each R module and assert it fails under Deno's classic-worker limitation. +/// +/// rdata1 and rcomm1 run R through webR, whose `WebR.init()` spawns the R interpreter on a *classic* Web Worker. +/// Deno implements only module workers, so the spawn throws `NotSupportedError: Classic workers are not +/// supported` and the module aborts in its `init()` default export -- before any WebSocket, peer-discovery, or +/// storage logic runs (so one runner suffices even for the comm-style rcomm1). Running them here still exercises +/// the fetch + module-graph load + entry-evaluation paths up to the failing `new Worker(...)`. We assert a +/// non-zero exit; a module that unexpectedly EXITS 0 is a real finding -- the runner gained classic-worker +/// support, so move it to `module_runs_successfully`. Both modules work in a real browser (classic workers + +/// the `SharedArrayBuffer` channel the ws-server's COOP/COEP headers enable). +#[rstest] +#[case::rdata1("et-ws-rdata1", Language::R)] +#[case::rcomm1("et-ws-rcomm1", Language::R)] +fn r_module_load_fails(#[case] module: &str, #[case] language: Language) { + if !mise_env_includes(language) { + println!( + "skipping {module}: requires the `{}` mise env, not loaded", + language.as_str() + ); + return; + } + let server = et_ws_test_server::start(); + // webR's classic-worker spawn throws promptly in init(), so the module exits well under this bound; the + // budget only bites if webR instead hangs before that point, in which case RUNNER_TIMEOUT kills it -- still + // the non-zero exit we assert. + let output = run_runner(module, &server.ws_url, 30); + #[cfg(feature = "coverage")] + collect_module_coverage(&server); + assert!( + !output.status.success(), + concat!( + "{} exited 0, but it was expected to fail: webR spawns a classic Worker, which Deno does not ", + "support. If the runner can now run it, move it to `module_runs_successfully`.\n", + "--- stdout ---\n{}\n--- stderr ---\n{}", + ), + module, + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); +} + /// Spawn one `et-ws-web-runner` against `ws_url` and return its captured output. /// /// `timeout_secs` is passed as `RUNNER_TIMEOUT` (humantime, e.g. `120s`); the From 9848cf116c1862970f836d1e83efbd3f9a56185e Mon Sep 17 00:00:00 2001 From: John Vandenberg Date: Tue, 21 Jul 2026 11:08:56 +0800 Subject: [PATCH 2/2] docstring tidy --- config/ast-grep/rules/doc-summary-ends-with-period.yaml | 4 ++++ services/ws-web-runner/tests/modules.rs | 2 ++ 2 files changed, 6 insertions(+) diff --git a/config/ast-grep/rules/doc-summary-ends-with-period.yaml b/config/ast-grep/rules/doc-summary-ends-with-period.yaml index 790d25d8..02d82b8f 100644 --- a/config/ast-grep/rules/doc-summary-ends-with-period.yaml +++ b/config/ast-grep/rules/doc-summary-ends-with-period.yaml @@ -28,11 +28,15 @@ files: - libs/edge-toolkit/tests/no_mise.rs - libs/test-helpers/src/lib.rs - libs/ws-runner-common/tests/config.rs + - services/websockify/src/lib.rs + - services/websockify/tests/relay.rs - services/ws-pyo3-runner/tests/modules.rs + - services/ws-server/src/lib.rs - services/ws-wasi-runner/tests/modules.rs - services/ws-wasi-runner/tests/otel_propagation.rs - services/ws-wasi-runner/tests/vector_otlp_relay.rs - services/ws-web-runner/build.rs + - services/ws-web-runner/tests/modules.rs - services/ws/tests/config.rs - utilities/int-gen/src/error.rs - utilities/int-gen/src/lib.rs diff --git a/services/ws-web-runner/tests/modules.rs b/services/ws-web-runner/tests/modules.rs index 03fb07fd..883b81ea 100644 --- a/services/ws-web-runner/tests/modules.rs +++ b/services/ws-web-runner/tests/modules.rs @@ -90,6 +90,8 @@ fn module_runs_successfully(#[case] module: &str, #[case] language: Language) { collect_module_coverage(&server); } +/// Probe for dotnet-data1's built `pkg/` wasm artifacts, logging a skip instead of failing when absent. +/// /// dotnet-data1's `pkg/` wasm artifacts only exist after `build-ws-dotnet-data1-module` has run on this /// host, so probe one and log a skip instead of failing on a checkout where the module wasn't built. The /// probe file is `dotnet.js` -- the one stably-named artifact `dotnet publish` emits (the rest carry