fix!: backport #9617 to release/v1.3 (#9619)

fix(pipeline)!: disable VRL host access by default (#9610)

fix(pipeline): disable VRL host access by default

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

Cargo.lock

@@ -10933,6 +10933,8 @@ dependencies = [
  "snafu 0.8.6",
  "sql",
  "table",
+ "temp-env",
+ "tempfile",
  "tokio",
  "urlencoding",
  "vrl",

config/config.md

@@ -266,6 +266,7 @@
 | `tracing.tokio_console_addr` | String | Unset | The tokio console address. |
 | `pipeline` | -- | -- | The pipeline options. |
 | `pipeline.cache_ttl` | String | `10s` | Time to live of the local pipeline cache. Default is `10s`. |
+| `pipeline.allow_vrl_host_access` | Bool | `false` | Allow VRL to access the server environment, filesystem, hostname, local timezone,<br/>network, and logs. Default is false. Enable only for trusted pipeline authors.<br/>Applies to pipeline creation, execution, and dry-runs. Requires a restart. |
 | `event_recorder` | -- | -- | Configuration options for the event recorder. |
 | `event_recorder.ttl` | String | `90d` | TTL for the events table that will be used to store the events. Default is `90d`. |
 | `event_recorder.event_types` | Array | All | Event types to record. Current available event types: `create_database`,<br/>`alter_database`, `drop_database`, `create_flow`, `drop_flow`,<br/>`create_table`, `create_logical_tables`, `alter_table`, `alter_logical_tables`,<br/>`drop_table`, `undrop_table`, `purge_dropped_table`, `truncate_table`,<br/>`create_view`, `drop_view`, `admin_function`, `reconcile_catalog`,<br/>`reconcile_database`, `reconcile_table`, `reconcile_logical_tables`.<br/>When omitted, all current and future event types are recorded.<br/>Set to an empty array to disable event recording. |
@@ -425,6 +426,7 @@
 | `memory.enable_heap_profiling` | Bool | `true` | Whether to enable heap profiling activation during startup.<br/>When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable<br/>is set to "prof:true,prof_active:false". The official image adds this env variable.<br/>Default is true. |
 | `pipeline` | -- | -- | The pipeline options. |
 | `pipeline.cache_ttl` | String | `10s` | Time to live of the frontend-local pipeline cache. A pipeline created or deleted on<br/>another frontend takes effect on this one after at most this duration. |
+| `pipeline.allow_vrl_host_access` | Bool | `false` | Allow VRL to access the server environment, filesystem, hostname, local timezone,<br/>network, and logs. Default is false. Enable only for trusted pipeline authors.<br/>Applies to pipeline creation, execution, and dry-runs. Requires a restart. |
 | `event_recorder` | -- | -- | Configuration options for the event recorder. |
 | `event_recorder.ttl` | String | `90d` | TTL for the events table that will be used to store the events. Default is `90d`. |
 | `event_recorder.event_types` | Array | All | Event types to record. Current available event type: `admin_function`.<br/>When omitted, all current and future event types are recorded.<br/>Set to an empty array to disable event recording. |

config/frontend.example.toml

@@ -504,6 +504,10 @@ enable_heap_profiling = true
 ## Time to live of the frontend-local pipeline cache. A pipeline created or deleted on
 ## another frontend takes effect on this one after at most this duration.
 cache_ttl = "10s"
+## Allow VRL to access the server environment, filesystem, hostname, local timezone,
+## network, and logs. Default is false. Enable only for trusted pipeline authors.
+## Applies to pipeline creation, execution, and dry-runs. Requires a restart.
+allow_vrl_host_access = false
 
 ## Configuration options for the event recorder.
 [event_recorder]

config/standalone.example.toml

@@ -1087,6 +1087,10 @@ ttl = "90d"
 [pipeline]
 ## Time to live of the local pipeline cache. Default is `10s`.
 cache_ttl = "10s"
+## Allow VRL to access the server environment, filesystem, hostname, local timezone,
+## network, and logs. Default is false. Enable only for trusted pipeline authors.
+## Applies to pipeline creation, execution, and dry-runs. Requires a restart.
+allow_vrl_host_access = false
 
 ## Configuration options for the event recorder.
 [event_recorder]

src/frontend/src/frontend.rs

@@ -297,6 +297,18 @@ backoff_multiplier = 3.0
         );
     }
 
