Skip to main content

plecto_host/
state.rs

1//! Per-request host state ([`HostState`]) and the host-API capability implementations
2//! (deny-by-default: only these are lent to a filter, ADR 000006 / 000011).
3
4use std::collections::BTreeMap;
5use std::sync::Arc;
6
7#[cfg(any(
8    feature = "outbound-http",
9    feature = "outbound-tcp",
10    feature = "fat-guest",
11    feature = "streaming-body"
12))]
13use wasmtime::component::Linker;
14use wasmtime::{StoreLimits, StoreLimitsBuilder};
15
16use crate::LogLevel;
17use crate::bindings;
18use crate::bindings::plecto::filter::{
19    host_clock, host_config, host_counter, host_kv, host_log, host_ratelimit,
20};
21#[cfg(feature = "outbound-http")]
22use crate::outbound_http;
23#[cfg(feature = "outbound-tcp")]
24use crate::outbound_tcp;
25use crate::quota::KvQuota;
26use crate::util::wall_now_ms;
27use crate::{Bucket, KvBackend};
28
29/// A log line captured from the host-log capability (test visibility / future tracing).
30#[derive(Debug, Clone, PartialEq, Eq)]
31pub struct LogLine {
32    pub level: LogLevel,
33    pub message: String,
34}
35
36/// Delimiter the host uses to namespace KV keys by filter identity. A filter can never
37/// remove the host-applied prefix, so it cannot reach another filter's namespace —
38/// capability isolation across a chain (ADR 000006 / 000011). Filter ids must not contain
39/// this byte (enforced by `Host::load`).
40pub(crate) const KV_NS_DELIM: char = '\u{1f}';
41
42// Primitive sub-namespace tags, so a filter's kv "x", counter "x", and bucket "x" never
43// collide in the shared backend keyspace.
44const TAG_KV: u8 = b'k';
45const TAG_COUNTER: u8 = b'c';
46const TAG_RATELIMIT: u8 = b'r';
47
48/// Largest value a filter may store under one KV key. A bigger `set` is dropped (fail-closed).
49const MAX_KV_VALUE_BYTES: usize = 256 * 1024;
50/// Largest filter-supplied key. A longer key is dropped (bounds the namespaced key itself).
51const MAX_KV_KEY_BYTES: usize = 1024;
52/// Per-request cap on host-log lines a filter may emit (CWE-770). The last slot is a
53/// single truncation marker so overflow stays observable.
54const MAX_LOG_LINES_PER_REQUEST: usize = 256;
55/// Per-line cap on a host-log message; a longer message is truncated on a char boundary.
56const MAX_LOG_MSG_BYTES: usize = 8 * 1024;
57
58/// Per-instance cap on total table elements (review f000003 #2), shared by the sync and
59/// streaming runtimes. `StoreLimits::memory_size` bounds linear memory but NOT `table.grow`;
60/// a guest growing a huge funcref table could eat host memory outside the linear-memory cap
61/// before the epoch deadline trips. This is generous for any reasonable filter and bounds the
62/// pathological case — cheap defense-in-depth.
63pub(crate) const MAX_TABLE_ELEMENTS: usize = 100_000;
64
65/// Neutralize a guest-supplied log message (CWE-117): truncate to `max_bytes` on a char
66/// boundary and replace control characters (CR/LF for log-line injection, C0/C1/ESC for terminal
67/// ANSI) with the replacement char. The filter is untrusted and may embed `Authorization` header
68/// bytes or escape sequences, so the host — the trust boundary — neutralizes before storing.
69/// Shared by every `LogBudget` impl (each brings its own `max_bytes`).
70pub(crate) fn sanitize_log_message(mut message: String, max_bytes: usize) -> String {
71    if message.len() > max_bytes {
72        let mut end = max_bytes;
73        while !message.is_char_boundary(end) {
74            end -= 1;
75        }
76        message.truncate(end);
77    }
78    if message.bytes().any(|b| b < 0x20 || b == 0x7f) {
79        message = message
80            .chars()
81            .map(|c| if c.is_control() { '\u{fffd}' } else { c })
82            .collect();
83    }
84    message
85}
86
87/// A per-request budget gate for guest-emitted log volume (CWE-770): decides whether/how to
88/// record one candidate line, sanitizing and truncating it in the process. Two independent
89/// policies implement this — `host-log`'s existing per-request LINE-COUNT cap
90/// ([`LineCountBudget`]) and the WASI stdio bridge's per-request BYTE budget (`ByteBudget`, ADR
91/// 000063) — sharing one interface so a chatty guest's stdout/stderr cannot starve its own
92/// explicit `host-log` calls of their budget, or vice versa, even though both funnel into the
93/// same [`HostState::logs`](HostState) → OTLP span-event path.
94pub(crate) trait LogBudget {
95    /// `level` is the line's natural level (the caller's level for a `host-log` call; a fixed
96    /// per-stream level for the stdio bridge). A truncation event always downgrades the stored
97    /// level to `LogLevel::Warn` regardless of `level`. Returns the exact `(level, message)` to
98    /// store, already sanitized/truncated, or `None` to drop this line silently (budget already
99    /// exhausted by an earlier call).
100    fn admit(&mut self, level: LogLevel, message: String) -> Option<(LogLevel, String)>;
101}
102
103/// [`LogBudget`] for `host-log`: caps the per-request LINE COUNT (not bytes) at `max_lines`,
104/// reserving the last slot for one truncation marker so overflow stays observable.
105pub(crate) struct LineCountBudget {
106    max_lines: usize,
107    count: usize,
108}
109
110impl LineCountBudget {
111    pub(crate) fn new(max_lines: usize) -> Self {
112        Self {
113            max_lines,
114            count: 0,
115        }
116    }
117
118    /// Reset for the next request (pooled/trusted instances reuse the same budget).
119    pub(crate) fn reset(&mut self) {
120        self.count = 0;
121    }
122}
123
124impl LogBudget for LineCountBudget {
125    fn admit(&mut self, level: LogLevel, message: String) -> Option<(LogLevel, String)> {
126        let last_slot = self.max_lines.saturating_sub(1);
127        if self.count < last_slot {
128            self.count += 1;
129            Some((level, sanitize_log_message(message, MAX_LOG_MSG_BYTES)))
130        } else if self.count == last_slot {
131            self.count += 1;
132            Some((
133                LogLevel::Warn,
134                "… host-log truncated (per-request line cap reached)".to_string(),
135            ))
136        } else {
137            None
138        }
139    }
140}
141
142/// Per-request host state: the capability handles lent to a filter plus request-scoped
143/// buffers. For untrusted filters a fresh one is built per request; for trusted filters
144/// the same one is reused with `begin_request` resetting the per-request fields, while the
145/// instance's init-derived linear memory persists (ADR 000011).
146pub struct HostState {
147    kv: Arc<dyn KvBackend>,
148    /// Host-owned prefix (`"{filter_id}\u{1f}"`) applied to every key. The filter cannot
149    /// observe or alter it.
150    kv_prefix: String,
151    /// Per-request host-log buffer. `pub(crate)`: `runtime.rs`'s `WasmtimeRuntime::take_logs`
152    /// drains it directly after a guest call (structural cross-module access, no behavior change).
153    pub(crate) logs: Vec<LogLine>,
154    /// `host-log`'s per-request [`LogBudget`] (line-count policy). Independent of the stdio
155    /// bridge's own [`ByteBudget`](crate::stdio::ByteBudget) (ADR 000063) so a chatty guest's
156    /// stdout/stderr cannot starve this filter's own explicit `host-log` calls, or vice versa.
157    host_log_budget: LineCountBudget,
158    /// Wall-clock ms captured once at request start: a stable per-request snapshot.
159    now_ms: u64,
160    /// Linear-memory / table / instance caps for this Store (ADR 000006). Wired via
161    /// `Store::limiter`; a grow past the cap is denied, bounding mis-allocation and runaway
162    /// growth even on the untrusted on-demand engine (which has no pooling reservation).
163    /// `pub(crate)`: `runtime.rs` wires `Store::limiter` to this field directly.
164    pub(crate) limits: StoreLimits,
165    /// This filter's host-configured token-bucket spec (manifest `[filter.ratelimit]`, ADR
166    /// 000026). `None` = no bucket configured → `host-ratelimit/try-acquire` fails closed. The
167    /// filter cannot supply or override it, so an untrusted filter cannot neuter its own limiter.
168    ratelimit_bucket: Option<Bucket>,
169    /// Shared per-namespace accounting + caps for host-held state. Charged on every
170    /// `set` / `increment` / `try_acquire` that grows the store; over-quota writes fail closed.
171    quota: Arc<KvQuota>,
172    /// Outbound capabilities (ADR 000036 HTTP / ADR 000060 TCP) and the fat-guest minimal WASI
173    /// grant (ADR 000063) share this minimal WASI base ctx and resource table. The relevant
174    /// interfaces are added to the Linker only for filters with an outbound policy or a `wasi =
175    /// "minimal"` declaration; for every other filter the guards below deny every call
176    /// (belt-and-suspenders, so a stray call still fails closed).
177    #[cfg(any(
178        feature = "outbound-http",
179        feature = "outbound-tcp",
180        feature = "fat-guest"
181    ))]
182    wasi: wasmtime_wasi::WasiCtx,
183    #[cfg(any(
184        feature = "outbound-http",
185        feature = "outbound-tcp",
186        feature = "fat-guest"
187    ))]
188    table: wasmtime::component::ResourceTable,
189    /// Outbound HTTP (ADR 000036): the `wasi:http` ctx and the SSRF-guarded send hooks.
190    #[cfg(feature = "outbound-http")]
191    http_ctx: wasmtime_wasi_http::WasiHttpCtx,
192    #[cfg(feature = "outbound-http")]
193    hooks: outbound_http::PlectoHttpHooks,
194    /// Outbound TCP (ADR 000060): the per-Store guard (pinned set + connect budget) shared by the
195    /// `socket_addr_check` closure inside `wasi` above and the host's ip-name-lookup impl.
196    #[cfg(feature = "outbound-tcp")]
197    tcp: outbound_tcp::TcpGuard,
198    /// Fat guest (ADR 000063): `Some` only for a filter this build lent the minimal WASI grant
199    /// AND that declared `wasi = "minimal"` — its stdout/stderr is bridged here, drained into
200    /// `logs` (above) by [`HostState::take_logs`]. `None` for every other filter, including a
201    /// fat-guest-capable build where this filter declared no such grant.
202    #[cfg(feature = "fat-guest")]
203    stdio_bridge: Option<crate::stdio::StdioBridge>,
204    /// This filter's manifest-declared business config (`[filter.config]`, ADR 000066) — a
205    /// read-only string map the filter reads back via `host-config`. The host never interprets it.
206    config: Arc<BTreeMap<String, String>>,
207}
208
209/// The always-present [`HostState::new`] fields, grouped so the constructor's argument count
210/// stays under clippy's threshold without an `#[allow]` — the outbound capabilities stay
211/// separate, cfg-gated trailing params, since they don't exist in every build.
212pub(crate) struct HostStateInit {
213    pub(crate) kv: Arc<dyn KvBackend>,
214    pub(crate) kv_prefix: String,
215    pub(crate) max_memory_bytes: u64,
216    pub(crate) ratelimit_bucket: Option<Bucket>,
217    pub(crate) quota: Arc<KvQuota>,
218    pub(crate) config: Arc<BTreeMap<String, String>>,
219    /// This filter's manifest `wasi = "minimal"` declaration (ADR 000063). Meaningless (but
220    /// harmless) when the `fat-guest` feature is off — the manifest validator rejects the
221    /// declaration on such a build before `HostState` is ever constructed.
222    #[cfg(feature = "fat-guest")]
223    pub(crate) wasi_minimal: bool,
224}
225
226impl HostState {
227    pub(crate) fn new(
228        init: HostStateInit,
229        #[cfg(feature = "outbound-http")] hooks: outbound_http::PlectoHttpHooks,
230        #[cfg(feature = "outbound-tcp")] tcp: outbound_tcp::TcpGuard,
231    ) -> Self {
232        let HostStateInit {
233            kv,
234            kv_prefix,
235            max_memory_bytes,
236            ratelimit_bucket,
237            quota,
238            config,
239            #[cfg(feature = "fat-guest")]
240            wasi_minimal,
241        } = init;
242        // The base WasiCtx builder: empty (no fs, no env, no preopens) until a lent capability
243        // configures its own slice. Outbound TCP installs its `socket_addr_check`; fat guest
244        // wires stdout/stderr into this filter's host-log (ADR 000063). Neither present leaves
245        // the builder's deny-all-sockets / discard-stdio defaults standing.
246        #[cfg(any(
247            feature = "outbound-http",
248            feature = "outbound-tcp",
249            feature = "fat-guest"
250        ))]
251        let mut wasi_builder = wasmtime_wasi::WasiCtxBuilder::new();
252        #[cfg(feature = "outbound-tcp")]
253        tcp.configure_wasi_ctx(&mut wasi_builder);
254        #[cfg(feature = "fat-guest")]
255        let stdio_bridge = if wasi_minimal {
256            let bridge = crate::stdio::StdioBridge::new();
257            wasi_builder.stdout(bridge.stream(crate::stdio::StdioKind::Stdout));
258            wasi_builder.stderr(bridge.stream(crate::stdio::StdioKind::Stderr));
259            Some(bridge)
260        } else {
261            None
262        };
263        Self {
264            kv,
265            kv_prefix,
266            logs: Vec::new(),
267            host_log_budget: LineCountBudget::new(MAX_LOG_LINES_PER_REQUEST),
268            now_ms: wall_now_ms(),
269            limits: StoreLimitsBuilder::new()
270                .memory_size(max_memory_bytes as usize)
271                .table_elements(MAX_TABLE_ELEMENTS)
272                .build(),
273            ratelimit_bucket,
274            quota,
275            config,
276            #[cfg(any(
277                feature = "outbound-http",
278                feature = "outbound-tcp",
279                feature = "fat-guest"
280            ))]
281            wasi: wasi_builder.build(),
282            #[cfg(any(
283                feature = "outbound-http",
284                feature = "outbound-tcp",
285                feature = "fat-guest"
286            ))]
287            table: wasmtime::component::ResourceTable::new(),
288            #[cfg(feature = "outbound-http")]
289            http_ctx: wasmtime_wasi_http::WasiHttpCtx::new(),
290            #[cfg(feature = "outbound-http")]
291            hooks,
292            #[cfg(feature = "outbound-tcp")]
293            tcp,
294            #[cfg(feature = "fat-guest")]
295            stdio_bridge,
296        }
297    }
298
299    /// Drain this request's host-log lines, merging in any lines the fat-guest stdio bridge
300    /// queued (ADR 000063) — both feed the same OTLP span-event path
301    /// (`observe::build_filter_span`), so a trapped guest's stdout/stderr shows up in the same
302    /// trace as the failing request. Called once per guest call, from `runtime.rs`.
303    pub(crate) fn take_logs(&mut self) -> Vec<LogLine> {
304        #[cfg(feature = "fat-guest")]
305        if let Some(bridge) = &self.stdio_bridge {
306            self.logs.extend(bridge.drain());
307        }
308        std::mem::take(&mut self.logs)
309    }
310
311    /// Like [`HostState::take_logs`], but for an instance about to be discarded — a trap, or a
312    /// fresh/untrusted instance's `Ok` arm (fresh instances are always single-use regardless of
313    /// outcome): this also flushes any still-unterminated stdio partial line (ADR 000063) — e.g.
314    /// a TinyGo panic message written to stderr with no trailing newline right before the trap —
315    /// instead of losing it along with the instance. `runtime.rs`/`pool.rs` call this instead of
316    /// `take_logs` whenever the instance will not be reused, so a discarded instance's own
317    /// diagnostic output still reaches `emit_span`, which is the whole point of the stdio bridge.
318    pub(crate) fn take_logs_final(&mut self) -> Vec<LogLine> {
319        #[cfg(feature = "fat-guest")]
320        if let Some(bridge) = &self.stdio_bridge {
321            self.logs.extend(bridge.drain_final());
322        }
323        std::mem::take(&mut self.logs)
324    }
325
326    /// Reset per-request state for a reused (trusted) instance. Clears the log buffer and
327    /// re-snapshots the clock; the WASM instance's linear memory (init-derived) is untouched.
328    pub(crate) fn begin_request(&mut self) {
329        self.logs.clear();
330        self.host_log_budget.reset();
331        #[cfg(feature = "fat-guest")]
332        if let Some(bridge) = &self.stdio_bridge {
333            bridge.begin_request();
334        }
335        self.now_ms = wall_now_ms();
336        // A reused instance's connect budget belongs to the request, not the instance.
337        #[cfg(feature = "outbound-tcp")]
338        self.tcp.begin_request();
339    }
340
341    /// The host's vetted `wasi:sockets/ip-name-lookup` view (ADR 000060): the Store's resource
342    /// table plus this filter's guard.
343    #[cfg(feature = "outbound-tcp")]
344    pub(crate) fn tcp_lookup(&mut self) -> outbound_tcp::TcpLookupView<'_> {
345        outbound_tcp::TcpLookupView {
346            table: &mut self.table,
347            guard: &self.tcp,
348        }
349    }
350
351    /// Namespace a filter-supplied key into `{filter_id}\u{1f}{tag}\u{1f}{key}` bytes.
352    fn ns_key(&self, tag: u8, key: &str) -> Vec<u8> {
353        let mut k = Vec::with_capacity(self.kv_prefix.len() + 2 + key.len());
354        k.extend_from_slice(self.kv_prefix.as_bytes());
355        k.push(tag);
356        k.push(KV_NS_DELIM as u8);
357        k.extend_from_slice(key.as_bytes());
358        k
359    }
360}
361
362// --- host-API capability implementations (deny-by-default: only these are lent) ---
363
364// Outbound capabilities (ADR 000036 / 000060) and the fat-guest grant (ADR 000063): the WASI
365// base + per-capability projections, added to the Linker only for filters with an outbound
366// policy or a `wasi = "minimal"` declaration. They share one resource table.
367#[cfg(any(
368    feature = "outbound-http",
369    feature = "outbound-tcp",
370    feature = "fat-guest"
371))]
372impl wasmtime_wasi::WasiView for HostState {
373    fn ctx(&mut self) -> wasmtime_wasi::WasiCtxView<'_> {
374        wasmtime_wasi::WasiCtxView {
375            ctx: &mut self.wasi,
376            table: &mut self.table,
377        }
378    }
379}
380
381#[cfg(feature = "outbound-http")]
382impl wasmtime_wasi_http::p2::WasiHttpView for HostState {
383    fn http(&mut self) -> wasmtime_wasi_http::p2::WasiHttpCtxView<'_> {
384        wasmtime_wasi_http::p2::WasiHttpCtxView {
385            ctx: &mut self.http_ctx,
386            table: &mut self.table,
387            hooks: &mut self.hooks,
388        }
389    }
390}
391
392/// Add the wasi:cli interfaces a std guest's runtime imports beyond the proxy slice (environment /
393/// exit / terminal-*), each inert under an empty `WasiCtx` (environment returns `[]`, exit traps,
394/// the terminals are not TTYs). Adds NO filesystem and NO sockets, so those capabilities stay
395/// denied (security audit F-002). Generic over the embedder state `T` so the sync host
396/// (`HostState`, here) and the experimental async streaming host (`streaming::Ctx`) share one
397/// definition instead of two copies of the same wiring; outbound-TCP filters (ADR 000060) get
398/// their `wasi:sockets` slice added separately, behind their own guard.
399#[cfg(any(
400    feature = "outbound-http",
401    feature = "outbound-tcp",
402    feature = "fat-guest",
403    feature = "streaming-body"
404))]
405pub(crate) fn add_cli_runtime<T: wasmtime_wasi::cli::WasiCliView>(
406    linker: &mut Linker<T>,
407) -> wasmtime::Result<()> {
408    use wasmtime_wasi::cli::{WasiCli, WasiCliView};
409    use wasmtime_wasi::p2::bindings::cli;
410    let getter = <T as WasiCliView>::cli;
411    cli::environment::add_to_linker::<T, WasiCli>(linker, getter)?;
412    cli::exit::add_to_linker::<T, WasiCli>(linker, getter)?;
413    cli::terminal_input::add_to_linker::<T, WasiCli>(linker, getter)?;
414    cli::terminal_output::add_to_linker::<T, WasiCli>(linker, getter)?;
415    cli::terminal_stdin::add_to_linker::<T, WasiCli>(linker, getter)?;
416    cli::terminal_stdout::add_to_linker::<T, WasiCli>(linker, getter)?;
417    cli::terminal_stderr::add_to_linker::<T, WasiCli>(linker, getter)?;
418    Ok(())
419}
420
421/// Link an EMPTY `wasi:filesystem` (`types` + `preopens`) for a fat guest declaring `wasi =
422/// "minimal"` (ADR 000063). Confirmed empirically (TinyGo 0.41.1): a wasip2 TinyGo guest
423/// unconditionally imports `wasi:filesystem/types` + `wasi:filesystem/preopens` as part of its
424/// runtime bootstrap, even for a program that never touches a file — so Tier B must LINK the
425/// interface for such a guest to instantiate at all. This lends ZERO real capability: the
426/// `WasiCtx`'s preopens list is never populated (no `WasiCtxBuilder::preopened_dir` call
427/// anywhere in the fat-guest path), so `preopens.get-directories()` returns `[]` and the guest
428/// can never obtain a `descriptor` resource to operate on — filesystem access stays structurally
429/// unreachable, exactly like `add_cli_runtime`'s inert `wasi:cli/environment`. Not part of
430/// `add_cli_runtime` itself: outbound-http/outbound-tcp/streaming-body guests are Rust-authored
431/// (no forced runtime import), so only the fat-guest path needs this.
432#[cfg(feature = "fat-guest")]
433pub(crate) fn add_inert_filesystem<T: wasmtime_wasi::filesystem::WasiFilesystemView>(
434    linker: &mut Linker<T>,
435) -> wasmtime::Result<()> {
436    use wasmtime_wasi::filesystem::{WasiFilesystem, WasiFilesystemView};
437    use wasmtime_wasi::p2::bindings::filesystem;
438    let getter = <T as WasiFilesystemView>::filesystem;
439    filesystem::types::add_to_linker::<T, WasiFilesystem>(linker, getter)?;
440    filesystem::preopens::add_to_linker::<T, WasiFilesystem>(linker, getter)?;
441    Ok(())
442}
443
444// `types` is a type-only interface (no functions); the generated `Host` trait is empty.
445impl bindings::plecto::filter::types::Host for HostState {}
446
447impl host_log::Host for HostState {
448    fn log(&mut self, level: LogLevel, message: String) {
449        // `LineCountBudget` bounds per-request log volume (CWE-770: a guest can loop `log` until
450        // its deadline) and `sanitize_log_message` (inside `admit`) neutralizes control bytes.
451        if let Some((level, message)) = self.host_log_budget.admit(level, message) {
452            self.logs.push(LogLine { level, message });
453        }
454    }
455}
456
457impl host_clock::Host for HostState {
458    fn now_ms(&mut self) -> u64 {
459        self.now_ms
460    }
461}
462
463impl host_kv::Host for HostState {
464    fn get(&mut self, key: String) -> Option<Vec<u8>> {
465        self.kv.get(&self.ns_key(TAG_KV, &key))
466    }
467    fn set(&mut self, key: String, value: Vec<u8>) {
468        // Per-key size limits + per-namespace/global quota. Over-limit writes are dropped
469        // (fail-closed): from the filter's view the host-API is infallible ("reads vanish").
470        if key.len() > MAX_KV_KEY_BYTES || value.len() > MAX_KV_VALUE_BYTES {
471            return;
472        }
473        let nskey = self.ns_key(TAG_KV, &key);
474        let kv = &self.kv;
475        let key_len = key.len();
476        let value_len = value.len();
477        // Charge the byte delta vs. any existing value (a new key also charges its key bytes + 1
478        // entry), then write — atomically with the quota decision (`KvQuota::charge_and_apply`),
479        // so a concurrent `set` on the same key (the pool runs many instances of one filter at
480        // once) cannot race the read-old-value step against this call's write. The closures run
481        // synchronously inside `charge_and_apply`, so they BORROW `nskey` / the backend (no
482        // per-call key clones or `Arc` bumps on this per-guest-call path — DECREE §4).
483        self.quota.charge_and_apply(
484            &self.kv_prefix,
485            &nskey,
486            || match kv.get(&nskey).map(|v| v.len()) {
487                None => (1isize, (key_len + value_len) as isize),
488                Some(old) => (0isize, value_len as isize - old as isize),
489            },
490            || kv.set(&nskey, value),
491        );
492    }
493    fn delete(&mut self, key: String) {
494        let nskey = self.ns_key(TAG_KV, &key);
495        let kv = &self.kv;
496        let key_len = key.len();
497        // Read-then-release must be atomic with the quota decision too: two concurrent deletes
498        // of the same key must not both observe `Some(old)` and both release — the second must
499        // see the first's delete has already happened and release nothing (see
500        // `KvQuota::charge_and_apply` doc for the race this closes).
501        self.quota.charge_and_apply(
502            &self.kv_prefix,
503            &nskey,
504            || match kv.get(&nskey).map(|v| v.len()) {
505                Some(old) => (-1isize, -((key_len + old) as isize)),
506                None => (0isize, 0isize),
507            },
508            || kv.delete(&nskey),
509        );
510    }
511}
512
513impl host_counter::Host for HostState {
514    fn increment(&mut self, key: String, delta: i64) -> i64 {
515        let nskey = self.ns_key(TAG_COUNTER, &key);
516        // A zero delta is a pure read (host-counter.get); it neither creates a key nor is charged.
517        if delta == 0 {
518            return self.kv.increment(&nskey, 0);
519        }
520        let kv = &self.kv;
521        let key_len = key.len();
522        // A counter is a fixed 8-byte value: only a NEW key grows the store, so charge one entry
523        // when first created and fail closed (report the current value, do not create) over quota
524        // — atomically with the increment itself (`KvQuota::charge_and_apply`), so two concurrent
525        // first-writes to the same new key cannot both observe "absent" and both charge an entry
526        // for what ends up being one logical key (the pool runs many concurrent instances of the
527        // same filter, all sharing this backend + quota).
528        self.quota
529            .charge_and_apply(
530                &self.kv_prefix,
531                &nskey,
532                || {
533                    if kv.get(&nskey).is_none() {
534                        (1isize, (key_len + 8) as isize)
535                    } else {
536                        (0isize, 0isize)
537                    }
538                },
539                || kv.increment(&nskey, delta),
540            )
541            .unwrap_or(0)
542    }
543    fn get(&mut self, key: String) -> i64 {
544        // increment-by-zero is an atomic read of the current value (and the canonical
545        // wasi:keyvalue/atomics idiom); keeps the counter encoding inside the backend.
546        self.kv.increment(&self.ns_key(TAG_COUNTER, &key), 0)
547    }
548}
549
550impl host_ratelimit::Host for HostState {
551    fn try_acquire(&mut self, key: String, cost: u64) -> host_ratelimit::Acquire {
552        // The bucket spec is host-configured per filter (manifest, ADR 000026); the filter cannot
553        // supply or override it. A filter with no configured bucket is denied (fail-closed) — it
554        // cannot opt out of its limiter.
555        let Some(spec) = self.ratelimit_bucket else {
556            return host_ratelimit::Acquire {
557                allowed: false,
558                remaining: 0,
559                retry_after_ms: 0,
560            };
561        };
562        let nskey = self.ns_key(TAG_RATELIMIT, &key);
563        let kv = &self.kv;
564        let key_len = key.len();
565        let now_ms = self.now_ms;
566        // A bucket is a fixed 16-byte value: charge one entry when first created. Over quota a
567        // filter cannot mint unbounded distinct-key buckets — deny (fail-closed), do not create.
568        // Reserving the entry and acquiring the bucket happen atomically
569        // (`KvQuota::charge_and_apply`), so two concurrent first-acquires on the same new key
570        // cannot both observe "absent" and both charge an entry for one logical bucket.
571        let result = self.quota.charge_and_apply(
572            &self.kv_prefix,
573            &nskey,
574            || {
575                if kv.get(&nskey).is_none() {
576                    (1isize, (key_len + 16) as isize)
577                } else {
578                    (0isize, 0isize)
579                }
580            },
581            || kv.try_acquire(&nskey, cost, spec, now_ms),
582        );
583        match result {
584            Some(r) => host_ratelimit::Acquire {
585                allowed: r.allowed,
586                remaining: r.remaining,
587                retry_after_ms: r.retry_after_ms,
588            },
589            None => host_ratelimit::Acquire {
590                allowed: false,
591                remaining: 0,
592                retry_after_ms: 0,
593            },
594        }
595    }
596}
597
598impl host_config::Host for HostState {
599    fn get(&mut self, key: String) -> Option<String> {
600        self.config.get(&key).cloned()
601    }
602}
603
604#[cfg(test)]
605mod tests {
606    //! Unit tests for the deny-by-default host-API implementations (ADR 000006 / 000011).
607    use super::*;
608    use crate::MemoryBackend;
609    use crate::options::DEFAULT_MAX_MEMORY_BYTES;
610    use host_clock::Host as ClockHost;
611    use host_config::Host as ConfigHost;
612    use host_counter::Host as CounterHost;
613    use host_kv::Host as KvHost;
614    use host_log::Host as LogHost;
615
616    /// A [`HostStateInit`] with test-friendly defaults: a fresh in-memory backend, no rate-limit
617    /// bucket, a fresh quota, and no business config. Callers override individual fields.
618    fn init_for(prefix: &str) -> HostStateInit {
619        HostStateInit {
620            kv: Arc::new(MemoryBackend::default()),
621            kv_prefix: prefix.to_string(),
622            max_memory_bytes: DEFAULT_MAX_MEMORY_BYTES,
623            ratelimit_bucket: None,
624            quota: Arc::new(KvQuota::new()),
625            config: Arc::new(BTreeMap::new()),
626            #[cfg(feature = "fat-guest")]
627            wasi_minimal: false,
628        }
629    }
630
631    fn state(prefix: &str) -> HostState {
632        HostState::new(
633            init_for(prefix),
634            #[cfg(feature = "outbound-http")]
635            outbound_http::PlectoHttpHooks::deny_all(),
636            #[cfg(feature = "outbound-tcp")]
637            crate::outbound_tcp::TcpGuard::deny_all(),
638        )
639    }
640
641    #[test]
642    fn config_get_is_a_readonly_passthrough_of_declared_keys() {
643        // host-config (ADR 000066): a declared `[filter.config]` key comes back verbatim, an
644        // undeclared one is None — the host never interprets or defaults anything.
645        let mut s = HostState::new(
646            HostStateInit {
647                config: Arc::new(BTreeMap::from([(
648                    "on_backend_error".to_string(),
649                    "deny".to_string(),
650                )])),
651                ..init_for("cfg\u{1f}")
652            },
653            #[cfg(feature = "outbound-http")]
654            outbound_http::PlectoHttpHooks::deny_all(),
655            #[cfg(feature = "outbound-tcp")]
656            crate::outbound_tcp::TcpGuard::deny_all(),
657        );
658        assert_eq!(
659            ConfigHost::get(&mut s, "on_backend_error".into()),
660            Some("deny".to_string())
661        );
662        assert_eq!(ConfigHost::get(&mut s, "window_secs".into()), None);
663    }
664
665    #[test]
666    fn kv_get_set_delete_roundtrip() {
667        let mut s = state("test\u{1f}");
668        assert_eq!(KvHost::get(&mut s, "k".into()), None);
669        KvHost::set(&mut s, "k".into(), b"v".to_vec());
670        assert_eq!(KvHost::get(&mut s, "k".into()), Some(b"v".to_vec()));
671        KvHost::delete(&mut s, "k".into());
672        assert_eq!(KvHost::get(&mut s, "k".into()), None);
673    }
674
675    #[test]
676    fn kv_is_namespaced_per_filter() {
677        // Two filters sharing one backing store must not see each other's keys
678        // (capability isolation across a chain, ADR 000006 / 000011).
679        let shared: Arc<dyn KvBackend> = Arc::new(MemoryBackend::default());
680        let mut a = HostState::new(
681            HostStateInit {
682                kv: shared.clone(),
683                ..init_for("filter-a\u{1f}")
684            },
685            #[cfg(feature = "outbound-http")]
686            outbound_http::PlectoHttpHooks::deny_all(),
687            #[cfg(feature = "outbound-tcp")]
688            crate::outbound_tcp::TcpGuard::deny_all(),
689        );
690        let mut b = HostState::new(
691            HostStateInit {
692                kv: shared.clone(),
693                ..init_for("filter-b\u{1f}")
694            },
695            #[cfg(feature = "outbound-http")]
696            outbound_http::PlectoHttpHooks::deny_all(),
697            #[cfg(feature = "outbound-tcp")]
698            crate::outbound_tcp::TcpGuard::deny_all(),
699        );
700
701        KvHost::set(&mut a, "count".into(), b"1".to_vec());
702        assert_eq!(
703            KvHost::get(&mut b, "count".into()),
704            None,
705            "b must not see a"
706        );
707        assert_eq!(KvHost::get(&mut a, "count".into()), Some(b"1".to_vec()));
708
709        // a key that embeds the delimiter still cannot escape a's namespace
710        KvHost::set(&mut a, format!("x{}count", '\u{1f}'), b"evil".to_vec());
711        assert_eq!(KvHost::get(&mut b, "count".into()), None);
712    }
713
714    #[test]
715    fn counter_is_namespaced_per_filter() {
716        // The counter primitive shares the backend keyspace with kv/ratelimit, so its per-filter
717        // isolation must hold too: one filter's `requests` counter must be invisible to another
718        // (cross-tenant leakage, CWE-200). Only the `_KV_` test covered this before.
719        let shared: Arc<dyn KvBackend> = Arc::new(MemoryBackend::default());
720        let mut a = HostState::new(
721            HostStateInit {
722                kv: shared.clone(),
723                ..init_for("filter-a\u{1f}")
724            },
725            #[cfg(feature = "outbound-http")]
726            outbound_http::PlectoHttpHooks::deny_all(),
727            #[cfg(feature = "outbound-tcp")]
728            crate::outbound_tcp::TcpGuard::deny_all(),
729        );
730        let mut b = HostState::new(
731            HostStateInit {
732                kv: shared.clone(),
733                ..init_for("filter-b\u{1f}")
734            },
735            #[cfg(feature = "outbound-http")]
736            outbound_http::PlectoHttpHooks::deny_all(),
737            #[cfg(feature = "outbound-tcp")]
738            crate::outbound_tcp::TcpGuard::deny_all(),
739        );
740
741        assert_eq!(CounterHost::increment(&mut a, "hits".into(), 5), 5);
742        assert_eq!(
743            CounterHost::get(&mut b, "hits".into()),
744            0,
745            "b must not observe a's counter"
746        );
747        assert_eq!(
748            CounterHost::increment(&mut b, "hits".into(), 1),
749            1,
750            "b's counter is independent of a's"
751        );
752        assert_eq!(
753            CounterHost::get(&mut a, "hits".into()),
754            5,
755            "a's counter is untouched by b"
756        );
757    }
758
759    #[test]
760    fn ratelimit_bucket_is_namespaced_per_filter() {
761        // A rate limiter is only a security control if one filter cannot drain — or be throttled
762        // by — another filter's bucket under the same key. The token bucket lives in the shared
763        // backend under a per-filter namespace; prove two filters' identical keys are independent.
764        use host_ratelimit::Host as RateLimitHost;
765        fn one_token_no_refill() -> Bucket {
766            Bucket {
767                capacity: 1,
768                refill_tokens: 0,
769                refill_interval_ms: 0,
770            }
771        }
772
773        // The bucket spec is host-configured (ADR 000026), so each filter's HostState carries it.
774        let shared: Arc<dyn KvBackend> = Arc::new(MemoryBackend::default());
775        let mut a = HostState::new(
776            HostStateInit {
777                kv: shared.clone(),
778                ratelimit_bucket: Some(one_token_no_refill()),
779                ..init_for("filter-a\u{1f}")
780            },
781            #[cfg(feature = "outbound-http")]
782            outbound_http::PlectoHttpHooks::deny_all(),
783            #[cfg(feature = "outbound-tcp")]
784            crate::outbound_tcp::TcpGuard::deny_all(),
785        );
786        let mut b = HostState::new(
787            HostStateInit {
788                kv: shared.clone(),
789                ratelimit_bucket: Some(one_token_no_refill()),
790                ..init_for("filter-b\u{1f}")
791            },
792            #[cfg(feature = "outbound-http")]
793            outbound_http::PlectoHttpHooks::deny_all(),
794            #[cfg(feature = "outbound-tcp")]
795            crate::outbound_tcp::TcpGuard::deny_all(),
796        );
797
798        // a drains its single-token bucket on key "k".
799        assert!(RateLimitHost::try_acquire(&mut a, "k".into(), 1).allowed);
800        assert!(
801            !RateLimitHost::try_acquire(&mut a, "k".into(), 1).allowed,
802            "a's bucket is now empty"
803        );
804        // b's bucket under the SAME key is a different namespace → still full.
805        assert!(
806            RateLimitHost::try_acquire(&mut b, "k".into(), 1).allowed,
807            "b's limiter must not share a's drained bucket"
808        );
809    }
810
811    #[test]
812    fn kv_and_counter_do_not_collide() {
813        // Same logical key under different primitives must stay distinct (tag sub-namespace).
814        let mut s = state("f\u{1f}");
815        KvHost::set(&mut s, "x".into(), b"bytes".to_vec());
816        assert_eq!(CounterHost::increment(&mut s, "x".into(), 7), 7);
817        assert_eq!(KvHost::get(&mut s, "x".into()), Some(b"bytes".to_vec()));
818        assert_eq!(CounterHost::get(&mut s, "x".into()), 7);
819    }
820
821    #[test]
822    fn counter_increment_and_read() {
823        let mut s = state("f\u{1f}");
824        assert_eq!(CounterHost::get(&mut s, "n".into()), 0);
825        assert_eq!(CounterHost::increment(&mut s, "n".into(), 1), 1);
826        assert_eq!(CounterHost::increment(&mut s, "n".into(), 2), 3);
827        assert_eq!(CounterHost::get(&mut s, "n".into()), 3);
828    }
829
830    #[test]
831    fn log_captures_lines() {
832        let mut s = state("test\u{1f}");
833        LogHost::log(&mut s, LogLevel::Info, "hello".into());
834        assert_eq!(s.logs.len(), 1);
835        assert_eq!(s.logs[0].message, "hello");
836    }
837
838    #[test]
839    fn begin_request_resets_logs_keeps_namespace() {
840        let mut s = state("test\u{1f}");
841        LogHost::log(&mut s, LogLevel::Info, "first".into());
842        s.begin_request();
843        assert!(s.logs.is_empty(), "logs reset for the next request");
844    }
845
846    #[test]
847    fn clock_returns_nonzero_wall_time() {
848        let mut s = state("test\u{1f}");
849        assert!(ClockHost::now_ms(&mut s) > 0);
850    }
851
852    #[test]
853    fn kv_value_over_cap_is_dropped_fail_closed() {
854        // a value past the per-key cap is dropped, not stored (host-API is infallible from
855        // the filter's view). A within-cap value stores normally.
856        let mut s = state("f\u{1f}");
857        KvHost::set(&mut s, "big".into(), vec![0u8; MAX_KV_VALUE_BYTES + 1]);
858        assert_eq!(
859            KvHost::get(&mut s, "big".into()),
860            None,
861            "an over-cap value is dropped"
862        );
863        KvHost::set(&mut s, "ok".into(), vec![0u8; 128]);
864        assert_eq!(KvHost::get(&mut s, "ok".into()), Some(vec![0u8; 128]));
865    }
866
867    #[test]
868    fn concurrent_delete_of_the_same_key_releases_quota_exactly_once() {
869        // Regression test: the trusted pool runs many concurrent instances of one filter, all
870        // sharing one backend + one KvQuota. A get-then-delete-then-release sequence done as
871        // three independent lock acquisitions lets N concurrent deletes of the SAME key all
872        // observe `Some(old)` and all release budget for a key that only existed once —
873        // permanently under-counting real usage. `KvQuota::charge_and_apply` closes this by
874        // making the whole read-decide-release sequence one atomic unit per call.
875        use std::sync::Barrier;
876        use std::thread;
877
878        let shared: Arc<dyn KvBackend> = Arc::new(MemoryBackend::default());
879        let quota = Arc::new(KvQuota::new());
880        let prefix = "f\u{1f}".to_string();
881
882        let mut seed = HostState::new(
883            HostStateInit {
884                kv: shared.clone(),
885                quota: quota.clone(),
886                ..init_for(&prefix)
887            },
888            #[cfg(feature = "outbound-http")]
889            outbound_http::PlectoHttpHooks::deny_all(),
890            #[cfg(feature = "outbound-tcp")]
891            crate::outbound_tcp::TcpGuard::deny_all(),
892        );
893        // Two distinct keys in the same namespace: "k1" (raced on below) and "k2" (untouched —
894        // its surviving budget is the tell-tale that a double-release didn't over-free the ns).
895        KvHost::set(&mut seed, "k1".into(), vec![0u8; 100]);
896        KvHost::set(&mut seed, "k2".into(), vec![0u8; 100]);
897        assert_eq!(
898            quota.usage_for_test(&prefix),
899            (2, 2 * (2 + 100)),
900            "both keys charged once each"
901        );
902
903        const RACERS: usize = 8;
904        let barrier = Arc::new(Barrier::new(RACERS));
905        let handles: Vec<_> = (0..RACERS)
906            .map(|_| {
907                let kv = shared.clone();
908                let q = quota.clone();
909                let p = prefix.clone();
910                let b = barrier.clone();
911                thread::spawn(move || {
912                    let mut s = HostState::new(
913                        HostStateInit {
914                            kv,
915                            quota: q,
916                            ..init_for(&p)
917                        },
918                        #[cfg(feature = "outbound-http")]
919                        outbound_http::PlectoHttpHooks::deny_all(),
920                        #[cfg(feature = "outbound-tcp")]
921                        crate::outbound_tcp::TcpGuard::deny_all(),
922                    );
923                    b.wait();
924                    KvHost::delete(&mut s, "k1".into());
925                })
926            })
927            .collect();
928        for h in handles {
929            h.join().unwrap();
930        }
931
932        // Exactly one release happened for "k1"; "k2" is untouched — the namespace must show
933        // precisely k2's own accounting, not zeroed-out-by-clamping evidence of over-release.
934        assert_eq!(
935            quota.usage_for_test(&prefix),
936            (1, 2 + 100),
937            "k1's single release must not consume k2's untouched budget"
938        );
939    }
940
941    #[test]
942    fn host_log_is_capped_and_sanitized() {
943        // control bytes are neutralized (no CRLF log-line injection / ANSI), and
944        // the per-request line count is bounded with a single truncation marker.
945        let mut s = state("f\u{1f}");
946        LogHost::log(&mut s, LogLevel::Info, "a\r\nInjected: x\u{1b}[31m".into());
947        assert!(
948            !s.logs[0].message.contains('\r') && !s.logs[0].message.contains('\n'),
949            "CR/LF are neutralized (no log-line injection)"
950        );
951        assert!(
952            !s.logs[0].message.contains('\u{1b}'),
953            "the ANSI escape is neutralized"
954        );
955
956        // a long message is truncated to the byte cap.
957        let mut s2 = state("f\u{1f}");
958        LogHost::log(&mut s2, LogLevel::Info, "x".repeat(MAX_LOG_MSG_BYTES * 2));
959        assert!(s2.logs[0].message.len() <= MAX_LOG_MSG_BYTES);
960
961        // the per-request line count is bounded, last slot is a truncation marker.
962        let mut s3 = state("f\u{1f}");
963        for i in 0..(MAX_LOG_LINES_PER_REQUEST + 50) {
964            LogHost::log(&mut s3, LogLevel::Info, format!("line {i}"));
965        }
966        assert_eq!(s3.logs.len(), MAX_LOG_LINES_PER_REQUEST);
967        assert!(
968            s3.logs.last().unwrap().message.contains("truncated"),
969            "the final retained line is the truncation marker"
970        );
971    }
972}