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}