allthingssecurity commented on code in PR #27678:
URL: https://github.com/apache/camel/pull/27678#discussion_r4249690190
##########
components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmProducer.java:
##########
@@ -16,76 +16,404 @@
*/
package org.apache.camel.component.wasm;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
-import org.apache.camel.Endpoint;
+import org.apache.camel.CamelExchangeException;
import org.apache.camel.Exchange;
-import org.apache.camel.spi.Resource;
-import org.apache.camel.spi.ResourceLoader;
+import org.apache.camel.ExchangeTimedOutException;
+import org.apache.camel.Message;
+import org.apache.camel.StreamCache;
import org.apache.camel.support.DefaultProducer;
-import org.apache.camel.support.PluginHelper;
-import org.apache.camel.wasm.WasmFunction;
-import org.apache.camel.wasm.WasmSupport;
+import org.apache.camel.support.ResourceHelper;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.URISupport;
+import org.apache.camel.wasm.WasiCommand;
+import org.apache.camel.wasm.Wasm;
+import run.endive.runtime.TrapException;
import run.endive.wasm.Parser;
import run.endive.wasm.WasmModule;
+/**
+ * Runs a WASI Preview 1 command module once per exchange: the message body is
the standard input of the module and its
+ * standard output becomes the message body.
+ * <p/>
+ * The module is parsed once, when the producer is initialized, and every
exchange runs on a new instance of it, so
+ * exchanges never share guest memory or globals and need no lock. The
instance is dropped after its run, whatever the
+ * outcome.
+ * <p/>
+ * Without a timeout the module runs on the calling thread and reads the body
as a stream. With a timeout the module
+ * runs on a thread of the producer, and the body is first read into memory on
the calling thread: the module then reads
+ * a private copy, so a module that is abandoned after a timeout never reads a
stream that the route is still using.
+ * <p/>
+ * When the exchange fails, the producer does not set a new body, but the
module may have read the old one: a body that
+ * is a stream has been read, partly or to its end. With stream caching (the
default) the body can be read again: Camel
+ * resets a {@link StreamCache}. When a timeout is set, a body that was a
stream but not a StreamCache is replaced by
+ * the bytes that were read from it, so that error handling still sees the
content.
+ */
public class WasmProducer extends DefaultProducer {
- private final String functionModule;
- private final String functionName;
+ private static final int MAX_STDERR_IN_MESSAGE = 256;
+ private static final String DEFAULT_PROGRAM_NAME = "module.wasm";
- private WasmModule module;
- private WasmFunction function;
+ private WasiCommand command;
+ private Map<String, String> environment = Collections.emptyMap();
+ private List<String> environmentHeaders = Collections.emptyList();
+ private volatile ExecutorService executor;
- public WasmProducer(Endpoint endpoint, String functionModule, String
functionName) throws Exception {
+ public WasmProducer(WasmEndpoint endpoint) {
super(endpoint);
+ }
- this.functionModule = functionModule;
- this.functionName = functionName;
+ @Override
+ public WasmEndpoint getEndpoint() {
+ return (WasmEndpoint) super.getEndpoint();
}
@Override
- public void doInit() throws Exception {
- final ResourceLoader rl =
PluginHelper.getResourceLoader(getEndpoint().getCamelContext());
- final Resource res = rl.resolveResource(this.functionModule);
+ protected void doInit() throws Exception {
+ super.doInit();
- try (InputStream is = res.getInputStream()) {
- this.module = Parser.parse(is);
+ WasmEndpoint endpoint = getEndpoint();
+ WasmConfiguration conf = endpoint.getConfiguration();
+
+ Map<String, String> env = new LinkedHashMap<>();
+ if (conf.getEnvironment() != null) {
+ for (Map.Entry<String, Object> e :
conf.getEnvironment().entrySet()) {
+ String value = e.getValue() != null ? e.getValue().toString()
: "";
+ env.put(checkName(e.getKey()), checkNoNul("environment
variable " + e.getKey(), value));
+ }
+ }
+ List<String> headers = new ArrayList<>();
+ if (ObjectHelper.isNotEmpty(conf.getEnvironmentHeaders())) {
+ for (String header : conf.getEnvironmentHeaders().split(",")) {
+ String name = checkName(header.trim());
+ if (env.containsKey(name)) {
+ throw new IllegalArgumentException(
+ "Environment variable " + name + " is configured
both in environment and environmentHeaders");
+ }
+ headers.add(name);
+ }
}
+ List<String> arguments = arguments(programName(endpoint.getModule()),
conf.getArgs());
+ for (String argument : arguments) {
+ checkNoNul("argument " + argument, argument);
+ }
+
+ WasmModule module;
+ try (InputStream is =
ResourceHelper.resolveMandatoryResourceAsInputStream(
+ endpoint.getCamelContext(), endpoint.getModule())) {
+ module = Parser.parse(is);
+ }
+
+ this.environment = Collections.unmodifiableMap(env);
+ this.environmentHeaders = List.copyOf(headers);
+ this.command = new WasiCommand(module, arguments,
conf.getMaxMemoryPages());
}
@Override
- public void doStart() throws Exception {
+ protected void doStart() throws Exception {
super.doStart();
- if (this.module != null && this.function == null) {
- this.function = new WasmFunction(this.module, this.functionName);
+ if (getEndpoint().getConfiguration().getTimeout() > 0 && executor ==
null) {
+ executor =
getEndpoint().getCamelContext().getExecutorServiceManager()
+ .newCachedThreadPool(this, "Wasm");
}
}
@Override
- public void doStop() throws Exception {
+ protected void doStop() throws Exception {
+ ExecutorService pool = executor;
+ executor = null;
+ if (pool != null) {
+ // interrupts the modules that are still running
+
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(pool);
+ }
+
super.doStop();
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ WasmConfiguration conf = getEndpoint().getConfiguration();
+ Message message = exchange.getMessage();
- if (this.function != null) {
- this.function.close();
+ // the result headers only ever describe this run: never a previous
one, nor a value sent with the message
+ message.removeHeader(Wasm.Headers.EXIT_CODE);
+ message.removeHeader(Wasm.Headers.STDERR);
+
+ Map<String, String> env = environment(message);
+ // no body is empty standard input; a body that cannot be read as a
stream fails the exchange instead of
+ // silently running the module with empty input
+ Object body = message.getBody();
+ InputStream in = body != null ?
message.getMandatoryBody(InputStream.class) : InputStream.nullInputStream();
+ byte[] restore = null;
+ if (conf.getTimeout() > 0 && body != null) {
+ // with a timeout the module runs on another thread, which goes on
for a moment after the timeout and is
+ // not stopped by an interrupt while it is blocked in a read of
the body. So the body is read here, on the
+ // calling thread, and the module gets a copy that no other thread
uses and that never blocks.
+ byte[] input = in.readAllBytes();
+ in = new ByteArrayInputStream(input);
Review Comment:
Fixed in e4309cfcd7a7. Checked first: for a `File`, `Path` and file-consumer
`GenericFile` body
`getMandatoryBody(InputStream.class)` opens a new file stream on every call
(`IOHelper.toInputStream`,
`GenericFileConverter.genericFileToInputStream`), and nothing closed it. The
producer now closes a stream that the
converter opened: right after `readAllBytes()` when a timeout is set (in a
`finally`), and in the outer `finally`
otherwise. Without a timeout the module runs on the calling thread (`run()`
calls `task.call()` directly), so it has
ended when the `finally` runs and no other thread reads the stream.
One difference from the suggestion: `in != body` alone is not enough to know
that the converter created the stream.
For a `StreamSource` body `XmlConverter.toInputStream(StreamSource)` returns
the stream the body holds (also for a
`StreamSourceCache`, which is a `StreamCache` but not an `InputStream`), and
closing it would close the caller's
stream. So the stream is closed when `in != body` and the body is neither a
`StreamCache` nor an XML `Source`.
With `in != body` only, the 4 runs of the new `streamOfTheCallerIsNotClosed`
fail.
Tests: `streamOpenedByTheTypeConverterIsClosed` (a body type whose converter
opens a recording stream; without and
with a timeout, on success and on a non-zero exit: exactly one stream,
closed), `fileBodyIsStandardInputAndCanBeDeletedAfterwards`
(a temp `File` body, without and with a timeout),
`streamOfTheCallerIsNotClosed` (a stream body and a `StreamSource`
body, same four cases). Without the fix the 4 runs of
`streamOpenedByTheTypeConverterIsClosed` fail. camel-wasm with
`install`: 65 tests, 0 failures (was 55). The PR description is updated.
_Claude Code on behalf of allthingssecurity_
##########
components/camel-wasm/src/main/docs/wasm-component.adoc:
##########
@@ -43,133 +58,169 @@ include::partial$component-endpoint-headers.adoc[]
== Usage
-=== Writing A Wasm processor
-
-In _Wasm_, sharing objects between the host, in this case the _JVM_, and the
_Wasm_ module is deliberately restricted and as of today, it requires a number
of steps:
-
-1. From the _host_, call a function inside the webassembly module that
allocates a block of memory and returns its address, then save it
-2. From the _host_, write the data that should be exchanged with the _Wasm_
module to the saved address
-3. From the _host_, invoke the required function passing both the address
where the data is written and its size
-4. From the _Wasm_ module, read the data and process it
-5. From the _host_, release the memory when done
+=== Writing a module
-==== Providing functions for memory management
-
-The module hosting the function *must* provide the functions to
allocate/deallocate memory that *must* be named `alloc` and `dealloc`
respectively.
-
-Here's an example of the mentioned functions implemented in
https://www.rust-lang.org[Rust]:
+A module is an ordinary program that reads its standard input and writes its
standard output. For example, in
+https://www.rust-lang.org[Rust]:
[source,rust]
----
-pub extern "C" fn alloc(size: u32) -> *mut u8 {
- let mut buf = Vec::with_capacity(size as usize);
- let ptr = buf.as_mut_ptr();
-
- // tell Rust not to clean this up
- mem::forget(buf);
+use std::io::{self, Read, Write};
- ptr
-}
-
-pub unsafe extern "C" fn dealloc(ptr: &mut u8, len: i32) {
- // Retakes the pointer which allows its memory to be freed.
- let _ = Vec::from_raw_parts(ptr, 0, len as usize);
+fn main() {
+ let mut input = String::new();
+ io::stdin().read_to_string(&mut input).unwrap();
+ io::stdout().write_all(input.to_uppercase().as_bytes()).unwrap();
}
----
-==== Data shapes
+Built with `rustup target add wasm32-wasip1` and `cargo build --release
--target wasm32-wasip1`.
-It is not possible to share a Java object with the Wasm module directly, and
as mentioned before,
-data exchange leverages Wasm's memory that can be accessed by both the host
and the guest runtimes.
-At this stage, the data structure that the component exchange with the Wasm
function is a subset of the Apache Camel Message,
-containing headers the body encoded as a base64 string:
+The same program in https://go.dev[Go]:
-._Java-only: data exchange wrapper class for Wasm module communication_
-[source,java]
+[source,go]
----
-public static class Wrapper {
- @JsonProperty
- public Map<String, String> headers = new HashMap<>();
-
- @JsonProperty
- public byte[] body;
-}
-----
-
-==== Data processing
-
-The component expects the processing function to have the following signature:
-
-[source,rust]
-----
-fn function(ptr: u32, len: u32) -> u64
-----
-
-* it accepts two 32bit unsigned integers arguments
-** a pointer to the memory location when the input data has been written
(`ptr`)
-** the size of the input data (`len`)
-* it returns a 64bit unsigned integer where:
-** the first 32bit represents a pointer to the return data
-** the last 31bit represents the size of the return data, which must have the
same data shape discussed above unless there is an error (see below)
-** the most significant bit of the returned data size is reserved to signal an
error, so if it is set, then the return data could contain an error
message/code/etc
-
-Here's an example of a complete function:
-
-[source,rust]
-----
-#[derive(Serialize, Deserialize)]
-struct Message {
- headers: HashMap<String, serde_json::Value>,
-
- #[serde(with = "Base64Standard")]
- body: Vec<u8>,
-}
-
-#[cfg_attr(all(target_arch = "wasm32"), export_name = "process")]
-#[no_mangle]
-pub extern fn process(ptr: u32, len: u32) -> u64 {
- let bytes = unsafe {
- slice::from_raw_parts_mut(
- ptr as *mut u8,
- len as usize)
- };
+package main
- let mut msg: Message = serde_json::from_slice(bytes).unwrap();
+import (
+ "bytes"
+ "io"
+ "os"
+)
-
- let out_vec = serde_json::to_vec(&msg).unwrap();
- let out_len = out_vec.len();
- let out_ptr = alloc(out_len as u32);
-
- unsafe {
- std::ptr::copy_nonoverlapping(
- out_vec.as_ptr(),
- out_ptr,
- out_len as usize)
- };
-
- return ((out_ptr as u64) << 32) | out_len as u64;
+func main() {
+ input, _ := io.ReadAll(os.Stdin)
+ os.Stdout.Write(bytes.ToUpper(input))
}
----
-==== Errors and module state
-
-The module is instantiated once per producer (each `to` of a `wasm` endpoint
in a route has its own instance) and that
-instance handles every message, one at a time, so data the module keeps in its
memory or globals carries over from one
-message to the next.
-
-* When the function returns with the error bit set, the error data is released
with `dealloc` using its size without
-the error bit, the error data becomes the message of the exception, and the
instance is kept.
-* When the call fails without returning, for example on a Rust `panic` or any
other Wasm trap, the instance is
-discarded and a new one is created from the module for the next message.
Whatever the failed call allocated can no
-longer be released, so keeping the instance would grow its memory with every
failure. Any state the module kept in
-the discarded instance is lost.
-On a route where every call fails this way, the module is instantiated again
for every message (its data segments
-are copied and any start function runs), which costs more than a call on a
kept instance.
+Built with `GOOS=wasip1 GOARCH=wasm go build -o upper.wasm`.
+
+=== Running a module
+
+For every exchange the component:
+
+. creates a new instance of the module (the module is parsed once, when the
producer starts),
+. runs its `_start` function with the message body as standard input, the
arguments configured with `args` and
+the environment variables configured with `environment` and
`environmentHeaders`,
+. sets the `CamelWasmExitCode` header to the exit code of the program: 0 when
`_start` returns, otherwise the code
+the program passed to `proc_exit`,
+. sets the `CamelWasmStderr` header to what the program wrote to its standard
error, if anything,
+. and replaces the message body with the standard output of the program, as a
`byte[]`.
+
+When the program exits with a non-zero code, the exchange fails with a
`WasmExitCodeException` (which holds the exit
+code and the standard error) and the standard output is not set as the body,
unless `failOnNonZeroExit` is `false`:
+then the standard output replaces the body as for any other exit code. The
other headers of the message are kept.
+See <<the-message-body-after-a-failure>> for what the body holds when the
exchange fails.
+
+Every exchange runs on its own instance, so exchanges never share the memory
or the globals of the module and can
+run in parallel without a lock. A module cannot keep state from one message to
the next.
+
+The `CamelWasmExitCode` and `CamelWasmStderr` headers always describe the
current run: a value already on the
+message is removed before the module runs.
+
+=== What the module can access
+
+* *Standard input, output and error*: the message body, the new body and the
`CamelWasmStderr` header. They are
+not terminals.
+* *Arguments*: the file name of the module as argument zero (`upper.wasm` for
`wasm:file:/opt/modules/upper.wasm`;
+the directories, the host, user info and query of the location are not passed
on), followed by the `args` option, for
+example `args=--level 2`. Use double quotes around an argument that contains
spaces. Arguments are fixed by the route.
+* *Environment variables*: only the ones configured with
`environment.NAME=value`, and the message headers listed in
+`environmentHeaders`, passed under the name they are listed with. Header names
are matched the way Camel matches
+them, ignoring case: `environmentHeaders=TENANT` also passes a header named
`tenant`, as the variable `TENANT`. A
+listed header that is not on the message is not set. Headers usually come from
whoever sent the message, so no
+header reaches the module unless it is listed. The environment of the JVM is
never passed on.
+* *Files and network*: none. The module gets no preopened directories, so it
cannot open files, and no sockets.
+* *Clock*: the one of the host.
+* *Random numbers*: WASI `random_get` (used by Go `crypto/rand` and Rust
`getrandom`) returns bytes from a
+`java.security.SecureRandom` of the host.
+
+=== Limits
+
+The following options bound what one run of a module can use. They are off by
default.
+
+* `maxMemoryPages`: the maximum size of the memory of the module, in pages of
64 KiB. Rust and Go programs declare
+no maximum by default, so the memory of one run can grow to gigabytes of JVM
heap. When the limit is reached, growing
+the memory fails inside the module, which then usually traps or exits with an
error.
+* `timeout`: the maximum time one run may take. The module then runs on a
thread of the producer while the route
+thread waits. When the time is up, the module is interrupted at its next
function call or loop iteration and the
+exchange fails with an `ExchangeTimedOutException`. Without a `timeout`, a
module that never ends blocks the route
+thread.
+* `maxOutputSize`: the maximum number of bytes the module may write to its
standard output, and separately to its
+standard error. Both are buffered in memory. A module that writes more is
stopped and the exchange fails.
+
+WARNING: Without `maxOutputSize` the standard output and the standard error
are buffered without a limit: a module
+that keeps writing fills the heap of the JVM, which affects the whole
application and not only one exchange. Set
+`maxOutputSize` unless the output size of the module is known and small, for
example `maxOutputSize=16777216`
+(16 MiB).
+
+With a `timeout`, the message body is read into memory, on the route thread,
before the module starts, and the module
+reads that copy. Without a `timeout` the body is streamed to the module. The
copy is what makes the timeout safe: the
+thread of a module that timed out can run for a moment longer, and it must not
read a stream that the route is using
+again. It also means that:
+
+* the whole body is on the heap for the time of the run, in addition to the
buffered output,
+* the time to read the body is not part of the `timeout`: a body stream that
delivers slowly, such as a slow network
+connection, holds the route thread as it does without a `timeout`,
+* the threads of the producer are never blocked on the body, so a module that
timed out gives its thread back at its
+next function call or loop iteration. The pool of these threads has no maximum
size and needs none: it holds one
+thread for each exchange that is in the endpoint at the same time.
+
+These limits are robustness controls, not a sandbox. A module is trusted code
chosen by the route author, like a
+processor or a bean, as described in the
xref:manual::security-model.adoc[security model]: the limits keep a
+misbehaving module from turning one message into an unbounded use of memory,
time or threads, but they are not a
+defence against a module written to attack the host.
+
+=== Errors
+
+[width="100%",cols="2,4",options="header"]
+|===
+| What happens | Result
+| The program exits with code 0 | The body is the standard output.
`CamelWasmExitCode` is 0.
+| The program exits with another code | A `WasmExitCodeException`, no new
body, `CamelWasmExitCode` and
+`CamelWasmStderr` set. With `failOnNonZeroExit=false`: the body is the
standard output.
+| The program traps, for example on a Rust `panic` | A
`CamelExchangeException` caused by the `TrapException`
+of Endive, with the start of the standard error in its message.
`CamelWasmStderr` is set, `CamelWasmExitCode` is not.
+| The program writes more than `maxOutputSize` | A `CamelExchangeException`.
`CamelWasmStderr` is set when the program
+wrote to its standard error, `CamelWasmExitCode` is not.
+| The program runs longer than `timeout` | An `ExchangeTimedOutException`.
Neither header is set.
+| The module is not a WASI Preview 1 command module, or the options are
invalid | The route fails to start.
+|===
+
+In every case, the instance the program ran on is discarded.
+
+[[the-message-body-after-a-failure]]
Review Comment:
Done in e4309cfcd7a7: the `[[the-message-body-after-a-failure]]` line is
removed and the link at line 114 is
`<<_the_message_body_after_a_failure>>`, in `wasm-component.adoc` and in the
catalog copy (the two files are
identical).
_Claude Code on behalf of allthingssecurity_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
