plecto-control 0.5.0

//! One filter to load (`[[filter]]`), pinned by OCI digest, plus its host-side rate-limit and
//! outbound-HTTP sub-config (ADR 000006 / 000026 / 000036).

use std::collections::BTreeMap;
use std::io::Read;
use std::path::Path;

use serde::{Deserialize, Serialize};

use crate::error::ControlError;

/// Host-side token-bucket spec for a filter's `host-ratelimit` (ADR 000026), set in the manifest
/// as an inline table `ratelimit = { capacity = .., refill_tokens = .., refill_interval_ms = .. }`.
/// The operator owns it, so an untrusted filter cannot supply or override its own limit (the WIT
/// `try-acquire` carries only `key` + `cost`).
#[derive(Debug, Clone, Copy, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct RateLimitConfig {
    /// Maximum tokens the bucket can hold.
    pub capacity: u64,
    /// Tokens added each refill interval.
    pub refill_tokens: u64,
    /// Refill interval in milliseconds (the host advances by whole intervals).
    pub refill_interval_ms: u64,
}

/// A URL scheme an outbound allowlist entry may name (ADR 000036). Defaults to `https`.
#[derive(
    Debug, Clone, Copy, Default, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq,
)]
#[serde(rename_all = "lowercase")]
pub enum SchemeKind {
    #[default]
    Https,
    Http,
}

impl SchemeKind {
    #[cfg(feature = "outbound-http")]
    pub(super) fn default_port(self) -> u16 {
        match self {
            SchemeKind::Https => 443,
            SchemeKind::Http => 80,
        }
    }
}

/// One allowed outbound destination — an exact `(scheme, host, port)` triple (ADR 000036). No
/// wildcards: the target endpoints (JWKS / introspection / ext_authz) are fixed and known.
#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct AllowDest {
    /// Exact host — a DNS name (matched case-insensitively) or an IP literal.
    pub host: String,
    /// Destination port. Defaults to the scheme's default (443/80) when omitted.
    pub port: Option<u16>,
    /// URL scheme (default `https`).
    #[serde(default)]
    pub scheme: SchemeKind,
}

/// A filter's outbound HTTP policy (ADR 000036), declared as `[filter.outbound_http]`. The operator
/// owns it — an untrusted filter cannot supply, widen, or override it (the
/// `wasi:http/outgoing-handler` import carries only the request). Absent = the filter is lent no
/// outbound HTTP capability.
///
/// `allow` is deny-by-default: only listed destinations are reachable. `allow_private` opts specific
/// RFC1918 / ULA CIDRs past the SSRF guard's private-range block (for internal ext_authz); it never
/// opens the always-blocked reserved floor (loopback / link-local / cloud-metadata). Timeouts and
/// sizes are host-clamped.
#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct OutboundHttpConfig {
    /// The exact destinations this filter may call. Must be non-empty when the section is present.
    pub allow: Vec<AllowDest>,
    /// Private/ULA CIDRs (e.g. `"10.1.0.0/16"`) this filter may reach despite the SSRF private-range
    /// block. Empty (default) leaves all private space blocked.
    #[serde(default)]
    pub allow_private: Vec<String>,
    /// TCP connect timeout (ms). Host default/ceiling apply when omitted/exceeded.
    pub connect_timeout_ms: Option<u64>,
    /// Whole-call wall-clock timeout (ms). Host default/ceiling apply when omitted/exceeded.
    pub total_timeout_ms: Option<u64>,
    /// Cap on the response body the host buffers back (bytes). Host default/ceiling apply.
    pub max_response_bytes: Option<u64>,
    /// Cap on concurrent in-flight outbound calls for this filter. Host default/ceiling apply.
    pub max_concurrent: Option<u32>,
}

/// One allowed outbound TCP destination — an exact `(host, port)` pair (ADR 000060). Unlike HTTP's
/// [`AllowDest`] there is no scheme and no default port: a raw TCP endpoint (Redis, memcached) has
/// no scheme to derive one from, so the operator states the port explicitly.
#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct TcpAllowDest {
    /// Exact host — a DNS name (matched case-insensitively) or an IP literal.
    pub host: String,
    /// Destination port (required — raw TCP has no scheme default).
    pub port: u16,
}

