fix(sinks): confinement follow-ups from #25820 codex review (#25830)

Author: pront

changelog.d/template_confinement.breaking.md

@@ -1,20 +1,33 @@
-Sinks that use `{{ field }}` references in routing templates now require a
-literal prefix in the template so Vector can enforce a routing boundary. For
-example, `key_prefix: "{{ host }}/"` is no longer accepted at startup because
-there is no fixed leading segment.
-
-The `file` sink additionally gains a `base_dir` config field to set the
-confinement root explicitly when the `path` template has no usable literal
-prefix.
+Sinks that accept `{{ field }}` references in routing templates now enforce a
+confinement boundary: the rendered value must stay within the literal prefix
+declared in the template. Templates with no literal prefix (e.g.
+`key_prefix: "{{ host }}/"`) are rejected at startup.
 
 Affected sinks: `aws_s3`, `azure_blob`, `gcp_cloud_storage`, `webhdfs`,
 `file`, `elasticsearch`, `kafka`, `http`, `splunk_hec_logs`,
-`splunk_hec_metrics`, `loki`, `clickhouse`, `redis`, `amqp`, `pulsar`,
-`mqtt`, `nats`, `greptimedb_logs`, `aws_cloudwatch_logs`,
-`gcp_stackdriver_logs`, `prometheus_remote_write`.
+`splunk_hec_metrics`, `humio_logs`, `humio_metrics`, `loki`, `clickhouse`,
+`redis`, `amqp`, `pulsar`, `mqtt`, `nats`, `greptimedb_logs`,
+`aws_cloudwatch_logs`, `gcp_stackdriver_logs`, `prometheus_remote_write`.
+
+The `file` sink gains a `base_dir` config field to set the confinement root
+explicitly when the `path` template has no usable literal prefix.
+
+**URI templates:** HTTP/HTTPS URI templates that use `{{ field }}` references
+must not contain `?`. A field-rendered value could smuggle additional query
+parameters into the request. Fully static URI templates (no `{{ }}`) with a
+query string are still accepted. Dynamic query segments (e.g.
+`https://api.internal/ingest?tenant={{ tenant }}`) are rejected at startup.
+
+**Opt-out:** set `dangerously_allow_unconfined_template_resolution: true` on
+the affected sink to disable all confinement checks for that sink — both at
+startup and at runtime. Vector logs a warning per template on startup and sets
+`vector_security_confinement_disabled{component_type=...}` to `1`.
 
-**To migrate:** please add a fixed prefix to the template, e.g. `key_prefix: "logs-{{ host }}/"`.
+**Observability:**
 
-**To restore previous behavior:** set `dangerously_allow_unconfined_template_resolution: true` on the affected sink.
+- `component_errors_total{error_type="confinement_failed"}` — increments on
+  each violation; events that trigger it are dropped.
+- `vector_security_confinement_disabled` — set to `1` while a sink is running
+  with confinement disabled.
 
 authors: pront

changelog.d/template_confinement.security.md

