vector/sources/dnstap/
mod.rs

1use std::path::PathBuf;
2
3use base64::prelude::{BASE64_STANDARD, Engine as _};
4use dnsmsg_parser::dns_message_parser::DnsParserOptions;
5use dnstap_parser::{
6    parser::DnstapParser,
7    schema::{DNSTAP_VALUE_PATHS, DnstapEventSchema},
8};
9use vector_lib::{
10    configurable::configurable_component,
11    event::{Event, LogEvent},
12    internal_event::{ByteSize, BytesReceived, InternalEventHandle, Protocol, Registered},
13    lookup::{owned_value_path, path},
14    tls::MaybeTlsSettings,
15};
16use vrl::{
17    path::{OwnedValuePath, PathPrefix},
18    value::{Kind, kind::Collection},
19};
20
21use super::util::framestream::{
22    FrameHandler, build_framestream_tcp_source, build_framestream_unix_source,
23};
24use crate::{
25    Result,
26    config::{DataType, SourceConfig, SourceContext, SourceOutput, log_schema},
27    internal_events::DnstapParseError,
28};
29
30pub mod tcp;
31#[cfg(unix)]
32pub mod unix;
33use vector_lib::{
34    config::{LegacyKey, LogNamespace},
35    lookup::lookup_v2::OptionalValuePath,
36};
37
38/// Configuration for the `dnstap` source.
39#[configurable_component(source("dnstap", "Collect DNS logs from a dnstap-compatible server."))]
40#[derive(Clone, Debug)]
41pub struct DnstapConfig {
42    #[serde(flatten)]
43    pub mode: Mode,
44
45    /// Maximum DNSTAP frame length that the source accepts.
46    ///
47    /// If any frame is longer than this, it is discarded.
48    #[serde(default = "default_max_frame_length")]
49    #[configurable(metadata(docs::type_unit = "bytes"))]
50    pub max_frame_length: usize,
51
52    /// Overrides the name of the log field used to add the source path to each event.
53    ///
54    /// The value is the socket path itself.
55    ///
56    /// By default, the [global `log_schema.host_key` option][global_host_key] is used.
57    ///
58    /// [global_host_key]: https://vector.dev/docs/reference/configuration/global-options/#log_schema.host_key
59    pub host_key: Option<OptionalValuePath>,
60
61    /// Whether or not to skip parsing or decoding of DNSTAP frames.
62    ///
63    /// If set to `true`, frames are not parsed or decoded. The raw frame data is set as a field on the event
64    /// (called `rawData`) and encoded as a base64 string.
65    pub raw_data_only: Option<bool>,
66
67    /// Whether or not to concurrently process DNSTAP frames.
68    pub multithreaded: Option<bool>,
69
70    /// Maximum number of frames that can be processed concurrently.
71    pub max_frame_handling_tasks: Option<usize>,
72
73    /// Whether to downcase all DNSTAP hostnames received for consistency
74    #[serde(default = "crate::serde::default_false")]
75    pub lowercase_hostnames: bool,
76
77    /// The namespace to use for logs. This overrides the global settings.
78    #[configurable(metadata(docs::hidden))]
79    #[serde(default)]
80    pub log_namespace: Option<bool>,
81}
82
83fn default_max_frame_length() -> usize {
84    bytesize::kib(100u64) as usize
85}
86
87/// Listening mode for the `dnstap` source.
88#[configurable_component]
89#[derive(Clone, Debug)]
90#[serde(tag = "mode", rename_all = "snake_case")]
91#[configurable(metadata(docs::enum_tag_description = "The type of dnstap socket to use."))]
92#[allow(clippy::large_enum_variant)] // just used for configuration
93pub enum Mode {
94    /// Listen on TCP.
95    Tcp(tcp::TcpConfig),
96
97    /// Listen on a Unix domain socket
98    #[cfg(unix)]
99    Unix(unix::UnixConfig),
100}
101
102impl DnstapConfig {
103    pub fn new(socket_path: PathBuf) -> Self {
104        Self {
105            mode: Mode::Unix(unix::UnixConfig::new(socket_path)),
106            ..Default::default()
107        }
108    }
109
110    fn log_namespace(&self) -> LogNamespace {
111        self.log_namespace.unwrap_or(false).into()
112    }
113
114    fn raw_data_only(&self) -> bool {
115        self.raw_data_only.unwrap_or(false)
116    }
117
118    pub fn schema_definition(&self, log_namespace: LogNamespace) -> vector_lib::schema::Definition {
119        let event_schema = DnstapEventSchema;
120
121        match self.log_namespace() {
122            LogNamespace::Legacy => {
123                let schema = vector_lib::schema::Definition::empty_legacy_namespace();
124
125                if self.raw_data_only()
126                    && let Some(message_key) = log_schema().message_key()
127                {
128                    return schema.with_event_field(message_key, Kind::bytes(), Some("message"));
129                }
130                event_schema.schema_definition(schema)
131            }
132            LogNamespace::Vector => {
133                let schema = vector_lib::schema::Definition::new_with_default_metadata(
134                    Kind::object(Collection::empty()),
135                    [log_namespace],
136                )
137                .with_standard_vector_source_metadata();
138
139                if self.raw_data_only() {
140                    schema.with_event_field(
141                        &owned_value_path!("message"),
142                        Kind::bytes(),
143                        Some("message"),
144                    )
145                } else {
146                    event_schema.schema_definition(schema)
147                }
148            }
149        }
150    }
151}
152
153impl Default for DnstapConfig {
154    fn default() -> Self {
155        Self {
156            #[cfg(unix)]
157            mode: Mode::Unix(unix::UnixConfig::default()),
158            #[cfg(not(unix))]
159            mode: Mode::Tcp(tcp::TcpConfig::from_address(std::net::SocketAddr::new(
160                std::net::IpAddr::V4(std::net::Ipv4Addr::new(0, 0, 0, 0)),
161                9000,
162            ))),
163            max_frame_length: default_max_frame_length(),
164            host_key: None,
165            raw_data_only: None,
166            multithreaded: None,
167            max_frame_handling_tasks: None,
168            lowercase_hostnames: false,
169            log_namespace: None,
170        }
171    }
172}
173
174impl_generate_config_from_default!(DnstapConfig);
175
176#[async_trait::async_trait]
177#[typetag::serde(name = "dnstap")]
178impl SourceConfig for DnstapConfig {
179    async fn build(&self, cx: SourceContext) -> Result<super::Source> {
180        let log_namespace = cx.log_namespace(self.log_namespace);
181        let common_frame_handler = CommonFrameHandler::new(self, log_namespace);
182        match &self.mode {
183            Mode::Tcp(config) => {
184                let tls_config = config.tls().as_ref().map(|tls| tls.tls_config.clone());
185
186                let tls = MaybeTlsSettings::from_config(tls_config.as_ref(), true)?;
187                let frame_handler = tcp::DnstapFrameHandler::new(
188                    config.clone(),
189                    tls,
190                    common_frame_handler,
191                    log_namespace,
192                );
193
194                build_framestream_tcp_source(frame_handler, cx.shutdown, cx.out)
195            }
196            #[cfg(unix)]
197            Mode::Unix(config) => {
198                let frame_handler =
199                    unix::DnstapFrameHandler::new(config.clone(), common_frame_handler);
200                build_framestream_unix_source(frame_handler, cx.shutdown, cx.out)
201            }
202        }
203    }
204
205    fn outputs(&self, global_log_namespace: LogNamespace) -> Vec<SourceOutput> {
206        let log_namespace = global_log_namespace.merge(Some(self.log_namespace()));
207        let schema_definition = self
208            .schema_definition(log_namespace)
209            .with_standard_vector_source_metadata();
210        vec![SourceOutput::new_maybe_logs(
211            DataType::Log,
212            schema_definition,
213        )]
214    }
215
216    fn can_acknowledge(&self) -> bool {
217        false
218    }
219}
220
221#[derive(Clone)]
222struct CommonFrameHandler {
223    max_frame_length: usize,
224    content_type: String,
225    raw_data_only: bool,
226    multithreaded: bool,
227    max_frame_handling_tasks: usize,
228    host_key: Option<OwnedValuePath>,
229    timestamp_key: Option<OwnedValuePath>,
230    source_type_key: Option<OwnedValuePath>,
231    bytes_received: Registered<BytesReceived>,
232    lowercase_hostnames: bool,
233    log_namespace: LogNamespace,
234}
235
236impl CommonFrameHandler {
237    pub fn new(config: &DnstapConfig, log_namespace: LogNamespace) -> Self {
238        let source_type_key = log_schema().source_type_key();
239        let timestamp_key = log_schema().timestamp_key();
240
241        let host_key = config
242            .host_key
243            .clone()
244            .map_or(log_schema().host_key().cloned(), |k| k.path);
245
246        Self {
247            max_frame_length: config.max_frame_length,
248            content_type: "protobuf:dnstap.Dnstap".to_string(),
249            raw_data_only: config.raw_data_only.unwrap_or(false),
250            multithreaded: config.multithreaded.unwrap_or(false),
251            max_frame_handling_tasks: config.max_frame_handling_tasks.unwrap_or(1000),
252            host_key,
253            timestamp_key: timestamp_key.cloned(),
254            source_type_key: source_type_key.cloned(),
255            bytes_received: register!(BytesReceived::from(Protocol::from("protobuf"))),
256            lowercase_hostnames: config.lowercase_hostnames,
257            log_namespace,
258        }
259    }
260}
261
262impl FrameHandler for CommonFrameHandler {
263    fn content_type(&self) -> String {
264        self.content_type.clone()
265    }
266
267    fn max_frame_length(&self) -> usize {
268        self.max_frame_length
269    }
270
271    fn handle_event(
272        &self,
273        received_from: Option<vrl::prelude::Bytes>,
274        frame: vrl::prelude::Bytes,
275    ) -> Option<vector_lib::event::Event> {
276        self.bytes_received.emit(ByteSize(frame.len()));
277
278        let mut log_event = LogEvent::default();
279
280        if let Some(host) = received_from {
281            self.log_namespace.insert_source_metadata(
282                DnstapConfig::NAME,
283                &mut log_event,
284                self.host_key.as_ref().map(LegacyKey::Overwrite),
285                path!("host"),
286                host,
287            );
288        }
289
290        if self.raw_data_only {
291            log_event.insert(
292                (PathPrefix::Event, &DNSTAP_VALUE_PATHS.raw_data),
293                BASE64_STANDARD.encode(&frame),
294            );
295        } else if let Err(err) = DnstapParser::parse(
296            &mut log_event,
297            frame,
298            DnsParserOptions {
299                lowercase_hostnames: self.lowercase_hostnames,
300            },
301        ) {
302            emit!(DnstapParseError {
303                error: format!("Dnstap protobuf decode error {err:?}.")
304            });
305            return None;
306        }
307
308        if self.log_namespace == LogNamespace::Vector {
309            // The timestamp is inserted by the parser which caters for the Legacy namespace.
310            self.log_namespace.insert_vector_metadata(
311                &mut log_event,
312                self.timestamp_key(),
313                path!("ingest_timestamp"),
314                chrono::Utc::now(),
315            );
316        }
317
318        self.log_namespace.insert_vector_metadata(
319            &mut log_event,
320            self.source_type_key(),
321            path!("source_type"),
322            DnstapConfig::NAME,
323        );
324
325        Some(Event::from(log_event))
326    }
327
328    fn multithreaded(&self) -> bool {
329        self.multithreaded
330    }
331
332    fn max_frame_handling_tasks(&self) -> usize {
333        self.max_frame_handling_tasks
334    }
335
336    fn host_key(&self) -> &Option<vrl::path::OwnedValuePath> {
337        &self.host_key
338    }
339
340    fn timestamp_key(&self) -> Option<&vrl::path::OwnedValuePath> {
341        self.timestamp_key.as_ref()
342    }
343
344    fn source_type_key(&self) -> Option<&vrl::path::OwnedValuePath> {
345        self.source_type_key.as_ref()
346    }
347}
348
349#[cfg(test)]
350mod tests {
351    use vector_lib::event::{Event, LogEvent};
352    use vrl::event_path;
353
354    use super::*;
355
356    #[test]
357    fn simple_matches_schema() {
358        let record = r#"{"dataType":"Message",
359                         "dataTypeId":1,
360                         "messageType":"ClientQuery",
361                         "messageTypeId":5,
362                         "requestData":{
363                           "fullRcode":0,
364                           "header":{
365                             "aa":false,
366                             "ad":true,
367                             "anCount":0,
368                             "arCount":1,
369                             "cd":false,
370                             "id":38339,
371                             "nsCount":0,
372                             "opcode":0,
373                             "qdCount":1,
374                             "qr":0,
375                             "ra":false,
376                             "rcode":0,
377                             "rd":true,
378                             "tc":false},
379                           "opt":{
380                             "do":false,
381                             "ednsVersion":0,
382                             "extendedRcode":0,
383                             "options":[{"optCode":10,
384                                         "optName":"Cookie",
385                                         "optValue":"5JiWq4VYa7U="}],
386                             "udpPayloadSize":1232},
387                           "question":[{"class":"IN","domainName":"whoami.example.org.","questionType":"A","questionTypeId":1}],
388                           "rcodeName":"NoError",
389                           "time":1667909880863224758,
390                           "timePrecision":"ns"},
391                         "serverId":"stephenwakely-Precision-5570",
392                         "serverVersion":"CoreDNS-1.10.0",
393                         "socketFamily":"INET",
394                         "socketProtocol":"UDP",
395                         "sourceAddress":"0.0.0.0",
396                         "sourcePort":54782,
397                         "source_type":"dnstap",
398                         "time":1667909880863224758,
399                         "timePrecision":"ns"
400                         }"#;
401
402        let json: serde_json::Value = serde_json::from_str(record).unwrap();
403        let mut event = Event::from(LogEvent::from(vrl::value::Value::from(json)));
404        event
405            .as_mut_log()
406            .insert(event_path!("timestamp"), chrono::Utc::now());
407
408        let definition = DnstapEventSchema;
409        let schema = vector_lib::schema::Definition::empty_legacy_namespace()
410            .with_standard_vector_source_metadata();
411
412        definition
413            .schema_definition(schema)
414            .assert_valid_for_event(&event)
415    }
416}
417
418#[cfg(all(test, feature = "dnstap-integration-tests"))]
419mod integration_tests {
420    #![allow(clippy::print_stdout)] // tests
421
422    use bollard::{
423        Docker,
424        exec::{CreateExecOptions, StartExecOptions},
425    };
426    use futures::StreamExt;
427    use serde_json::json;
428    use tokio::time;
429    use vector_lib::{event::Event, lookup::lookup_v2::OptionalValuePath};
430    use vrl::event_path;
431
432    use self::unix::UnixConfig;
433    use super::*;
434    use crate::{
435        SourceSender,
436        event::Value,
437        test_util::{
438            components::{SOURCE_TAGS, assert_source_compliance},
439            wait_for,
440        },
441    };
442
443    async fn test_dnstap(raw_data: bool, query_type: &'static str) {
444        assert_source_compliance(&SOURCE_TAGS, async {
445            let (sender, mut recv) = SourceSender::new_test();
446
447            tokio::spawn(async move {
448                let socket = get_socket(raw_data, query_type);
449
450                DnstapConfig {
451                    mode: Mode::Unix(UnixConfig {
452                        socket_path: socket,
453                        socket_file_mode: Some(511),
454                        socket_receive_buffer_size: Some(10485760),
455                        socket_send_buffer_size: Some(10485760),
456                    }),
457                    max_frame_length: 102400,
458                    host_key: Some(OptionalValuePath::from(owned_value_path!("key"))),
459                    raw_data_only: Some(raw_data),
460                    multithreaded: Some(false),
461                    max_frame_handling_tasks: Some(100000),
462                    lowercase_hostnames: false,
463                    log_namespace: None,
464                }
465                .build(SourceContext::new_test(sender, None))
466                .await
467                .unwrap()
468                .await
469                .unwrap()
470            });
471
472            send_query(raw_data, query_type);
473
474            let event = time::timeout(time::Duration::from_secs(10), recv.next())
475                .await
476                .expect("fetch dnstap source event timeout")
477                .expect("failed to get dnstap source event from a stream");
478            let mut events = vec![event];
479            loop {
480                match time::timeout(time::Duration::from_secs(1), recv.next()).await {
481                    Ok(Some(event)) => events.push(event),
482                    Ok(None) => {
483                        println!("None: No event");
484                        break;
485                    }
486                    Err(e) => {
487                        println!("Error: {e}");
488                        break;
489                    }
490                }
491            }
492
493            verify_events(raw_data, query_type, &events);
494        })
495        .await;
496    }
497
498    fn send_query(raw_data: bool, query_type: &'static str) {
499        tokio::spawn(async move {
500            let socket_path = get_socket(raw_data, query_type);
501            let (query_port, control_port) = get_bind_ports(raw_data, query_type);
502
503            // Wait for the source to create its respective socket before telling BIND to reload, causing it to open
504            // that new socket file.
505            wait_for(move || {
506                let path = socket_path.clone();
507                async move { path.exists() }
508            })
509            .await;
510
511            // Now instruct BIND to reopen its DNSTAP socket file and execute the given query.
512            reload_bind_dnstap_socket(control_port).await;
513
514            match query_type {
515                "query" => {
516                    nslookup(query_port).await;
517                }
518                "update" => {
519                    nsupdate().await;
520                }
521                _ => (),
522            }
523        });
524    }
525
526    fn verify_events(raw_data: bool, query_event: &'static str, events: &[Event]) {
527        if raw_data {
528            assert_eq!(events.len(), 2);
529            assert!(
530                events
531                    .iter()
532                    .all(|v| v.as_log().get(event_path!("rawData")).is_some()),
533                "No rawData field!"
534            );
535        } else if query_event == "query" {
536            assert_eq!(events.len(), 2);
537            assert!(
538                events
539                    .iter()
540                    .any(|v| v.as_log().get(event_path!("messageType"))
541                        == Some(&Value::Bytes("ClientQuery".into()))),
542                "No ClientQuery event!"
543            );
544            assert!(
545                events
546                    .iter()
547                    .any(|v| v.as_log().get(event_path!("messageType"))
548                        == Some(&Value::Bytes("ClientResponse".into()))),
549                "No ClientResponse event!"
550            );
551        } else if query_event == "update" {
552            assert_eq!(events.len(), 4);
553            assert!(
554                events
555                    .iter()
556                    .any(|v| v.as_log().get(event_path!("messageType"))
557                        == Some(&Value::Bytes("UpdateQuery".into()))),
558                "No UpdateQuery event!"
559            );
560            assert!(
561                events
562                    .iter()
563                    .any(|v| v.as_log().get(event_path!("messageType"))
564                        == Some(&Value::Bytes("UpdateResponse".into()))),
565                "No UpdateResponse event!"
566            );
567            assert!(
568                events
569                    .iter()
570                    .any(|v| v.as_log().get(event_path!("messageType"))
571                        == Some(&Value::Bytes("AuthQuery".into()))),
572                "No UpdateQuery event!"
573            );
574            assert!(
575                events
576                    .iter()
577                    .any(|v| v.as_log().get(event_path!("messageType"))
578                        == Some(&Value::Bytes("AuthResponse".into()))),
579                "No UpdateResponse event!"
580            );
581        }
582
583        for event in events {
584            let json = serde_json::to_value(event.as_log().all_event_fields().unwrap()).unwrap();
585            match query_event {
586                "query" => {
587                    if json["messageType"] == json!("ClientQuery") {
588                        assert_eq!(
589                            json["requestData.question[0].domainName"],
590                            json!("h1.example.com.")
591                        );
592                        assert_eq!(json["requestData.rcodeName"], json!("NoError"));
593                    } else if json["messageType"] == json!("ClientResponse") {
594                        assert_eq!(
595                            json["responseData.answers[0].domainName"],
596                            json!("h1.example.com.")
597                        );
598                        assert_eq!(json["responseData.answers[0].rData"], json!("10.0.0.11"));
599                        assert_eq!(json["responseData.rcodeName"], json!("NoError"));
600                    }
601                }
602                "update" => {
603                    if json["messageType"] == json!("UpdateQuery") {
604                        assert_eq!(
605                            json["requestData.update[0].domainName"],
606                            json!("dh1.example.com.")
607                        );
608                        assert_eq!(json["requestData.update[0].rData"], json!("10.0.0.21"));
609                        assert_eq!(json["requestData.rcodeName"], json!("NoError"));
610                    } else if json["messageType"] == json!("UpdateResponse") {
611                        assert_eq!(json["responseData.rcodeName"], json!("NoError"));
612                    }
613                }
614                _ => (),
615            }
616        }
617    }
618
619    fn get_container() -> String {
620        std::env::var("CONTAINER_NAME").unwrap_or_else(|_| "vector_dnstap".into())
621    }
622
623    fn get_socket(raw_data: bool, query_type: &'static str) -> PathBuf {
624        let socket_folder = std::env::var("BIND_SOCKET")
625            .map(PathBuf::from)
626            .expect("BIND socket directory must be specified via BIND_SOCKET");
627
628        match query_type {
629            "query" if raw_data => socket_folder.join("dnstap.sock1"),
630            "query" => socket_folder.join("dnstap.sock2"),
631            "update" => socket_folder.join("dnstap.sock3"),
632            _ => unreachable!("no other test variants should exist"),
633        }
634    }
635
636    fn get_bind_ports(raw_data: bool, query_type: &'static str) -> (&'static str, &'static str) {
637        // Returns the query port and control port, respectively, for the given BIND instance.
638        match query_type {
639            "query" if raw_data => ("8001", "9001"),
640            "query" => ("8002", "9002"),
641            "update" => ("8003", "9003"),
642            _ => unreachable!("no other test variants should exist"),
643        }
644    }
645
646    async fn dnstap_exec(cmd: Vec<&str>) {
647        let docker = Docker::connect_with_defaults().expect("failed binding to docker socket");
648        let config = CreateExecOptions {
649            cmd: Some(cmd),
650            attach_stdout: Some(true),
651            attach_stderr: Some(true),
652            ..Default::default()
653        };
654        let result = docker
655            .create_exec(get_container().as_str(), config)
656            .await
657            .expect("failed to execute command");
658        docker
659            .start_exec(&result.id, None::<StartExecOptions>)
660            .await
661            .expect("failed to execute command");
662    }
663
664    async fn reload_bind_dnstap_socket(control_port: &str) {
665        dnstap_exec(vec![
666            "/usr/sbin/rndc",
667            "-p",
668            control_port,
669            "dnstap",
670            "-reopen",
671        ])
672        .await
673    }
674
675    async fn nslookup(port: &str) {
676        dnstap_exec(vec![
677            "nslookup",
678            "-type=A",
679            format!("-port={port}").as_str(),
680            "h1.example.com",
681            "localhost",
682        ])
683        .await
684    }
685
686    async fn nsupdate() {
687        dnstap_exec(vec!["nsupdate", "-v", "/bind3/etc/bind/nsupdate.txt"]).await
688    }
689
690    #[tokio::test]
691    async fn test_dnstap_raw_event() {
692        test_dnstap(true, "query").await;
693    }
694
695    #[tokio::test]
696    async fn test_dnstap_query_event() {
697        test_dnstap(false, "query").await;
698    }
699
700    #[tokio::test]
701    async fn test_dnstap_update_event() {
702        test_dnstap(false, "update").await;
703    }
704}
vector/sources/dnstap/mod.rs

vector/sources/dnstap/
mod.rs