/// A filter's outbound TCP policy (ADR 000060), declared as `[filter.outbound_tcp]`. Lends the
/// `wasi:sockets` TCP-connect vocabulary behind the same three checks as outbound HTTP: the
/// deny-by-default allowlist, the SSRF guard on host-resolved addresses (name-lookup vetting +
/// IP pin), and host-clamped resource bounds. UDP and listen are never lent.
#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct OutboundTcpConfig {
    /// The exact destinations this filter may connect to. Must be non-empty when present.
    pub allow: Vec<TcpAllowDest>,
    /// Private/ULA CIDRs (e.g. `"10.1.0.0/16"`) this filter may reach despite the SSRF
    /// private-range block. Empty (default) leaves all private space blocked.
    #[serde(default)]
    pub allow_private: Vec<String>,
    /// Per-request budget of TCP connects (a connection held across requests costs only its
    /// opening request). Host default/ceiling apply when omitted/exceeded.
    pub max_connections: Option<u32>,
    /// Wall-clock ceiling (ms) on each guest hook call for this filter — the host-side I/O
    /// deadline epoch interruption cannot provide while the guest blocks in socket I/O
    /// (connect *and* read hangs). Host default/ceiling apply when omitted/exceeded.
    pub io_deadline_ms: Option<u64>,
}

/// One filter to load, pinned by OCI digest.
#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize)]
#[serde(deny_unknown_fields)]
pub struct FilterEntry {
    /// Host-assigned identity; namespaces the filter's KV (ADR 000011) and names it in chains.
    ///
    /// KV continuity is **per-id and survives reload** (f000004 #4): reloading the same `id`
    /// with a new `digest` keeps the same KV namespace, so the new version inherits the old
    /// version's bytes (good for rate-limit counters — a reload doesn't reset them). If a new
    /// version changes its state encoding, handle the migration or use a new `id` (= a fresh
    /// namespace).
    pub id: String,
    /// Manifest-relative path to the local OCI image-layout for this filter.
    pub source: String,
    /// Pinned OCI image-manifest digest, `sha256:...` (reproducibility / supply chain).
    pub digest: String,
    #[serde(default)]
    pub isolation: IsolationKind,
    pub init_deadline_ms: Option<u64>,
    pub request_deadline_ms: Option<u64>,
    pub max_memory_bytes: Option<u64>,
    /// Trusted pool capacity (max concurrent reusable instances, ADR 000012), `pool_size`.
    /// Only valid with `isolation = "trusted"` (the untrusted lifecycle is fresh-per-request);
    /// the host clamps to its ceiling. Absent = the host default.
    pub pool_size: Option<usize>,
    /// Bounded wait (ms) for a free trusted-pool instance under saturation before failing closed
    /// (ADR 000012), `checkout_timeout_ms`. Trusted-only, like `pool_size`; `0` = fail
    /// immediately at saturation. Absent = the host default.
    pub checkout_timeout_ms: Option<u64>,
    /// Recycle a trusted instance (discard + rebuild) after this many requests, bounding
    /// linear-memory state accumulation (ADR 000012), `max_requests_per_instance`.
    /// Trusted-only, like `pool_size`. Absent = the host default.
    pub max_requests_per_instance: Option<u64>,
    /// Host-side token bucket for this filter's `host-ratelimit` (ADR 000026). Absent = the filter
    /// has no limiter (its `try-acquire` fails closed). Operator-configured so an untrusted filter
    /// cannot override its own limit.
    pub ratelimit: Option<RateLimitConfig>,
    /// This filter's outbound HTTP policy (ADR 000036), `[filter.outbound_http]`. Absent = no
    /// outbound HTTP capability. Requires the `outbound-http` build; otherwise a present section is
    /// rejected at validate (fail-closed).
    #[serde(default)]
    pub outbound_http: Option<OutboundHttpConfig>,
    /// This filter's outbound TCP policy (ADR 000060), `[filter.outbound_tcp]`: the wasi:sockets
    /// lend behind the same deny-by-default allowlist + SSRF floor shape as outbound HTTP. Absent =
    /// no outbound TCP capability. Requires the `outbound-tcp` build; otherwise a present section
    /// is rejected at validate (fail-closed).
    #[serde(default)]
    pub outbound_tcp: Option<OutboundTcpConfig>,
    /// This filter's WASI grant (ADR 000063), `wasi = "none" | "minimal"`. Absent/`"none"` (the
    /// default) keeps the filter zero-WASI (Tier A). Requires the `fat-guest` build; otherwise
    /// `"minimal"` is rejected at validate (fail-closed).
    #[serde(default)]
    pub wasi: WasiKind,
    /// This filter's business config (ADR 000066), `[filter.config]`: an arbitrary string→string
    /// map the filter reads back via `host-config::get`. The host does not interpret keys or
    /// values — only the filter does (e.g. Redis host/port, window seconds, `on_backend_error`).
    /// Absent = an empty map (every `get` reads `None`). Unlike `ratelimit` / `outbound_*`, this is
    /// a base capability with no feature gate.
    #[serde(default)]
    pub config: Option<BTreeMap<String, String>>,
    /// File-indirected config values (field report §3.2), `[filter.config_files]`: the same
    /// string→string key space as `config`, but each value is a **path** — absolute, or
    /// relative to the manifest's directory — whose file content (UTF-8, whitespace-trimmed at
    /// both ends, ≤ 1 MiB) is served through the same `host-config::get` keys. The shape
    /// container secrets arrive in (`/run/secrets/<name>` mounts, credential directories):
    /// the manifest stays a static, checked-in single source while the secret material never
    /// appears in it. Resolved at load/reload time, so a SIGHUP picks up a rotated secret
    /// file; symlinks are followed (orchestrator secret mounts rotate via symlink swaps). A
    /// key set in both `config` and `config_files` is a validate/load error, and a missing or
    /// unreadable file fails the load — same fail-closed posture as trust keys and TLS certs.
    #[serde(default)]
    pub config_files: Option<BTreeMap<String, String>>,
}