@@ -15,6 +15,8 @@ pub mod error_type {
     // When a condition for the event to be valid failed.
     // This is used for example when a field is missing or should be a string.
     pub const CONDITION_FAILED: &str = "condition_failed";
+    // When a templated routing value fell outside its configured confinement boundary.
+    pub const CONFINEMENT_FAILED: &str = "confinement_failed";
     // When the component received a request with missing or invalid authentication credentials.
     pub const AUTHENTICATION_FAILED: &str = "authentication_failed";
     // When the component or the service on which it depends is not configured properly.

lib/vector-common/src/internal_event/prelude.rs

@@ -10,22 +10,28 @@ use vector_lib::{counter, histogram};
 
 #[derive(Debug, NamedInternalEvent)]
 pub struct KeyOutsideBasePrefixError<'a> {
-    pub key: &'a str,
+    /// Bounded preview of the rejected key — never the full rendered value,
+    /// so attacker-controlled input can't amplify into logs and secrets in
+    /// templated header/query fields don't leak.
+    pub key_preview: &'a str,
+    /// Full byte length of the rejected rendered value.
+    pub key_len: usize,
     pub message: &'a str,
 }
 
 impl InternalEvent for KeyOutsideBasePrefixError<'_> {
     fn emit(self) {
         error!(
             message = "Rendered key is outside the configured base prefix; dropping event.",
-            key = self.key,
+            key_preview = self.key_preview,
+            key_len = self.key_len,
             error = self.message,
-            error_type = error_type::CONDITION_FAILED,
+            error_type = error_type::CONFINEMENT_FAILED,
             stage = error_stage::PROCESSING,
         );
         counter!(
             CounterName::ComponentErrorsTotal,
-            "error_type" => error_type::CONDITION_FAILED,
+            "error_type" => error_type::CONFINEMENT_FAILED,
             "stage" => error_stage::PROCESSING,
         )
         .increment(1);

src/internal_events/common.rs

@@ -122,12 +122,12 @@ impl InternalEvent for FilePathOutsideBaseDirError<'_> {
             path = ?self.path,
             base_dir = ?self.base_dir,
             error = %self.error,
-            error_type = error_type::CONDITION_FAILED,
+            error_type = error_type::CONFINEMENT_FAILED,
             stage = error_stage::PROCESSING,
         );
         counter!(
             CounterName::ComponentErrorsTotal,
-            "error_type" => error_type::CONDITION_FAILED,
+            "error_type" => error_type::CONFINEMENT_FAILED,
             "stage" => error_stage::PROCESSING,
         )
         .increment(1);

src/internal_events/file.rs

@@ -15,35 +15,86 @@ pub struct TemplateRenderingError<'a> {
 
 impl InternalEvent for TemplateRenderingError<'_> {
     fn emit(self) {
-        let mut msg = "Failed to render template".to_owned();
+        let confined = matches!(
+            self.error,
+            crate::template::TemplateRenderingError::Confined { .. }
+        );
+
+        // Message wording tracks BOTH the error class (confinement vs render
+        // failure) AND whether the caller is dropping the event. A caller
+        // like the KeyPartitioner falling back to a dead-letter key emits
+        // `Confined` + `drop_event: false` — we still want to log the
+        // confinement violation, but claiming "dropping event" would be a
+        // lie because the event still routes to the dead-letter.
+        let mut msg = match (confined, self.drop_event) {
+            (true, true) => {
+                "Templated routing value was outside the configured confinement; dropping event"
+                    .to_owned()
+            }
+            (true, false) => {
+                "Templated routing value was outside the configured confinement".to_owned()
+            }
+            (false, _) => "Failed to render template".to_owned(),
+        };
         if let Some(field) = self.field {
             use std::fmt::Write;
             _ = write!(msg, " for \"{field}\"");
         }
         msg.push('.');
 
-        let confined = matches!(
-            self.error,
-            crate::template::TemplateRenderingError::Confined { .. }
-        );
-
-        if self.drop_event {
+        // A `Confined` error is always alert-worthy: an attacker attempted
+        // to steer routing via event data. Some callers legitimately don't
+        // drop the event (e.g. a partitioner falling back to a dead-letter
+        // key), but the confinement fire itself must still surface in logs
+        // and `component_errors_total` regardless of the caller's
+        // drop_event decision.
+        //
+        // `check-events` requires `error_type` counter tag values to be
+        // constants, so the two error-class branches are split
+        // explicitly. Confined renders always surface at `error!` +
+        // `component_errors_total` (they're security-relevant even when
+        // the caller doesn't drop the event); non-confined renders only
+        // surface on the drop path.
+        if confined {
+            error!(
+                message = %msg,
+                error = %self.error,
+                error_type = error_type::CONFINEMENT_FAILED,
+                stage = error_stage::PROCESSING,
+            );
+            counter!(
+                CounterName::ComponentErrorsTotal,
+                "error_type" => error_type::CONFINEMENT_FAILED,
+                "stage" => error_stage::PROCESSING,
+            )
+            .increment(1);
+        } else if self.drop_event {
             error!(
                 message = %msg,
                 error = %self.error,
                 error_type = error_type::TEMPLATE_FAILED,
                 stage = error_stage::PROCESSING,
             );
-
             counter!(
                 CounterName::ComponentErrorsTotal,
                 "error_type" => error_type::TEMPLATE_FAILED,
                 "stage" => error_stage::PROCESSING,
             )
             .increment(1);
+        } else {
+            warn!(
+                message = %msg,
+                error = %self.error,
+                error_type = error_type::TEMPLATE_FAILED,
+                stage = error_stage::PROCESSING,
+            );
+        }
 
-            // Confinement violations are intentional security discards.
-            // All other template render failures are unintentional drops.
+        // Only emit `ComponentEventsDropped` when the caller actually
+        // dropped the event. `Confined` + `drop_event: false` (e.g. the
+        // dead-letter fallback) doesn't count as a drop — the event still
+        // lands somewhere, just at the operator-authored fallback key.
+        if self.drop_event {
             if confined {
                 emit!(ComponentEventsDropped::<INTENTIONAL> {
                     count: 1,
@@ -55,13 +106,6 @@ impl InternalEvent for TemplateRenderingError<'_> {
                     reason: "Failed to render template.",
                 });
             }
-        } else {
-            warn!(
-                message = %msg,
-                error = %self.error,
-                error_type = error_type::TEMPLATE_FAILED,
-                stage = error_stage::PROCESSING,
-            );
         }
     }
 }

