vector/sinks/http/
config.rs

1//! Configuration for the `http` sink.
2
3use std::{collections::BTreeMap, path::PathBuf};
4
5#[cfg(feature = "aws-core")]
6use aws_config::meta::region::ProvideRegion;
7#[cfg(feature = "aws-core")]
8use aws_types::region::Region;
9use http::{HeaderName, HeaderValue, Method, Request, StatusCode, header::AUTHORIZATION};
10use hyper::Body;
11use vector_lib::codecs::{
12    CharacterDelimitedEncoderConfig, LengthDelimitedEncoderConfig,
13    encoding::{Framer, FramingConfig, SerializerConfig},
14};
15#[cfg(feature = "aws-core")]
16use vector_lib::config::proxy::ProxyConfig;
17
18use super::{
19    encoder::HttpEncoder, request_builder::HttpRequestBuilder, service::HttpSinkRequestBuilder,
20    sink::HttpSink,
21};
22#[cfg(feature = "aws-core")]
23use crate::aws::AwsAuthentication;
24#[cfg(feature = "aws-core")]
25use crate::sinks::util::http::SigV4Config;
26use crate::{
27    codecs::{EncodingConfigWithFraming, SinkType},
28    config::{DynValidatedSink, ValidatedSink},
29    http::{Auth, HttpClient, MaybeAuth},
30    sinks::{
31        prelude::*,
32        util::{
33            RealtimeSizeBasedDefaultBatchSettings, TowerRequestSettings, UriSerde,
34            http::{
35                HttpService, OrderedHeaderName, RequestConfig, RetryStrategy,
36                http_response_retry_logic,
37            },
38        },
39    },
40    template::ConfinementConfig,
41};
42
43const CONTENT_TYPE_TEXT: &str = "text/plain";
44const CONTENT_TYPE_NDJSON: &str = "application/x-ndjson";
45const CONTENT_TYPE_JSON: &str = "application/json";
46
47/// Configuration for the `http` sink.
48#[configurable_component(sink("http", "Deliver observability event data to an HTTP server."))]
49#[derive(Clone, Debug)]
50#[serde(deny_unknown_fields)]
51pub struct HttpSinkConfig {
52    /// The full URI to make HTTP requests to.
53    ///
54    /// This should include the protocol and host, but can also include the port, path, and any other valid part of a URI.
55    #[configurable(metadata(docs::examples = "https://10.22.212.22:9000/endpoint"))]
56    pub uri: Template,
57
58    /// The HTTP method to use when making the request.
59    #[serde(default)]
60    pub method: HttpMethod,
61
62    #[configurable(derived)]
63    pub auth: Option<Auth>,
64
65    #[configurable(derived)]
66    #[serde(default)]
67    pub compression: Compression,
68
69    #[serde(flatten)]
70    pub encoding: EncodingConfigWithFraming,
71
72    /// A string to prefix the payload with.
73    ///
74    /// This option is ignored if the encoding is not character delimited JSON.
75    ///
76    /// If specified, the `payload_suffix` must also be specified and together they must produce a valid JSON object.
77    #[configurable(metadata(docs::examples = "{\"data\":"))]
78    #[serde(default)]
79    pub payload_prefix: String,
80
81    /// A string to suffix the payload with.
82    ///
83    /// This option is ignored if the encoding is not character delimited JSON.
84    ///
85    /// If specified, the `payload_prefix` must also be specified and together they must produce a valid JSON object.
86    #[configurable(metadata(docs::examples = "}"))]
87    #[serde(default)]
88    pub payload_suffix: String,
89
90    #[configurable(derived)]
91    #[serde(default)]
92    pub batch: BatchConfig<RealtimeSizeBasedDefaultBatchSettings>,
93
94    #[configurable(derived)]
95    #[serde(default)]
96    pub request: RequestConfig,
97
98    #[configurable(derived)]
99    pub tls: Option<TlsConfig>,
100
101    #[configurable(derived)]
102    #[serde(
103        default,
104        deserialize_with = "crate::serde::bool_or_struct",
105        skip_serializing_if = "crate::serde::is_default"
106    )]
107    pub acknowledgements: AcknowledgementsConfig,
108
109    #[configurable(derived)]
110    #[serde(default)]
111    pub retry_strategy: RetryStrategy,
112
113    #[serde(flatten)]
114    pub confinement: ConfinementConfig,
115}
116
117/// HTTP method.
118///
119/// A subset of the HTTP methods described in [RFC 9110, section 9.1][rfc9110] are supported.
120///
121/// [rfc9110]: https://datatracker.ietf.org/doc/html/rfc9110#section-9.1
122#[configurable_component]
123#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
124#[serde(rename_all = "snake_case")]
125pub enum HttpMethod {
126    /// GET.
127    Get,
128
129    /// HEAD.
130    Head,
131
132    /// POST.
133    #[default]
134    Post,
135
136    /// PUT.
137    Put,
138
139    /// DELETE.
140    Delete,
141
142    /// OPTIONS.
143    Options,
144
145    /// TRACE.
146    Trace,
147
148    /// PATCH.
149    Patch,
150}
151
152impl From<HttpMethod> for Method {
153    fn from(http_method: HttpMethod) -> Self {
154        match http_method {
155            HttpMethod::Head => Self::HEAD,
156            HttpMethod::Get => Self::GET,
157            HttpMethod::Post => Self::POST,
158            HttpMethod::Put => Self::PUT,
159            HttpMethod::Patch => Self::PATCH,
160            HttpMethod::Delete => Self::DELETE,
161            HttpMethod::Options => Self::OPTIONS,
162            HttpMethod::Trace => Self::TRACE,
163        }
164    }
165}
166
167impl HttpSinkConfig {
168    fn build_http_client(&self, cx: &SinkContext) -> crate::Result<HttpClient> {
169        let tls = TlsSettings::from_options(self.tls.as_ref())?;
170        Ok(HttpClient::new(tls, cx.proxy())?)
171    }
172
173    #[cfg(test)]
174    pub(super) fn build_encoder(&self) -> crate::Result<Encoder<Framer>> {
175        let (framer, serializer) = self.encoding.build(SinkType::MessageBased)?;
176        Ok(Encoder::<Framer>::new(framer, serializer))
177    }
178}
179
180impl GenerateConfig for HttpSinkConfig {
181    fn generate_config() -> serde_json::Value {
182        serde_yaml::from_str(indoc::indoc! {
183            r#"uri: https://10.22.212.22:9000/endpoint
184            encoding:
185              codec: json"#,
186        })
187        .unwrap()
188    }
189}
190
191async fn healthcheck(uri: UriSerde, auth: Option<Auth>, client: HttpClient) -> crate::Result<()> {
192    let auth = auth.choose_one(&uri.auth)?;
193    let uri = uri.with_default_parts();
194    let mut request = Request::head(&uri.uri).body(Body::empty()).unwrap();
195
196    if let Some(auth) = auth {
197        auth.apply(&mut request);
198    }
199
200    let response = client.send(request).await?;
201
202    match response.status() {
203        StatusCode::OK => Ok(()),
204        status => Err(HealthcheckError::UnexpectedStatus { status }.into()),
205    }
206}
207
208pub(super) fn validate_headers(
209    headers: &BTreeMap<String, String>,
210    configures_auth: bool,
211) -> crate::Result<BTreeMap<OrderedHeaderName, HeaderValue>> {
212    let headers = crate::sinks::util::http::validate_headers(headers)?;
213
214    for name in headers.keys() {
215        if configures_auth && name.inner() == AUTHORIZATION {
216            return Err("Authorization header can not be used with defined auth options".into());
217        }
218    }
219
220    Ok(headers)
221}
222
223/// Returns the effective framing configuration for the `http` sink (which is
224/// message-based): the explicit framing if set, otherwise the default for the
225/// serializer. This mirrors the framer selection in
226/// `EncodingConfigWithFraming::build` without building the serializer (which
227/// may read files for codecs such as protobuf).
228fn effective_framer_config(encoding: &EncodingConfigWithFraming) -> FramingConfig {
229    match encoding.config().0 {
230        Some(framing) => framing.clone(),
231        None => match encoding.config().1 {
232            SerializerConfig::Json(_) => {
233                FramingConfig::CharacterDelimited(CharacterDelimitedEncoderConfig::new(b','))
234            }
235            SerializerConfig::Avro { .. } | SerializerConfig::Native => {
236                FramingConfig::LengthDelimited(LengthDelimitedEncoderConfig::default())
237            }
238            SerializerConfig::Gelf(_) => {
239                FramingConfig::CharacterDelimited(CharacterDelimitedEncoderConfig::new(0))
240            }
241            SerializerConfig::Protobuf(_) => {
242                FramingConfig::LengthDelimited(LengthDelimitedEncoderConfig::default())
243            }
244            SerializerConfig::Cef(_)
245            | SerializerConfig::Csv(_)
246            | SerializerConfig::Logfmt
247            | SerializerConfig::NativeJson
248            | SerializerConfig::RawMessage
249            | SerializerConfig::Text(_) => FramingConfig::NewlineDelimited,
250            #[cfg(feature = "codecs-syslog")]
251            SerializerConfig::Syslog(_) => FramingConfig::NewlineDelimited,
252            #[cfg(feature = "codecs-opentelemetry")]
253            SerializerConfig::Otlp => FramingConfig::Bytes,
254        },
255    }
256}
257
258pub(super) fn validate_payload_wrapper(
259    payload_prefix: &str,
260    payload_suffix: &str,
261    serializer: &SerializerConfig,
262    framer: &FramingConfig,
263) -> crate::Result<(String, String)> {
264    let payload = [payload_prefix, "{}", payload_suffix].join("");
265    match (
266        serializer,
267        framer,
268        serde_json::from_str::<serde_json::Value>(&payload),
269    ) {
270        (SerializerConfig::Json(_), FramingConfig::CharacterDelimited(cfg), Err(_))
271            if cfg.character_delimited.delimiter == b',' =>
272        {
273            Err("Payload prefix and suffix wrapper must produce a valid JSON object.".into())
274        }
275        _ => Ok((payload_prefix.to_owned(), payload_suffix.to_owned())),
276    }
277}
278
279#[async_trait]
280#[typetag::serde(name = "http")]
281impl SinkConfig for HttpSinkConfig {
282    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
283        Some(&self.confinement)
284    }
285
286    fn input(&self) -> Input {
287        Input::new(self.encoding.config().1.input_type())
288    }
289
290    fn files_to_watch(&self) -> Vec<&PathBuf> {
291        let mut files = Vec::new();
292        if let Some(tls) = &self.tls {
293            if let Some(crt_file) = &tls.crt_file {
294                files.push(crt_file)
295            }
296            if let Some(key_file) = &tls.key_file {
297                files.push(key_file)
298            }
299        };
300        files
301    }
302
303    fn acknowledgements(&self) -> &AcknowledgementsConfig {
304        &self.acknowledgements
305    }
306
307    fn as_dyn_validated(&self) -> Option<&dyn DynValidatedSink> {
308        Some(self)
309    }
310}
311
312#[derive(Clone, Debug)]
313pub struct ValidatedHttp {
314    batch_settings: BatcherSettings,
315    transformer: Transformer,
316    template_headers: BTreeMap<String, Template>,
317    payload_prefix: String,
318    payload_suffix: String,
319    content_type: Option<String>,
320    content_encoding: Option<String>,
321    converted_static_headers: BTreeMap<OrderedHeaderName, HeaderValue>,
322    request_limits: TowerRequestSettings,
323}
324
325#[async_trait::async_trait]
326impl ValidatedSink for HttpSinkConfig {
327    type Validated = ValidatedHttp;
328
329    fn validate(&self) -> crate::Result<ValidatedHttp> {
330        let batch_settings = self.batch.validate()?.into_batcher_settings()?;
331
332        let serializer_config = self.encoding.config().1;
333        let framer_config = effective_framer_config(&self.encoding);
334        let transformer = self.encoding.transformer();
335
336        let request = self.request.clone();
337
338        validate_headers(&request.headers, self.auth.is_some())?;
339        let (static_headers, template_headers) = request.split_headers();
340
341        // Pure confinement checks for the URI and templated headers. The actual
342        // confinement (with the component name threaded from the
343        // `opentelemetry`/`axiom` delegations) happens in `build_from_validated`;
344        // running the checks here lets `vector validate --no-environment` catch
345        // unconfined routing templates. Skipped under the full opt-out, where
346        // `confine` only emits a warning.
347        if !self
348            .confinement
349            .dangerously_allow_unconfined_template_resolution
350        {
351            self.uri
352                .clone()
353                .confine(&self.confinement, Self::NAME, "uri")?;
354            for tpl in template_headers.values() {
355                tpl.clone()
356                    .confine(&self.confinement, Self::NAME, "request.headers")?;
357            }
358        }
359
360        // `Template::default()` — produced by delegating sinks such as
361        // `opentelemetry` before the user supplies a URI — yields an empty
362        // template whose `is_static` is false, so `is_dynamic()` reports true
363        // even though there is nothing to render. Reject the empty URI up
364        // front rather than deferring a guaranteed per-request failure.
365        if self.uri.is_empty() {
366            return Err("uri must not be empty, e.g. `https://example.com/endpoint`"
367                .to_string()
368                .into());
369        }
370
371        // A static URI can be parsed and checked for embedded credentials up
372        // front; dynamic URIs are only validated at render time.
373        if !self.uri.is_dynamic() {
374            let uri_serde: UriSerde = self.uri.get_ref().parse()?;
375            self.auth.choose_one(&uri_serde.auth)?;
376            if uri_serde.uri.scheme().is_none() || uri_serde.uri.authority().is_none() {
377                return Err(format!(
378                    "uri must include a scheme and host, e.g. `https://example.com/endpoint`; got `{}`",
379                    self.uri.get_ref()
380                )
381                .into());
382            }
383        }
384
385        let (payload_prefix, payload_suffix) = validate_payload_wrapper(
386            &self.payload_prefix,
387            &self.payload_suffix,
388            serializer_config,
389            &framer_config,
390        )?;
391
392        let content_type = {
393            use FramingConfig::*;
394            use SerializerConfig::*;
395            match (serializer_config, &framer_config) {
396                (RawMessage | Text(_), _) => Some(CONTENT_TYPE_TEXT.to_owned()),
397                (Json(_), NewlineDelimited) => Some(CONTENT_TYPE_NDJSON.to_owned()),
398                (Json(_), CharacterDelimited(cfg)) if cfg.character_delimited.delimiter == b',' => {
399                    Some(CONTENT_TYPE_JSON.to_owned())
400                }
401                #[cfg(feature = "codecs-opentelemetry")]
402                (Otlp, _) => Some("application/x-protobuf".to_owned()),
403                _ => None,
404            }
405        };
406
407        let content_encoding = self.compression.is_compressed().then(|| {
408            self.compression
409                .content_encoding()
410                .expect("Encoding should be specified for compression.")
411                .to_string()
412        });
413
414        let converted_static_headers = static_headers
415            .into_iter()
416            .map(|(name, value)| -> crate::Result<_> {
417                let header_name =
418                    HeaderName::from_bytes(name.as_bytes()).map(OrderedHeaderName::from)?;
419                let header_value = HeaderValue::try_from(value)?;
420                Ok((header_name, header_value))
421            })
422            .collect::<Result<BTreeMap<_, _>, _>>()?;
423
424        let request_limits = self.request.tower.into_settings();
425
426        Ok(ValidatedHttp {
427            batch_settings,
428            transformer,
429            template_headers,
430            payload_prefix,
431            payload_suffix,
432            content_type,
433            content_encoding,
434            converted_static_headers,
435            request_limits,
436        })
437    }
438
439    async fn build(
440        &self,
441        validated: &ValidatedHttp,
442        cx: SinkContext,
443    ) -> crate::Result<(VectorSink, Healthcheck)> {
444        self.build_from_validated(validated, cx, Self::NAME).await
445    }
446}
447
448impl HttpSinkConfig {
449    /// Builds the sink from the validated state. Confinement of the URI and
450    /// templated headers happens here (not in `validate`) because the
451    /// `component_name` threaded from the `opentelemetry`/`axiom` delegations
452    /// must appear in per-template security warnings.
453    pub(crate) async fn build_from_validated(
454        &self,
455        validated: &ValidatedHttp,
456        cx: SinkContext,
457        component_name: &'static str,
458    ) -> crate::Result<(VectorSink, Healthcheck)> {
459        let ValidatedHttp {
460            batch_settings,
461            transformer,
462            template_headers,
463            payload_prefix,
464            payload_suffix,
465            content_type,
466            content_encoding,
467            converted_static_headers,
468            request_limits,
469        } = validated;
470
471        let client = self.build_http_client(&cx)?;
472
473        let healthcheck = match cx.healthcheck.uri {
474            Some(healthcheck_uri) => {
475                healthcheck(healthcheck_uri, self.auth.clone(), client.clone()).boxed()
476            }
477            None => future::ok(()).boxed(),
478        };
479
480        let (framer, serializer) = self.encoding.build(SinkType::MessageBased)?;
481        let encoder = Encoder::<Framer>::new(framer, serializer);
482
483        let request_builder = HttpRequestBuilder {
484            encoder: HttpEncoder::new(
485                encoder,
486                transformer.clone(),
487                payload_prefix.clone(),
488                payload_suffix.clone(),
489            ),
490            compression: self.compression,
491        };
492
493        let http_sink_request_builder = HttpSinkRequestBuilder::new(
494            self.method,
495            self.auth.clone(),
496            converted_static_headers.clone(),
497            content_type.clone(),
498            content_encoding.clone(),
499        );
500
501        let service = match &self.auth {
502            #[cfg(feature = "aws-core")]
503            Some(Auth::Aws { auth, service }) => {
504                let default_region = crate::aws::region_provider(&ProxyConfig::default(), None)?
505                    .region()
506                    .await;
507                let region = (match &auth {
508                    AwsAuthentication::AccessKey { region, .. } => region.clone(),
509                    AwsAuthentication::File { .. } => None,
510                    AwsAuthentication::Role { region, .. } => region.clone(),
511                    AwsAuthentication::Default { region, .. } => region.clone(),
512                })
513                .map_or(default_region, |r| Some(Region::new(r.to_string())))
514                .expect("Region must be specified");
515
516                HttpService::new_with_sig_v4(
517                    client,
518                    http_sink_request_builder,
519                    SigV4Config {
520                        shared_credentials_provider: auth
521                            .credentials_provider(region.clone(), &ProxyConfig::default(), None)
522                            .await?,
523                        region: region.clone(),
524                        service: service.clone(),
525                    },
526                )
527            }
528            _ => HttpService::new(client, http_sink_request_builder),
529        };
530
531        let service = ServiceBuilder::new()
532            .settings(
533                request_limits.clone(),
534                http_response_retry_logic(self.retry_strategy.clone()),
535            )
536            .service(service);
537
538        let uri = self
539            .uri
540            .clone()
541            .confine(&self.confinement, component_name, "uri")?;
542
543        // Confine every templated header value. Header-based routing
544        // (e.g. `X-Scope-OrgID: "{{ tenant }}"`) is as steerable as URI
545        // routing — an event that controls the header field picks the
546        // destination tenant unless we confine the header template too.
547        let template_headers = template_headers
548            .clone()
549            .into_iter()
550            .map(|(name, tpl)| {
551                tpl.confine(&self.confinement, component_name, "request.headers")
552                    .map(|tpl| (name, tpl))
553            })
554            .collect::<crate::Result<BTreeMap<_, _>>>()?;
555
556        let sink = HttpSink::new(
557            service,
558            uri,
559            template_headers,
560            *batch_settings,
561            request_builder,
562        );
563
564        Ok((VectorSink::from_event_streamsink(sink), healthcheck))
565    }
566}
567
568#[cfg(test)]
569mod tests {
570    use vector_lib::codecs::encoding::format::JsonSerializerOptions;
571
572    use super::*;
573    use crate::components::validation::prelude::*;
574    use crate::template::{ConfinementConfig, Template};
575
576    impl ValidatableComponent for HttpSinkConfig {
577        fn validation_configuration() -> ValidationConfiguration {
578            use std::str::FromStr;
579
580            use vector_lib::{
581                codecs::{JsonSerializerConfig, MetricTagValues},
582                config::LogNamespace,
583            };
584
585            let endpoint = "http://127.0.0.1:9000/endpoint";
586            let uri = UriSerde::from_str(endpoint).expect("should never fail to parse");
587
588            let config = HttpSinkConfig {
589                uri: Template::try_from(endpoint).expect("should never fail to parse"),
590                method: HttpMethod::Post,
591                encoding: EncodingConfigWithFraming::new(
592                    None,
593                    JsonSerializerConfig::new(
594                        MetricTagValues::Full,
595                        JsonSerializerOptions::default(),
596                    )
597                    .into(),
598                    Transformer::default(),
599                ),
600                auth: None,
601                compression: Compression::default(),
602                batch: BatchConfig::default(),
603                request: RequestConfig::default(),
604                tls: None,
605                acknowledgements: AcknowledgementsConfig::default(),
606                payload_prefix: String::new(),
607                payload_suffix: String::new(),
608                retry_strategy: RetryStrategy::default(),
609                confinement: ConfinementConfig::default(),
610            };
611
612            let external_resource = ExternalResource::new(
613                ResourceDirection::Push,
614                HttpResourceConfig::from_parts(uri.uri, Some(config.method.into())),
615                config.encoding.clone(),
616            );
617
618            ValidationConfiguration::from_sink(
619                Self::NAME,
620                LogNamespace::Legacy,
621                vec![ComponentTestCaseConfig::from_sink(
622                    config,
623                    None,
624                    Some(external_resource),
625                )],
626            )
627        }
628    }
629
630    register_validatable_component!(HttpSinkConfig);
631
632    #[test]
633    fn validate_rejects_static_uri_with_auth_conflict() {
634        use crate::config::ValidatedSink;
635        let config: HttpSinkConfig = serde_yaml::from_str(
636            r#"
637            uri: "http://user:pass@localhost:9000/endpoint"
638            auth:
639              strategy: basic
640              user: user
641              password: pass
642            encoding:
643              codec: json
644            "#,
645        )
646        .unwrap();
647        assert!(
648            config.validate().is_err(),
649            "embedded credentials plus auth should fail validation"
650        );
651    }
652
653    #[test]
654    fn validate_accepts_static_uri() {
655        use crate::config::ValidatedSink;
656        let config: HttpSinkConfig = serde_yaml::from_str(
657            r#"
658            uri: "http://localhost:9000/endpoint"
659            encoding:
660              codec: json
661            "#,
662        )
663        .unwrap();
664        config.validate().expect("valid static uri should validate");
665    }
666
667    #[test]
668    fn validate_accepts_dynamic_uri() {
669        use crate::config::ValidatedSink;
670        let config: HttpSinkConfig = serde_yaml::from_str(
671            r#"
672            uri: "http://example.com/{{ path }}"
673            encoding:
674              codec: json
675            "#,
676        )
677        .unwrap();
678        config
679            .validate()
680            .expect("dynamic uri validation is deferred to render time");
681    }
682
683    #[test]
684    fn validate_rejects_empty_default_uri() {
685        use crate::config::ValidatedSink;
686        // `Template::default()` — produced by delegating sinks such as
687        // `opentelemetry` before the user supplies a URI — is empty but reports
688        // `is_dynamic() == true` (the derived default leaves `is_static` false),
689        // so it must be rejected explicitly rather than deferred as a dynamic
690        // template that can never render.
691        let mut config: HttpSinkConfig = serde_yaml::from_str(
692            r#"
693            uri: "http://localhost:9000/endpoint"
694            encoding:
695              codec: json
696            "#,
697        )
698        .unwrap();
699        config.uri = Template::default();
700        assert!(
701            config.validate().is_err(),
702            "empty default uri should fail validation"
703        );
704    }
705
706    #[test]
707    fn validate_rejects_relative_static_uri() {
708        use crate::config::ValidatedSink;
709        let config: HttpSinkConfig = serde_yaml::from_str(
710            r#"
711            uri: "/ingest"
712            encoding:
713              codec: json
714            "#,
715        )
716        .unwrap();
717        assert!(
718            config.validate().is_err(),
719            "relative static uri should fail validation"
720        );
721    }
722
723    #[test]
724    fn confinement_rejects_unconfined_uri() {
725        let template: Template = "{{ endpoint }}".try_into().unwrap();
726        let err = template
727            .confine(&ConfinementConfig::default(), "http", "uri")
728            .unwrap_err();
729        assert!(
730            err.to_string().contains("no literal string prefix"),
731            "unexpected error: {err}"
732        );
733    }
734
735    #[test]
736    fn confinement_opt_out_allows_unconfined_uri() {
737        let cfg = ConfinementConfig {
738            dangerously_allow_unconfined_template_resolution: true,
739        };
740        let template: Template = "{{ endpoint }}".try_into().unwrap();
741        assert!(template.confine(&cfg, "http", "uri").is_ok());
742    }
743
744    #[test]
745    fn confinement_blocks_host_redirect_at_render() {
746        use crate::event::Event;
747        use vector_lib::event::LogEvent;
748        use vrl::event_path;
749
750        let template: Template = "https://logs.example.com/ingest/{{ tenant }}"
751            .try_into()
752            .unwrap();
753        let template = template
754            .confine(&ConfinementConfig::default(), "http", "uri")
755            .unwrap();
756
757        // Attacker tries to redirect to a different host via the tenant field.
758        let mut event = Event::Log(LogEvent::from("x"));
759        event
760            .as_mut_log()
761            .insert(event_path!("tenant"), "../../evil.com/steal?data=");
762        assert!(template.render_string(&event).is_err());
763    }
764
765    #[test]
766    fn validate_returns_usable_values() {
767        let config: HttpSinkConfig = serde_yaml::from_str(
768            r#"
769            uri: "http://127.0.0.1:9000/endpoint"
770            encoding:
771              codec: json
772            "#,
773        )
774        .unwrap();
775
776        let validated = config.validate().expect("validation should succeed");
777        // JSON + newline-delimited framing maps to the NDJSON content type.
778        assert_eq!(validated.content_type.as_deref(), Some(CONTENT_TYPE_JSON));
779        assert_eq!(validated.payload_prefix, "");
780        assert_eq!(validated.payload_suffix, "");
781        assert!(validated.template_headers.is_empty());
782        assert!(validated.converted_static_headers.is_empty());
783    }
784}
vector/sinks/http/config.rs

vector/sinks/http/
config.rs
