diff --git a/.cargo/audit.toml b/.cargo/audit.toml index 591d3b5..5132935 100644 --- a/.cargo/audit.toml +++ b/.cargo/audit.toml @@ -8,7 +8,7 @@ ignore = [ # client (core, connectors/amqp, connectors/mq); connectors/amqp also uses # rustls 0.22 directly for amqps://. nats 0.25 is the last release of the # sync client, so this line gets no patched version; moving to async-nats - # removes all six. Ignores go by advisory ID, so these would also hide + # removes all seven. Ignores go by advisory ID, so these would also hide # 0346/0349 on rustls 0.23: keep core on >= 0.23.46. # None is reachable the way the crates are used here: @@ -28,6 +28,13 @@ ignore = [ # not require TLS stays plaintext). "RUSTSEC-2026-0098", "RUSTSEC-2026-0099", + # Certificate paths through the distrusted emSign intermediates: rustls-webpki + # 0.102 has no distrust list. Reachable only on a tls:// NATS_URL, from a + # network position that can impersonate the NATS server, with a chain the + # machine's OS trust store accepts (nats 0.25 loads the native roots). A NATS + # server on a plain or private-CA connection is outside it. This ID also + # covers rustls-webpki 0.103: keep core on >= 0.103.16, webpki-roots >= 1.0.10. + "RUSTSEC-2026-0382", # idna 0.1.5 accepts Punycode labels that decode to plain ASCII. It comes # through url 1.7 <- amq-protocol 1.4 <- amiquip 0.4.2 (the last amiquip diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 07452a2..58a8671 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -90,6 +90,12 @@ jobs: run: e2e/api-write-gate.sh - name: Governed mode gates every write (ADR-0024/R7 — Finding B regression) run: e2e/governed-gate.sh + - name: Secrets stay write-only in shadow runs (ADR-0008, S4) + run: e2e/governed-agent.sh + - name: An emit subject cannot carry protocol + run: e2e/emit-subject.sh + - name: http-in survives a burst and a thread limit + run: e2e/http-in-bound.sh - name: Docs book builds (broken includes fail here, not on deploy) run: | curl -sL https://github.com/rust-lang/mdBook/releases/download/v0.5.4/mdbook-v0.5.4-x86_64-unknown-linux-gnu.tar.gz | tar xz diff --git a/CHANGELOG.md b/CHANGELOG.md index 330037f..81fe622 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,61 @@ is `0`, minor versions may carry breaking changes — they are called out here. ## [Unreleased] +### Security +- **An event field in an `emit` subject could inject NATS protocol.** The subject + reached the wire as it was, so a line break in a field of an event turned the + rest of that field into protocol commands, within the permissions of the + runtime's NATS user. Through `http-in` this needed no credential: a flow that + builds a subject from an event field let an anonymous POST publish anywhere or + purge a stream. No shipped flow or template does this; check the flows of your + own that do. `emit` now refuses a subject with whitespace or a control + character: the run fails and the message reaches the DLQ. `http-in` answers 400 + on such a suffix. +- **Secret values left through every path that shows a run.** In v0.3.6 + preview, `vejas_run_flow`, flow-as-tool, shadow-replay, time-travel, canary + and golden capture evaluated flows with the real secret store and returned + the result. `GET /preview`, which needs no token, returned any variable + holding a secret. In governed mode, `vejas_time_travel` handed the value to an + agent holding only `VEJAS_TOKEN`, through a candidate that emits it. These + paths now evaluate `secret()` to the placeholder ``, + fail-closed as before, and a real run still gets the value. Rotate any secret + that a shared or agent-facing deployment could have exposed this way. A branch + comparing with a secret now takes the not-equal path in shadow. +- **rustls-webpki 0.103.15 → 0.103.16 and webpki-roots 1.0.9 → 1.0.11**, for + RUSTSEC-2026-0382 and -0381 (certificate paths through distrusted emSign + intermediates). The TLS stack of the runtime's HTTP client and of MQTT over + TLS rejects them again. The sync `nats` client keeps rustls-webpki 0.102 and + its OS trust store, so -0382 stays ignored there, with the reason in + `.cargo/audit.toml`, until the move to async-nats. + +### Fixed +- **A burst of connections could silently kill `http-in`.** When the OS refused + a thread (a pids limit, memory), `thread::spawn` panicked in the accept loop: + the connector stopped for good, while `/healthz` said 200 and the topology said + `running`, `restarts: 0`, `last_error: null`. Reproduced on the v0.3.6 image + with `--pids-limit=64` and 300 idle connections. `http-in` now holds at most + `MAX_CONNECTIONS` (default 512) and answers 503 past it, gives a request 30 s + to arrive, and drops one connection when no thread is available. The panel + port has the same bound (`VEJAS_HTTP_MAX_INFLIGHT`, default 256). Refusals + count in `vejas_http_rejected_total`. +- **A panic in a unit no longer ends its supervision.** It takes the path of + any other failure: `restarting`, the message in `last_error`, backoff, another + attempt. Supervisor threads carry the unit's name, so a panic in the log says + which unit it was. +- **The panel could not approve when `VEJAS_TOKEN` was set.** Approve and Reject + got a 401 from the write gate, because the panel holds the approval token and + never the agent's. A valid `X-Approval-Token` now answers the write gate on + the routes that belong to the human. + +### Changed +- **Governed mode: replaying and discarding dead letters are human acts.** + `vejas_dlq_replay` and `vejas_dlq_purge` are refused, and `POST /dlq/replay` + and `/dlq/purge` answer 409 without `X-Approval-Token`. They used to accept + `VEJAS_TOKEN`, which the agent holds. The last step of the self-healing loop + moves to the panel in governed mode (ADR-0024 Amendment A). Outside governed + mode nothing changes. +- Replays and purges are now recorded in `VEJAS_AUDIT`, as the guide already said. + ### Corrected - The production line (README, vejas.dev, launch kit) claimed two production customers collecting NIS2 supplier-compliance evidence across eight EU diff --git a/connectors/amqp/Cargo.lock b/connectors/amqp/Cargo.lock index 91d5292..e3dd235 100644 --- a/connectors/amqp/Cargo.lock +++ b/connectors/amqp/Cargo.lock @@ -1647,14 +1647,14 @@ version = "0.26.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" dependencies = [ - "webpki-roots 1.0.9", + "webpki-roots 1.0.11", ] [[package]] name = "webpki-roots" -version = "1.0.9" +version = "1.0.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a" +checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171" dependencies = [ "rustls-pki-types", ] diff --git a/core/Cargo.lock b/core/Cargo.lock index a5e379c..326f87f 100644 --- a/core/Cargo.lock +++ b/core/Cargo.lock @@ -1345,7 +1345,7 @@ dependencies = [ "once_cell", "ring", "rustls-pki-types", - "rustls-webpki 0.103.15", + "rustls-webpki 0.103.16", "subtle", "zeroize", ] @@ -1394,9 +1394,9 @@ dependencies = [ [[package]] name = "rustls-webpki" -version = "0.103.15" +version = "0.103.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" +checksum = "4443b1b6d39d7c663139942cc4558262562d2c84123bb6282ef177009ca8b281" dependencies = [ "ring", "rustls-pki-types", @@ -2051,14 +2051,14 @@ version = "0.26.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" dependencies = [ - "webpki-roots 1.0.9", + "webpki-roots 1.0.11", ] [[package]] name = "webpki-roots" -version = "1.0.9" +version = "1.0.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a" +checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171" dependencies = [ "rustls-pki-types", ] diff --git a/core/src/connectors.rs b/core/src/connectors.rs index b6a0328..2d2fde7 100644 --- a/core/src/connectors.rs +++ b/core/src/connectors.rs @@ -15,8 +15,8 @@ // bus. These native drivers are the batteries included. use std::io::{Read, Write}; -use std::net::TcpListener; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::net::{TcpListener, TcpStream}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; use std::thread; use std::time::{Duration, Instant}; @@ -188,10 +188,15 @@ impl Driver for HttpIn { "source:webhook" } fn about(&self) -> &'static str { - "HTTP webhook: POST /ingest/ publishes the JSON body on vx.. Config: PORT, ALLOW (optional list of permitted subject suffixes/prefixes — this port is unauthenticated, so restrict it to your ingest subjects; without ALLOW any vx.* subject is reachable)." + "HTTP webhook: POST /ingest/ publishes the JSON body on vx.. Config: PORT, MAX_CONNECTIONS (default 512, past it a 503), ALLOW (optional list of permitted subject suffixes/prefixes — this port is unauthenticated, so restrict it to your ingest subjects; without ALLOW any vx.* subject is reachable)." } fn run(&self, ctx: &Ctx) -> Result<(), String> { let port = ctx.config.u64_or("PORT", 8787) as u16; + // One OS thread per connection on an unauthenticated port: bounded, so a + // burst costs a 503 rather than the thread budget of the whole runtime. + let max_connections = ctx.config.u64_or("MAX_CONNECTIONS", 512).max(1) as usize; + let open = Arc::new(AtomicUsize::new(0)); + let mut refused = 0u64; // nc for publish_confirmed (direct flush past the 5ms flusher floor — // measured: /ingest p50 5.2ms→0.77ms, no throughput regression at conc=32); // _js keeps the stream ensured. @@ -220,11 +225,40 @@ impl Driver for HttpIn { } match s { Ok(mut sock) => { + let Some(slot) = Slot::try_take(&open, max_connections) else { + crate::metrics::inc_rejected(&ctx.name); + refused += 1; + if refused.is_power_of_two() { + eprintln!( + "[{}] {refused} connection(s) refused so far: MAX_CONNECTIONS is {max_connections}", + ctx.name + ); + } + refuse(sock); + continue; + }; let nc = nc.clone(); let root = ctx.subj_root.clone(); let name = ctx.name.clone(); let allow = allow.clone(); - thread::spawn(move || handle_http_in(&name, &mut sock, &nc, &root, &allow)); + // `thread::spawn` panics when the OS has no thread to give (a + // pids limit, memory), which here killed the connector and left + // it reported as running. The closure drops with the error, + // which closes this one connection and frees its slot. + let spawned = thread::Builder::new().name(format!("{name}:conn")).spawn( + move || { + let _slot = slot; + handle_http_in(&name, &mut sock, &nc, &root, &allow) + }, + ); + if let Err(e) = spawned { + crate::metrics::inc_rejected(&ctx.name); + refused += 1; + if refused.is_power_of_two() { + eprintln!("[{}] {refused} connection(s) refused so far: {e}", ctx.name); + } + thread::sleep(Duration::from_millis(20)); + } } Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { // Poll responsively (stop is re-checked each turn) without a @@ -247,6 +281,36 @@ fn suffix_allowed(suffix: &str, allow: &[String]) -> bool { .any(|e| suffix == e || suffix.starts_with(&format!("{e}."))) } +/// One place in a bounded pool of per-connection or per-request threads, given +/// back when the thread ends, a panic included. +pub struct Slot(Arc); + +impl Slot { + pub fn try_take(open: &Arc, max: usize) -> Option { + open.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| (n < max).then_some(n + 1)) + .ok() + .map(|_| Slot(open.clone())) + } +} + +impl Drop for Slot { + fn drop(&mut self) { + self.0.fetch_sub(1, Ordering::SeqCst); + } +} + +/// Past the bound: tell the client, without a thread, then close. +fn refuse(mut sock: TcpStream) { + sock.set_write_timeout(Some(Duration::from_millis(200))).ok(); + let _ = sock.write_all( + b"HTTP/1.1 503 Service Unavailable\r\nRetry-After: 1\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + ); +} + +/// How long one request may take to arrive. The 5 s read timeout restarts on +/// every byte, so without this a client trickling bytes keeps its slot for good. +const REQUEST_READ_LIMIT: Duration = Duration::from_secs(30); + fn handle_http_in( name: &str, sock: &mut std::net::TcpStream, @@ -268,7 +332,12 @@ fn handle_http_in( let mut tmp = [0u8; 4096]; let mut header_end = None; let mut eof = false; + let deadline = Instant::now() + REQUEST_READ_LIMIT; loop { + if Instant::now() >= deadline { + eof = true; + break; + } match sock.read(&mut tmp) { Ok(0) => { eof = true; @@ -314,6 +383,8 @@ fn handle_http_in( let suffix = path.trim_start_matches("/ingest/").trim_matches('/'); if suffix.is_empty() { ("400 Bad Request", "{\"error\":\"missing subject suffix\"}".to_string()) + } else if !crate::vjs::valid_subject(suffix) { + ("400 Bad Request", "{\"error\":\"invalid subject suffix\"}".to_string()) } else if serde_json::from_slice::(body).is_err() { ("400 Bad Request", "{\"error\":\"body must be JSON\"}".to_string()) } else if !allow.is_empty() && !suffix_allowed(suffix, allow) { diff --git a/core/src/main.rs b/core/src/main.rs index c0a8690..1092efd 100644 --- a/core/src/main.rs +++ b/core/src/main.rs @@ -265,7 +265,7 @@ fn replay_literal( }; let pkg = pkg_of_path(&path); let run_one = |prog: &vjs::Program, ev: &Value| -> Value { - let mut engine = vjs::Engine::new(root.to_path_buf(), pkg.clone()); + let mut engine = vjs::Engine::shadow(root.to_path_buf(), pkg.clone()); match vjs::run(prog, ev, &mut engine) { Ok(ctx) => json!({ "emits": ctx.emits.iter().map(|(s, p)| json!({"subject": s, "payload": p})).collect::>(), @@ -304,7 +304,8 @@ fn replay_literal( /// side-by-side emit diff joined by the source message's stream sequence. This is /// `replay_literal` generalised from a literal patch to an arbitrary new source. /// The shadow invariant holds by construction — `vjs::run` collects emits, it -/// never publishes, so nothing a candidate emits reaches a real subject. +/// never publishes, so nothing a candidate emits reaches a real subject — and +/// `Engine::shadow` holds no secret value, so a candidate cannot read one back. fn time_travel(root: &Path, file: &str, candidate_src: &str, n: usize) -> Result { let path = guard_path(root, file).ok_or("path not allowed")?; let baseline = fs::read_to_string(&path).map_err(|e| e.to_string())?; @@ -325,7 +326,7 @@ fn time_travel(root: &Path, file: &str, candidate_src: &str, n: usize) -> Result .ok_or("flow has no bus source to replay")?; let rows = connectors::hydrate_recent(&js, &stream, &source_subj, n)?; let run_one = |prog: &vjs::Program, ev: &Value| -> Value { - let mut engine = vjs::Engine::new(root.to_path_buf(), pkg.clone()); + let mut engine = vjs::Engine::shadow(root.to_path_buf(), pkg.clone()); match vjs::run(prog, ev, &mut engine) { Ok(ctx) => json!({ "emits": ctx.emits.iter().map(|(s, p)| json!({"subject": s, "payload": p})).collect::>(), @@ -382,7 +383,7 @@ fn canaries() -> &'static Mutex> { /// Run one event through a parsed version in shadow, returning its emits (never /// published — vjs::run only collects them). fn shadow_emits(prog: &vjs::Program, ev: &Value, root: &Path, pkg: &str) -> Value { - let mut engine = vjs::Engine::new(root.to_path_buf(), pkg.to_string()); + let mut engine = vjs::Engine::shadow(root.to_path_buf(), pkg.to_string()); match vjs::run(prog, ev, &mut engine) { Ok(ctx) => json!({ "emits": ctx.emits.iter().map(|(s, p)| json!({"subject": s, "payload": p})).collect::>(), @@ -633,6 +634,28 @@ fn write_token() -> Option { env::var("VEJAS_TOKEN").ok().filter(|t| !t.is_empty()) } +/// Does the request carry the human approval credential? Constant-time. +fn approval_authorized(request: &tiny_http::Request) -> bool { + approval_token().is_some_and(|tok| { + request + .headers() + .iter() + .any(|h| h.field.equiv("x-approval-token") && ct_eq(h.value.as_str(), &tok)) + }) +} + +/// The doors that belong to the human, not to the agent: deciding a proposal, +/// and replaying or discarding dead letters. The panel holds the approval token +/// and never the agent's `VEJAS_TOKEN`, so on these routes a valid approval +/// token answers the write gate too. Otherwise a deployment with both tokens +/// set, the documented production posture, leaves the human unable to approve +/// from the panel. +fn is_human_route(path: &str) -> bool { + matches!(path, "/dlq/replay" | "/dlq/purge") + || (path.starts_with("/proposals/") + && (path.ends_with("/approve") || path.ends_with("/reject"))) +} + /// Does the request carry a valid `Authorization: Bearer `? Constant-time. fn bearer_authorized(request: &tiny_http::Request, token: &str) -> bool { request.headers().iter().any(|h| { @@ -809,6 +832,33 @@ fn decide_proposal( Ok(json!({"ok": true, "proposal": updated, "applied": result})) } +const DLQ_HUMAN_ACT: &str = "approval required: replaying or discarding dead letters is a human \ + action in governed mode: the panel, or X-Approval-Token (ADR-0024)"; + +/// Replaying or discarding dead letters in governed mode. No proposal carries +/// them, and an agent holds `VEJAS_TOKEN`, so they sit on the human credential: +/// a replay re-runs side effects, and a purge destroys what the agent would be +/// judged on. A dead letter is also untrusted content the agent reads, so these +/// are the last things it should be able to reach. +fn dlq_agent_gate() -> Result<(), String> { + if require_approval() { + Err(DLQ_HUMAN_ACT.into()) + } else { + Ok(()) + } +} + +/// The same rule on the HTTP door, where the human can present the credential. +fn dlq_http_gate(request: &tiny_http::Request) -> Result<(), (u16, String)> { + if !require_approval() || approval_authorized(request) { + return Ok(()); + } + if request.headers().iter().any(|h| h.field.equiv("x-approval-token")) { + return Err((401, "unauthorized: X-Approval-Token required".into())); + } + Err((409, DLQ_HUMAN_ACT.into())) +} + /// Governed-change mode (ADR-0024): when `VEJAS_REQUIRE_APPROVAL=1`, a direct /// mutation is refused — the caller submits a proposal a human approves instead. /// Gates EVERY mutating path (MCP tools AND raw HTTP), not just the MCP door: an @@ -1060,7 +1110,7 @@ fn supervise_vjs(handle: Arc, root: PathBuf) { if !RUNNING.load(Ordering::SeqCst) || handle.stop.load(Ordering::SeqCst) { break; } - let attempt = (|| -> Result<(), String> { + let attempt = contained(|| -> Result<(), String> { let baseline = fs::read_to_string(&handle.spec.path).map_err(|e| e.to_string())?; // Cluster-wide overlay (ADR-0021): a promoted version overrides the // file iff it forked from exactly this baseline; a stale overlay @@ -1162,7 +1212,7 @@ fn supervise_vjs(handle: Arc, root: PathBuf) { "[vejas] {} (vjs, native) consuming {source}", handle.spec.name ); - let mut engine = vjs::Engine::new(root.clone(), handle.spec.pkg.clone()); + let mut engine = vjs::Engine::live(root.clone(), handle.spec.pkg.clone()); // One pull request is always in flight ahead of the round being // processed. Without it every round began with ~2 ms of waiting for // its first message — request, round-trip, server delivery — while @@ -1354,7 +1404,7 @@ fn supervise_vjs(handle: Arc, root: PathBuf) { processed_at.elapsed().as_secs_f64(), ); } - })(); + }); match attempt { Ok(()) => break 'outer, Err(e) => { @@ -1427,7 +1477,7 @@ fn supervise_detect(handle: Arc, _root: PathBuf) { if !RUNNING.load(Ordering::SeqCst) || handle.stop.load(Ordering::SeqCst) { break; } - let attempt = (|| -> Result<(), String> { + let attempt = contained(|| -> Result<(), String> { let src = fs::read_to_string(&handle.spec.path).map_err(|e| e.to_string())?; let mut program = varpulis_engine::Program::compile(&src).map_err(|e| e.to_string())?; @@ -1470,6 +1520,7 @@ fn supervise_detect(handle: Arc, _root: PathBuf) { // into one channel. Stop is shared: the unit's flag plus one for // this attempt, so a restart never leaves a stale reader behind. let attempt_stop = Arc::new(AtomicBool::new(false)); + let _stop_reader = StopOnDrop(attempt_stop.clone()); let (tx, rx) = std::sync::mpsc::channel::(); let durable = handle.spec.name.replace([':', '.'], "_"); let filters: Vec = sources.iter().map(|(s, _)| s.clone()).collect(); @@ -1791,9 +1842,8 @@ fn supervise_detect(handle: Arc, _root: PathBuf) { } } })(); - attempt_stop.store(true, Ordering::SeqCst); result - })(); + }); match attempt { Ok(()) => break 'outer, Err(e) => { @@ -1863,6 +1913,32 @@ fn pkg_of_path(path: &Path) -> String { "default".to_string() } +/// Run one supervised attempt with its panics caught. Each unit's supervisor is +/// the thread that runs it, so a panic used to end the supervision too: the unit +/// stopped for good while its status stayed `running`, `restarts` 0 and +/// `last_error` null. Caught here, a panic takes the path of any other failure: +/// `restarting`, the error on record, backoff, another attempt. +fn contained(attempt: impl FnOnce() -> Result<(), String>) -> Result<(), String> { + std::panic::catch_unwind(std::panic::AssertUnwindSafe(attempt)).unwrap_or_else(|p| { + let text = p + .downcast_ref::<&str>() + .map(|s| s.to_string()) + .or_else(|| p.downcast_ref::().cloned()) + .unwrap_or_else(|| "unknown panic".into()); + Err(format!("panic: {text}")) + }) +} + +/// Raises a flag when dropped, so that a helper thread stops on every way out of +/// an attempt, a panic included. +struct StopOnDrop(Arc); + +impl Drop for StopOnDrop { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } +} + fn start_proc(registry: &Registry, spec: Spec, root: &Path) { let handle = Arc::new(Handle { spec: spec.clone(), @@ -1876,16 +1952,18 @@ fn start_proc(registry: &Registry, spec: Spec, root: &Path) { }); registry.lock().unwrap().insert(spec.name.clone(), handle.clone()); let root = root.to_path_buf(); - match spec.kind { - Kind::Flow => { - thread::spawn(move || supervise_vjs(handle, root)); - } - Kind::Connector => { - thread::spawn(move || supervise_connector(handle, root)); - } - Kind::Detect => { - thread::spawn(move || supervise_detect(handle, root)); - } + let supervised = handle.clone(); + let spawned = thread::Builder::new().name(spec.name.clone()).spawn(move || match spec.kind { + Kind::Flow => supervise_vjs(handle, root), + Kind::Connector => supervise_connector(handle, root), + Kind::Detect => supervise_detect(handle, root), + }); + if let Err(e) = spawned { + eprintln!("[vejas] {}: cannot start its thread: {e}", supervised.spec.name); + set_state(&supervised, |st| { + st.status = "failed".into(); + st.last_error = Some(format!("cannot start its thread: {e}")); + }); } } @@ -1901,7 +1979,7 @@ fn supervise_connector(handle: Arc, root: PathBuf) { if !RUNNING.load(Ordering::SeqCst) || handle.stop.load(Ordering::SeqCst) { break; } - let attempt = (|| -> Result<(), String> { + let attempt = contained(|| -> Result<(), String> { let src = fs::read_to_string(&handle.spec.path).map_err(|e| e.to_string())?; let prog = vjs::parse(&src)?; let driver_name = prog.driver.clone().ok_or("no `driver` declaration")?; @@ -1911,7 +1989,7 @@ fn supervise_connector(handle: Arc, root: PathBuf) { // config value resolves to the real value, fail-closed). The // manifest's UPPERCASE variables are the driver's config. let mut cfg_engine = - vjs::Engine::new(root.clone(), handle.spec.pkg.clone()); + vjs::Engine::live(root.clone(), handle.spec.pkg.clone()); let cfg_ctx = vjs::run(&prog, &Value::Object(serde_json::Map::new()), &mut cfg_engine)?; let mut config = serde_json::Map::new(); for (k, v) in cfg_ctx.vars { @@ -1987,7 +2065,7 @@ fn supervise_connector(handle: Arc, root: PathBuf) { None }; driver.run(&ctx) - })(); + }); stop.store(false, Ordering::SeqCst); match attempt { Ok(()) => { @@ -2438,7 +2516,7 @@ fn preview_json(root: &Path, file: &str) -> Result { ) .map_err(|e| format!("fixture: {e}"))?; // package context: services resolve relative to the file's package - let mut engine = vjs::Engine::new(root.to_path_buf(), pkg_of_path(&path)); + let mut engine = vjs::Engine::shadow(root.to_path_buf(), pkg_of_path(&path)); let (emits, error, pipeline) = match vjs::run(&prog, &event, &mut engine) { Ok(ctx) => { let emits: Vec = ctx @@ -2545,6 +2623,8 @@ Rules: - A flow file's first line is `# flow: `; it lives under flows/ (or packages//flows/). - Its sample input lives at flows/fixtures/.json (or packages//fixtures/) — one JSON event. - A flow is triggered by ONE of: `source "vx…"` (bus), `tool "…"` (MCP), or `api "VERB /path"` (HTTP). An `api` flow answers with `respond , {body}`; the request's JSON body, {path params} and `query` are all in the event. The whole API is described at GET /api/openapi.json. +- An `emit` subject must be a valid subject: no whitespace, no control character. A subject built from an event field is checked when the flow runs, and a value that breaks the rule fails the run, so the message reaches the DLQ after its deliveries. Map such a field through a table rather than trusting it. +- `secret()` never yields the value where a run is SHOWN instead of performed (preview, vejas_run_flow, flow-as-tool, shadow-replay, time-travel, canary, golden capture): it stays fail-closed and evaluates to the placeholder ``, so a branch comparing with a secret takes the not-equal path there. - A connector manifest's first line is `# connector: `, then `driver ""` (catalog: vejas_drivers) and UPPERCASE literal config; any credential uses secret("path/key"), never a literal."#; const CONTRACT_VJS: &str = r#"You write ONE VejasScript file for the Vejas integration platform. Reply with ONLY the file content (no markdown fences, no commentary). @@ -2865,7 +2945,7 @@ fn probe_connector(root: &Path, file: &str) -> Value { let driver_name = prog.driver.clone().ok_or("not a connector manifest (no `driver`)")?; let driver = connectors::driver_for(&driver_name) .ok_or_else(|| format!("unknown driver {driver_name:?}"))?; - let mut engine = vjs::Engine::new(root.to_path_buf(), pkg_of_path(&path)); + let mut engine = vjs::Engine::live(root.to_path_buf(), pkg_of_path(&path)); let cfg_ctx = vjs::run(&prog, &Value::Object(serde_json::Map::new()), &mut engine) .map_err(|e| connectors::humanize(&e))?; let mut config = serde_json::Map::new(); @@ -2932,7 +3012,7 @@ fn sap_rpc_request(root: &Path, op: &Value) -> Result { if prog.driver.as_deref() != Some("exec-rpc") { continue; } - let mut engine = vjs::Engine::new(root.to_path_buf(), pkg_of_path(&path)); + let mut engine = vjs::Engine::live(root.to_path_buf(), pkg_of_path(&path)); let cfg = vjs::run(&prog, &Value::Object(serde_json::Map::new()), &mut engine) .map_err(|e| connectors::humanize(&e))?; for (k, v) in cfg.vars { @@ -2994,11 +3074,30 @@ fn dlq_list(limit: usize) -> Result, String> { Ok(out) } +/// Record a replay or a purge: one summary per call, best-effort like the +/// live-promote path. A purge in particular leaves no other trace. +fn dlq_audit( + js: &nats::jetstream::JetStream, + kind: &str, + actor: &str, + seq: Option, + unit: Option<&str>, + count: u64, +) { + let record = json!({ + "ts": now_secs(), "actor": actor, "kind": kind, + "unit": unit, "seq": seq, "count": count, + }); + if let Err(e) = connectors::to_audit(js, unit.unwrap_or("dlq"), &record) { + eprintln!("[vejas] {kind}: audit not written: {e}"); + } +} + /// Replay dead letters: re-publish each to its ORIGINAL subject (where the /// corrected flow reprocesses it), then remove it from the DLQ. Publish-before- /// delete, so a failed re-inject keeps the dead letter. By `seq`, by `unit`, or /// all (both None). -fn dlq_replay(seq: Option, unit: Option<&str>) -> Result { +fn dlq_replay(seq: Option, unit: Option<&str>, actor: &str) -> Result { let js = dlq_js()?; let mut targets: Vec<(u64, String, Vec)> = Vec::new(); let take = |env: &Value| -> (u64, String, Vec) { @@ -3035,14 +3134,16 @@ fn dlq_replay(seq: Option, unit: Option<&str>) -> Result { Err(e) => errs.push(format!("seq {s}: {e}")), } } + dlq_audit(&js, "dlq-replay", actor, seq, unit, n); Ok(json!({"replayed": n, "errors": errs})) } /// Discard dead letters without replaying — by `seq`, by `unit`, or all. -fn dlq_purge(seq: Option, unit: Option<&str>) -> Result { +fn dlq_purge(seq: Option, unit: Option<&str>, actor: &str) -> Result { let js = dlq_js()?; if seq.is_none() && unit.is_none() { let r = js.purge_stream(connectors::DLQ_STREAM).map_err(|e| e.to_string())?; + dlq_audit(&js, "dlq-purge", actor, None, None, r.purged); return Ok(json!({"purged": r.purged})); } let mut n = 0u64; @@ -3060,6 +3161,7 @@ fn dlq_purge(seq: Option, unit: Option<&str>) -> Result { } } } + dlq_audit(&js, "dlq-purge", actor, seq, unit, n); Ok(json!({"purged": n})) } @@ -3122,7 +3224,7 @@ fn run_flow_on_input(root: &Path, file: &str, input: &Value) -> Result Result Value { json!({"name": "sap_call", "description": "Call a SAP function module (BAPI/RFC) on the live system and return its outputs. `import` values may be scalars, structures (a JSON object) or tables (an array of row objects); every EXPORT/CHANGING scalar & structure and every TABLES parameter comes back auto-marshalled from metadata. `max_rows` caps table output. E.g. func=\"RFC_READ_TABLE\", import={\"QUERY_TABLE\":\"T000\"}.", "inputSchema": obj(json!({"func": {"type": "string"}, "import": {"type": "object"}, "max_rows": {"type": "integer"}}), vec!["func"])}), json!({"name": "sap_send_idoc", "description": "Send an IDoc INTO SAP via transactional RFC (exactly-once). `control` is the EDI_DC40 control record (e.g. {TABNAM:\"EDI_DC40\", IDOCTYP:\"MATMAS05\", MESTYP:\"MATMAS\", SNDPRN, SNDPRT:\"LS\", RCVPRN, RCVPRT:\"LS\", DIRECT:\"2\"}); `data` is the array of EDI_DD40 segments ({SEGNAM, SDATA}). Pass a stable `tid` (24 chars) derived from an idempotency key for dedup across retries; omit for a fresh one. Returns the transaction id.", "inputSchema": obj(json!({"control": {"type": "object"}, "data": {"type": "array"}, "tid": {"type": "string"}}), vec!["control", "data"])}), json!({"name": "vejas_dlq", "description": "List dead letters — poison messages parked in the DLQ (ADR-0015) instead of dropped: unit, original subject, attempts, last error, payload, each with a `seq` for replay/purge. Newest first.", "inputSchema": obj(json!({"limit": {"type": "integer"}}), vec![])}), - json!({"name": "vejas_dlq_replay", "description": "Replay dead letters — re-inject each to its ORIGINAL subject so the (now corrected) flow reprocesses it, then remove it from the DLQ. Target one by `seq`, a whole `unit`, or all (omit both). Do this AFTER fixing the cause (vejas_set_literal, previewed with vejas_replay_literal).", "inputSchema": obj(json!({"seq": {"type": "integer"}, "unit": {"type": "string"}}), vec![])}), - json!({"name": "vejas_dlq_purge", "description": "Discard dead letters without replaying — by `seq`, by `unit`, or all (omit both).", "inputSchema": obj(json!({"seq": {"type": "integer"}, "unit": {"type": "string"}}), vec![])}), + json!({"name": "vejas_dlq_replay", "description": "Replay dead letters — re-inject each to its ORIGINAL subject so the (now corrected) flow reprocesses it, then remove it from the DLQ. Target one by `seq`, a whole `unit`, or all (omit both). Do this AFTER fixing the cause (vejas_set_literal, previewed with vejas_replay_literal). Refused in governed mode: a human replays them in the panel, after approving the fix.", "inputSchema": obj(json!({"seq": {"type": "integer"}, "unit": {"type": "string"}}), vec![])}), + json!({"name": "vejas_dlq_purge", "description": "Discard dead letters without replaying — by `seq`, by `unit`, or all (omit both). Refused in governed mode: a human discards them in the panel.", "inputSchema": obj(json!({"seq": {"type": "integer"}, "unit": {"type": "string"}}), vec![])}), ]; // generation-by-prompt shells out to the agent CLI: only advertised where // one exists (the stock container has none — external agents write .vjs) @@ -3707,14 +3809,16 @@ fn mcp_call(root: &Path, registry: &Registry, name: &str, args: &Value) -> Resul text(json!({"dead_letters": dlq_list(limit)?}).to_string()) } "vejas_dlq_replay" => { + dlq_agent_gate()?; let seq = args.get("seq").and_then(|v| v.as_u64()); let unit = args.get("unit").and_then(|v| v.as_str()); - text(dlq_replay(seq, unit)?.to_string()) + text(dlq_replay(seq, unit, "mcp")?.to_string()) } "vejas_dlq_purge" => { + dlq_agent_gate()?; let seq = args.get("seq").and_then(|v| v.as_u64()); let unit = args.get("unit").and_then(|v| v.as_str()); - text(dlq_purge(seq, unit)?.to_string()) + text(dlq_purge(seq, unit, "mcp")?.to_string()) } other => { // flow-as-tool: run the declaring flow on the arguments @@ -3800,7 +3904,10 @@ fn handle_request(mut request: tiny_http::Request, registry: Registry, root: Pat // an emitting `api "GET /x"` slipped past — a bus write with no token.) let wtoken = write_token(); if let Some(tok) = &wtoken { - if !is_read_method(&method) && !bearer_authorized(&request, tok) { + if !is_read_method(&method) + && !bearer_authorized(&request, tok) + && !(is_human_route(&path_only) && approval_authorized(&request)) + { return respond( request, 401, @@ -3876,14 +3983,10 @@ fn handle_request(mut request: tiny_http::Request, registry: Registry, root: Pat "approval requires VEJAS_APPROVAL_TOKEN (a credential distinct from the agent's VEJAS_TOKEN)".into(), "text/plain"); } - Some(tok) => { - let ok = request.headers().iter().any(|h| { - h.field.equiv("x-approval-token") && ct_eq(h.value.as_str(), &tok) - }); - if !ok { - return respond(request, 401, "unauthorized: X-Approval-Token required".into(), "text/plain"); - } + Some(_) if !approval_authorized(&request) => { + return respond(request, 401, "unauthorized: X-Approval-Token required".into(), "text/plain"); } + Some(_) => {} } return match decide_proposal(®istry, &root, id, approve, &actor) { Ok(v) => respond(request, 200, v.to_string(), "application/json"), @@ -3979,16 +4082,22 @@ fn handle_request(mut request: tiny_http::Request, registry: Registry, root: Pat Err(e) => respond(request, 500, e, "text/plain"), }, (tiny_http::Method::Post, "/dlq/replay") => { + if let Err((code, e)) = dlq_http_gate(&request) { + return respond(request, code, e, "text/plain"); + } let body = read_body(&mut request); - let r = dlq_replay(body.get("seq").and_then(|v| v.as_u64()), body.get("unit").and_then(|v| v.as_str())); + let r = dlq_replay(body.get("seq").and_then(|v| v.as_u64()), body.get("unit").and_then(|v| v.as_str()), &actor); match r { Ok(v) => respond(request, 200, v.to_string(), "application/json"), Err(e) => respond(request, 422, e, "text/plain"), } } (tiny_http::Method::Post, "/dlq/purge") => { + if let Err((code, e)) = dlq_http_gate(&request) { + return respond(request, code, e, "text/plain"); + } let body = read_body(&mut request); - let r = dlq_purge(body.get("seq").and_then(|v| v.as_u64()), body.get("unit").and_then(|v| v.as_str())); + let r = dlq_purge(body.get("seq").and_then(|v| v.as_u64()), body.get("unit").and_then(|v| v.as_str()), &actor); match r { Ok(v) => respond(request, 200, v.to_string(), "application/json"), Err(e) => respond(request, 422, e, "text/plain"), @@ -4373,10 +4482,31 @@ fn serve_http(registry: Registry, root: PathBuf, addr: String) { } }; eprintln!("[vejas] panel + monitoring on http://{addr}"); + // One thread per request, so a slow /flows/new never blocks monitoring, and a + // bound on them, so a burst costs a 503 instead of this loop. + let max_inflight = env::var("VEJAS_HTTP_MAX_INFLIGHT") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(256) + .max(1); + let inflight = Arc::new(std::sync::atomic::AtomicUsize::new(0)); for request in server.incoming_requests() { + let Some(slot) = connectors::Slot::try_take(&inflight, max_inflight) else { + metrics::inc_rejected("panel"); + respond(request, 503, "too many requests in flight".into(), "text/plain"); + continue; + }; let registry = registry.clone(); let root = root.clone(); - thread::spawn(move || handle_request(request, registry, root)); + // On failure the request drops with the closure and tiny_http answers 500. + let spawned = thread::Builder::new().name("http-request".into()).spawn(move || { + let _slot = slot; + handle_request(request, registry, root) + }); + if let Err(e) = spawned { + metrics::inc_rejected("panel"); + eprintln!("[vejas] no thread for a request: {e}"); + } if !RUNNING.load(Ordering::SeqCst) { break; } @@ -4438,7 +4568,7 @@ fn main() { serde_json::from_str(&fs::read_to_string(&args[3]).expect("read fixture")) .expect("fixture json"); let root = PathBuf::from(env::var("VEJAS_ROOT").unwrap_or_else(|_| ".".into())); - let mut engine = vjs::Engine::new(root, pkg_of_path(&PathBuf::from(&args[2]))); + let mut engine = vjs::Engine::live(root, pkg_of_path(&PathBuf::from(&args[2]))); match vjs::run(&prog, &event, &mut engine) { Ok(ctx) => { for (s, p) in &ctx.emits { @@ -4482,7 +4612,7 @@ fn main() { .and_then(|src| vjs::parse(&src)) .and_then(|prog| { let mut engine = - vjs::Engine::new(root.clone(), pkg_of_path(&PathBuf::from(flow))); + vjs::Engine::shadow(root.clone(), pkg_of_path(&PathBuf::from(flow))); vjs::run(&prog, &input, &mut engine) .map(|ctx| ctx.emits) }); @@ -4599,6 +4729,32 @@ fn main() { thread::sleep(Duration::from_millis(800)); } +#[cfg(test)] +mod supervision_tests { + use super::*; + + #[test] + fn a_panic_in_an_attempt_comes_back_as_an_error() { + assert_eq!(contained(|| Ok(())), Ok(())); + assert_eq!(contained(|| Err("plain".into())), Err("plain".to_string())); + let err = contained(|| panic!("boom {}", 7)).unwrap_err(); + assert_eq!(err, "panic: boom 7"); + } + + #[test] + fn a_slot_comes_back_when_its_thread_panics() { + let open = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let slot = connectors::Slot::try_take(&open, 1).unwrap(); + assert!(connectors::Slot::try_take(&open, 1).is_none(), "the bound holds"); + let _ = thread::spawn(move || { + let _slot = slot; + panic!("in the handler"); + }) + .join(); + assert!(connectors::Slot::try_take(&open, 1).is_some(), "the slot came back"); + } +} + #[cfg(test)] mod tests { #[test] diff --git a/core/src/metrics.rs b/core/src/metrics.rs index 0c98ba7..d23b04e 100644 --- a/core/src/metrics.rs +++ b/core/src/metrics.rs @@ -22,6 +22,7 @@ struct Unit { err: u64, emits: u64, dead: u64, + rejected: u64, /// Pull rounds the consumer loop made, and messages they returned. /// `messages / rounds` is the mean batch: the number that says whether a /// unit is paying one pull round-trip per message or per sixty-four. @@ -131,6 +132,13 @@ pub fn inc_dead_letter(unit: &str) { map.entry(unit.to_string()).or_default().dead += 1; } +/// A connection or request refused because a bound was reached, or because the +/// OS would not give a thread: the visible side of keeping the listener alive. +pub fn inc_rejected(unit: &str) { + let mut map = units().lock().unwrap(); + map.entry(unit.to_string()).or_default().rejected += 1; +} + fn esc(s: &str) -> String { s.replace('\\', "\\\\").replace('"', "\\\"").replace('\n', "\\n") } @@ -258,6 +266,15 @@ pub fn render(gauges: &[(String, String, u64)]) -> String { )); } + out.push_str("# HELP vejas_http_rejected_total Connections or requests refused past a bound, or for want of a thread.\n"); + out.push_str("# TYPE vejas_http_rejected_total counter\n"); + for (unit, u) in map.iter().filter(|(_, u)| u.rejected > 0) { + out.push_str(&format!( + "vejas_http_rejected_total{{unit=\"{}\"}} {}\n", + esc(unit), + u.rejected + )); + } out.push_str("# HELP vejas_dead_letters_total Messages a unit dead-lettered (ADR-0015).\n"); out.push_str("# TYPE vejas_dead_letters_total counter\n"); for (unit, u) in map.iter() { @@ -428,12 +445,14 @@ mod tests { observe("flow:t", true, 2, 0.003); observe("flow:t", false, 0, 0.02); inc_dead_letter("flow:t"); + inc_rejected("flow:t"); let text = render(&[("flow:t".into(), "flow".into(), 1)]); assert!(text.contains("vejas_up 1")); assert!(text.contains("vejas_events_processed_total{unit=\"flow:t\",result=\"ok\"} 1")); assert!(text.contains("vejas_events_processed_total{unit=\"flow:t\",result=\"error\"} 1")); assert!(text.contains("vejas_emits_published_total{unit=\"flow:t\"} 2")); assert!(text.contains("vejas_dead_letters_total{unit=\"flow:t\"} 1")); + assert!(text.contains("vejas_http_rejected_total{unit=\"flow:t\"} 1")); assert!(text.contains("vejas_event_duration_seconds_count{unit=\"flow:t\"} 2")); assert!(text.contains("le=\"+Inf\"} 2")); assert!(text.contains("vejas_flow_restarts{unit=\"flow:t\"} 1")); diff --git a/core/src/panel.html b/core/src/panel.html index e8f7672..020e17a 100644 --- a/core/src/panel.html +++ b/core/src/panel.html @@ -904,7 +904,26 @@

Secrets

loadEvents(); setInterval(loadEvents, 4000); -async function dlqAction(path, body) { try { await post(path, body); toast('Done — reloaded'); await loadDlq(); } catch (e) { toast(e.message.slice(0,160), true); } } +// In governed mode a replay or a discard is a human act (ADR-0024): the server +// answers 409 or 401 and the approval token goes with the retry. +async function dlqAction(path, body) { + const send = tok => fetch(path, { method: 'POST', + headers: { 'Content-Type': 'application/json', ...(tok ? { 'X-Approval-Token': tok } : {}) }, + body: JSON.stringify(body) }); + try { + let tok = ($('#approval-token').value || '').trim(); + let res = await send(tok); + if ((res.status === 409 || res.status === 401) && !tok) { + tok = (prompt('Approval token (governed mode)') || '').trim(); + if (!tok) return; + $('#approval-token').value = tok; + res = await send(tok); + } + if (!res.ok) throw new Error(await res.text()); + toast('Done — reloaded'); + await loadDlq(); + } catch (e) { toast(e.message.slice(0, 160), true); } +} async function loadDlq() { try { const items = (await fetch('/dlq').then(r => r.json())).dead_letters || []; diff --git a/core/src/secrets.rs b/core/src/secrets.rs index a5e22d6..1495c62 100644 --- a/core/src/secrets.rs +++ b/core/src/secrets.rs @@ -53,6 +53,33 @@ pub trait SecretStore: Send + Sync { } } +/// What every path that SHOWS a run instead of performing it resolves through: +/// preview, run on a fixture, flow-as-tool, shadow-replay, time-travel, canary, +/// golden capture. `secret("a/b")` stays fail-closed (a missing secret aborts as +/// in a real run) and evaluates to the placeholder ``, never to the +/// value. Those results go to an agent, into a model's context, or to an open +/// GET, and a time-travel candidate is arbitrary code that could emit the value +/// or transform it, so masking the output afterwards cannot work: the value must +/// never enter the run. The price is that a branch comparing with a secret takes +/// the not-equal path in shadow. +pub struct RedactedStore(Arc); + +impl RedactedStore { + pub fn new(inner: Arc) -> Self { + RedactedStore(inner) + } +} + +impl SecretStore for RedactedStore { + fn kind(&self) -> &'static str { + "redacted" + } + fn get(&self, reference: &str) -> Result { + self.0.get(reference)?; + Ok(format!("")) + } +} + /// The store the runtime uses, selected at startup: Vault if configured, else /// the writable file backend if configured, else the dev env backend. pub fn default_store() -> Arc { diff --git a/core/src/vjs.rs b/core/src/vjs.rs index a275305..18ce902 100644 --- a/core/src/vjs.rs +++ b/core/src/vjs.rs @@ -1327,9 +1327,20 @@ pub struct Engine { } impl Engine { - pub fn new(project_root: PathBuf, caller_pkg: String) -> Self { + /// An engine for a REAL run: `secret()` yields the value. Only what performs + /// a flow or a connector takes this; anything that returns a run's result to + /// a caller takes `shadow`. There is deliberately no `new`, so that a new + /// proof path cannot inherit real secrets by default. + pub fn live(project_root: PathBuf, caller_pkg: String) -> Self { Self::with_secrets(project_root, caller_pkg, crate::secrets::default_store()) } + + /// An engine for a SHADOW run: `secret()` stays fail-closed and yields a + /// placeholder, never the value (see `secrets::RedactedStore`). + pub fn shadow(project_root: PathBuf, caller_pkg: String) -> Self { + let store = crate::secrets::RedactedStore::new(crate::secrets::default_store()); + Self::with_secrets(project_root, caller_pkg, Arc::new(store)) + } pub fn with_secrets( project_root: PathBuf, caller_pkg: String, @@ -1528,6 +1539,9 @@ fn exec_block(stmts: &[Stmt], ctx: &mut Ctx, engine: &mut Engine) -> Result<(), Value::String(s) => s, other => return Err(format!("emit subject must be a string, got {other}")), }; + if !valid_subject(&s) { + return Err(format!("emit subject {s:?} is not a valid subject")); + } let p = eval(payload, ctx, engine)?; ctx.emits.push((s, p)); } @@ -1541,6 +1555,15 @@ fn exec_block(stmts: &[Stmt], ctx: &mut Ctx, engine: &mut Engine) -> Result<(), Ok(()) } +/// Can this string go on the wire as a subject? The NATS protocol is line-based +/// and space-separated, so a line break in a subject is a second protocol +/// command, and a subject built from an event field hands that to whoever sends +/// the event. Checked where a flow emits: the run fails, the message takes the +/// poison path to the DLQ, and a shadow run shows the error a live run would get. +pub fn valid_subject(subject: &str) -> bool { + !subject.is_empty() && !subject.chars().any(|c| c.is_whitespace() || c.is_control()) +} + fn set_path(root: &mut Value, path: &[String], v: Value) { if !root.is_object() { *root = Value::Object(Map::new()); @@ -1927,7 +1950,7 @@ mod tests { fn eval_flow(src: &str, event: Value) -> Result, String> { let prog = parse(src)?; - let mut engine = Engine::new(std::env::temp_dir(), "default".into()); + let mut engine = Engine::live(std::env::temp_dir(), "default".into()); run(&prog, &event, &mut engine).map(|c| c.emits) } @@ -2168,6 +2191,39 @@ mod tests { } } + #[test] + fn an_emit_subject_that_cannot_go_on_the_wire_fails_the_run() { + let prog = parse("emit f\"vx.{t}\", {a: 1}\n").unwrap(); + let run_with = |t: &str| { + let mut engine = Engine::live(std::env::temp_dir(), "default".into()); + run(&prog, &json!({"t": t}), &mut engine).map(|ctx| ctx.emits.len()) + }; + assert_eq!(run_with("orders.eu-west_1"), Ok(1)); + for bad in ["a b", "a\tb", "a\nb", "a\r\nPUB b 0", "a\u{0}b", "a\u{a0}b"] { + let err = run_with(bad).unwrap_err(); + assert!(err.contains("not a valid subject"), "{bad:?}: {err}"); + } + assert!(!valid_subject("")); + } + + #[test] + fn shadow_never_sees_a_secret_value() { + let mut map = std::collections::HashMap::new(); + map.insert("a/b".to_string(), "s3cret".to_string()); + let store = std::sync::Arc::new(MapStore(map)); + let shadow = std::sync::Arc::new(crate::secrets::RedactedStore::new(store)); + let prog = parse("v = secret(\"a/b\")\nemit \"o\", {v: v, up: upper(v)}\n").unwrap(); + let mut engine = Engine::with_secrets(std::env::temp_dir(), "default".into(), shadow.clone()); + let ctx = run(&prog, &json!({}), &mut engine).unwrap(); + let shown = serde_json::to_string(&ctx.emits).unwrap() + &format!("{:?}", ctx.vars); + assert!(!shown.contains("s3cret") && !shown.contains("S3CRET"), "{shown}"); + assert!(shown.contains(""), "{shown}"); + // fail-closed like a real run: a missing secret still aborts + let prog = parse("v = secret(\"missing/x\")\n").unwrap(); + let mut engine = Engine::with_secrets(std::env::temp_dir(), "default".into(), shadow); + assert!(run(&prog, &json!({}), &mut engine).is_err()); + } + #[test] fn secret_resolves_and_fails_closed() { let mut m = std::collections::HashMap::new(); @@ -2209,7 +2265,7 @@ mod tests { std::fs::write(pkg.join("services").join("secret.vjs"), "y = 1\n").unwrap(); std::fs::write(pkg.join("package.vjs"), "ENABLED = true\nEXPORTS = [\"fmt\"]\n").unwrap(); - let mut engine = Engine::new(root.clone(), "default".into()); + let mut engine = Engine::live(root.clone(), "default".into()); // merge form let prog = parse("invoke double(n: 21)\nemit \"vx.t\", {t: twice}").unwrap(); let ctx = run(&prog, &json!({}), &mut engine).unwrap(); @@ -2240,7 +2296,7 @@ mod tests { ) .unwrap(); std::fs::write(pkg.join("package.vjs"), "EXPORTS = [\"outer\"]\n").unwrap(); - let mut engine = Engine::new(root.clone(), "default".into()); + let mut engine = Engine::live(root.clone(), "default".into()); // outer is exported; inner is private but called FROM WITHIN pkg n -> allowed let prog = parse("invoke n:outer(v: 9)\n").unwrap(); let ctx = run(&prog, &json!({}), &mut engine).unwrap(); diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 11a8cd5..65c65e5 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -46,7 +46,13 @@ The runtime is a single OS process. Inside it: consumer on `vx.slack.out` → webhook or DRY-RUN). - **the HTTP server thread** — serves the panel, the JSON introspection endpoints, the editing endpoints, and the MCP endpoint. One request = one - thread (so a slow `/flows/new` never blocks monitoring). + thread (so a slow `/flows/new` never blocks monitoring), at most + `VEJAS_HTTP_MAX_INFLIGHT` at once; past it the request gets a 503. + +A panic in a unit's supervisor is caught around each attempt and takes the path +of any other failure: `restarting`, the error in `last_error`, backoff, another +attempt. `http-in` holds one thread per connection, at most `MAX_CONNECTIONS`, +and when the OS refuses a thread it drops that one connection. There is no database. NATS JetStream holds the stream `VEJAS` bound to `vx.>`, the durable consumers, and (later) KV/object state. diff --git a/docs/adr/0008-secrets-vault.md b/docs/adr/0008-secrets-vault.md index 3d23184..ff89237 100644 --- a/docs/adr/0008-secrets-vault.md +++ b/docs/adr/0008-secrets-vault.md @@ -47,6 +47,15 @@ a sidecar. paths (a package should not read another's secrets — mirrors ADR-0004's private-by-default stance); audit of secret access. +**Amendment A (2026-11-06, normative):** a secret value never enters a run +that is shown rather than performed. v0.3.6 evaluated preview, time-travel, +shadow-replay and canary with the real store, so their results carried the +value: a time-travel candidate gave it to an agent holding only `VEJAS_TOKEN`, +and `GET /preview` gave it to anyone who reached the port. Those paths now use +`Engine::shadow`, where `secret()` stays fail-closed and yields +``. `Engine::live` is the only way to a value and there is no +default constructor, so a new proof path has to choose. + ## Alternatives considered - **Env vars only** (today's stopgap): fine for one Slack webhook, doesn't diff --git a/docs/adr/0024-proposal-queue.md b/docs/adr/0024-proposal-queue.md index 91942a7..040e902 100644 --- a/docs/adr/0024-proposal-queue.md +++ b/docs/adr/0024-proposal-queue.md @@ -103,6 +103,18 @@ the on-call human hear about a pending proposal? The platform's own answer: a flow — route it to Slack, PagerDuty, email with the connectors that already exist. No notification subsystem; dogfood. +**Amendment A (2026-11-06, normative):** dead letters are in scope of "every +mutating path". Replay and purge had stayed on `VEJAS_TOKEN` as operational +actions, which left them in the agent's hands: a replay re-runs side effects, a +purge destroys what the agent's work is judged on, and dead letters are +untrusted content that agent reads. No proposal kind carries them, so in +governed mode they sit on `VEJAS_APPROVAL_TOKEN`: refused over MCP, and +`POST /dlq/replay|purge` want the `X-Approval-Token` header. On those routes +and on approve/reject, a valid approval token also answers the write gate. The +panel holds that token and never `VEJAS_TOKEN`, so with both set, the documented +production posture, the human could not approve from the panel (401 from the +write gate) until this change. + ## The loop this enables (the acte-6 recipe) A message dies (DLQ, version-tagged) → an agent reads `vejas_dlq`, drafts a diff --git a/docs/book/src/concepts/agents-mcp.md b/docs/book/src/concepts/agents-mcp.md index a8f23b0..ed8e82d 100644 --- a/docs/book/src/concepts/agents-mcp.md +++ b/docs/book/src/concepts/agents-mcp.md @@ -25,4 +25,5 @@ MCP tool — the platform grows its own tool surface as flows are written. And the self-healing loop closes the circle: a dead letter → the agent reads `vejas_dlq`, drafts a candidate, proves it on real traffic (`vejas_time_travel`, `vejas_canary_start`), proposes with evidence → human -approves → cluster-wide promote in 60 ms → `vejas_dlq_replay`. +approves → cluster-wide promote in 60 ms → replay, from the panel in governed +mode, `vejas_dlq_replay` otherwise. diff --git a/docs/book/src/concepts/secrets.md b/docs/book/src/concepts/secrets.md index 7f937cc..6e97645 100644 --- a/docs/book/src/concepts/secrets.md +++ b/docs/book/src/concepts/secrets.md @@ -23,3 +23,11 @@ What keeps the rule real, rather than aspirational: a credential-shaped env key must be a `${VAR:?}` reference, never a value. - Secret **paths** are listed (`GET /secrets`); values are never returned. +- **Shadow runs hold no value.** Whatever shows a run instead of performing + it (preview, `vejas_run_flow`, flow-as-tool, shadow-replay, time-travel, + canary, golden capture) resolves `secret()` to the placeholder + ``. It stays fail-closed: a missing secret still aborts. + A candidate is arbitrary code, so it cannot be left to a filter on the + output, and the value never enters the run. One consequence to know when + reading a diff: a branch that compares with a secret takes the not-equal + path in shadow. diff --git a/docs/book/src/guides/dlq-replay.md b/docs/book/src/guides/dlq-replay.md index 41cb49a..f78068d 100644 --- a/docs/book/src/guides/dlq-replay.md +++ b/docs/book/src/guides/dlq-replay.md @@ -17,7 +17,11 @@ post-mortem knows *which* logic failed, even after a promote. - Panel: the DLQ card; API: `GET /dlq`; agent: `vejas_dlq`. - Replay is **explicit** — `POST /dlq/replay` (`vejas_dlq_replay`), never automatic: if the logic was wrong, replaying before fixing just kills the - message again. Purge (`/dlq/purge`) is equally explicit and audited. + message again. Purge (`/dlq/purge`) is equally explicit. Both are recorded + in `VEJAS_AUDIT` with the actor and the count. +- In [governed mode](governed-mode.md) both are human acts: `vejas_dlq_replay` + and `vejas_dlq_purge` are refused, and the HTTP routes want the + `X-Approval-Token` header, which the panel sends. ## The loop that makes it self-healing @@ -28,7 +32,8 @@ post-mortem knows *which* logic failed, even after a promote. (canary). 3. In governed mode it *proposes* with that evidence; you approve — the promote fans out cluster-wide in 60 ms. -4. `vejas_dlq_replay` — the dead messages pass under the new version, and - their envelopes record the transition. +4. Replay from the panel, or `vejas_dlq_replay` outside governed mode — the + dead messages pass under the new version, and their envelopes record the + transition. The human owns exactly one step: the meaning. diff --git a/docs/book/src/guides/governed-mode.md b/docs/book/src/guides/governed-mode.md index 7ae7376..a573cac 100644 --- a/docs/book/src/guides/governed-mode.md +++ b/docs/book/src/guides/governed-mode.md @@ -31,6 +31,11 @@ executing and answers: Reads stay open. A direct write is never silently accepted; it is turned into a request. +Two things cannot be proposed, so they sit on the approval token instead: +replaying and discarding dead letters. `vejas_dlq_replay` and +`vejas_dlq_purge` are refused, and `POST /dlq/replay|purge` answer 409 without +the `X-Approval-Token` header. The panel asks for the token when it needs it. + ## The flow 1. **Agent proposes** — `vejas_propose` with the change and its *evidence* @@ -38,7 +43,8 @@ into a request. lists the queue. 2. **Human decides** — in the panel's **Approval queue** card, or `POST /proposals/{id}/approve|reject` with the `X-Approval-Token` - header. The card shows the evidence; a proposal with none is flagged + header, which is enough on its own even when `VEJAS_TOKEN` is set: the + panel never holds the agent's token. The card shows the evidence; a proposal with none is flagged **"⚠ No evidence"** loud — approving blind is a deliberate act. 3. **On approve** — the change executes exactly as a normal promote would (hot-reload, or a [cluster-wide version](change-safely.md) in 60 ms) and diff --git a/docs/book/src/reference/env-vars.md b/docs/book/src/reference/env-vars.md index de0a3f6..1deb4a0 100644 --- a/docs/book/src/reference/env-vars.md +++ b/docs/book/src/reference/env-vars.md @@ -19,6 +19,7 @@ directory is a complete system. | `VEJAS_TENANT` | *(unset)* | Tenant label for provisioned packages. | | `VEJAS_BUNDLE` | *(unset)* | Bundle path for provisioning (`vejas_provision`). | | `VEJAS_AGENT_CMD` | *(unset)* | Command the panel uses for its agent-assist box. | +| `VEJAS_HTTP_MAX_INFLIGHT` | `256` | Requests handled at once on the panel/MCP port, one thread each. Past it: 503. | ## Secrets (ADR-0008) diff --git a/docs/book/src/reference/http-api.md b/docs/book/src/reference/http-api.md index 7d59636..3203fb9 100644 --- a/docs/book/src/reference/http-api.md +++ b/docs/book/src/reference/http-api.md @@ -45,7 +45,7 @@ answer with a didactic refusal pointing at the proposal queue. | Route | Purpose | |---|---| | `GET /dlq` | Dead letters with death envelopes (version-tagged). | -| `POST /dlq/replay` · `POST /dlq/purge` | Explicit replay / purge (ADR-0015). | +| `POST /dlq/replay` · `POST /dlq/purge` | Explicit replay / purge (ADR-0015), audited. In governed mode they want `X-Approval-Token`. | ## Versions, time-travel, canary (ADR-0021) @@ -76,4 +76,4 @@ answer with a didactic refusal pointing at the proposal queue. | Route | Purpose | |---|---| | `POST /mcp` | JSON-RPC 2.0 — the runtime **is** the MCP server. See [MCP tools](mcp-tools.md). | -| `POST /ingest/` *(http-in connector, own port)* | Webhook ingestion → `vx.` on the bus, 202 after the JetStream pub-ack. | +| `POST /ingest/` *(http-in connector, own port)* | Webhook ingestion → `vx.` on the bus, 202 after the JetStream pub-ack. At most `MAX_CONNECTIONS` at once (default 512), then 503. | diff --git a/docs/book/src/reference/mcp-tools.md b/docs/book/src/reference/mcp-tools.md index 29fc3dd..b097d45 100644 --- a/docs/book/src/reference/mcp-tools.md +++ b/docs/book/src/reference/mcp-tools.md @@ -107,11 +107,11 @@ List dead letters — poison messages parked in the DLQ (ADR-0015) instead of ## `vejas_dlq_replay` -Replay dead letters — re-inject each to its ORIGINAL subject so the (now corrected) flow reprocesses it, then remove it from the DLQ. Target one by `seq`, a whole `unit`, or all (omit both). Do this AFTER fixing the cause (vejas_set_literal, previewed with vejas_replay_literal). +Replay dead letters — re-inject each to its ORIGINAL subject so the (now corrected) flow reprocesses it, then remove it from the DLQ. Target one by `seq`, a whole `unit`, or all (omit both). Do this AFTER fixing the cause (vejas_set_literal, previewed with vejas_replay_literal). Refused in governed mode: a human replays them in the panel, after approving the fix. ## `vejas_dlq_purge` -Discard dead letters without replaying — by `seq`, by `unit`, or all (omit both). +Discard dead letters without replaying — by `seq`, by `unit`, or all (omit both). Refused in governed mode: a human discards them in the panel. ## `vejas_new_flow` diff --git a/docs/book/src/reference/vejascript.md b/docs/book/src/reference/vejascript.md index 81b1df8..346a0ab 100644 --- a/docs/book/src/reference/vejascript.md +++ b/docs/book/src/reference/vejascript.md @@ -40,6 +40,8 @@ Rules: - A flow file's first line is `# flow: `; it lives under flows/ (or packages//flows/). - Its sample input lives at flows/fixtures/.json (or packages//fixtures/) — one JSON event. - A flow is triggered by ONE of: `source "vx…"` (bus), `tool "…"` (MCP), or `api "VERB /path"` (HTTP). An `api` flow answers with `respond , {body}`; the request's JSON body, {path params} and `query` are all in the event. The whole API is described at GET /api/openapi.json. +- An `emit` subject must be a valid subject: no whitespace, no control character. A subject built from an event field is checked when the flow runs, and a value that breaks the rule fails the run, so the message reaches the DLQ after its deliveries. Map such a field through a table rather than trusting it. +- `secret()` never yields the value where a run is SHOWN instead of performed (preview, vejas_run_flow, flow-as-tool, shadow-replay, time-travel, canary, golden capture): it stays fail-closed and evaluates to the placeholder ``, so a branch comparing with a secret takes the not-equal path there. - A connector manifest's first line is `# connector: `, then `driver ""` (catalog: vejas_drivers) and UPPERCASE literal config; any credential uses secret("path/key"), never a literal. ``` diff --git a/docs/security-audit-scope.md b/docs/security-audit-scope.md index 777ca29..9e0442f 100644 --- a/docs/security-audit-scope.md +++ b/docs/security-audit-scope.md @@ -16,7 +16,7 @@ invariant is wrong, that is the finding we want. |---|---|---| | **Operator** | The write API (panel, `/mcp`, HTTP mutations) with `VEJAS_TOKEN` | Everything: write flows, promote, set secrets — this is the admin | | **Agent** | `/mcp` with `VEJAS_TOKEN` | Propose and (outside governed mode) mutate; in governed mode, *propose only* — never approve | -| **Approver (human)** | `/proposals/{id}/approve` with `VEJAS_APPROVAL_TOKEN` | Approve/reject in governed mode; nothing else | +| **Approver (human)** | `/proposals/{id}/approve` and `/dlq/replay`, `/dlq/purge` with `VEJAS_APPROVAL_TOKEN` | Approve/reject in governed mode, and replay or discard dead letters; nothing else | | **Event source** | `http-in` on its own port, **no auth** | Publish a JSON body to one subject — nothing else | | **Bus peer** | NATS/JetStream | Governed by NATS's own auth (deployment-owned) | @@ -100,10 +100,16 @@ pressure-test. in-process HTTP (ureq) carries them in headers, never a subprocess argv; the standalone connectors that shell out use a `0600` config file, never argv (ADR-0008). A literal is never accepted where a `secret()` is required. + Whatever shows a run instead of performing it uses `Engine::shadow`, where + `secret()` yields the placeholder ``; `Engine::live` is the + only way to the value, and there is no default constructor. - **How to test:** grep every response body and log line for a set secret; set a secret and confirm `GET /secrets` shows status only, never the value; inspect argv and env of every spawned process; attempt to store a credential - as a plain literal and confirm refusal. + as a plain literal and confirm refusal; `e2e/governed-agent.sh` checks every + shadow path, a candidate included, and that a live run still gets the value. + **Break it:** a surface returning a run's result that does not go through + `Engine::shadow`. - **Honest bound (found this pass — S4-1).** `secret()` protects the **store**: no endpoint returns a stored secret, and secrets are references, never literals. But a flow *author* can copy a resolved secret onto the **data @@ -128,8 +134,13 @@ pressure-test. is therefore "inject any bus event," not "publish to a source only." The optional `ALLOW` list restricts the surface to declared suffixes/prefixes (recommended, defence in depth); without it the port is any unauthenticated - webhook — it belongs behind the operator's ingress / trust boundary. -- **How to test:** confirm `http-in` reaches no mutation route or secret; + webhook — it belongs behind the operator's ingress / trust boundary. A burst + costs refusals and never the connector: at most `MAX_CONNECTIONS` threads, a + request has 30 s to arrive, and when the OS refuses a thread that connection + drops. It is not rate limiting: a client holding every slot blocks ingestion + until it stops, so a hostile network still needs the ingress in front. +- **How to test:** `e2e/http-in-bound.sh` holds the bound and survives a pids + limit; confirm `http-in` reaches no mutation route or secret; with `ALLOW` set, confirm a suffix outside it is refused (`403`) and a permitted one still publishes (`e2e` covers this); without `ALLOW`, confirm the honest bound above is documented, not surprising; fuzz the suffix for @@ -242,8 +253,20 @@ re-deriving these — they are fixed and regression-tested; instead, try to connector (an exec driver = arbitrary `CMD`) with no human approval: governance bypass **and** RCE. Fixed to mirror `/flows/new`; `e2e/governed-gate.sh` in CI. Sibling-checked: the HTTP twin, all mutating - MCP tools, and `/provision` are gated; `dlq/replay`, `dlq/purge` and - `connectors/test` are operational (write-token, not approval) by design. + MCP tools, and `/provision` are gated; `connectors/test` is operational + (write-token, not approval) by design. `dlq/replay` and `dlq/purge` were + too until 2026-11-06, when ADR-0024 Amendment A moved them to the approval + token: an agent that reads dead letters reads untrusted content, and must + not hold what destroys them. + +- **G — protocol injection through an `emit` subject (HIGH, fixed).** A subject + built from an event field reached the NATS wire as it was, so a line break in + the field became protocol commands with the runtime's NATS permissions, and + `http-in` makes that field anonymous. Fixed where a flow emits, by + `vjs::valid_subject`; `http-in` applies it to its suffix, which had no injection + of its own (the path cannot carry the CRLF a payload must end with) but could + drop the connection. `e2e/emit-subject.sh` in CI. **Break it:** a subject that + reaches the bus from outside without passing `valid_subject`. ## Known minor points to confirm or harden (not blockers) diff --git a/e2e/emit-subject.sh b/e2e/emit-subject.sh new file mode 100755 index 0000000..d1a9313 --- /dev/null +++ b/e2e/emit-subject.sh @@ -0,0 +1,77 @@ +#!/bin/bash +# An emit subject cannot carry protocol. A subject built from an event field used +# to reach the NATS wire as it was: a line break in the field became protocol +# commands, and through http-in that needed no credential. A refused subject fails +# the run like any other error, so the message reaches the DLQ. +# e2e/emit-subject.sh (needs the nats CLI) +set -uo pipefail +cd "$(dirname "$0")/.." +BIN="$PWD/core/target/release/vejas-runtime" +[ -x "$BIN" ] || { echo "build first" >&2; exit 1; } +S=$(mktemp -d); R=$(mktemp -d) +RT=""; NP=""; SUB=""; EVIL="" +trap 'kill $RT $NP $SUB $EVIL 2>/dev/null; wait $RT $NP $SUB $EVIL 2>/dev/null; rm -rf "$S" "$R"' EXIT +fail=0 +ok(){ echo " ✓ $1"; } +ko(){ echo " ✗ $1"; fail=1; } + +URL=nats://127.0.0.1:4281; H="http://127.0.0.1:8753"; IN=18794 +mkdir -p "$R/flows" "$R/connectors" +printf '# flow: dyn\nsource "vx.dyn.in"\nemit f"vx.out.{tenant}", {id: id}\n' > "$R/flows/dyn.vjs" +printf '# connector: http_in\ndriver "http-in"\nPORT = %s\n' $IN > "$R/connectors/http_in.vjs" + +nats-server -js -sd "$S/n" -a 127.0.0.1 -p 4281 > /dev/null 2>&1 & NP=$! +sleep 0.5 +NATS_URL=$URL VEJAS_ROOT="$R" VEJAS_HTTP_ADDR=127.0.0.1:8753 VEJAS_ACK_WAIT_SECS=1 \ + "$BIN" > "$S/rt.log" 2>&1 & RT=$! +until curl -sf -o /dev/null "$H/healthz"; do sleep 0.1; done +post(){ curl -s -m 5 -o /dev/null -w '%{http_code}' -X POST -H 'Content-Type: application/json' \ + --data-binary "$1" "http://127.0.0.1:$IN/ingest/dyn.in"; } +for _ in $(seq 1 100); do [ "$(post '{"id":0,"tenant":"warmup"}')" = "202" ] && break; sleep 0.1; done +sleep 1.5 +messages(){ nats -s $URL stream info VEJAS --json 2>/dev/null \ + | python3 -c 'import json,sys; print(json.load(sys.stdin)["state"]["messages"])'; } +hostile(){ python3 -c 'import json,sys; print(json.dumps({"id": 1, "tenant": sys.argv[1].encode().decode("unicode_escape")}))' "$1"; } + +nats -s $URL sub 'vx.out.>' --raw > "$S/bus" 2>/dev/null & SUB=$! +nats -s $URL sub vx.evil --raw > "$S/evil" 2>/dev/null & EVIL=$! +sleep 0.5 + +echo "── a clean field still routes" +C=$(post '{"id":1,"tenant":"acme"}') +for _ in $(seq 1 50); do grep -q '"id":1}' "$S/bus" && break; sleep 0.1; done +grep -q '"id":1}' "$S/bus" && ok "the flow emitted for tenant acme ($C)" || ko "the clean emit is gone" + +echo "── a hostile field carries no protocol" +BEFORE=$(messages) +post "$(hostile 'x 0\r\n\r\nPUB vx.evil 6\r\ninject\r\nPUB vx.pad')" > /dev/null +post "$(hostile 'x 0\r\n\r\nPUB $JS.API.STREAM.PURGE.VEJAS 0\r\n\r\nPUB vx.pad')" > /dev/null +sleep 2 +[ -s "$S/evil" ] && ko "a field published on the bus" || ok "nothing was published by the field" +[ "$(messages)" -ge "$BEFORE" ] && ok "the stream was not purged" || ko "the stream lost messages" + +echo "── the refused run takes the poison path" +letters(){ curl -s -m 5 "$H/dlq" | python3 -c ' +import json, sys +print(len([d for d in json.load(sys.stdin)["dead_letters"] if "not a valid subject" in d.get("last_error", "")]))'; } +for _ in $(seq 1 150); do [ "$(letters)" = "2" ] && break; sleep 0.1; done +[ "$(letters)" = "2" ] && ok "both messages reached the DLQ with the reason" || ko "$(letters) of 2 reached the DLQ" + +echo "── a shadow run shows the same error" +curl -s -m 10 -H 'Content-Type: application/json' -d \ + '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"vejas_run_flow","arguments":{"file":"flows/dyn.vjs","input":{"id":1,"tenant":"a b"}}}}' \ + "$H/mcp" | grep -q 'not a valid subject' && ok "vejas_run_flow reports it" || ko "the shadow run hides it" + +echo "── http-in refuses such a suffix" +C=$(python3 -c ' +import socket, sys +s = socket.create_connection(("127.0.0.1", int(sys.argv[1])), timeout=5) +s.sendall(b"POST /ingest/x\t0\n\nPUB\tvx.evil\t6 HTTP/1.1\r\nContent-Length: 2\r\n\r\n{}") +s.settimeout(4) +print(s.recv(100).split(b" ")[1].decode())' $IN) +[ "$C" = "400" ] && ok "a tab and a bare line feed in the path: $C" || ko "the suffix got $C" +C=$(post '{"id":3,"tenant":"acme"}') +[ "$C" = "202" ] && ok "ingestion carries on ($C)" || ko "ingestion broke ($C)" + +[ $fail -eq 0 ] && echo "EMIT SUBJECT HOLDS" || echo "EMIT SUBJECT REGRESSION — do not ship" +exit $fail diff --git a/e2e/governed-agent.sh b/e2e/governed-agent.sh new file mode 100755 index 0000000..f425737 --- /dev/null +++ b/e2e/governed-agent.sh @@ -0,0 +1,154 @@ +#!/bin/bash +# What an agent holding only VEJAS_TOKEN can reach in governed mode, and what an +# open GET can read at all (ADR-0008, ADR-0024). Secret values are write-only: +# every path that SHOWS a run evaluates it in shadow, so a secret comes back as +# the placeholder , while a real run still gets the value. A +# time-travel candidate is arbitrary code, so this cannot be a filter on the +# output (v0.3.6 returned the value in the diff, and over GET /preview with no +# token at all). +# e2e/governed-agent.sh (needs the nats CLI) +set -uo pipefail +cd "$(dirname "$0")/.." +BIN="$PWD/core/target/release/vejas-runtime" +[ -x "$BIN" ] || { echo "build first" >&2; exit 1; } +S=$(mktemp -d); R=$(mktemp -d) +trap 'kill $RT $NP 2>/dev/null; wait $RT $NP 2>/dev/null; rm -rf "$S" "$R"' EXIT +fail=0 +ok(){ echo " ✓ $1"; } +ko(){ echo " ✗ $1"; fail=1; } + +SECRET="canary-s3cret-$$" +AGENT="agent-tok-$$" +HUMAN="human-tok-$$" +H="http://127.0.0.1:8749" +URL=nats://127.0.0.1:4278 +PLACEHOLDER='' + +mkdir -p "$R/flows/fixtures" +cat > "$R/flows/sdemo.vjs" <<'VJS' +# flow: sdemo +source "vx.sdemo.in" +tool "secret demo" +CUTOFF = 1 +api_key = secret("demo/key") +emit "vx.sdemo.out", {id: id, auth: f"Bearer {api_key}"} +VJS +echo '{"id": 1}' > "$R/flows/fixtures/sdemo.json" + +nats-server -js -sd "$S/n" -a 127.0.0.1 -p 4278 > /dev/null 2>&1 & NP=$! +sleep 0.5 +NATS_URL=$URL VEJAS_ROOT="$R" VEJAS_HTTP_ADDR=127.0.0.1:8749 \ + VEJAS_REQUIRE_APPROVAL=1 VEJAS_TOKEN=$AGENT VEJAS_APPROVAL_TOKEN=$HUMAN \ + VEJAS_SECRET_DEMO_KEY=$SECRET "$BIN" > "$S/rt.log" 2>&1 & RT=$! +until curl -sf -o /dev/null "$H/healthz"; do sleep 0.1; done + +mcp(){ # tool, arguments as JSON + curl -s -m 20 -H 'Content-Type: application/json' -H "Authorization: Bearer $AGENT" \ + -d "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"$1\",\"arguments\":$2}}" \ + "$H/mcp" +} +# The stream and the durable exist only once the flow runs: an event published +# before that is lost, and replay and time-travel then have nothing to show. +for _ in $(seq 1 100); do + mcp vejas_topology '{}' | grep -q 'running' && break + sleep 0.1 +done +shadowed(){ # label, output: the placeholder is there, the value is not + if echo "$2" | grep -q "$SECRET"; then ko "$1 returned the secret VALUE" + elif echo "$2" | grep -q "$PLACEHOLDER"; then ok "$1 shows the placeholder" + else ko "$1 answered neither value nor placeholder: $(echo "$2" | cut -c1-160)"; fi +} + +echo "── a real run still gets the real value" +nats -s $URL sub vx.sdemo.out --count=1 --raw > "$S/live.out" 2>/dev/null & SUB=$! +sleep 0.5 +nats -s $URL pub vx.sdemo.in '{"id": 7}' > /dev/null 2>&1 +for _ in $(seq 1 50); do [ -s "$S/live.out" ] && break; sleep 0.1; done +kill $SUB 2>/dev/null +grep -q "$SECRET" "$S/live.out" && ok "the live emit carries the value" \ + || ko "the live run lost its secret: $(cut -c1-120 "$S/live.out")" + +echo "── every path that shows a run is shadowed" +shadowed "GET /preview (no token)" "$(curl -s -m 10 "$H/preview?file=flows/sdemo.vjs")" +shadowed "vejas_preview" "$(mcp vejas_preview '{"file":"flows/sdemo.vjs"}')" +shadowed "vejas_run_flow" "$(mcp vejas_run_flow '{"file":"flows/sdemo.vjs","input":{"id":1}}')" +shadowed "flow-as-tool" "$(mcp flow_sdemo '{"id":1}')" +shadowed "vejas_replay_literal" "$(mcp vejas_replay_literal '{"file":"flows/sdemo.vjs","name":"CUTOFF","key":"-","value":2,"n":5}')" + +CAND='# flow: sdemo\nsource \"vx.sdemo.in\"\nv = secret(\"demo/key\")\nemit \"vx.sdemo.out\", {id: id, leak: v, loud: upper(v)}\n' +UPPER=$(echo "$SECRET" | tr '[:lower:]' '[:upper:]') +OUT=$(mcp vejas_time_travel "{\"file\":\"flows/sdemo.vjs\",\"candidate\":\"$CAND\",\"n\":5}") +shadowed "vejas_time_travel" "$OUT" +echo "$OUT" | grep -q "$UPPER" && ko "a transformed value left through time-travel" \ + || ok "a candidate cannot transform the value out either" + +mcp vejas_canary_start "{\"file\":\"flows/sdemo.vjs\",\"candidate\":\"$CAND\"}" > /dev/null +sleep 1 +nats -s $URL pub vx.sdemo.in '{"id": 8}' > /dev/null 2>&1 +OUT="" +for _ in $(seq 1 50); do + OUT=$(mcp vejas_canary_status '{"file":"flows/sdemo.vjs"}') + echo "$OUT" | grep -q 'leak' && break + sleep 0.1 +done +shadowed "vejas_canary_status" "$OUT" +mcp vejas_canary_stop '{"file":"flows/sdemo.vjs"}' > /dev/null + +echo "── shadow stays fail-closed" +CAND='# flow: sdemo\nsource \"vx.sdemo.in\"\nv = secret(\"nope/x\")\nemit \"vx.sdemo.out\", {v: v}\n' +mcp vejas_time_travel "{\"file\":\"flows/sdemo.vjs\",\"candidate\":\"$CAND\",\"n\":1}" | grep -q 'nope/x' \ + && ok "a missing secret aborts a shadow run" || ko "a missing secret did not abort in shadow" + +grep -q "$SECRET" "$S/rt.log" && ko "the value reached the runtime log" || ok "the value is not in the log" + +echo "── dead letters: the agent reads them, a human acts on them" +code(){ curl -s -o /dev/null -w "%{http_code}" -m 10 "$@"; } +dead_letters(){ curl -s -m 10 "$H/dlq"; } +bury(){ # a dead letter on the bus, as the runtime parks one + nats -s $URL pub vxdlq.vx.sdemo.in \ + "{\"unit\":\"flow:sdemo\",\"original_subject\":\"vx.sdemo.in\",\"payload\":\"{\\\"id\\\":$1}\",\"error\":\"boom\",\"attempts\":5}" \ + > /dev/null 2>&1 +} +BEARER="Authorization: Bearer $AGENT" +JSON='Content-Type: application/json' +dead_letters > /dev/null # ensures the stream +bury 9 +mcp vejas_dlq '{"limit":5}' | grep -q 'flow:sdemo' && ok "the agent reads them" || ko "vejas_dlq shows nothing" +mcp vejas_dlq_purge '{"unit":"flow:sdemo"}' | grep -q 'human action' \ + && ok "vejas_dlq_purge refused" || ko "vejas_dlq_purge NOT governed" +mcp vejas_dlq_replay '{"unit":"flow:sdemo"}' | grep -q 'human action' \ + && ok "vejas_dlq_replay refused" || ko "vejas_dlq_replay NOT governed" +C=$(code -X POST -H "$BEARER" -H "$JSON" -d '{}' "$H/dlq/purge") +[ "$C" = "409" ] && ok "POST /dlq/purge with the agent's token: $C" || ko "/dlq/purge NOT governed ($C)" +C=$(code -X POST -H "$BEARER" -H "$JSON" -d '{}' "$H/dlq/replay") +[ "$C" = "409" ] && ok "POST /dlq/replay with the agent's token: $C" || ko "/dlq/replay NOT governed ($C)" +C=$(code -X POST -H "X-Approval-Token: wrong" -H "$JSON" -d '{}' "$H/dlq/purge") +[ "$C" = "401" ] && ok "a wrong approval token: $C" || ko "a wrong approval token got $C" +C=$(code -X POST -H "$JSON" -d '{}' "$H/dlq/purge") +[ "$C" = "401" ] && ok "no credential at all: $C" || ko "no credential at all got $C" +dead_letters | grep -q 'flow:sdemo' && ok "the dead letter is still there" || ko "the refused calls lost the dead letter" + +# what the panel sends: the approval token, never the agent's +OUT=$(curl -s -m 10 -X POST -H "X-Approval-Token: $HUMAN" -H "$JSON" -d '{"unit":"flow:sdemo"}' "$H/dlq/replay") +echo "$OUT" | grep -q '"replayed":1' && ok "the human replays with the approval token alone" \ + || ko "human replay: $OUT" +bury 10 +OUT=$(curl -s -m 10 -X POST -H "X-Approval-Token: $HUMAN" -H "$JSON" -d '{"unit":"flow:sdemo"}' "$H/dlq/purge") +echo "$OUT" | grep -q '"purged":1' && ok "the human purges with the approval token alone" \ + || ko "human purge: $OUT" +AUDIT=$(for seq in 1 2 3 4 5 6; do + nats -s $URL stream get VEJAS_AUDIT $seq --json 2>/dev/null \ + | python3 -c 'import base64,json,sys; print(base64.b64decode(json.load(sys.stdin)["data"]).decode())' 2>/dev/null +done) +echo "$AUDIT" | grep -q '"dlq-replay"' && ok "the replay is audited" || ko "no audit record for the replay" +echo "$AUDIT" | grep -q '"dlq-purge"' && ok "the purge is audited" || ko "no audit record for the purge" + +echo "── the human can decide from the panel with both tokens set" +C=$(code -X POST -H "X-Approval-Token: $HUMAN" "$H/proposals/nope/approve") +[ "$C" = "422" ] && ok "approve passes the write gate on the approval token alone ($C: no such proposal)" \ + || ko "approve got $C" +C=$(code -X POST -H "$BEARER" "$H/proposals/nope/approve") +[ "$C" = "401" ] && ok "the agent's token still cannot approve ($C)" || ko "the agent's token got $C on approve" + +[ $fail -eq 0 ] && echo "GOVERNED AGENT SURFACE HOLDS" || echo "GOVERNED AGENT SURFACE REGRESSION — do not ship" +exit $fail diff --git a/e2e/http-in-bound.sh b/e2e/http-in-bound.sh new file mode 100755 index 0000000..d7f2625 --- /dev/null +++ b/e2e/http-in-bound.sh @@ -0,0 +1,112 @@ +#!/bin/bash +# http-in is an unauthenticated port with one OS thread per connection. A burst +# must cost 503s, never the connector: in v0.3.6 `thread::spawn` panicked when +# the OS refused a thread, the accept loop died with it and nothing restarted +# it, while /healthz said 200 and the topology said running, restarts 0, +# last_error null. Part 1 holds the bound. Part 2 takes the thread budget away +# for real, with a pids limit, and needs docker (skipped without it). +# e2e/http-in-bound.sh +set -uo pipefail +cd "$(dirname "$0")/.." +BIN="$PWD/core/target/release/vejas-runtime" +[ -x "$BIN" ] || { echo "build first" >&2; exit 1; } +S=$(mktemp -d); R=$(mktemp -d); R2=$(mktemp -d) +CT=vejas-http-in-bound-$$ +RT=""; NP="" +cleanup(){ + kill $RT $NP 2>/dev/null; wait $RT $NP 2>/dev/null + command -v docker > /dev/null && docker rm -f "$CT" > /dev/null 2>&1 + rm -rf "$S" "$R" "$R2" +} +trap cleanup EXIT +fail=0 +ok(){ echo " ✓ $1"; } +ko(){ echo " ✗ $1"; fail=1; } + +NP_PORT=4279; URL=nats://127.0.0.1:$NP_PORT +IN=18789; H="http://127.0.0.1:8750" +hold(){ # port, connections, seconds: idle sockets, as a burst leaves them + python3 -c ' +import socket, sys, time +port, n, secs = int(sys.argv[1]), int(sys.argv[2]), float(sys.argv[3]) +socks = [socket.create_connection(("127.0.0.1", port), timeout=5) for _ in range(n)] +time.sleep(secs) +' "$1" "$2" "$3" & +} +ingest(){ curl -s -m 3 -o /dev/null -w '%{http_code}' -X POST \ + -H 'Content-Type: application/json' -d '{"a":1}' "http://127.0.0.1:$1/ingest/t"; } +topology(){ # the unit as JSON: the MCP result carries it as an escaped string + curl -s -m 5 -H 'Content-Type: application/json' \ + -d '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"vejas_topology","arguments":{}}}' "$1/mcp" \ + | python3 -c ' +import json, sys +def units(v): + if isinstance(v, dict): + if "restarts" in v: yield v + for x in v.values(): yield from units(x) + elif isinstance(v, list): + for x in v: yield from units(x) +doc = json.loads(json.load(sys.stdin)["result"]["content"][0]["text"]) +print(json.dumps([u for u in units(doc) if "http_in" in u.get("name", "")]))' +} +threads(){ awk '/^Threads:/{print $2}' "/proc/$1/status"; } +wait_ingest(){ # port: until it answers 202 + for _ in $(seq 1 100); do [ "$(ingest "$1")" = "202" ] && return 0; sleep 0.1; done + return 1 +} + +mkdir -p "$R/connectors" +printf '# connector: http_in\ndriver "http-in"\nPORT = %s\nMAX_CONNECTIONS = 20\n' $IN > "$R/connectors/http_in.vjs" +nats-server -js -sd "$S/n" -a 127.0.0.1 -p $NP_PORT > /dev/null 2>&1 & NP=$! +sleep 0.5 +NATS_URL=$URL VEJAS_ROOT="$R" VEJAS_HTTP_ADDR=127.0.0.1:8750 "$BIN" > "$S/rt.log" 2>&1 & RT=$! +until curl -sf -o /dev/null "$H/healthz"; do sleep 0.1; done +wait_ingest $IN || ko "http-in never came up" + +echo "── part 1: past MAX_CONNECTIONS a client gets a refusal, the thread count stays bounded" +BEFORE=$(threads $RT) +hold $IN 100 4; HOLD=$! +sleep 1.5 +GROWTH=$(( $(threads $RT) - BEFORE )) +[ "$GROWTH" -le 22 ] && ok "100 connections, $GROWTH more threads (bound 20)" \ + || ko "$GROWTH more threads for a bound of 20" +C=$(ingest $IN) +case "$C" in + 503|000) ok "a request past the bound is refused ($C)" ;; + *) ko "a request past the bound got $C" ;; +esac +wait $HOLD 2>/dev/null +wait_ingest $IN && ok "ingestion is back once the connections are gone" || ko "http-in did not recover" +curl -s -m 5 "$H/metrics" | grep -q 'vejas_http_rejected_total{unit="connector:http_in"}' \ + && ok "the refusals are on /metrics" || ko "no vejas_http_rejected_total on /metrics" +topology "$H" | grep -q 'running' && ok "the connector still reports running" || ko "the connector left running" + +kill $RT 2>/dev/null; wait $RT 2>/dev/null; RT="" + +echo "── part 2: the OS has no more threads to give" +if ! { command -v docker > /dev/null && docker info > /dev/null 2>&1; }; then + echo " - skipped: no docker" +else + IN2=18790; H2="http://127.0.0.1:8751" + mkdir -p "$R2/connectors" + printf '# connector: http_in\ndriver "http-in"\nPORT = %s\n' $IN2 > "$R2/connectors/http_in.vjs" + chmod -R a+rX "$R2" + docker run -d --name "$CT" --network host --pids-limit=64 \ + -e NATS_URL=$URL -e VEJAS_ROOT=/vroot -e VEJAS_HTTP_ADDR=127.0.0.1:8751 \ + -v "$BIN":/usr/local/bin/vejas-runtime:ro -v "$R2":/vroot \ + debian:trixie-slim vejas-runtime > /dev/null + for _ in $(seq 1 100); do curl -sf -o /dev/null "$H2/healthz" && break; sleep 0.1; done + wait_ingest $IN2 || ko "http-in never came up in the container" + hold $IN2 300 3; HOLD=$! + sleep 1 + wait $HOLD 2>/dev/null + wait_ingest $IN2 && ok "http-in survives 300 connections on a pids limit of 64" \ + || ko "http-in died on the pids limit" + P=$(docker logs "$CT" 2>&1 | grep -c panicked) + [ "$P" = "0" ] && ok "no panic in the log" || ko "$P panic(s) in the log" + UNIT=$(topology "$H2") + echo "$UNIT" | grep -q '"restarts": 0' && ok "no restart was needed" || ko "the connector restarted: $UNIT" +fi + +[ $fail -eq 0 ] && echo "HTTP-IN BOUND HOLDS" || echo "HTTP-IN BOUND REGRESSION — do not ship" +exit $fail