/// Per-file ceiling for `[filter.config_files]` reads: far beyond any sane config value, small
/// enough that a mis-pointed path (a device file, a dump) fails fast instead of ballooning the
/// load. Matches the magnitude credential-passing systems cap secrets at.
const MAX_CONFIG_FILE_BYTES: u64 = 1024 * 1024;

/// Manifest spelling of the host's `Isolation`. Defaults to `untrusted` (fail-closed).
#[derive(
    Debug, Clone, Copy, Default, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq,
)]
#[serde(rename_all = "lowercase")]
pub enum IsolationKind {
    #[default]
    Untrusted,
    Trusted,
}

/// A filter's WASI grant (ADR 000063). `minimal` lends the fixed, non-configurable slice —
/// `wasi:io` / `wasi:clocks` / `wasi:random` / `wasi:cli` (no filesystem, no sockets), stdout/
/// stderr bridged into this filter's `host-log` — for guests whose runtime assumes WASI is
/// present (TinyGo/Go and similar "fat" guests). Unlike `outbound_http`/`outbound_tcp` there is
/// no allowlist to declare: the grant is fixed, so this is a bare enum, not a sub-table. Requires
/// the `fat-guest` build; otherwise `minimal` is rejected at validate (fail-closed).
#[derive(
    Debug, Clone, Copy, Default, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq,
)]
#[serde(rename_all = "lowercase")]
pub enum WasiKind {
    #[default]
    None,
    Minimal,
}