+    #[test]
+    fn test_pipeline_host_access_config() {
+        let defaults: FrontendOptions = toml::from_str("").unwrap();
+        assert!(!defaults.pipeline.allow_vrl_host_access);
+        let options: FrontendOptions =
+            toml::from_str("[pipeline]\nallow_vrl_host_access = true").unwrap();
+        assert!(options.pipeline.allow_vrl_host_access);
+        let serialized = toml::to_string(&options).unwrap();
+        let restored: FrontendOptions = toml::from_str(&serialized).unwrap();
+        assert_eq!(options.pipeline, restored.pipeline);
+    }
+
     #[test]
     fn test_logical_batcher_protocols_from_env() {
         temp_env::with_vars(

src/frontend/src/instance/log_handler.rs

@@ -23,7 +23,6 @@ use auth::{
 use client::Output;
 use common_error::ext::BoxedError;
 use datatypes::timestamp::TimestampNanosecond;
-use pipeline::pipeline_operator::PipelineOperator;
 use pipeline::{Pipeline, PipelineInfo, PipelineVersion};
 use servers::error::{
     AuthSnafu, Error as ServerError, ExecuteGrpcRequestSnafu, PipelineSnafu, Result as ServerResult,
@@ -148,7 +147,9 @@ impl PipelineHandler for Instance {
     }
 
     fn build_pipeline(&self, pipeline: &str) -> ServerResult<Pipeline> {
-        PipelineOperator::build_pipeline(pipeline).context(PipelineSnafu)
+        self.pipeline_operator
+            .build_pipeline(pipeline)
+            .context(PipelineSnafu)
     }
 
     async fn get_pipeline_str(

src/pipeline/Cargo.toml

@@ -60,6 +60,8 @@ yaml-rust = "0.4"
 [dev-dependencies]
 catalog = { workspace = true, features = ["testing"] }
 criterion = { workspace = true, features = ["html_reports"] }
+temp-env.workspace = true
+tempfile.workspace = true
 
 [[bench]]
 name = "processor"

src/pipeline/src/etl.rs

@@ -42,8 +42,8 @@ use crate::etl::transform::transformer::greptime::{
 };
 use crate::tablesuffix::TableSuffixTemplate;
 use crate::{
-    ContextOpt, GreptimeTransformer, IdentityTimeIndex, PipelineContext, SchemaInfo,
-    unwrap_or_continue_if_err,
+    ContextOpt, GreptimeTransformer, IdentityTimeIndex, PipelineContext, PipelineOptions,
+    SchemaInfo, unwrap_or_continue_if_err,
 };
 
 const DESCRIPTION: &str = "description";
@@ -60,6 +60,11 @@ pub enum Content<'a> {
 }
 
 pub fn parse(input: &Content) -> Result<Pipeline> {
+    parse_with_options(input, &PipelineOptions::default())
+}
+
+/// Parses a pipeline using the server's compilation policy.
+pub fn parse_with_options(input: &Content, options: &PipelineOptions) -> Result<Pipeline> {
     match input {
         Content::Yaml(str) => {
             let docs = YamlLoader::load_from_str(str).context(YamlLoadSnafu)?;
@@ -73,7 +78,7 @@ pub fn parse(input: &Content) -> Result<Pipeline> {
             let doc_version = (&doc[DOC_VERSION]).try_into()?;
 
             let processors = if let Some(v) = doc[PROCESSORS].as_vec() {
-                v.try_into()?
+                Processors::from_yaml(v, options.allow_vrl_host_access)?
             } else {
                 Processors::default()
             };

src/pipeline/src/etl/README.md

@@ -2,6 +2,27 @@
 
 ETL capability
 
+## VRL host access
+
+VRL host access is disabled by default. The server configuration option
+`[pipeline].allow_vrl_host_access = true` restores the full VRL standard library
+for trusted pipeline authors. It applies to both stored pipelines and dry-runs,
+requires a server restart, and cannot be overridden in a pipeline definition.
+
+With host access disabled, environment, hostname, system timezone, filesystem,
+network, and logging functions are unavailable. This includes `get_env_var`,
+`get_hostname`, `get_timezone_name`, `http_request`, `dns_lookup`, `reverse_dns`,
+`parse_proto`, `encode_proto`, `parse_etld`, `validate_json_schema`, and `log`.
+Grok parsing supports inline
+aliases but rejects nonempty `alias_sources`. Timestamp parsing and formatting
+use UTC by default; an explicit timezone must be a constant named timezone such
+as `Asia/Shanghai`, rather than `local`, an empty string, or an event-derived
+value. Ordinary time and random-value functions remain available.
+
+Existing pipelines using restricted functions or arguments fail compilation
+under the default policy, including when loaded from storage. This option
+restricts VRL capabilities; it does not impose CPU or memory limits.
+
 ## processors
 
 refer [elastic ingest processor][elastic-ingest-processor] for detail

src/pipeline/src/etl/processor.rs

@@ -51,7 +51,7 @@ use urlencoding::UrlEncodingProcessor;
 use vrl::value::Value as VrlValue;
 
 use crate::error::{
-    Error, FailedParseFieldFromStringSnafu, FieldMustBeTypeSnafu, InvalidFieldRenameSnafu,
+    FailedParseFieldFromStringSnafu, FieldMustBeTypeSnafu, InvalidFieldRenameSnafu,
     ProcessorKeyMustBeStringSnafu, ProcessorMustBeMapSnafu, ProcessorMustHaveStringKeySnafu,
     Result, UnsupportedProcessorSnafu,
 };
@@ -173,13 +173,12 @@ impl std::ops::DerefMut for Processors {
     }
 }
 
-impl TryFrom<&Vec<yaml_rust::Yaml>> for Processors {
-    type Error = Error;
-
-    fn try_from(vec: &Vec<yaml_rust::Yaml>) -> Result<Self> {
+impl Processors {
+    /// Parses processors using the server's VRL host access policy.
+    pub fn from_yaml(vec: &[yaml_rust::Yaml], allow_vrl_host_access: bool) -> Result<Self> {
         let mut processors_builders = vec![];
         for doc in vec {
-            let processor = parse_processor(doc)?;
+            let processor = parse_processor(doc, allow_vrl_host_access)?;
             processors_builders.push(processor);
         }
         Ok(Processors {
@@ -188,7 +187,7 @@ impl TryFrom<&Vec<yaml_rust::Yaml>> for Processors {
     }
 }
 
-fn parse_processor(doc: &yaml_rust::Yaml) -> Result<ProcessorKind> {
+fn parse_processor(doc: &yaml_rust::Yaml, allow_vrl_host_access: bool) -> Result<ProcessorKind> {
     let map = doc.as_hash().context(ProcessorMustBeMapSnafu)?;
 
     let key = map.keys().next().context(ProcessorMustHaveStringKeySnafu)?;
@@ -227,7 +226,9 @@ fn parse_processor(doc: &yaml_rust::Yaml) -> Result<ProcessorKind> {
         json_parse::PROCESSOR_JSON_PARSE => {
             ProcessorKind::JsonParse(JsonParseProcessor::try_from(value)?)
         }
-        vrl_processor::PROCESSOR_VRL => ProcessorKind::Vrl(VrlProcessor::try_from(value)?),
+        vrl_processor::PROCESSOR_VRL => {
+            ProcessorKind::Vrl(VrlProcessor::from_yaml(value, allow_vrl_host_access)?)
+        }
         select::PROCESSOR_SELECT => ProcessorKind::Select(SelectProcessor::try_from(value)?),
         filter::PROCESSOR_FILTER => ProcessorKind::Filter(FilterProcessor::try_from(value)?),
         _ => return UnsupportedProcessorSnafu { processor: str_key }.fail(),

src/pipeline/src/etl/processor/vrl_processor.rs

@@ -21,11 +21,14 @@ use snafu::{OptionExt, ensure};
 use vrl::compiler::runtime::Runtime;
 use vrl::compiler::{Program, TargetValue, compile};
 use vrl::diagnostic::Formatter;
-use vrl::prelude::TimeZone;
+use vrl::prelude::{
+    ArgumentList, Compiled, Example, Function, FunctionCompileContext, Parameter, TimeZone,
+    TypeState, function,
+};
 use vrl::value::{Kind, Secrets, Value as VrlValue};
 
 use crate::error::{
-    CompileVrlSnafu, Error, ExecuteVrlSnafu, KeyMustBeStringSnafu, Result, VrlRegexValueSnafu,
+    CompileVrlSnafu, ExecuteVrlSnafu, KeyMustBeStringSnafu, Result, VrlRegexValueSnafu,
     VrlReturnValueSnafu,
 };
 use crate::etl::processor::yaml_string;
@@ -46,8 +49,8 @@ pub struct VrlProcessor {
 }
 
 impl VrlProcessor {
-    pub fn new(source: String) -> Result<Self> {
-        let fns = vrl::stdlib::all();
+    pub fn new(source: String, allow_host_access: bool) -> Result<Self> {
+        let fns = functions(allow_host_access);
 
         let compile_result = compile(&source, &fns).map_err(|e| {
             CompileVrlSnafu {
@@ -101,10 +104,9 @@ impl VrlProcessor {
     }
 }
 
-impl TryFrom<&yaml_rust::yaml::Hash> for VrlProcessor {
-    type Error = Error;
-
-    fn try_from(value: &yaml_rust::yaml::Hash) -> Result<Self> {
+impl VrlProcessor {
+    /// Parses VRL source using the server's host access policy.
+    pub fn from_yaml(value: &yaml_rust::yaml::Hash, allow_host_access: bool) -> Result<Self> {
         let mut source = String::new();
         for (k, v) in value.iter() {
             let key = k
@@ -114,8 +116,111 @@ impl TryFrom<&yaml_rust::yaml::Hash> for VrlProcessor {
                 source = yaml_string(v, SOURCE)?;
             }
         }
-        let processor = VrlProcessor::new(source)?;
-        Ok(processor)
+        Self::new(source, allow_host_access)
+    }
+}
+
+fn functions(allow_host_access: bool) -> Vec<Box<dyn Function>> {
+    let mut functions = vrl::stdlib::all();
+    if allow_host_access {
+        return functions;
+    }
+
+    // These functions bypass VRL's event target, including file reads during compilation.
+    // When upgrading VRL, re-audit this denylist and the argument restrictions below;
+    // new host-access functions are otherwise enabled by default.
+    functions.retain(|function| {
+        !matches!(
+            function.identifier(),
+            "get_env_var"
+                | "get_hostname"
+                | "get_timezone_name"
+                | "http_request"
+                | "dns_lookup"
+                | "reverse_dns"
+                | "encode_proto"
+                | "parse_proto"
+                | "parse_etld"
+                | "validate_json_schema"
+                | "log"
+        )
+    });
+    functions
+        .into_iter()
+        .map(|function| match function.identifier() {
+            "parse_groks" | "parse_timestamp" | "format_timestamp" => {
+                Box::new(RestrictedFunction(function)) as Box<dyn Function>
+            }
+            _ => function,
+        })
+        .collect()
+}
+
+/// Retains data-only uses of functions that also accept host-access arguments.
+#[derive(Debug)]
+struct RestrictedFunction(Box<dyn Function>);
+
+impl Function for RestrictedFunction {
+    fn identifier(&self) -> &'static str {
+        self.0.identifier()
+    }
+
+    fn usage(&self) -> &'static str {
+        self.0.usage()
+    }
+
+    fn category(&self) -> &'static str {
+        self.0.category()
+    }
+
+    fn return_kind(&self) -> u16 {
+        self.0.return_kind()
+    }
+
+    fn examples(&self) -> &'static [Example] {
+        self.0.examples()
+    }
+
+    fn parameters(&self) -> &'static [Parameter] {
+        self.0.parameters()
+    }
+
+    fn pure(&self) -> bool {
+        self.0.pure()
+    }
+
+    fn compile(
+        &self,
+        state: &TypeState,
+        ctx: &mut FunctionCompileContext,
+        arguments: ArgumentList,
+    ) -> Compiled {
+        if self.identifier() == "parse_groks" {
+            if arguments
+                .optional_array("alias_sources")?
+                .is_some_and(|sources| !sources.is_empty())
+            {
+                return Err(function::Error::InvalidArgument {
+                    keyword: "alias_sources",
+                    value: "alias_sources".into(),
+                    error: "file imports require pipeline.allow_vrl_host_access",
+                }
+                .into());
+            }
+        } else if let Some(timezone) = arguments.optional_literal("timezone", state)? {
+            // A static named timezone cannot resolve to the host's local timezone at runtime.
+            if !matches!(&timezone, VrlValue::Bytes(bytes)
+                if matches!(TimeZone::parse(&String::from_utf8_lossy(bytes)), Some(TimeZone::Named(_))))
+            {
+                return Err(function::Error::InvalidArgument {
+                    keyword: "timezone",
+                    value: timezone,
+                    error: "a named timezone is required when pipeline.allow_vrl_host_access is false",
+                }
+                .into());
+            }
+        }
+        self.0.compile(state, ctx, arguments)
     }
 }
 
@@ -163,6 +268,141 @@ mod tests {
 
     use super::*;
 
+    #[test]
+    fn test_host_functions_require_opt_in() {
+        let calls = [
+            ("get_env_var", r#"get_env_var!("GREPTIME_VRL_TEST_SECRET")"#),
+            ("get_hostname", "get_hostname!()"),
+            ("get_timezone_name", "get_timezone_name!()"),
+            ("http_request", r#"http_request!("http://127.0.0.1:9/")"#),
+            ("dns_lookup", r#"dns_lookup!("localhost")"#),
+            ("reverse_dns", r#"reverse_dns!("127.0.0.1")"#),
+            (
+                "parse_proto",
+                r#"parse_proto!("", "missing.desc", "Message")"#,
+            ),
+            (
+                "encode_proto",
+                r#"encode_proto!({}, "missing.desc", "Message")"#,
+            ),
+            ("parse_etld", r#"parse_etld!("example.com")"#),
+            (
+                "validate_json_schema",
+                r#"validate_json_schema!("{}", "missing.json")"#,
+            ),
+            ("log", r#"log("host log write")"#),
+        ];
+        let unrestricted = functions(true);
+        for (name, call) in calls {
+            assert!(
+                unrestricted
+                    .iter()
+                    .any(|function| function.identifier() == name)
+            );
+            let source = format!(".result = {call}\n.");
+            let error = VrlProcessor::new(source, false).unwrap_err().to_string();
+            assert!(error.contains("undefined function"), "{name}: {error}");
+        }
+
+        temp_env::with_var("GREPTIME_VRL_TEST_SECRET", Some("decoy"), || {
+            let source = ".secret = get_env_var!(\"GREPTIME_VRL_TEST_SECRET\")\n.";
+            let processor = VrlProcessor::new(source.to_string(), true).unwrap();
+            let result = processor
+                .resolve(VrlValue::Object(BTreeMap::new()))
+                .unwrap();
+            assert_eq!(
+                result.as_object().unwrap()["secret"],
+                VrlValue::from("decoy")
+            );
+            assert!(VrlProcessor::new(source.to_string(), false).is_err());
+        });
+    }
+
+    #[test]
+    fn test_grok_file_import_requires_opt_in() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("aliases.json");
+        std::fs::write(&path, r#"{"custom_word":"%{word}"}"#).unwrap();
+        let path = serde_json::to_string(&path).unwrap();
+        let source = format!(
+            r#".parsed = parse_groks!("hello", patterns: ["%{{custom_word:message}}"], alias_sources: [{path}])
+."#
+        );
+        let error = VrlProcessor::new(source.clone(), false)
+            .unwrap_err()
+            .to_string();
+        assert!(error.contains("file imports require pipeline.allow_vrl_host_access"));
+        let result = VrlProcessor::new(source, true)
+            .unwrap()
+            .resolve(VrlValue::Object(BTreeMap::new()))
+            .unwrap();
+        assert_eq!(
+            result.as_object().unwrap()["parsed"].as_object().unwrap()["message"],
+            VrlValue::from("hello")
+        );
+
+        let source = r#".parsed = parse_groks!("hello", patterns: ["%{custom_word:message}"], aliases: {"custom_word": "%{word}"}, alias_sources: [])
+."#;
+        assert!(VrlProcessor::new(source.to_string(), false).is_ok());
+    }
+
+    #[test]
+    fn test_local_and_dynamic_timezones_require_opt_in() {
+        for timezone in [r#""local""#, r#""""#, "string!(.timezone)"] {
+            for call in [
+                format!(
+                    r#"format_timestamp!(t'2020-01-01T00:00:00Z', format: "%z", timezone: {timezone})"#
+                ),
+                format!(
+                    r#"parse_timestamp!("2020-01-01", format: "%Y-%m-%d", timezone: {timezone})"#
+                ),
+            ] {
+                let source = format!(".result = {call}\n.");
+                assert!(
+                    VrlProcessor::new(source.clone(), false).is_err(),
+                    "{source}"
+                );
+                assert!(VrlProcessor::new(source.clone(), true).is_ok(), "{source}");
+            }
+        }
+
+        let source = r#".parsed = parse_timestamp!("2020-01-01 00:00:00", format: "%Y-%m-%d %H:%M:%S")
+.utc = format_timestamp!(.parsed, format: "%z")
+.named = format_timestamp!(.parsed, format: "%z", timezone: "Asia/Shanghai")
+."#;
+        let result = VrlProcessor::new(source.to_string(), false)
+            .unwrap()
+            .resolve(VrlValue::Object(BTreeMap::new()))
+            .unwrap();
+        assert_eq!(result.as_object().unwrap()["utc"], VrlValue::from("+0000"));
+        assert_eq!(
+            result.as_object().unwrap()["named"],
+            VrlValue::from("+0800")
+        );
+    }
+
+    #[test]
+    fn test_pipeline_host_policy_is_server_controlled() {
+        let yaml = r#"
+allow_vrl_host_access: true
+processors:
+  - vrl:
+      allow_vrl_host_access: true
+      source: |
+        .secret = get_env_var!("GREPTIME_VRL_TEST_SECRET")
+        .
+transform:
+  - field: secret
+    type: string
+"#;
+        assert!(crate::parse(&crate::Content::Yaml(yaml)).is_err());
+        let options = crate::PipelineOptions {
+            allow_vrl_host_access: true,
+            ..Default::default()
+        };
+        assert!(crate::parse_with_options(&crate::Content::Yaml(yaml), &options).is_ok());
+    }
+
     #[test]
     fn test_vrl() {
         let source = r#"
@@ -173,7 +413,7 @@ del(.user_info)
 .
 "#;
 
-        let v = VrlProcessor::new(source.to_string());
+        let v = VrlProcessor::new(source.to_string(), false);
         assert!(v.is_ok());
         let v = v.unwrap();
 
@@ -226,7 +466,7 @@ processors:
             .and_then(|x| x.as_hash())
             .unwrap();
 
-        let vrl = VrlProcessor::try_from(vrl_processor_yaml);
+        let vrl = VrlProcessor::from_yaml(vrl_processor_yaml, false);
         assert!(vrl.is_ok());
         let vrl = vrl.unwrap();
 
@@ -245,7 +485,7 @@ del(.re)
 .
 "#;
 
-        let v = VrlProcessor::new(source.to_string());
+        let v = VrlProcessor::new(source.to_string(), false);
         assert!(v.is_err());
     }
 }

src/pipeline/src/lib.rs

@@ -27,7 +27,7 @@ pub use etl::transform::transformer::greptime::{GreptimePipelineParams, SchemaIn
 pub use etl::transform::transformer::identity_pipeline;
 pub use etl::{
     Content, DispatchedTo, Pipeline, PipelineExecOutput, PipelineProcessOutput, TransformedOutput,
-    TransformerMode, parse,
+    TransformerMode, parse, parse_with_options,
 };
 pub use manager::{
     GREPTIME_INTERNAL_IDENTITY_PIPELINE_NAME, GREPTIME_INTERNAL_TRACE_PIPELINE_V1_NAME,

src/pipeline/src/manager/pipeline_operator.rs

@@ -14,7 +14,7 @@
 
 use std::collections::HashMap;
 use std::sync::{Arc, RwLock};
-use std::time::{Duration, Instant};
+use std::time::Instant;
 
 use api::v1::CreateTableExpr;
 use catalog::{CatalogManagerRef, RegisterSystemTableRequest};
@@ -56,7 +56,7 @@ pub struct PipelineOperator {
     catalog_manager: CatalogManagerRef,
     query_engine: QueryEngineRef,
     tables: RwLock<HashMap<String, PipelineTableRef>>,
-    cache_ttl: Duration,
+    options: PipelineOptions,
 }
 
 impl PipelineOperator {
@@ -99,7 +99,7 @@ impl PipelineOperator {
                 self.statement_executor.clone(),
                 table,
                 self.query_engine.clone(),
-                self.cache_ttl,
+                &self.options,
             )),
         );
     }
@@ -183,7 +183,7 @@ impl PipelineOperator {
             catalog_manager,
             tables: RwLock::new(HashMap::new()),
             query_engine,
-            cache_ttl: options.cache_ttl,
+            options: options.clone(),
         }
     }
 
@@ -281,7 +281,7 @@ impl PipelineOperator {
     }
 
     /// Compile a pipeline.
-    pub fn build_pipeline(pipeline: &str) -> Result<Pipeline> {
-        PipelineTable::compile_pipeline(pipeline)
+    pub fn build_pipeline(&self, pipeline: &str) -> Result<Pipeline> {
+        crate::parse_with_options(&crate::Content::Yaml(pipeline), &self.options)
     }
 }

src/pipeline/src/manager/table.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 use std::sync::Arc;
-use std::time::Duration;
 
 use api::v1::value::ValueData;
 use api::v1::{
@@ -46,10 +45,11 @@ use crate::error::{
     ExecuteInternalStatementSnafu, InsertPipelineSnafu, InvalidPipelineVersionSnafu,
     MultiPipelineWithDiffSchemaSnafu, PipelineNotFoundSnafu, RecordBatchLenNotMatchSnafu, Result,
 };
-use crate::etl::{Content, Pipeline, parse};
+use crate::etl::{Content, Pipeline, parse_with_options};
 use crate::manager::pipeline_cache::{PipelineCache, PipelineContent};
 use crate::manager::{PipelineInfo, PipelineVersion};
 use crate::metrics::METRIC_PIPELINE_TABLE_FIND_COUNT;
+use crate::options::PipelineOptions;
 use crate::util::prepare_dataframe_conditions;
 
 pub(crate) const PIPELINE_TABLE_NAME: &str = "pipelines";
@@ -68,6 +68,7 @@ pub struct PipelineTable {
     table: TableRef,
     query_engine: QueryEngineRef,
     cache: PipelineCache,
+    options: PipelineOptions,
 }
 
 impl PipelineTable {
@@ -77,14 +78,15 @@ impl PipelineTable {
         statement_executor: StatementExecutorRef,
         table: TableRef,
         query_engine: QueryEngineRef,
-        cache_ttl: Duration,
+        options: &PipelineOptions,
     ) -> Self {
         Self {
             inserter,
             statement_executor,
             table,
             query_engine,
-            cache: PipelineCache::new(cache_ttl),
+            cache: PipelineCache::new(options.cache_ttl),
+            options: options.clone(),
         }
     }
 
@@ -198,9 +200,9 @@ impl PipelineTable {
     }
 
     /// Compile a pipeline from a string.
-    pub fn compile_pipeline(pipeline: &str) -> Result<Pipeline> {
+    pub fn compile_pipeline(&self, pipeline: &str) -> Result<Pipeline> {
         let yaml_content = Content::Yaml(pipeline);
-        parse(&yaml_content)
+        parse_with_options(&yaml_content, &self.options)
     }
 
     /// Insert a pipeline into the pipeline table.
@@ -267,7 +269,7 @@ impl PipelineTable {
         self.cache
             .get_pipeline_with(schema, name, input_version, async {
                 let pipeline_content = self.get_pipeline_str(schema, name, input_version).await?;
-                Ok(Arc::new(Self::compile_pipeline(&pipeline_content.content)?))
+                Ok(Arc::new(self.compile_pipeline(&pipeline_content.content)?))
             })
             .await
     }
@@ -373,7 +375,7 @@ impl PipelineTable {
         content_type: &str,
         pipeline: &str,
     ) -> Result<PipelineInfo> {
-        let compiled_pipeline = Arc::new(Self::compile_pipeline(pipeline)?);
+        let compiled_pipeline = Arc::new(self.compile_pipeline(pipeline)?);
         // we will use the version in the future
         let version = self
             .insert_pipeline_to_pipeline_table(name, content_type, pipeline)

src/pipeline/src/options.rs

@@ -24,12 +24,17 @@ pub struct PipelineOptions {
     /// duration. Default: "10s".
     #[serde(with = "humantime_serde")]
     pub cache_ttl: Duration,
+    /// Allows VRL programs to access the server environment, filesystem, hostname,
+    /// local timezone, network, and logs. Enable only for trusted pipeline authors.
+    /// Defaults to false. Changes require a restart.
+    pub allow_vrl_host_access: bool,
 }
 
 impl Default for PipelineOptions {
     fn default() -> Self {
         Self {
             cache_ttl: Duration::from_secs(10),
+            allow_vrl_host_access: false,
         }
     }
 }

src/standalone/src/options.rs

@@ -238,6 +238,19 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_pipeline_host_access_config_forwarding() {
+        let defaults: StandaloneOptions = toml::from_str("").unwrap();
+        assert!(!defaults.pipeline.allow_vrl_host_access);
+        assert!(!defaults.frontend_options().pipeline.allow_vrl_host_access);
+        let options: StandaloneOptions =
+            toml::from_str("[pipeline]\nallow_vrl_host_access = true").unwrap();
+        assert!(options.frontend_options().pipeline.allow_vrl_host_access);
+        let serialized = toml::to_string(&options).unwrap();
+        let restored: StandaloneOptions = toml::from_str(&serialized).unwrap();
+        assert_eq!(options.pipeline, restored.pipeline);
+    }
+
     #[test]
     fn test_logical_batcher_config_forwarding() {
         let opts: StandaloneOptions = toml::from_str("[pending_rows_batcher]\nprotocols = ['influxdb']\n[pending_rows_batcher.logical_table]\nprotocols = ['otlp', 'prom']\npending_rows_flush_interval = '10ms'").unwrap();

tests-integration/Cargo.toml

@@ -56,6 +56,7 @@ meta-srv = { workspace = true, features = ["mock"] }
 mito2.workspace = true
 object-store.workspace = true
 operator.workspace = true
+pipeline.workspace = true
 prost.workspace = true
 query.workspace = true
 rand.workspace = true
@@ -107,7 +108,6 @@ opentelemetry-proto.workspace = true
 operator = { workspace = true, features = ["testing"] }
 otel-arrow-rust.workspace = true
 paste.workspace = true
-pipeline.workspace = true
 plugins.workspace = true
 prost.workspace = true
 rand.workspace = true

tests-integration/src/standalone.rs

@@ -92,6 +92,7 @@ pub struct GreptimeDbStandaloneBuilder {
     logical_batcher: Option<BatcherOptions>,
     table_batcher: BatcherOptions,
     mito_config: Option<mito2::config::MitoConfig>,
+    pipeline_options: pipeline::PipelineOptions,
 }
 
 impl GreptimeDbStandaloneBuilder {
@@ -115,6 +116,7 @@ impl GreptimeDbStandaloneBuilder {
             logical_batcher: None,
             table_batcher: BatcherOptions::default(),
             mito_config: None,
+            pipeline_options: pipeline::PipelineOptions::default(),
         }
     }
 
@@ -132,6 +134,13 @@ impl GreptimeDbStandaloneBuilder {
         self
     }
 
+    /// Configures pipeline compilation and caching for this test instance.
+    #[must_use]
+    pub fn with_pipeline_options(mut self, options: pipeline::PipelineOptions) -> Self {
+        self.pipeline_options = options;
+        self
+    }
+
     /// Configures ordinary-table batching for protocol integration tests.
     #[must_use]
     pub fn with_table_batcher(mut self, options: BatcherOptions) -> Self {
@@ -405,6 +414,7 @@ impl GreptimeDbStandaloneBuilder {
             event_recorder: self.event_recorder_options.clone(),
             auto_create_table: self.auto_create_table,
             experimental_metric_export: self.experimental_metric_export,
+            pipeline: self.pipeline_options.clone(),
             pending_rows_batcher: PendingRowsBatcherOptions {
                 logical_table: self.logical_batcher.clone(),
                 table: self.table_batcher.clone(),

tests-integration/tests/http.rs

@@ -200,6 +200,7 @@ macro_rules! http_tests {
                 test_pipeline_suffix_template,
                 test_pipeline_context,
                 test_pipeline_with_vrl,
+                test_pipeline_vrl_host_access,
                 test_pipeline_with_hint_vrl,
                 test_pipeline_one_to_many_vrl,
                 test_pipeline_2,
@@ -3063,6 +3064,7 @@ enable_heap_profiling = true
 
 [pipeline]
 cache_ttl = "10s"
+allow_vrl_host_access = false
 
 [event_recorder]
 ttl = "2months 29days 2h 52m 48s"
@@ -6555,6 +6557,72 @@ table_suffix: _${type}
     guard.remove_all().await;
 }
 
+pub async fn test_pipeline_vrl_host_access(storage_type: StorageType) {
+    let pipeline = r#"
+processors:
+  - vrl:
+      source: |
+        .value = is_string(get_env_var("GREPTIME_VRL_HOST_ACCESS_TEST") ?? "missing")
+        .
+transform:
+  - field: value
+    type: boolean
+"#;
+
+    for allow_vrl_host_access in [false, true] {
+        let mut instance = tests_integration::standalone::GreptimeDbStandaloneBuilder::new(
+            "test_pipeline_vrl_host_access",
+        )
+        .with_default_store_type(storage_type)
+        .with_pipeline_options(pipeline::PipelineOptions {
+            // Force named dry-runs to reload and compile the stored definition.
+            cache_ttl: Duration::ZERO,
+            allow_vrl_host_access,
+        })
+        .build()
+        .await;
+        let server = HttpServerBuilder::new(HttpOptions::default())
+            .with_log_ingest_handler(instance.fe_instance().clone(), None, None)
+            .build();
+        let client = TestClient::new(server.build(server.make_app()).unwrap()).await;
+
+        let response = client
+            .post("/v1/pipelines/_dryrun")
+            .header("Content-Type", "application/json")
+            .body(json!({"pipeline": pipeline, "data": "[{}]"}).to_string())
+            .send()
+            .await;
+        let expected_status = if allow_vrl_host_access {
+            StatusCode::OK
+        } else {
+            StatusCode::BAD_REQUEST
+        };
+        assert_eq!(response.status(), expected_status);
+        if !allow_vrl_host_access {
+            assert!(response.text().await.contains("undefined function"));
+        }
+
+        let response = client
+            .post("/v1/pipelines/host_policy")
+            .header("Content-Type", "application/x-yaml")
+            .body(pipeline)
+            .send()
+            .await;
+        assert_eq!(response.status(), expected_status);
+
+        if allow_vrl_host_access {
+            let response = client
+                .post("/v1/pipelines/_dryrun")
+                .header("Content-Type", "application/json")
+                .body(json!({"pipeline_name": "host_policy", "data": "[{}]"}).to_string())
+                .send()
+                .await;
+            assert_eq!(response.status(), StatusCode::OK);
+        }
+        instance.guard.remove_all().await;
+    }
+}
+
 pub async fn test_pipeline_with_vrl(storage_type: StorageType) {
     common_telemetry::init_default_ut_logging();
     let (app, mut guard) =