src/internal_events/template.rs

@@ -145,6 +145,7 @@ impl SinkConfig for AmqpSinkConfig {
             .transpose()?;
         let sink = AmqpSink::new(config).await?;
         let hc = healthcheck(sink.channels.clone()).boxed();
+        self.confinement.set_confinement_gauge("sink", Self::NAME);
         Ok((VectorSink::from_event_streamsink(sink), hc))
     }
 

src/sinks/amqp/config.rs

@@ -241,6 +241,7 @@ impl SinkConfig for CloudwatchLogsSinkConfig {
             service: svc,
         };
 
+        self.confinement.set_confinement_gauge("sink", Self::NAME);
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 

src/sinks/aws_cloudwatch_logs/config.rs

@@ -232,6 +232,7 @@ impl SinkConfig for S3SinkConfig {
         let service = self.create_service(&cx.proxy).await?;
         let healthcheck = self.build_healthcheck(service.client())?;
         let sink = self.build_processor(service, cx)?;
+        self.confinement.set_confinement_gauge("sink", Self::NAME);
         Ok((sink, healthcheck))
     }
 

src/sinks/aws_s3/config.rs

@@ -19,6 +19,7 @@ use crate::{
             http::{RequestConfig, RetryStrategy},
         },
     },
+    template::ConfinementConfig,
     tls::TlsConfig,
 };
 
@@ -129,6 +130,9 @@ pub struct AxiomConfig {
     #[configurable(derived)]
     #[serde(default)]
     pub retry_strategy: RetryStrategy,
+
+    #[serde(flatten)]
+    pub confinement: ConfinementConfig,
 }
 
 impl GenerateConfig for AxiomConfig {
@@ -186,10 +190,17 @@ impl SinkConfig for AxiomConfig {
             payload_prefix: "".into(), // Always newline delimited JSON
             payload_suffix: "".into(), // Always newline delimited JSON
             retry_strategy: self.retry_strategy.clone(),
-            confinement: Default::default(),
+            confinement: self.confinement.clone(),
         };
 
-        http_sink_config.build(cx).await
+        // Route through the HTTP builder that doesn't emit the gauge, so
+        // per-template warnings and the sink-level gauge both carry
+        // `component_type=axiom` — not `http`.
+        let result = http_sink_config
+            .build_without_confinement_gauge(cx, Self::NAME)
+            .await?;
+        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        Ok(result)
     }
 
     fn input(&self) -> Input {