impl FilterEntry {
    /// Merge `[filter.config]` with the file-resolved `[filter.config_files]` values (field
    /// report §3.2), or `None` when there is nothing to resolve (the inline map alone keeps
    /// flowing through `load_options()`, ADR 000066 behaviour unchanged). Runs at every
    /// load/reload — a SIGHUP re-reads rotated secret files — and inside `validate_manifest`,
    /// where it joins the artifact-free fail-closed file loads (trust keys, TLS certs).
    ///
    /// Fail-closed rules, all surfaced as errors rather than silent fallbacks: a key set in
    /// both maps (the `X` / `X_file` pairs convention treats them as mutually exclusive), a
    /// missing/unreadable file, non-UTF-8 content (`host-config::get` serves `string`), and a
    /// file past [`MAX_CONFIG_FILE_BYTES`]. Content is whitespace-trimmed at both ends — the
    /// trailing-newline problem of `echo`-written secret files; no standard fixes a posture,
    /// so this one is deliberate and documented. Symlinks are followed: orchestrator secret
    /// mounts rotate atomically via symlink swaps, and the manifest (operator input) is
    /// trusted to point where it says — unlike an artifact blob, which stays symlink-refused.
    pub fn resolved_config(
        &self,
        base_dir: &Path,
    ) -> Result<Option<BTreeMap<String, String>>, ControlError> {
        let Some(files) = &self.config_files else {
            return Ok(None);
        };
        let mut merged = self.config.clone().unwrap_or_default();
        for (key, path) in files {
            if merged.contains_key(key) {
                return Err(ControlError::InvalidFilterConfig {
                    id: self.id.clone(),
                    reason: format!(
                        "config key {key} is set in both [filter.config] and \
                         [filter.config_files] — they are mutually exclusive"
                    ),
                });
            }
            // `join` on an absolute value keeps it absolute — `/run/secrets/<name>` works as
            // written, a bare name resolves beside the manifest.
            let full = base_dir.join(path);
            let file = std::fs::File::open(&full).map_err(|e| ControlError::IoAt {
                path: full.clone(),
                source: e,
            })?;
            // Bounded single-open read (no metadata-then-read race): one byte past the cap is
            // enough to detect an oversized file without buffering it.
            let mut bytes = Vec::new();
            file.take(MAX_CONFIG_FILE_BYTES + 1)
                .read_to_end(&mut bytes)
                .map_err(|e| ControlError::IoAt {
                    path: full.clone(),
                    source: e,
                })?;
            if bytes.len() as u64 > MAX_CONFIG_FILE_BYTES {
                return Err(ControlError::InvalidFilterConfig {
                    id: self.id.clone(),
                    reason: format!(
                        "config file for key {key} ({}) exceeds the {MAX_CONFIG_FILE_BYTES}-byte \
                         cap",
                        full.display()
                    ),
                });
            }
            let text = String::from_utf8(bytes).map_err(|_| ControlError::InvalidFilterConfig {
                id: self.id.clone(),
                reason: format!(
                    "config file for key {key} ({}) is not valid UTF-8 — host-config serves \
                     strings",
                    full.display()
                ),
            })?;
            merged.insert(key.clone(), text.trim().to_string());
        }
        Ok(Some(merged))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::manifest::Manifest;

    #[test]
    fn parses_filters_and_chain_with_defaults() {
        let m = Manifest::from_toml(
            r#"
[[filter]]
id = "auth"
source = "artifacts/auth"
digest = "sha256:abc"

[[filter]]
id = "rl"
source = "artifacts/rl"
digest = "sha256:def"
isolation = "trusted"
request_deadline_ms = 25

[chain]
filters = ["auth", "rl"]
"#,
        )
        .unwrap();

        assert_eq!(m.filters.len(), 2);
        assert_eq!(m.filters[0].isolation, IsolationKind::Untrusted); // default
        assert_eq!(m.filters[1].isolation, IsolationKind::Trusted);
        assert_eq!(m.filters[1].request_deadline_ms, Some(25));
        assert_eq!(m.chain.filters, vec!["auth".to_string(), "rl".to_string()]);
    }

    const OUTBOUND_TOML: &str = r#"
[[filter]]
id = "extauthz"
source = "oci/extauthz"
digest = "sha256:abc"

[filter.outbound_http]
allow = [
  { host = "authz.example.com", port = 8443, scheme = "https" },
  { host = "jwks.example.com" },
]
allow_private = ["10.1.0.0/16"]
connect_timeout_ms = 1500
"#;

    #[test]
    fn outbound_section_parses() {
        let m = Manifest::from_toml(OUTBOUND_TOML).unwrap();
        let ob = m.filters[0]
            .outbound_http
            .as_ref()
            .expect("outbound_http present");
        assert_eq!(ob.allow.len(), 2);
        assert_eq!(ob.allow[0].host, "authz.example.com");
        assert_eq!(ob.allow[0].port, Some(8443));
        assert_eq!(ob.allow[0].scheme, SchemeKind::Https);
        assert_eq!(ob.allow[1].port, None); // defaulted at lowering time
        assert_eq!(ob.allow_private, vec!["10.1.0.0/16".to_string()]);
        assert_eq!(ob.connect_timeout_ms, Some(1500));
    }

    #[cfg(feature = "outbound-http")]
    #[test]
    fn outbound_validates_and_lowers_to_policy() {
        let m = Manifest::from_toml(OUTBOUND_TOML).unwrap();
        let entry = &m.filters[0];
        entry.validate().expect("valid outbound section");

        let opts = entry.load_options();
        let policy = opts
            .outbound_http
            .expect("outbound_http lowered into LoadOptions");
        assert_eq!(policy.allow.len(), 2);
        // exact allowlist matching + scheme default port
        assert!(policy.allows(plecto_host::Scheme::Https, "authz.example.com", 8443));
        assert!(policy.allows(plecto_host::Scheme::Https, "jwks.example.com", 443));
        assert!(!policy.allows(plecto_host::Scheme::Http, "authz.example.com", 8443));
        // allow_private opt-in reaches the SSRF classifier
        assert_eq!(
            policy.classify("10.1.2.3".parse().unwrap()),
            plecto_host::AddrVerdict::Allowed
        );
        assert_eq!(
            policy.classify("169.254.169.254".parse().unwrap()),
            plecto_host::AddrVerdict::BlockedReserved
        );
        // small value passes through; the connect timeout was 1500ms
        assert_eq!(
            policy.connect_timeout,
            std::time::Duration::from_millis(1500)
        );
    }

    #[cfg(feature = "outbound-http")]
    #[test]
    fn outbound_clamps_oversized_values() {
        let toml = r#"
[[filter]]
id = "x"
source = "s"
digest = "sha256:abc"
[filter.outbound_http]
allow = [{ host = "a.example.com" }]
total_timeout_ms = 999999999
max_concurrent = 100000
"#;
        let m = Manifest::from_toml(toml).unwrap();
        let policy = m.filters[0].load_options().outbound_http.unwrap();
        assert!(policy.total_timeout <= std::time::Duration::from_secs(30));
        assert!(policy.max_concurrent <= 64);
    }

    #[cfg(feature = "outbound-http")]
    #[test]
    fn outbound_rejects_bad_config() {
        let cases = [
            // unparseable CIDR
            (
                "allow = [{ host = \"a\" }]\nallow_private = [\"not-a-cidr\"]",
                "bad CIDR",
            ),
            // zero timeout
            (
                "allow = [{ host = \"a\" }]\nconnect_timeout_ms = 0",
                "zero connect timeout",
            ),
        ];
        for (body, why) in cases {
            let toml = format!(
                "[[filter]]\nid = \"x\"\nsource = \"s\"\ndigest = \"sha256:abc\"\n[filter.outbound_http]\n{body}\n"
            );
            let m = Manifest::from_toml(&toml).unwrap();
            assert!(m.filters[0].validate().is_err(), "{why} must be rejected");
        }
    }

    /// Empty `allow` is deny-by-default (no reachable destination) but still links wasi:http —
    /// required by wasip2 guests that import the interface even when unused (filter-jwt static path).
    #[cfg(feature = "outbound-http")]
    #[test]
    fn outbound_allows_empty_allowlist_as_deny_all() {
        let toml = r#"
[[filter]]
id = "x"
source = "s"
digest = "sha256:abc"
[filter.outbound_http]
allow = []
"#;
        let m = Manifest::from_toml(toml).unwrap();
        m.filters[0]
            .validate()
            .expect("empty allow is deny-all, not invalid");
        let opts = m.filters[0].load_options();
        assert!(opts.outbound_http.is_some());
        assert!(opts.outbound_http.unwrap().allow.is_empty());
    }

    const OUTBOUND_TCP_TOML: &str = r#"
[[filter]]
id = "ratelimit-redis"
source = "oci/ratelimit-redis"
digest = "sha256:abc"

[filter.outbound_tcp]
allow = [{ host = "redis.internal", port = 6379 }]
allow_private = ["10.1.0.0/16"]
max_connections = 2
io_deadline_ms = 1500
"#;

    #[test]
    fn outbound_tcp_section_parses() {
        let m = Manifest::from_toml(OUTBOUND_TCP_TOML).unwrap();
        let ob = m.filters[0]
            .outbound_tcp
            .as_ref()
            .expect("outbound_tcp present");
        assert_eq!(ob.allow.len(), 1);
        assert_eq!(ob.allow[0].host, "redis.internal");
        assert_eq!(ob.allow[0].port, 6379);
        assert_eq!(ob.allow_private, vec!["10.1.0.0/16".to_string()]);
        assert_eq!(ob.max_connections, Some(2));
        assert_eq!(ob.io_deadline_ms, Some(1500));
    }

    #[test]
    fn outbound_tcp_port_is_required() {
        // No scheme → no default port; an entry without one must fail to parse, not silently
        // default.
        let toml = r#"
[[filter]]
id = "x"
source = "s"
digest = "sha256:abc"
[filter.outbound_tcp]
allow = [{ host = "redis.internal" }]
"#;
        assert!(Manifest::from_toml(toml).is_err(), "port is mandatory");
    }

    #[cfg(feature = "outbound-tcp")]
    #[test]
    fn outbound_tcp_validates_and_lowers_to_policy() {
        let m = Manifest::from_toml(OUTBOUND_TCP_TOML).unwrap();
        let entry = &m.filters[0];
        entry.validate().expect("valid outbound_tcp section");

        let opts = entry.load_options();
        let policy = opts
            .outbound_tcp
            .expect("outbound_tcp lowered into LoadOptions");
        assert_eq!(policy.allow.len(), 1);
        assert!(policy.allows_name("REDIS.internal")); // DNS case-insensitive
        assert!(!policy.allows_name("evil.internal"));
        // allow_private opt-in reaches the shared SSRF classifier; the floor stays closed
        assert_eq!(
            policy.classify("10.1.2.3".parse().unwrap()),
            plecto_host::AddrVerdict::Allowed
        );
        assert_eq!(
            policy.classify("169.254.169.254".parse().unwrap()),
            plecto_host::AddrVerdict::BlockedReserved
        );
        assert_eq!(policy.max_connections, 2);
        assert_eq!(policy.io_deadline, std::time::Duration::from_millis(1500));
    }

    #[cfg(feature = "outbound-tcp")]
    #[test]
    fn outbound_tcp_clamps_oversized_values() {
        let toml = r#"
[[filter]]
id = "x"
source = "s"
digest = "sha256:abc"
[filter.outbound_tcp]
allow = [{ host = "redis.internal", port = 6379 }]
max_connections = 100000
io_deadline_ms = 999999999
"#;
        let m = Manifest::from_toml(toml).unwrap();
        let policy = m.filters[0].load_options().outbound_tcp.unwrap();
        assert!(policy.max_connections <= 64);
        assert!(policy.io_deadline <= std::time::Duration::from_secs(30));
    }

    #[cfg(feature = "outbound-tcp")]
    #[test]
    fn outbound_tcp_rejects_bad_config() {
        let cases = [
            ("allow = []", "empty allowlist"),
            ("allow = [{ host = \"a\", port = 0 }]", "port 0"),
            (
                "allow = [{ host = \"a\", port = 6379 }]\nallow_private = [\"not-a-cidr\"]",
                "bad CIDR",
            ),
            (
                "allow = [{ host = \"a\", port = 6379 }]\nmax_connections = 0",
                "zero connect budget",
            ),
            (
                "allow = [{ host = \"a\", port = 6379 }]\nio_deadline_ms = 0",
                "zero io deadline",
            ),
        ];
        for (body, why) in cases {
            let toml = format!(
                "[[filter]]\nid = \"x\"\nsource = \"s\"\ndigest = \"sha256:abc\"\n[filter.outbound_tcp]\n{body}\n"
            );
            let m = Manifest::from_toml(&toml).unwrap();
            assert!(m.filters[0].validate().is_err(), "{why} must be rejected");
        }
    }

    #[cfg(not(feature = "outbound-tcp"))]
    #[test]
    fn outbound_tcp_rejected_without_feature() {
        // A manifest that asks for outbound TCP must fail closed on a build that cannot provide it.
        let m = Manifest::from_toml(OUTBOUND_TCP_TOML).unwrap();
        assert!(
            m.filters[0].validate().is_err(),
            "outbound_tcp requires the outbound-tcp build"
        );
    }

    #[cfg(not(feature = "outbound-http"))]
    #[test]
    fn outbound_rejected_without_feature() {
        // A manifest that asks for outbound must fail closed on a build that cannot provide it.
        let m = Manifest::from_toml(OUTBOUND_TOML).unwrap();
        assert!(
            m.filters[0].validate().is_err(),
            "outbound_http requires the outbound-http build"
        );
    }

    const WASI_MINIMAL_TOML: &str = r#"
[[filter]]
id = "hello-go"
source = "oci/hello-go"
digest = "sha256:abc"
wasi = "minimal"
"#;

    #[test]
    fn wasi_defaults_to_none() {
        let m = Manifest::from_toml(
            r#"
[[filter]]
id = "x"
source = "s"
digest = "sha256:abc"
"#,
        )
        .unwrap();
        assert_eq!(m.filters[0].wasi, WasiKind::None);
    }

    #[test]
    fn wasi_minimal_parses() {
        let m = Manifest::from_toml(WASI_MINIMAL_TOML).unwrap();
        assert_eq!(m.filters[0].wasi, WasiKind::Minimal);
    }

    #[cfg(feature = "fat-guest")]
    #[test]
    fn wasi_minimal_validates_and_lowers_to_load_options() {
        let m = Manifest::from_toml(WASI_MINIMAL_TOML).unwrap();
        let entry = &m.filters[0];
        entry.validate().expect("valid wasi = \"minimal\" section");
        assert!(entry.load_options().wasi_minimal);
    }

    #[cfg(not(feature = "fat-guest"))]
    #[test]
    fn wasi_minimal_rejected_without_feature() {
        // A manifest that asks for the fat-guest grant must fail closed on a build that cannot
        // provide it (same rule as outbound_http/outbound_tcp).
        let m = Manifest::from_toml(WASI_MINIMAL_TOML).unwrap();
        assert!(
            m.filters[0].validate().is_err(),
            "wasi = \"minimal\" requires the fat-guest build"
        );
    }

    #[test]
    fn config_section_parses_as_string_map() {
        // `[filter.config]` (ADR 000066) is an arbitrary string→string map — no feature gate,
        // no `deny_unknown_fields`, the host never interprets it.
        let m = Manifest::from_toml(
            r#"
[[filter]]
id = "ratelimit-redis"
source = "oci/ratelimit-redis"
digest = "sha256:abc"

[filter.config]
on_backend_error = "deny"
redis_host = "redis.internal"
redis_port = "6379"
window_seconds = "60"
limit = "1000"
route_tag = "api-v1"
"#,
        )
        .unwrap();
        let cfg = m.filters[0].config.as_ref().expect("config present");
        assert_eq!(
            cfg.get("on_backend_error").map(String::as_str),
            Some("deny")
        );
        assert_eq!(
            cfg.get("redis_host").map(String::as_str),
            Some("redis.internal")
        );
        assert_eq!(cfg.len(), 6);
    }

    #[test]
    fn config_files_resolve_into_the_config_map_trimmed() {
        // `[filter.config_files]` (field report §3.2): container secrets arrive as files
        // (`/run/secrets/...`), so each value is a path — absolute, or relative to the
        // manifest's directory — whose file content becomes the value served through the
        // existing `host-config::get` key space. Content is whitespace-trimmed at both ends
        // (the `echo`-appended trailing newline problem; the `*_file` sibling-key convention's
        // usual posture) — a deliberate, documented choice since no standard fixes one.
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("hmac_key"), "s3cret-value\n").unwrap();
        let m = Manifest::from_toml(
            r#"
[[filter]]
id = "session-auth"
source = "oci/session-auth"
digest = "sha256:abc"

[filter.config]
cookie_name = "session"

[filter.config_files]
hmac_key = "hmac_key"
"#,
        )
        .unwrap();
        let cfg = m.filters[0]
            .resolved_config(dir.path())
            .unwrap()
            .expect("config_files present resolves to a merged map");
        assert_eq!(
            cfg.get("hmac_key").map(String::as_str),
            Some("s3cret-value")
        );
        assert_eq!(cfg.get("cookie_name").map(String::as_str), Some("session"));
    }

