vector/sinks/splunk_hec/logs/
sink.rs

1use std::{fmt, sync::Arc};
2
3use vector_lib::{
4    config::{LogNamespace, log_schema},
5    lookup::{OwnedValuePath, PathPrefix, event_path, lookup_v2::OptionalTargetPath},
6    schema::meaning,
7};
8use vrl::path::OwnedTargetPath;
9
10use super::request_builder::HecLogsRequestBuilder;
11use crate::{
12    internal_events::{
13        SplunkEventTimestampInvalidType, SplunkEventTimestampMissing, TemplateRenderingError,
14    },
15    sinks::{
16        prelude::*,
17        splunk_hec::common::{
18            EndpointTarget, INDEX_FIELD, SOURCE_FIELD, SOURCETYPE_FIELD, render_template_string,
19            request::HecRequest,
20        },
21        util::processed_event::ProcessedEvent,
22    },
23    template::Template,
24};
25
26// NOTE: The `OptionalTargetPath`s are wrapped in an `Option` in order to distinguish between a true
27//       `None` type and an empty string. This is necessary because `OptionalTargetPath` deserializes an
28//       empty string to a `None` path internally.
29pub struct HecLogsSink<S> {
30    pub service: S,
31    pub request_builder: HecLogsRequestBuilder,
32    pub batch_settings: BatcherSettings,
33    pub sourcetype: Option<Template>,
34    pub source: Option<Template>,
35    pub index: Option<Template>,
36    pub indexed_fields: Vec<OwnedValuePath>,
37    pub host_key: Option<OptionalTargetPath>,
38    pub timestamp_nanos_key: Option<String>,
39    pub timestamp_key: Option<OptionalTargetPath>,
40    pub endpoint_target: EndpointTarget,
41    pub auto_extract_timestamp: bool,
42}
43
44pub struct HecLogData<'a> {
45    pub sourcetype: Option<&'a Template>,
46    pub source: Option<&'a Template>,
47    pub index: Option<&'a Template>,
48    pub indexed_fields: &'a [OwnedValuePath],
49    pub host_key: Option<OptionalTargetPath>,
50    pub timestamp_nanos_key: Option<&'a String>,
51    pub timestamp_key: Option<OptionalTargetPath>,
52    pub endpoint_target: EndpointTarget,
53    pub auto_extract_timestamp: bool,
54}
55
56impl<S> HecLogsSink<S>
57where
58    S: Service<HecRequest> + Send + 'static,
59    S::Future: Send + 'static,
60    S::Response: DriverResponse + Send + 'static,
61    S::Error: fmt::Debug + Into<crate::Error> + Send,
62{
63    async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
64        let data = HecLogData {
65            sourcetype: self.sourcetype.as_ref(),
66            source: self.source.as_ref(),
67            index: self.index.as_ref(),
68            indexed_fields: self.indexed_fields.as_slice(),
69            host_key: self.host_key.clone(),
70            timestamp_nanos_key: self.timestamp_nanos_key.as_ref(),
71            timestamp_key: self.timestamp_key.clone(),
72            endpoint_target: self.endpoint_target,
73            auto_extract_timestamp: self.auto_extract_timestamp,
74        };
75        let batch_settings = self.batch_settings;
76
77        // Clones for the confined-event pre-filter below.  Templates are also
78        // cloned into the EventPartitioner further down; both clones are needed
79        // because `EventPartitioner::partition` has no way to signal "drop this
80        // event" — a `None` key still sends the event without metadata.
81        let source_check = self.source.clone();
82        let sourcetype_check = self.sourcetype.clone();
83        let index_check = self.index.clone();
84
85        input
86            // Pre-check partition templates for confinement violations
87            // BEFORE `process_log` mutates the event. `process_log` extracts
88            // and removes fields like `timestamp` from the event body, so
89            // running this check on the processed event turned a
90            // `Confined` render (e.g. `index: "idx-{{ timestamp }}"` with a
91            // `../../evil` value) into a `MissingKeys` render — the security
92            // drop was silently downgraded to "field missing", and the
93            // attack event shipped without an index.
94            //
95            // For the Raw endpoint a None partition key still routes to
96            // Splunk (without metadata), so we must drop here rather than
97            // inside `partition`. For the Event endpoint the metadata is
98            // embedded in the event body; a Confined render would silently
99            // omit the field, so we drop here for both endpoint types.
100            .filter_map(move |event| {
101                future::ready(
102                    if has_confined_partition_error(
103                        &event,
104                        source_check.as_ref(),
105                        sourcetype_check.as_ref(),
106                        index_check.as_ref(),
107                    ) {
108                        None
109                    } else {
110                        Some(event)
111                    },
112                )
113            })
114            .map(move |event| process_log(event, &data))
115            .batched_partitioned(
116                if self.endpoint_target == EndpointTarget::Raw {
117                    // We only need to partition by the metadata fields for the raw endpoint since those fields
118                    // are sent via query parameters in the request.
119                    EventPartitioner::new(
120                        self.sourcetype.clone(),
121                        self.source.clone(),
122                        self.index.clone(),
123                        self.host_key.clone(),
124                    )
125                } else {
126                    EventPartitioner::new(None, None, None, None)
127                },
128                batch_settings.timeout,
129                |_| batch_settings.as_byte_size_config(),
130            )
131            .request_builder(
132                default_request_builder_concurrency_limit(),
133                self.request_builder,
134            )
135            .filter_map(|request| async move {
136                match request {
137                    Err(e) => {
138                        error!("Failed to build HEC Logs request: {:?}.", e);
139                        None
140                    }
141                    Ok(req) => Some(req),
142                }
143            })
144            .into_driver(self.service)
145            .run()
146            .await
147    }
148}
149
150#[async_trait]
151impl<S> StreamSink<Event> for HecLogsSink<S>
152where
153    S: Service<HecRequest> + Send + 'static,
154    S::Future: Send + 'static,
155    S::Response: DriverResponse + Send + 'static,
156    S::Error: fmt::Debug + Into<crate::Error> + Send,
157{
158    async fn run(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
159        self.run_inner(input).await
160    }
161}
162
163/// Returns `true` if any of the given partition templates produce a
164/// `Confined` render error for this event, emitting the appropriate internal
165/// event for each violation.  Used to pre-filter events before batching so
166/// that confinement violations result in event drops rather than silent
167/// metadata omission.
168fn has_confined_partition_error(
169    event: &Event,
170    source: Option<&Template>,
171    sourcetype: Option<&Template>,
172    index: Option<&Template>,
173) -> bool {
174    let mut confined = false;
175    for (tpl, field) in [
176        (source, SOURCE_FIELD),
177        (sourcetype, SOURCETYPE_FIELD),
178        (index, INDEX_FIELD),
179    ] {
180        if let Some(error) = tpl
181            .and_then(|t| t.render_string(event).err())
182            .filter(|e| matches!(e, crate::template::TemplateRenderingError::Confined { .. }))
183        {
184            emit!(TemplateRenderingError {
185                error,
186                field: Some(field),
187                // Count the drop once — subsequent violations on the same event
188                // should not increment ComponentEventsDropped again.
189                drop_event: !confined,
190            });
191            confined = true;
192        }
193    }
194    confined
195}
196
197#[derive(Clone, Debug, PartialEq, Hash, Eq)]
198pub(super) struct Partitioned {
199    pub(super) token: Option<Arc<str>>,
200    pub(super) source: Option<String>,
201    pub(super) sourcetype: Option<String>,
202    pub(super) index: Option<String>,
203    pub(super) host: Option<String>,
204}
205
206#[derive(Default)]
207struct EventPartitioner {
208    pub sourcetype: Option<Template>,
209    pub source: Option<Template>,
210    pub index: Option<Template>,
211    pub host_key: Option<OptionalTargetPath>,
212}
213
214impl EventPartitioner {
215    const fn new(
216        sourcetype: Option<Template>,
217        source: Option<Template>,
218        index: Option<Template>,
219        host_key: Option<OptionalTargetPath>,
220    ) -> Self {
221        Self {
222            sourcetype,
223            source,
224            index,
225            host_key,
226        }
227    }
228}
229
230impl Partitioner for EventPartitioner {
231    type Item = HecProcessedEvent;
232    type Key = Option<Partitioned>;
233
234    fn partition(&self, item: &Self::Item) -> Self::Key {
235        let emit_err = |error, field| {
236            emit!(TemplateRenderingError {
237                error,
238                field: Some(field),
239                drop_event: false,
240            })
241        };
242
243        let source = self.source.as_ref().and_then(|source| {
244            source
245                .render_string(&item.event)
246                .map_err(|error| emit_err(error, SOURCE_FIELD))
247                .ok()
248        });
249
250        let sourcetype = self.sourcetype.as_ref().and_then(|sourcetype| {
251            sourcetype
252                .render_string(&item.event)
253                .map_err(|error| emit_err(error, SOURCETYPE_FIELD))
254                .ok()
255        });
256
257        let index = self.index.as_ref().and_then(|index| {
258            index
259                .render_string(&item.event)
260                .map_err(|error| emit_err(error, INDEX_FIELD))
261                .ok()
262        });
263
264        let host = user_or_namespaced_path(
265            &item.event,
266            self.host_key.as_ref(),
267            meaning::HOST,
268            log_schema().host_key_target_path(),
269        )
270        .and_then(|path| item.event.get(&path))
271        .and_then(|value| value.as_str().map(|s| s.to_string()));
272
273        Some(Partitioned {
274            token: item.event.metadata().splunk_hec_token(),
275            source,
276            sourcetype,
277            index,
278            host,
279        })
280    }
281}
282
283#[derive(PartialEq, Default, Clone, Debug)]
284pub struct HecLogsProcessedEventMetadata {
285    pub sourcetype: Option<String>,
286    pub source: Option<String>,
287    pub index: Option<String>,
288    pub host: Option<Value>,
289    pub timestamp: Option<f64>,
290    pub fields: LogEvent,
291    pub endpoint_target: EndpointTarget,
292}
293
294impl ByteSizeOf for HecLogsProcessedEventMetadata {
295    fn allocated_bytes(&self) -> usize {
296        self.sourcetype.allocated_bytes()
297            + self.source.allocated_bytes()
298            + self.index.allocated_bytes()
299            + self.host.allocated_bytes()
300            + self.fields.allocated_bytes()
301    }
302}
303
304pub type HecProcessedEvent = ProcessedEvent<LogEvent, HecLogsProcessedEventMetadata>;
305
306// determine the path for a field from one of the following use cases:
307// 1. user provided a path in the config settings
308//     a. If the path provided was an empty string, None is returned
309// 2. namespaced path ("default")
310//     a. if Legacy namespace, use the provided path from the global log schema
311//     b. if Vector namespace, use the semantically defined path
312fn user_or_namespaced_path(
313    log: &LogEvent,
314    user_key: Option<&OptionalTargetPath>,
315    semantic: &str,
316    legacy_path: Option<&OwnedTargetPath>,
317) -> Option<OwnedTargetPath> {
318    match user_key {
319        Some(maybe_key) => maybe_key.path.clone(),
320        None => match log.namespace() {
321            LogNamespace::Vector => log.find_key_by_meaning(semantic).cloned(),
322            LogNamespace::Legacy => legacy_path.cloned(),
323        },
324    }
325}
326
327pub fn process_log(event: Event, data: &HecLogData) -> HecProcessedEvent {
328    let mut log = event.into_log();
329
330    let sourcetype = data
331        .sourcetype
332        .and_then(|sourcetype| render_template_string(sourcetype, &log, SOURCETYPE_FIELD));
333
334    let source = data
335        .source
336        .and_then(|source| render_template_string(source, &log, SOURCE_FIELD));
337
338    let index = data
339        .index
340        .and_then(|index| render_template_string(index, &log, INDEX_FIELD));
341
342    let host = user_or_namespaced_path(
343        &log,
344        data.host_key.as_ref(),
345        meaning::HOST,
346        log_schema().host_key_target_path(),
347    )
348    .and_then(|path| log.get(&path))
349    .cloned();
350
351    // only extract the timestamp if this is the Event endpoint, and if the setting
352    // `auto_extract_timestamp` is false (because that indicates that we should leave
353    // the timestamp in the event as-is, and let Splunk do the extraction).
354    let timestamp = if EndpointTarget::Event == data.endpoint_target && !data.auto_extract_timestamp
355    {
356        user_or_namespaced_path(
357            &log,
358            data.timestamp_key.as_ref(),
359            meaning::TIMESTAMP,
360            log_schema().timestamp_key_target_path(),
361        )
362        .and_then(|timestamp_path| {
363            match log.remove(&timestamp_path) {
364                Some(Value::Timestamp(ts)) => {
365                    // set nanos in log if valid timestamp in event and timestamp_nanos_key is configured
366                    if let Some(key) = data.timestamp_nanos_key {
367                        log.try_insert(event_path!(key), ts.timestamp_subsec_nanos() % 1_000_000);
368                    }
369                    Some((ts.timestamp_millis() as f64) / 1000f64)
370                }
371                Some(value) => {
372                    emit!(SplunkEventTimestampInvalidType {
373                        r#type: value.kind_str()
374                    });
375                    None
376                }
377                None => {
378                    emit!(SplunkEventTimestampMissing {});
379                    None
380                }
381            }
382        })
383    } else {
384        None
385    };
386
387    let fields = data
388        .indexed_fields
389        .iter()
390        .filter_map(|field| {
391            log.get((PathPrefix::Event, field))
392                .map(|value| (field.to_string(), value.clone()))
393        })
394        .collect::<LogEvent>();
395
396    let metadata = HecLogsProcessedEventMetadata {
397        sourcetype,
398        source,
399        index,
400        host,
401        timestamp,
402        fields,
403        endpoint_target: data.endpoint_target,
404    };
405
406    ProcessedEvent {
407        event: log,
408        metadata,
409    }
410}
411
412impl EventCount for HecProcessedEvent {
413    fn event_count(&self) -> usize {
414        // A HecProcessedEvent is mapped one-to-one with an event.
415        1
416    }
417}
418
419#[cfg(test)]
420mod pre_check_tests {
421    use super::*;
422    use crate::template::ConfinementConfig;
423    use vector_lib::event::LogEvent;
424    use vrl::event_path;
425
426    /// Regression for issue #25829: `process_log` extracts + removes fields
427    /// like `timestamp` from the event body. If the confinement pre-check
428    /// runs *after* `process_log`, an attacker-controlled `timestamp: "../..."`
429    /// value renders as `Confined` inside `process_log`, then the field is
430    /// removed, and the pre-check sees `MissingKeys` instead of `Confined` —
431    /// the attack event ships without an index. The check must run on the
432    /// original event before any mutation.
433    #[test]
434    fn confinement_detects_attack_before_process_log_mutates_event() {
435        let index_template = Template::try_from("idx-{{ timestamp }}")
436            .unwrap()
437            .confine(&ConfinementConfig::default(), "splunk_hec_logs", "index")
438            .unwrap();
439
440        // Attacker-controlled `timestamp` field with a traversal payload.
441        let mut log = LogEvent::from("attack payload");
442        log.insert(event_path!("timestamp"), "../../evil");
443        let event = Event::Log(log);
444
445        // Running the pre-check against the original event catches the
446        // `Confined` render before any downstream mutation gets a chance
447        // to convert it to `MissingKeys`.
448        assert!(has_confined_partition_error(
449            &event,
450            None,
451            None,
452            Some(&index_template),
453        ));
454    }
455}
vector/sinks/splunk_hec/logs/sink.rs

vector/sinks/splunk_hec/logs/
sink.rs
