diff --git a/Cargo.lock b/Cargo.lock index cbd87363918..5bb44bf45b2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10933,6 +10933,8 @@ dependencies = [ "snafu 0.8.6", "sql", "table", + "temp-env", + "tempfile", "tokio", "urlencoding", "vrl", diff --git a/config/config.md b/config/config.md index dbd72b5fb1c..4e8b844da53 100644 --- a/config/config.md +++ b/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,
network, and logs. Default is false. Enable only for trusted pipeline authors.
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`,
`alter_database`, `drop_database`, `create_flow`, `drop_flow`,
`create_table`, `create_logical_tables`, `alter_table`, `alter_logical_tables`,
`drop_table`, `undrop_table`, `purge_dropped_table`, `truncate_table`,
`create_view`, `drop_view`, `admin_function`, `reconcile_catalog`,
`reconcile_database`, `reconcile_table`, `reconcile_logical_tables`.
When omitted, all current and future event types are recorded.
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.
When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable
is set to "prof:true,prof_active:false". The official image adds this env variable.
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
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,
network, and logs. Default is false. Enable only for trusted pipeline authors.
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`.
When omitted, all current and future event types are recorded.
Set to an empty array to disable event recording. | diff --git a/config/frontend.example.toml b/config/frontend.example.toml index 0e7c0482fa3..68d25396b35 100644 --- a/config/frontend.example.toml +++ b/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] diff --git a/config/standalone.example.toml b/config/standalone.example.toml index cebbf8ac265..99c9e28e15d 100644 --- a/config/standalone.example.toml +++ b/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] diff --git a/src/frontend/src/frontend.rs b/src/frontend/src/frontend.rs index 4608f70c093..8ee10903710 100644 --- a/src/frontend/src/frontend.rs +++ b/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( diff --git a/src/frontend/src/instance/log_handler.rs b/src/frontend/src/instance/log_handler.rs index 44738b9dcc1..7b6a8fcb8ed 100644 --- a/src/frontend/src/instance/log_handler.rs +++ b/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 { - PipelineOperator::build_pipeline(pipeline).context(PipelineSnafu) + self.pipeline_operator + .build_pipeline(pipeline) + .context(PipelineSnafu) } async fn get_pipeline_str( diff --git a/src/pipeline/Cargo.toml b/src/pipeline/Cargo.toml index 5a9cb3e9aa6..38a5a79996d 100644 --- a/src/pipeline/Cargo.toml +++ b/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" diff --git a/src/pipeline/src/etl.rs b/src/pipeline/src/etl.rs index f78a515fb59..ba68fde79ee 100644 --- a/src/pipeline/src/etl.rs +++ b/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 { + 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 { match input { Content::Yaml(str) => { let docs = YamlLoader::load_from_str(str).context(YamlLoadSnafu)?; @@ -73,7 +78,7 @@ pub fn parse(input: &Content) -> Result { 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() }; diff --git a/src/pipeline/src/etl/README.md b/src/pipeline/src/etl/README.md index e512ea31f7b..020bcdc304d 100644 --- a/src/pipeline/src/etl/README.md +++ b/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 diff --git a/src/pipeline/src/etl/processor.rs b/src/pipeline/src/etl/processor.rs index 761bcd6cba4..8c999cd13ee 100644 --- a/src/pipeline/src/etl/processor.rs +++ b/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> for Processors { - type Error = Error; - - fn try_from(vec: &Vec) -> Result { +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 { 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> for Processors { } } -fn parse_processor(doc: &yaml_rust::Yaml) -> Result { +fn parse_processor(doc: &yaml_rust::Yaml, allow_vrl_host_access: bool) -> Result { 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 { 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(), diff --git a/src/pipeline/src/etl/processor/vrl_processor.rs b/src/pipeline/src/etl/processor/vrl_processor.rs index ee3452523d7..d950834a2a8 100644 --- a/src/pipeline/src/etl/processor/vrl_processor.rs +++ b/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 { - let fns = vrl::stdlib::all(); + pub fn new(source: String, allow_host_access: bool) -> Result { + 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 { +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 { 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> { + 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 + } + _ => function, + }) + .collect() +} + +/// Retains data-only uses of functions that also accept host-access arguments. +#[derive(Debug)] +struct RestrictedFunction(Box); + +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()); } } diff --git a/src/pipeline/src/lib.rs b/src/pipeline/src/lib.rs index 15954b18f98..086758fa9cc 100644 --- a/src/pipeline/src/lib.rs +++ b/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, diff --git a/src/pipeline/src/manager/pipeline_operator.rs b/src/pipeline/src/manager/pipeline_operator.rs index c9731b36f20..8c6f6c7e2f9 100644 --- a/src/pipeline/src/manager/pipeline_operator.rs +++ b/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>, - 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 { - PipelineTable::compile_pipeline(pipeline) + pub fn build_pipeline(&self, pipeline: &str) -> Result { + crate::parse_with_options(&crate::Content::Yaml(pipeline), &self.options) } } diff --git a/src/pipeline/src/manager/table.rs b/src/pipeline/src/manager/table.rs index 0b81493b14c..ab675b6cea5 100644 --- a/src/pipeline/src/manager/table.rs +++ b/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 { + pub fn compile_pipeline(&self, pipeline: &str) -> Result { 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 { - 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) diff --git a/src/pipeline/src/options.rs b/src/pipeline/src/options.rs index 6e34f0ce933..ad1c72a430b 100644 --- a/src/pipeline/src/options.rs +++ b/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, } } } diff --git a/src/standalone/src/options.rs b/src/standalone/src/options.rs index af72cabef09..ea0b85e3efb 100644 --- a/src/standalone/src/options.rs +++ b/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(); diff --git a/tests-integration/Cargo.toml b/tests-integration/Cargo.toml index 2c90f4ee2ae..79efd9ed3a0 100644 --- a/tests-integration/Cargo.toml +++ b/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 diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index d0354477fac..368f1584c40 100644 --- a/tests-integration/src/standalone.rs +++ b/tests-integration/src/standalone.rs @@ -92,6 +92,7 @@ pub struct GreptimeDbStandaloneBuilder { logical_batcher: Option, table_batcher: BatcherOptions, mito_config: Option, + 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(), diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 8b608fe7694..2fa92a4e608 100644 --- a/tests-integration/tests/http.rs +++ b/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) =