    #[test]
    fn config_files_reject_a_key_collision_with_config() {
        // `X` and `X_file`-style pairs are conventionally mutually exclusive — both set is an
        // operator mistake, surfaced as an error rather than a silent precedence rule.
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("hmac_key"), "x").unwrap();
        let m = Manifest::from_toml(
            r#"
[[filter]]
id = "session-auth"
source = "oci/session-auth"
digest = "sha256:abc"

[filter.config]
hmac_key = "inline"

[filter.config_files]
hmac_key = "hmac_key"
"#,
        )
        .unwrap();
        let err = m.filters[0].resolved_config(dir.path()).unwrap_err();
        assert!(
            err.to_string().contains("hmac_key"),
            "the collision names the key, got: {err}"
        );
    }

    #[test]
    fn config_files_fail_closed_on_a_missing_file() {
        // Same posture as a missing trust key or TLS cert: a secret that cannot be read is a
        // load error, never a silently absent config key.
        let dir = tempfile::tempdir().unwrap();
        let m = Manifest::from_toml(
            r#"
[[filter]]
id = "session-auth"
source = "oci/session-auth"
digest = "sha256:abc"

[filter.config_files]
hmac_key = "no-such-file"
"#,
        )
        .unwrap();
        assert!(m.filters[0].resolved_config(dir.path()).is_err());
    }

    #[test]
    fn config_files_fail_closed_on_non_utf8_and_oversize() {
        // `host-config::get` serves `string`, so the file must decode as UTF-8; and a config
        // value has no business being megabytes — both are load errors, fail-closed.
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("binary"), [0xFFu8, 0xFE, 0x00]).unwrap();
        std::fs::write(dir.path().join("huge"), vec![b'a'; 1024 * 1024 + 1]).unwrap();
        for file in ["binary", "huge"] {
            let m = Manifest::from_toml(&format!(
                r#"
[[filter]]
id = "session-auth"
source = "oci/session-auth"
digest = "sha256:abc"

[filter.config_files]
hmac_key = "{file}"
"#
            ))
            .unwrap();
            assert!(
                m.filters[0].resolved_config(dir.path()).is_err(),
                "{file} must be rejected"
            );
        }
    }

    #[test]
    fn resolved_config_is_none_without_config_files() {
        // No `[filter.config_files]` → nothing to resolve; the inline `[filter.config]` keeps
        // flowing through `load_options()` untouched (ADR 000066 behaviour, unchanged).
        let m = Manifest::from_toml(
            r#"
[[filter]]
id = "x"
source = "s"
digest = "sha256:abc"

[filter.config]
k = "v"
"#,
        )
        .unwrap();
        assert!(
            m.filters[0]
                .resolved_config(Path::new("."))
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn config_section_absent_by_default() {
        let m = Manifest::from_toml(
            r#"
[[filter]]
id = "x"
source = "s"
digest = "sha256:abc"
"#,
        )
        .unwrap();
        assert_eq!(m.filters[0].config, None);
    }
}