security: v0.3.6 fixes for secrets, emit subjects, governed mode and http-in

.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

.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

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 `<secret:path/key>`,
+  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

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/<suffix> publishes the JSON body on vx.<suffix>. 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/<suffix> publishes the JSON body on vx.<suffix>. 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<AtomicUsize>);
+
+impl Slot {
+    pub fn try_take(open: &Arc<AtomicUsize>, max: usize) -> Option<Slot> {
+        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::<Value>(body).is_err() {
                     ("400 Bad Request", "{\"error\":\"body must be JSON\"}".to_string())
                 } else if !allow.is_empty() && !suffix_allowed(suffix, allow) {