pipeline/etl/processor/
vrl_processor.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::cell::RefCell;
16use std::collections::BTreeMap;
17
18use chrono_tz::Tz;
19use once_cell::sync::Lazy;
20use snafu::{OptionExt, ensure};
21use vrl::compiler::runtime::Runtime;
22use vrl::compiler::{Program, TargetValue, compile};
23use vrl::diagnostic::Formatter;
24use vrl::prelude::{
25    ArgumentList, Compiled, Example, Function, FunctionCompileContext, Parameter, TimeZone,
26    TypeState, function,
27};
28use vrl::value::{Kind, Secrets, Value as VrlValue};
29
30use crate::error::{
31    CompileVrlSnafu, ExecuteVrlSnafu, KeyMustBeStringSnafu, Result, VrlRegexValueSnafu,
32    VrlReturnValueSnafu,
33};
34use crate::etl::processor::yaml_string;
35
36pub(crate) const PROCESSOR_VRL: &str = "vrl";
37const SOURCE: &str = "source";
38
39static UTC_TIMEZONE: Lazy<TimeZone> = Lazy::new(|| TimeZone::Named(Tz::UTC));
40
41thread_local! {
42    static VRL_RUNTIME: RefCell<Runtime> = RefCell::new(Runtime::default());
43}
44
45#[derive(Debug)]
46pub struct VrlProcessor {
47    source: String,
48    program: Program,
49}
50
51impl VrlProcessor {
52    pub fn new(source: String, allow_host_access: bool) -> Result<Self> {
53        let fns = functions(allow_host_access);
54
55        let compile_result = compile(&source, &fns).map_err(|e| {
56            CompileVrlSnafu {
57                msg: Formatter::new(&source, e).to_string(),
58            }
59            .build()
60        })?;
61
62        let program = compile_result.program;
63
64        // check if the return value is have regex
65        let result_def = program.final_type_info().result;
66        let kind = result_def.kind();
67        // Check if the return type could possibly be an object or array.
68        // We use contains_* methods since VRL type inference may return
69        // a Kind that represents multiple possible types.
70        ensure!(
71            kind.contains_object() || kind.contains_array(),
72            VrlReturnValueSnafu {
73                result_kind: kind.clone(),
74            }
75        );
76        check_regex_output(kind)?;
77
78        Ok(Self { source, program })
79    }
80
81    pub fn resolve(&self, value: VrlValue) -> Result<VrlValue> {
82        let mut target = TargetValue {
83            value,
84            metadata: VrlValue::Object(BTreeMap::new()),
85            secrets: Secrets::default(),
86        };
87
88        let re = VRL_RUNTIME
89            .with(|runtime| {
90                let mut runtime = runtime.borrow_mut();
91                runtime.clear();
92                let result = runtime.resolve(&mut target, &self.program, &UTC_TIMEZONE);
93                runtime.clear();
94                result
95            })
96            .map_err(|e| {
97                ExecuteVrlSnafu {
98                    msg: e.get_expression_error().to_string(),
99                }
100                .build()
101            })?;
102
103        Ok(re)
104    }
105}
106
107impl VrlProcessor {
108    /// Parses VRL source using the server's host access policy.
109    pub fn from_yaml(value: &yaml_rust::yaml::Hash, allow_host_access: bool) -> Result<Self> {
110        let mut source = String::new();
111        for (k, v) in value.iter() {
112            let key = k
113                .as_str()
114                .with_context(|| KeyMustBeStringSnafu { k: k.clone() })?;
115            if key == SOURCE {
116                source = yaml_string(v, SOURCE)?;
117            }
118        }
119        Self::new(source, allow_host_access)
120    }
121}
122
123fn functions(allow_host_access: bool) -> Vec<Box<dyn Function>> {
124    let mut functions = vrl::stdlib::all();
125    if allow_host_access {
126        return functions;
127    }
128
129    // These functions bypass VRL's event target, including file reads during compilation.
130    // When upgrading VRL, re-audit this denylist and the argument restrictions below;
131    // new host-access functions are otherwise enabled by default.
132    functions.retain(|function| {
133        !matches!(
134            function.identifier(),
135            "get_env_var"
136                | "get_hostname"
137                | "get_timezone_name"
138                | "http_request"
139                | "dns_lookup"
140                | "reverse_dns"
141                | "encode_proto"
142                | "parse_proto"
143                | "parse_etld"
144                | "validate_json_schema"
145                | "log"
146        )
147    });
148    functions
149        .into_iter()
150        .map(|function| match function.identifier() {
151            "parse_groks" | "parse_timestamp" | "format_timestamp" => {
152                Box::new(RestrictedFunction(function)) as Box<dyn Function>
153            }
154            _ => function,
155        })
156        .collect()
157}
158
159/// Retains data-only uses of functions that also accept host-access arguments.
160#[derive(Debug)]
161struct RestrictedFunction(Box<dyn Function>);
162
163impl Function for RestrictedFunction {
164    fn identifier(&self) -> &'static str {
165        self.0.identifier()
166    }
167
168    fn usage(&self) -> &'static str {
169        self.0.usage()
170    }
171
172    fn category(&self) -> &'static str {
173        self.0.category()
174    }
175
176    fn return_kind(&self) -> u16 {
177        self.0.return_kind()
178    }
179
180    fn examples(&self) -> &'static [Example] {
181        self.0.examples()
182    }
183
184    fn parameters(&self) -> &'static [Parameter] {
185        self.0.parameters()
186    }
187
188    fn pure(&self) -> bool {
189        self.0.pure()
190    }
191
192    fn compile(
193        &self,
194        state: &TypeState,
195        ctx: &mut FunctionCompileContext,
196        arguments: ArgumentList,
197    ) -> Compiled {
198        if self.identifier() == "parse_groks" {
199            if arguments
200                .optional_array("alias_sources")?
201                .is_some_and(|sources| !sources.is_empty())
202            {
203                return Err(function::Error::InvalidArgument {
204                    keyword: "alias_sources",
205                    value: "alias_sources".into(),
206                    error: "file imports require pipeline.allow_vrl_host_access",
207                }
208                .into());
209            }
210        } else if let Some(timezone) = arguments.optional_literal("timezone", state)? {
211            // A static named timezone cannot resolve to the host's local timezone at runtime.
212            if !matches!(&timezone, VrlValue::Bytes(bytes)
213                if matches!(TimeZone::parse(&String::from_utf8_lossy(bytes)), Some(TimeZone::Named(_))))
214            {
215                return Err(function::Error::InvalidArgument {
216                    keyword: "timezone",
217                    value: timezone,
218                    error: "a named timezone is required when pipeline.allow_vrl_host_access is false",
219                }
220                .into());
221            }
222        }
223        self.0.compile(state, ctx, arguments)
224    }
225}
226
227impl crate::etl::processor::Processor for VrlProcessor {
228    fn kind(&self) -> &str {
229        PROCESSOR_VRL
230    }
231
232    fn ignore_missing(&self) -> bool {
233        true
234    }
235
236    fn exec_mut(&self, val: VrlValue) -> Result<VrlValue> {
237        self.resolve(val)
238    }
239}
240
241fn check_regex_output(output_kind: &Kind) -> Result<()> {
242    if output_kind.is_regex() {
243        return VrlRegexValueSnafu.fail();
244    }
245
246    if let Some(arr) = output_kind.as_array() {
247        let k = arr.known();
248        for v in k.values() {
249            check_regex_output(v)?
250        }
251    }
252
253    if let Some(obj) = output_kind.as_object() {
254        let k = obj.known();
255        for v in k.values() {
256            check_regex_output(v)?
257        }
258    }
259
260    Ok(())
261}
262
263#[cfg(test)]
264mod tests {
265
266    use vrl::prelude::Bytes;
267    use vrl::value::KeyString;
268
269    use super::*;
270
271    #[test]
272    fn test_host_functions_require_opt_in() {
273        let calls = [
274            ("get_env_var", r#"get_env_var!("GREPTIME_VRL_TEST_SECRET")"#),
275            ("get_hostname", "get_hostname!()"),
276            ("get_timezone_name", "get_timezone_name!()"),
277            ("http_request", r#"http_request!("http://127.0.0.1:9/")"#),
278            ("dns_lookup", r#"dns_lookup!("localhost")"#),
279            ("reverse_dns", r#"reverse_dns!("127.0.0.1")"#),
280            (
281                "parse_proto",
282                r#"parse_proto!("", "missing.desc", "Message")"#,
283            ),
284            (
285                "encode_proto",
286                r#"encode_proto!({}, "missing.desc", "Message")"#,
287            ),
288            ("parse_etld", r#"parse_etld!("example.com")"#),
289            (
290                "validate_json_schema",
291                r#"validate_json_schema!("{}", "missing.json")"#,
292            ),
293            ("log", r#"log("host log write")"#),
294        ];
295        let unrestricted = functions(true);
296        for (name, call) in calls {
297            assert!(
298                unrestricted
299                    .iter()
300                    .any(|function| function.identifier() == name)
301            );
302            let source = format!(".result = {call}\n.");
303            let error = VrlProcessor::new(source, false).unwrap_err().to_string();
304            assert!(error.contains("undefined function"), "{name}: {error}");
305        }
306
307        temp_env::with_var("GREPTIME_VRL_TEST_SECRET", Some("decoy"), || {
308            let source = ".secret = get_env_var!(\"GREPTIME_VRL_TEST_SECRET\")\n.";
309            let processor = VrlProcessor::new(source.to_string(), true).unwrap();
310            let result = processor
311                .resolve(VrlValue::Object(BTreeMap::new()))
312                .unwrap();
313            assert_eq!(
314                result.as_object().unwrap()["secret"],
315                VrlValue::from("decoy")
316            );
317            assert!(VrlProcessor::new(source.to_string(), false).is_err());
318        });
319    }
320
321    #[test]
322    fn test_grok_file_import_requires_opt_in() {
323        let dir = tempfile::tempdir().unwrap();
324        let path = dir.path().join("aliases.json");
325        std::fs::write(&path, r#"{"custom_word":"%{word}"}"#).unwrap();
326        let path = serde_json::to_string(&path).unwrap();
327        let source = format!(
328            r#".parsed = parse_groks!("hello", patterns: ["%{{custom_word:message}}"], alias_sources: [{path}])
329."#
330        );
331        let error = VrlProcessor::new(source.clone(), false)
332            .unwrap_err()
333            .to_string();
334        assert!(error.contains("file imports require pipeline.allow_vrl_host_access"));
335        let result = VrlProcessor::new(source, true)
336            .unwrap()
337            .resolve(VrlValue::Object(BTreeMap::new()))
338            .unwrap();
339        assert_eq!(
340            result.as_object().unwrap()["parsed"].as_object().unwrap()["message"],
341            VrlValue::from("hello")
342        );
343
344        let source = r#".parsed = parse_groks!("hello", patterns: ["%{custom_word:message}"], aliases: {"custom_word": "%{word}"}, alias_sources: [])
345."#;
346        assert!(VrlProcessor::new(source.to_string(), false).is_ok());
347    }
348
349    #[test]
350    fn test_local_and_dynamic_timezones_require_opt_in() {
351        for timezone in [r#""local""#, r#""""#, "string!(.timezone)"] {
352            for call in [
353                format!(
354                    r#"format_timestamp!(t'2020-01-01T00:00:00Z', format: "%z", timezone: {timezone})"#
355                ),
356                format!(
357                    r#"parse_timestamp!("2020-01-01", format: "%Y-%m-%d", timezone: {timezone})"#
358                ),
359            ] {
360                let source = format!(".result = {call}\n.");
361                assert!(
362                    VrlProcessor::new(source.clone(), false).is_err(),
363                    "{source}"
364                );
365                assert!(VrlProcessor::new(source.clone(), true).is_ok(), "{source}");
366            }
367        }
368
369        let source = r#".parsed = parse_timestamp!("2020-01-01 00:00:00", format: "%Y-%m-%d %H:%M:%S")
370.utc = format_timestamp!(.parsed, format: "%z")
371.named = format_timestamp!(.parsed, format: "%z", timezone: "Asia/Shanghai")
372."#;
373        let result = VrlProcessor::new(source.to_string(), false)
374            .unwrap()
375            .resolve(VrlValue::Object(BTreeMap::new()))
376            .unwrap();
377        assert_eq!(result.as_object().unwrap()["utc"], VrlValue::from("+0000"));
378        assert_eq!(
379            result.as_object().unwrap()["named"],
380            VrlValue::from("+0800")
381        );
382    }
383
384    #[test]
385    fn test_pipeline_host_policy_is_server_controlled() {
386        let yaml = r#"
387allow_vrl_host_access: true
388processors:
389  - vrl:
390      allow_vrl_host_access: true
391      source: |
392        .secret = get_env_var!("GREPTIME_VRL_TEST_SECRET")
393        .
394transform:
395  - field: secret
396    type: string
397"#;
398        assert!(crate::parse(&crate::Content::Yaml(yaml)).is_err());
399        let options = crate::PipelineOptions {
400            allow_vrl_host_access: true,
401            ..Default::default()
402        };
403        assert!(crate::parse_with_options(&crate::Content::Yaml(yaml), &options).is_ok());
404    }
405
406    #[test]
407    fn test_vrl() {
408        let source = r#"
409.name.a = .user_info.name
410.name.b = .user_info.name
411del(.user_info)
412.timestamp = now()
413.
414"#;
415
416        let v = VrlProcessor::new(source.to_string(), false);
417        assert!(v.is_ok());
418        let v = v.unwrap();
419
420        let mut n = BTreeMap::new();
421        n.insert(
422            KeyString::from("name"),
423            VrlValue::Bytes(Bytes::from("certain_name")),
424        );
425
426        let mut m = BTreeMap::new();
427        m.insert(KeyString::from("user_info"), VrlValue::Object(n));
428
429        let re = v.resolve(VrlValue::Object(m));
430        assert!(re.is_ok());
431        let re = re.unwrap();
432
433        assert!(matches!(re, VrlValue::Object(_)));
434        let re = re.as_object().unwrap();
435        assert!(re.get("name").is_some());
436        let name = re.get("name").unwrap();
437        let name = name.as_object().unwrap();
438        assert!(matches!(name.get("a").unwrap(), VrlValue::Bytes(x) if x == "certain_name"));
439        assert!(matches!(name.get("b").unwrap(), VrlValue::Bytes(x) if x == "certain_name"));
440        assert!(re.get("timestamp").is_some());
441        let timestamp = re.get("timestamp").unwrap();
442        assert!(matches!(timestamp, VrlValue::Timestamp(_)));
443    }
444
445    #[test]
446    fn test_yaml_to_vrl() {
447        let yaml = r#"
448processors:
449  - vrl:
450      source: |
451        .name.a = .user_info.name
452        .name.b = .user_info.name
453        del(.user_info)
454        .timestamp = now()
455        .
456"#;
457        let y = yaml_rust::YamlLoader::load_from_str(yaml).unwrap();
458        let vrl_processor_yaml = y
459            .first()
460            .and_then(|x| x.as_hash())
461            .and_then(|x| x.get(&yaml_rust::Yaml::String("processors".to_string())))
462            .and_then(|x| x.as_vec())
463            .and_then(|x| x.first())
464            .and_then(|x| x.as_hash())
465            .and_then(|x| x.get(&yaml_rust::Yaml::String("vrl".to_string())))
466            .and_then(|x| x.as_hash())
467            .unwrap();
468
469        let vrl = VrlProcessor::from_yaml(vrl_processor_yaml, false);
470        assert!(vrl.is_ok());
471        let vrl = vrl.unwrap();
472
473        assert_eq!(
474            vrl.source,
475            ".name.a = .user_info.name\n.name.b = .user_info.name\ndel(.user_info)\n.timestamp = now()\n.\n"
476        );
477    }
478
479    #[test]
480    fn test_regex() {
481        let source = r#"
482.re = r'(?i)^Hello, World!$'
483del(.re)
484.re = r'(?i)^Hello, World!$'
485.
486"#;
487
488        let v = VrlProcessor::new(source.to_string(), false);
489        assert!(v.is_err());
490    }
491}
pipeline/etl/processor/vrl_processor.rs

pipeline/etl/processor/
vrl_processor.rs
