rivet/config/mod.rs
1pub mod cursor;
2mod destination;
3mod export;
4mod format;
5mod lints;
6mod notifications;
7pub mod resolve;
8pub mod schema;
9mod source;
10
11pub use cursor::IncrementalCursorMode;
12pub use destination::*;
13pub use export::*;
14pub use format::*;
15pub use notifications::*;
16#[allow(unused_imports)]
17pub(crate) use resolve::resolve_env_vars;
18pub use resolve::{parse_file_size, resolve_vars};
19pub use schema::generate_config_schema_pretty;
20pub use source::*;
21
22use schemars::JsonSchema;
23use serde::Deserialize;
24
25/// Top-level Rivet configuration root.
26///
27/// Operators write this struct as YAML (typically `rivet.yaml`). The
28/// `JsonSchema` derive is the source of truth for the `schemas/rivet.schema.json`
29/// artifact and the `rivet schema config` command's output (v0.7.3 P0).
30#[derive(Debug, Deserialize, JsonSchema, Clone)]
31#[serde(deny_unknown_fields)]
32pub struct Config {
33 pub source: SourceConfig,
34 pub exports: Vec<ExportConfig>,
35 #[serde(default)]
36 pub notifications: Option<NotificationsConfig>,
37 #[serde(default)]
38 pub parallel_exports: bool,
39 #[serde(default)]
40 pub parallel_export_processes: bool,
41 /// Reserved for a downstream, first-party warehouse loader: the load target
42 /// lives here so ONE config can drive both the export and a
43 /// downstream load. Rivet stops at "file in a bucket" and **does not
44 /// interpret** this block — it is accepted and ignored, present only so
45 /// `rivet check` / `run` / `apply` don't reject a config that carries it.
46 // Deliberately never read by OSS — it is a reserved passthrough for a
47 // downstream loader; only serde (accept) and schemars (document) touch it.
48 #[allow(dead_code)]
49 #[serde(default)]
50 pub load: Option<serde_json::Value>,
51}
52
53impl Config {
54 pub fn load(path: &str) -> crate::error::Result<Self> {
55 Self::load_with_params(path, None)
56 }
57
58 pub fn load_with_params(
59 path: &str,
60 params: Option<&std::collections::HashMap<String, String>>,
61 ) -> crate::error::Result<Self> {
62 // F11 (0.7.5 audit): raw `std::io::Error` lost the path on
63 // not-found. Wrap with the file path + a hint so the operator
64 // can see *which* config the tool could not open.
65 let contents = std::fs::read_to_string(path).map_err(|e| {
66 if e.kind() == std::io::ErrorKind::NotFound {
67 anyhow::anyhow!(
68 "config file '{}' not found.\n Hint: check the path, or run `rivet init` to generate one.",
69 path
70 )
71 } else {
72 anyhow::anyhow!("cannot read config file '{}': {}", path, e)
73 }
74 })?;
75 // Warn about typo'd `--param` keys once per CLI invocation, using the
76 // un-resolved YAML as the haystack so the placeholders are still there.
77 // We pass the raw `contents` (not `resolved`) on purpose: after
78 // resolution the placeholders are gone, and every key would look unused.
79 resolve::warn_unused_params(&contents, params);
80 let resolved = resolve_vars(&contents, params)?;
81 // F12 (0.7.5 audit): YAML parse errors did not name the config
82 // file. When loading from disk we know the path — thread it
83 // into the parse error.
84 // `.context` (not `anyhow!("…{:#}", e)`, which stringifies into a fresh
85 // error) so a typed `CodedError` raised by validation survives the chain —
86 // `error::error_code` downcasts it for the `[CODE]` prefix. `{e:#}` in
87 // `main` still renders the full "config file '…': <message>" chain.
88 Self::from_yaml(&resolved).map_err(|e| e.context(format!("config file '{}'", path)))
89 }
90
91 pub fn from_yaml(yaml: &str) -> crate::error::Result<Self> {
92 // Intercept the unquoted-`{partition}` footgun before the raw-YAML
93 // pre-scans below (they `from_str::<Value>` too and would re-emit the
94 // same cryptic libyaml message). A document that does not even parse as
95 // a `Value` is malformed; if the failure looks like a flow-mapping
96 // scanner error *and* the source carries an unquoted brace value, point
97 // straight at the quoting fix. On a valid config this `Value` parse
98 // succeeds and the block is skipped, so the success path is unchanged.
99 if let Err(e) = serde_yaml_ng::from_str::<serde_yaml_ng::Value>(yaml) {
100 let m = e.to_string();
101 if let Some(hint) = Self::unquoted_template_brace_hint(yaml, &m) {
102 return Err(anyhow::anyhow!("{e}\n {hint}"));
103 }
104 // A TAB in indentation is the most common beginner YAML mistake; it
105 // trips libyaml before serde ever sees a field, so it must be caught
106 // here in the raw scan, not in `enhance_parse_error`.
107 if let Some(hint) = Self::tab_indent_hint(yaml, &m) {
108 return Err(anyhow::anyhow!("{e}\n {hint}"));
109 }
110 }
111 Self::check_misplaced_tuning_fields(yaml)?;
112 Self::check_csv_compression(yaml)?;
113 Self::check_tls_mode_downgrade(yaml)?;
114 let config: Config = serde_yaml_ng::from_str(yaml).map_err(|e| {
115 // A well-formed flow map (`prefix: {partition}`) parses as a YAML
116 // value but serde then rejects it with `invalid type: map, expected
117 // a string`. That is the same unquoted-brace footgun, surfacing one
118 // layer later than the scanner errors caught above — so try the same
119 // hint here before falling back to the generic field-typo enhancer.
120 if let Some(hint) = Self::unquoted_template_brace_hint(yaml, &e.to_string()) {
121 anyhow::anyhow!("{e}\n {hint}")
122 } else {
123 lints::enhance_parse_error(e)
124 }
125 })?;
126 config.validate()?;
127 Ok(config)
128 }
129
130 /// Detect the unquoted-`{partition}` (or `{date}`, …) template footgun and
131 /// return an actionable quoting hint, or `None` when the error is unrelated.
132 ///
133 /// A YAML value that *starts* a flow mapping — `prefix: {partition}` — is
134 /// the common copy-paste mistake: `{partition}` is the required token for
135 /// `partition_by`, but unquoted it parses as a YAML map (or, with trailing
136 /// text, trips the libyaml scanner). serde then emits a cryptic
137 /// `did not find expected ',' or '}'` / `while parsing a flow mapping` /
138 /// `invalid type: map, expected a string` with no hint that a pair of
139 /// quotes is the fix.
140 ///
141 /// Two guards keep this from firing on unrelated parse errors: the error
142 /// message must carry one of the flow-mapping symptoms, AND the raw source
143 /// must actually contain an unquoted `{…}` value. Both must hold, so a
144 /// valid config (every brace value quoted) never sees the hint, and a
145 /// genuine map-typed field error elsewhere is left alone.
146 /// A YAML document indented with a TAB trips libyaml with `found character
147 /// that cannot start any token`. Point straight at the fix (spaces, not
148 /// tabs) — but only when the cited line really begins with a tab, so an
149 /// unrelated scanner error is left with its original message.
150 fn tab_indent_hint(yaml: &str, err_msg: &str) -> Option<String> {
151 if !err_msg.contains("tab character") {
152 return None;
153 }
154 // serde_yaml_ng reports `found a tab character … at line N column C …`;
155 // the FIRST `line N` is where the offending tab is.
156 let line_no: usize = err_msg
157 .split_once("line ")
158 .and_then(|(_, rest)| rest.split([' ', ',']).next())
159 .and_then(|n| n.parse().ok())?;
160 let line = yaml.lines().nth(line_no.checked_sub(1)?)?;
161 let leading = &line[..line.len() - line.trim_start().len()];
162 leading.contains('\t').then(|| {
163 format!(
164 "line {line_no} is indented with a TAB — YAML requires spaces. Replace the tab(s) with spaces."
165 )
166 })
167 }
168
169 fn unquoted_template_brace_hint(yaml: &str, err_msg: &str) -> Option<String> {
170 const FLOW_SYMPTOMS: &[&str] = &[
171 "did not find expected ',' or '}'",
172 "while parsing a flow mapping",
173 // A bare `key: {token}` parses as a map, then serde rejects the
174 // map where it wanted a scalar — same root cause, later layer.
175 "invalid type: map, expected a string",
176 // `key: {token}/more` runs the flow map into block context.
177 "did not find expected key",
178 ];
179 if !FLOW_SYMPTOMS.iter().any(|s| err_msg.contains(s)) {
180 return None;
181 }
182 if !yaml.lines().any(line_has_unquoted_brace_value) {
183 return None;
184 }
185 Some(
186 "a YAML value containing { } (such as {partition} or {date}) must be quoted, \
187 e.g. prefix: \"exports/{partition}/\""
188 .to_string(),
189 )
190 }
191
192 /// Reject `format: csv` paired with an explicitly-requested compression
193 /// codec (Finding #10). The CSV writer has no compression encoder, so the
194 /// codec is silently dropped on write while the run manifest still records
195 /// it — a degraded, dishonest no-op. We reject loudly at config-validate
196 /// time so `rivet check` / `rivet doctor` catch it before any run.
197 ///
198 /// This is a raw-YAML scan (like [`Self::check_misplaced_tuning_fields`])
199 /// rather than a `validate_export` check on purpose: `ExportConfig.
200 /// compression` is `#[serde(default)]` and `CompressionType::default()` is
201 /// `Zstd`, so a parsed export cannot distinguish "user asked for zstd" from
202 /// "user omitted the field". Only a user who *wrote* `compression:`/
203 /// `compression_profile:` is asking for something the CSV writer cannot
204 /// honour; the bare-`format: csv` default writes uncompressed and is fine.
205 fn check_csv_compression(yaml: &str) -> crate::error::Result<()> {
206 let root: serde_yaml_ng::Value = serde_yaml_ng::from_str(yaml)?;
207 let Some(exports) = root.get("exports").and_then(|e| e.as_sequence()) else {
208 return Ok(());
209 };
210 for export in exports {
211 if export.get("format").and_then(|f| f.as_str()) != Some("csv") {
212 continue;
213 }
214 let name = export
215 .get("name")
216 .and_then(|n| n.as_str())
217 .unwrap_or("<unnamed>");
218
219 // Explicit `compression:` codec that the CSV writer cannot apply.
220 // An unrecognised label is left for serde to reject during the real
221 // parse; we only act on a codec we understand and that CSV cannot
222 // honour (everything but `none`).
223 if let Some(codec) = export.get("compression").and_then(|c| c.as_str())
224 && let Some(ct) = CompressionType::from_label(codec)
225 && !format::compression_supported(FormatType::Csv, ct)
226 {
227 anyhow::bail!(
228 "export '{}': CSV output does not support compression: {}. \
229 CSV has no compression encoder, so the codec would be silently dropped \
230 while the manifest records it.\n \
231 Hint: use `format: parquet` for compression, or set `compression: none`.",
232 name,
233 codec,
234 );
235 }
236
237 // A `compression_profile:` other than `none` resolves to a real
238 // codec too (fast→snappy, balanced/compact→zstd) — same no-op.
239 if let Some(profile) = export.get("compression_profile").and_then(|c| c.as_str())
240 && profile != CompressionProfile::None.label()
241 {
242 anyhow::bail!(
243 "export '{}': CSV output does not support compression_profile: {} \
244 (it resolves to a compression codec the CSV writer cannot apply).\n \
245 Hint: use `format: parquet` for compression, or set `compression_profile: none`.",
246 name,
247 profile,
248 );
249 }
250 }
251 Ok(())
252 }
253
254 /// V13: reject a `source.tls` block that pairs an *explicitly chosen*
255 /// enforced `mode:` with a verification-disabling danger knob
256 /// (`accept_invalid_certs` / `accept_invalid_hostnames`). `mode: verify-full`
257 /// promises chain + hostname verification, but the knob silently downgrades
258 /// it to "trust anything" — a MITM exposure that contradicts the stated
259 /// intent (see `src/source/tls.rs::build_native_tls`, whose comment claims
260 /// this is warned about at config-time but is not).
261 ///
262 /// Like [`Self::check_csv_compression`], this is a raw-YAML scan rather than
263 /// a `validate` check on purpose: `TlsMode` is `#[serde(default)]` and the
264 /// default is `VerifyFull`, so a parsed config cannot distinguish "user
265 /// wrote `mode: verify-full`" (a contradiction to flag) from "user omitted
266 /// `mode:`" (the common dev-container case `tls: { accept_invalid_certs:
267 /// true }` against a loopback self-signed cert — which must keep working).
268 /// Only an *explicit* enforced `mode:` next to a danger knob is the footgun.
269 fn check_tls_mode_downgrade(yaml: &str) -> crate::error::Result<()> {
270 let root: serde_yaml_ng::Value = serde_yaml_ng::from_str(yaml)?;
271 let Some(tls) = root.get("source").and_then(|s| s.get("tls")) else {
272 return Ok(());
273 };
274
275 // Only an explicitly written `mode:` is a deliberate, contradicted
276 // choice; an omitted mode is the dev-container default path.
277 let Some(mode) = tls.get("mode").and_then(|m| m.as_str()) else {
278 return Ok(());
279 };
280 // `disable` carries no verification promise to contradict; the danger
281 // knobs are a no-op there. Flag only the enforced modes.
282 if mode == "disable" {
283 return Ok(());
284 }
285
286 let knob = if tls
287 .get("accept_invalid_certs")
288 .and_then(|v| v.as_bool())
289 .unwrap_or(false)
290 {
291 Some("accept_invalid_certs")
292 } else if tls
293 .get("accept_invalid_hostnames")
294 .and_then(|v| v.as_bool())
295 .unwrap_or(false)
296 {
297 Some("accept_invalid_hostnames")
298 } else {
299 None
300 };
301
302 if let Some(knob) = knob {
303 anyhow::bail!(
304 "source.tls: {} disables certificate verification, silently downgrading the \
305 chosen `mode: {}` to trust-anything (MITM exposure — credentials and rows \
306 readable/forgeable on the wire).\n \
307 Hint: drop the danger knob and trust a private CA with `tls.ca_file: <pem>`; \
308 only use a danger knob for a loopback self-signed dev container, and then omit \
309 the explicit `mode:` so the contradiction is gone.",
310 knob,
311 mode,
312 );
313 }
314 Ok(())
315 }
316
317 /// Detect tuning-related fields placed directly under `source:` or an
318 /// `exports[]` entry instead of inside the `tuning:` sub-key. Without this
319 /// check serde silently ignores unknown keys and the user gets unexpected
320 /// defaults (e.g. batch_size=10 000 instead of the intended 1 000).
321 fn check_misplaced_tuning_fields(yaml: &str) -> crate::error::Result<()> {
322 const TUNING_FIELDS: &[&str] = &[
323 "batch_size",
324 "batch_size_memory_mb",
325 "throttle_ms",
326 "statement_timeout_s",
327 "max_retries",
328 "retry_backoff_ms",
329 "lock_timeout_s",
330 "memory_threshold_mb",
331 "profile",
332 ];
333
334 let root: serde_yaml_ng::Value = serde_yaml_ng::from_str(yaml)?;
335
336 if let Some(source) = root.get("source") {
337 let misplaced: Vec<&str> = TUNING_FIELDS
338 .iter()
339 .copied()
340 .filter(|&f| source.get(f).is_some())
341 .collect();
342 if !misplaced.is_empty() {
343 anyhow::bail!(
344 "source: field(s) [{}] belong under 'source.tuning:', not directly under 'source:'. \
345 Example:\n source:\n tuning:\n {}: <value>",
346 misplaced.join(", "),
347 misplaced[0],
348 );
349 }
350 }
351
352 if let Some(exports) = root.get("exports").and_then(|e| e.as_sequence()) {
353 for (i, export) in exports.iter().enumerate() {
354 let name = export
355 .get("name")
356 .and_then(|n| n.as_str())
357 .unwrap_or("<unnamed>");
358 let misplaced: Vec<&str> = TUNING_FIELDS
359 .iter()
360 .copied()
361 .filter(|&f| export.get(f).is_some())
362 .collect();
363 if !misplaced.is_empty() {
364 anyhow::bail!(
365 "export '{}' (index {}): field(s) [{}] belong under 'exports[].tuning:', \
366 not directly in the export. Example:\n exports:\n - name: {}\n tuning:\n {}: <value>",
367 name,
368 i,
369 misplaced.join(", "),
370 name,
371 misplaced[0],
372 );
373 }
374 }
375 }
376
377 Ok(())
378 }
379
380 /// Reject a config before any plan/connect step. The body is split into
381 /// three cohesive validators so each can be read — and unit-tested — on its
382 /// own: the export-list shape, the source connection block, and the
383 /// per-export rules. The end-to-end surface (`Config::from_yaml`) is
384 /// covered by `config/tests/{validation,secops}.rs`; the split additionally
385 /// lets a rule be exercised directly via `validate_export`.
386 fn validate(&self) -> crate::error::Result<()> {
387 self.validate_exports_list()?;
388 self.validate_source_connection()?;
389 for export in &self.exports {
390 self.validate_export(export)?;
391 }
392 self.validate_cdc_resource_conflicts()?;
393 self.validate_non_sql_source_modes()?;
394 Ok(())
395 }
396
397 /// Non-SQL sources (MongoDB) support only `mode: full` today. Chunked /
398 /// incremental / keyset / time-window all build SQL predicates over an
399 /// introspectable columnar schema, and CDC (change streams) is a separate,
400 /// not-yet-implemented seam. Rejecting the other modes here is precisely
401 /// what makes the SQL-only builders' `unreachable!` arms
402 /// (`sql::quote_ident`, `query::cursor_rhs`, …) provably unreachable for a
403 /// document-store source.
404 fn validate_non_sql_source_modes(&self) -> crate::error::Result<()> {
405 if self.source.source_type.is_sql() {
406 return Ok(());
407 }
408 for e in &self.exports {
409 if !matches!(e.mode, ExportMode::Full | ExportMode::Cdc) {
410 crate::config_bail!(
411 crate::error::codes::CONFIG_SOURCE_MODE_UNSUPPORTED,
412 "export '{}': source type '{:?}' supports `mode: full` (batch) and \
413 `mode: cdc` (change streams) (got `mode: {:?}`). MongoDB has no SQL, so \
414 chunked / incremental / keyset / time-window are not available; every \
415 document exports as `_id` + a `document` JSON column.",
416 e.name,
417 self.source.source_type,
418 e.mode
419 );
420 }
421 // An impossible combination must be a config error, not a silent
422 // behavior downgrade: the parallel `_id`-range path keeps NO keyset
423 // checkpoint, so `resume: true` was silently ignored — the whole
424 // collection re-read every run with no warning (bug-hunt find).
425 if e.parallel > 1 && self.source.mongo.as_ref().is_some_and(|m| m.resume) {
426 crate::config_bail!(
427 crate::error::codes::CONFIG_SOURCE_MODE_UNSUPPORTED,
428 "export '{}': `source.mongo.resume: true` cannot be combined with \
429 `parallel: {}` — the parallel `_id`-range fan-out keeps no keyset \
430 checkpoint, so resume would be silently ignored. Drop `parallel` \
431 to keep cross-run resume, or drop `resume` to keep the fan-out.",
432 e.name,
433 e.parallel
434 );
435 }
436 }
437 Ok(())
438 }
439
440 /// CDC stream resources are per-export and their **defaults collide**: two
441 /// PostgreSQL cdc exports without an explicit `slot:` both resolve to
442 /// `rivet_slot` — each export's ack advances `confirmed_flush_lsn` past
443 /// changes the *other* never read (mutual, silent data loss). Two MySQL
444 /// exports both default to `server_id: 4271` — the server kills the older
445 /// replica connection. A shared `checkpoint:` path makes exports overwrite
446 /// each other's resume position on any engine. All three are config bugs a
447 /// naive multi-table CDC config hits by default, so reject them at load, on
448 /// the RESOLVED values (defaults included). SQL Server `capture_instance`
449 /// sharing is deliberately allowed: the change-table poll is read-only and
450 /// resume state lives in the per-export checkpoint.
451 fn validate_cdc_resource_conflicts(&self) -> crate::error::Result<()> {
452 use std::collections::HashMap;
453
454 let mut slots: HashMap<String, &str> = HashMap::new();
455 let mut server_ids: HashMap<u32, &str> = HashMap::new();
456 let mut checkpoints: HashMap<&str, &str> = HashMap::new();
457
458 for e in self.exports.iter().filter(|e| e.mode == ExportMode::Cdc) {
459 let cdc = e.cdc.as_ref();
460 match self.source.source_type {
461 SourceType::Postgres => {
462 let slot = cdc
463 .and_then(|c| c.slot.clone())
464 .unwrap_or_else(|| DEFAULT_PG_SLOT.to_string());
465 if let Some(prev) = slots.insert(slot.clone(), &e.name) {
466 crate::config_bail!(
467 crate::error::codes::CONFIG_CDC_RESOURCE_CONFLICT,
468 "exports '{prev}' and '{}': same PostgreSQL slot '{slot}' — a slot \
469 has ONE consumer; each export's ack would advance it past changes \
470 the other never read (silent data loss). Set a distinct `cdc.slot:` \
471 per export (the default is '{DEFAULT_PG_SLOT}').",
472 e.name
473 );
474 }
475 }
476 SourceType::Mysql => {
477 let sid = cdc
478 .and_then(|c| c.server_id)
479 .unwrap_or(DEFAULT_MYSQL_SERVER_ID);
480 if let Some(prev) = server_ids.insert(sid, &e.name) {
481 crate::config_bail!(
482 crate::error::codes::CONFIG_CDC_RESOURCE_CONFLICT,
483 "exports '{prev}' and '{}': same MySQL server_id {sid} — the server \
484 kills the older replica connection when a new one registers with \
485 the same id. Set a distinct `cdc.server_id:` per export (the \
486 default is {DEFAULT_MYSQL_SERVER_ID}).",
487 e.name
488 );
489 }
490 }
491 SourceType::Mssql => {}
492 // MongoDB change streams watch the whole database — no per-export
493 // slot or server_id to collide. Two Mongo CDC exports sharing a
494 // `checkpoint:` path IS still a conflict, caught by the shared
495 // checkpoint check below.
496 SourceType::Mongo => {}
497 }
498 if let Some(ckpt) = cdc.and_then(|c| c.checkpoint.as_deref())
499 && let Some(prev) = checkpoints.insert(ckpt, &e.name)
500 {
501 crate::config_bail!(
502 crate::error::codes::CONFIG_CDC_RESOURCE_CONFLICT,
503 "exports '{prev}' and '{}': same checkpoint path '{ckpt}' — each export \
504 must own its resume position or they overwrite each other's. Set a \
505 distinct `cdc.checkpoint:` per export.",
506 e.name
507 );
508 }
509 }
510 Ok(())
511 }
512
513 /// Whole-config shape: at least one export, names unique.
514 fn validate_exports_list(&self) -> crate::error::Result<()> {
515 // An empty `exports:` list is almost always a typo (wrong config file,
516 // dropped anchor, merged doc with the anchor section missing). Running
517 // with zero exports is a silent no-op that looks like success in CI;
518 // reject fast instead. See QA backlog Task 5.1.
519 if self.exports.is_empty() {
520 crate::config_bail!(
521 crate::error::codes::CONFIG_NO_EXPORTS,
522 "exports: at least one export must be defined (got empty list)"
523 );
524 }
525
526 // Duplicate export names break state tracking: `export_state`,
527 // `file_log`, and `chunk_run` are all keyed by `export_name`, so
528 // two configs with the same name silently share cursor/file-log rows.
529 // QA backlog Task 5.1.
530 let mut seen: std::collections::HashSet<&str> =
531 std::collections::HashSet::with_capacity(self.exports.len());
532 for e in &self.exports {
533 if !seen.insert(e.name.as_str()) {
534 crate::config_bail!(
535 crate::error::codes::CONFIG_DUPLICATE_EXPORT,
536 "exports: duplicate export name '{}' (each export must have a unique name; state is keyed by name)",
537 e.name
538 );
539 }
540 }
541 Ok(())
542 }
543
544 /// Source connection block: exactly one connection method, well-formed,
545 /// and the source-level tuning that is shared by every export.
546 fn validate_source_connection(&self) -> crate::error::Result<()> {
547 if let Some(t) = &self.source.tuning
548 && t.batch_size.is_some()
549 && t.batch_size_memory_mb.is_some()
550 {
551 anyhow::bail!(
552 "tuning: batch_size and batch_size_memory_mb are mutually exclusive. \
553 Prefer batch_size_memory_mb (rivet sizes the batch to a memory budget, \
554 adapting to row width); set batch_size only to pin an exact row count."
555 );
556 }
557
558 if !self.source.has_url_fields() && !self.source.has_structured_fields() {
559 // First-run footgun: a config that forgot the source block
560 // entirely. Show the recommended path (`url_env`) up-front;
561 // operators who actually want structured fields know to look
562 // for them.
563 anyhow::bail!(
564 "source: no connection method configured. Add one of:\n url_env: DATABASE_URL (URL from env var — recommended)\n url: 'postgresql://user:pass@host:5432/db' (inline — not recommended for committed configs)\n url_file: /etc/rivet/source.url (URL from file — rotation-friendly)\n host/user/database/... (structured fields under `source:`)"
565 );
566 }
567
568 if self.source.has_url_fields() {
569 let url_count = [
570 &self.source.url,
571 &self.source.url_env,
572 &self.source.url_file,
573 ]
574 .iter()
575 .filter(|u| u.is_some())
576 .count();
577 if url_count > 1 {
578 anyhow::bail!(
579 "source: specify exactly one of 'url', 'url_env', or 'url_file' (got {} set).\n Hint: pick one — `url_env` is recommended so credentials never enter the YAML.",
580 url_count
581 );
582 }
583 }
584
585 if self.source.has_url_fields() && self.source.has_structured_fields() {
586 anyhow::bail!(
587 "source: pick either URL-based config (url/url_env/url_file) OR structured fields (host/user/database/port/password_env), not both.\n Hint: remove whichever block you don't want; mixing the two is ambiguous."
588 );
589 }
590
591 if self.source.has_structured_fields() {
592 if self.source.host.is_none() {
593 anyhow::bail!(
594 "source: structured config is missing 'host'.\n Hint: add `host: localhost` (or your DB host) under `source:` in rivet.yaml.\n Or switch to URL-based config: `url_env: DATABASE_URL`."
595 );
596 }
597 if self.source.user.is_none() {
598 anyhow::bail!(
599 "source: structured config is missing 'user'.\n Hint: add `user: <username>` under `source:` in rivet.yaml."
600 );
601 }
602 if self.source.database.is_none() {
603 anyhow::bail!(
604 "source: structured config is missing 'database'.\n Hint: add `database: <dbname>` under `source:` in rivet.yaml."
605 );
606 }
607 if self.source.password.is_some() && self.source.password_env.is_some() {
608 anyhow::bail!(
609 "source: specify 'password' OR 'password_env', not both.\n Hint: prefer `password_env: DB_PASSWORD` so credentials never enter the YAML."
610 );
611 }
612 }
613 Ok(())
614 }
615
616 /// Per-export rules: effective tuning, query source, `query_file` SecOps,
617 /// destination auth, compression, and the mode/chunk matrix. Takes `&self`
618 /// because effective tuning merges the source-level block.
619 fn validate_export(&self, export: &ExportConfig) -> crate::error::Result<()> {
620 // V5: `name` is keyed into output paths, file logs, and on-disk state,
621 // yet is otherwise free-form. A traversal (`../../etc/x`), absolute or
622 // slash-bearing (`/abs/x`, `sub/dir`), leading-dot, or NUL-bearing name
623 // escapes the intended output tree and corrupts name-keyed state.
624 // Mirror the `query_file` `..`/absolute guard: reject at config-load,
625 // accepting only a filename-safe charset.
626 if !is_filename_safe_name(&export.name) {
627 anyhow::bail!(
628 "export name '{}' is not filename-safe: it must not be absolute, contain \
629 '/', '\\', '..', a NUL, or start with '.' (the name is used in output paths \
630 and state keys). Use a plain identifier like `orders` or `daily_events`.",
631 export.name.escape_default(),
632 );
633 }
634
635 let merged =
636 crate::tuning::merge_tuning_config(self.source.tuning.as_ref(), export.tuning.as_ref());
637 if let Some(t) = merged
638 && t.batch_size.is_some()
639 && t.batch_size_memory_mb.is_some()
640 {
641 anyhow::bail!(
642 "export '{}': effective tuning has both batch_size and batch_size_memory_mb (mutually exclusive)",
643 export.name
644 );
645 }
646 if let Some(et) = &export.tuning
647 && et.batch_size.is_some()
648 && et.batch_size_memory_mb.is_some()
649 {
650 anyhow::bail!(
651 "export '{}': tuning.batch_size and tuning.batch_size_memory_mb are mutually exclusive",
652 export.name
653 );
654 }
655
656 // Before the generic exactly-one counting: the `table:`/`tables:` pair
657 // gets its specific message (the generic one doesn't mention `tables`).
658 if export.table.is_some() && export.tables.is_some() {
659 anyhow::bail!(
660 "export '{}': `table:` and `tables:` are mutually exclusive — use \
661 `tables: [a, b]` for a multi-table stream",
662 export.name
663 );
664 }
665
666 let set_count = [
667 export.query.is_some(),
668 export.query_file.is_some(),
669 export.table.is_some(),
670 // A multi-table CDC stream (`tables:`) is that export's source
671 // specification — the CDC arm below owns its shape rules.
672 export.tables.is_some(),
673 ]
674 .iter()
675 .filter(|b| **b)
676 .count();
677 if set_count == 0 {
678 anyhow::bail!(
679 "export '{}': specify exactly one of 'query', 'query_file', 'table', or 'tables'. \
680 Use table: <name> for a whole table (enables PK auto-chunking); \
681 query: \"SELECT …\" for an inline one-liner; \
682 query_file: <path> for SQL you keep in version control.",
683 export.name
684 );
685 }
686 if set_count > 1 {
687 anyhow::bail!(
688 "export '{}': specify exactly one of 'query', 'query_file', 'table', or 'tables' (got {} set)",
689 export.name,
690 set_count
691 );
692 }
693 // SecOps: syntactic `query_file` checks must run at config-validate
694 // time so `rivet check` / `rivet doctor` catch them before any
695 // plan step. The same checks repeat (with a canonicalize-based
696 // symlink probe) in `ExportConfig::resolve_query` because the
697 // file may have been swapped between validation and read.
698 if let Some(file) = &export.query_file {
699 let p = std::path::Path::new(file);
700 if p.is_absolute() {
701 anyhow::bail!(
702 "export '{}': query_file must be a relative path: '{}'",
703 export.name,
704 file
705 );
706 }
707 if p.components().any(|c| c == std::path::Component::ParentDir) {
708 anyhow::bail!(
709 "export '{}': query_file path must not contain '..': '{}'",
710 export.name,
711 file
712 );
713 }
714 }
715 // V2/V12: a custom cloud `endpoint` is handed straight to the opendal
716 // S3/GCS/Azure builder with no validation, so a committed config can
717 // silently redirect every upload to an attacker host (exfiltration) or
718 // send credentials + rows over cleartext `http://`. The legitimate use
719 // is a local emulator (Minio / Azurite / fake-gcs on `127.0.0.1`), so
720 // accept a loopback host (any scheme), and otherwise accept a remote
721 // endpoint only when the operator has explicitly opted into anonymous
722 // (emulator) mode. Reject every other custom endpoint at config-load.
723 if matches!(
724 export.destination.destination_type,
725 DestinationType::S3 | DestinationType::Gcs | DestinationType::Azure
726 ) && let Some(endpoint) = &export.destination.endpoint
727 {
728 // Loopback emulator (Minio/Azurite/fake-gcs) is the legitimate
729 // local-dev path — accept any scheme. A non-loopback (or
730 // unparseable) custom endpoint is only accepted when the operator
731 // has explicitly opted into anonymous (emulator) mode, where no
732 // credentials are sent. Everything else is rejected.
733 let loopback = endpoint_host(endpoint).is_some_and(|host| is_loopback_host(&host));
734 if !loopback && !export.destination.allow_anonymous {
735 anyhow::bail!(
736 "export '{}': destination.endpoint '{}' points at a non-loopback host. \
737 A custom endpoint redirects every upload there — committing one is a \
738 data-exfiltration / cleartext-credential risk.\n \
739 Hint: drop `endpoint:` to use the provider default, point it at a \
740 loopback emulator (e.g. http://127.0.0.1:9000 with allow_anonymous: true \
741 for Minio/Azurite), or set `allow_anonymous: true` for an anonymous \
742 emulator.",
743 export.name,
744 endpoint,
745 );
746 }
747 }
748
749 // V15: a `type: local` destination `path` (or `prefix`) is written
750 // verbatim to the filesystem. A `..` component lets a committed config
751 // climb out of the intended output tree (`../../../../tmp/x`) — mirror
752 // the `query_file` traversal guard and reject it at config-load.
753 //
754 // Absolute paths are deliberately *not* rejected: `path: /output` is a
755 // legitimate Docker volume-mount pattern (see `examples/rivet.yaml`) and
756 // an explicit operator choice, not a hidden escape. The `..` climb is
757 // the unambiguous traversal footgun.
758 if export.destination.destination_type == DestinationType::Local {
759 for (field, value) in [
760 ("path", export.destination.path.as_deref()),
761 ("prefix", export.destination.prefix.as_deref()),
762 ] {
763 let Some(value) = value else { continue };
764 if std::path::Path::new(value)
765 .components()
766 .any(|c| c == std::path::Component::ParentDir)
767 {
768 anyhow::bail!(
769 "export '{}': local destination {} must not contain a '..' component: \
770 '{}' (a parent-dir climb writes outside the output tree).",
771 export.name,
772 field,
773 value
774 );
775 }
776 }
777 }
778
779 if export.destination.destination_type == DestinationType::S3 {
780 let ak = export.destination.access_key_env.is_some();
781 let sk = export.destination.secret_key_env.is_some();
782 if ak != sk {
783 anyhow::bail!(
784 "export '{}': S3 requires both access_key_env and secret_key_env, or neither (use default AWS credential chain)",
785 export.name
786 );
787 }
788 }
789
790 if export.destination.destination_type == DestinationType::Gcs
791 && export.destination.allow_anonymous
792 && export.destination.credentials_file.is_some()
793 {
794 anyhow::bail!(
795 "export '{}': GCS allow_anonymous cannot be used together with credentials_file",
796 export.name
797 );
798 }
799
800 if export.destination.destination_type == DestinationType::Azure {
801 let has_name = export.destination.account_name.is_some();
802 let has_key = export.destination.account_key_env.is_some();
803 let has_sas = export.destination.sas_token_env.is_some();
804 if export.destination.allow_anonymous {
805 if has_name || has_key || has_sas {
806 anyhow::bail!(
807 "export '{}': Azure allow_anonymous cannot be combined with account_name/account_key_env/sas_token_env",
808 export.name
809 );
810 }
811 } else if has_key && has_sas {
812 anyhow::bail!(
813 "export '{}': Azure account_key_env and sas_token_env are mutually exclusive — pick one auth mode",
814 export.name
815 );
816 } else if !has_name {
817 anyhow::bail!(
818 "export '{}': Azure requires account_name (plus account_key_env or sas_token_env), or allow_anonymous: true for Azurite",
819 export.name
820 );
821 } else if !has_key && !has_sas {
822 anyhow::bail!(
823 "export '{}': Azure requires account_key_env or sas_token_env (or allow_anonymous: true for Azurite)",
824 export.name
825 );
826 }
827 }
828
829 if let Some(cred_path) = &export.destination.credentials_file
830 && !std::path::Path::new(cred_path).exists()
831 {
832 anyhow::bail!(
833 "export '{}': credentials_file '{}' does not exist",
834 export.name,
835 cred_path
836 );
837 }
838
839 if let Some(ref size_str) = export.max_file_size {
840 parse_file_size(size_str).map_err(|_| {
841 anyhow::anyhow!(
842 "export '{}': invalid max_file_size '{}'",
843 export.name,
844 size_str
845 )
846 })?;
847 }
848
849 if let Some(level) = export.compression_level {
850 match export.compression {
851 CompressionType::Zstd => {
852 if !(1..=22).contains(&level) {
853 anyhow::bail!(
854 "export '{}': zstd compression_level must be 1..22, got {}",
855 export.name,
856 level
857 );
858 }
859 }
860 CompressionType::Gzip => {
861 if level > 10 {
862 anyhow::bail!(
863 "export '{}': gzip compression_level must be 0..10, got {}",
864 export.name,
865 level
866 );
867 }
868 }
869 _ => {
870 anyhow::bail!(
871 "export '{}': compression_level is only supported for zstd and gzip",
872 export.name
873 );
874 }
875 }
876 }
877
878 match export.mode {
879 ExportMode::Incremental => {
880 if export.cursor_column.is_none() {
881 anyhow::bail!(
882 "export '{}': incremental mode requires cursor_column",
883 export.name
884 );
885 }
886 match export.incremental_cursor_mode {
887 IncrementalCursorMode::Coalesce => {
888 if export.cursor_fallback_column.is_none() {
889 anyhow::bail!(
890 "export '{}': incremental_cursor_mode: coalesce requires cursor_fallback_column",
891 export.name
892 );
893 }
894 }
895 IncrementalCursorMode::SingleColumn => {
896 if export.cursor_fallback_column.is_some() {
897 anyhow::bail!(
898 "export '{}': cursor_fallback_column is only valid with incremental_cursor_mode: coalesce",
899 export.name
900 );
901 }
902 }
903 }
904 }
905 ExportMode::Chunked => {
906 // `chunk_column` is mandatory unless the user used the `table:`
907 // shortcut on a Postgres source — in that case it is auto-resolved
908 // from the table's single-integer PK at plan-build time (see
909 // `crate::plan::build::resolve_chunk_column`).
910 if export.chunk_column.is_none() && export.table.is_none() {
911 anyhow::bail!(
912 "export '{}': chunked mode needs a chunking strategy. Pick one:\n \
913 chunk_column: <int col> range chunks on an integer column (most common)\n \
914 chunk_by_key: <unique col> keyset pagination when there's no integer PK\n \
915 chunk_count: <N> split the range into N equal chunks\n \
916 chunk_by_days: <D> time-bucketed chunks (needs a date/timestamp column)\n \
917 Or use the `table:` shortcut on a single table — rivet auto-resolves the column from the primary key.",
918 export.name
919 );
920 }
921 // chunk_size == 0 would divide the range into zero-width
922 // slices and (before the saturating fix in generate_chunks)
923 // either infinite-loop or produce no progress. QA backlog
924 // Task 5.1.
925 if export.chunk_size == 0 {
926 anyhow::bail!(
927 "export '{}': chunked mode requires chunk_size >= 1 (got 0)",
928 export.name
929 );
930 }
931 // parallel == 0 means "spawn zero workers". Claiming tasks
932 // with no workers stalls the pipeline. QA backlog Task 5.1.
933 if export.parallel == 0 {
934 anyhow::bail!(
935 "export '{}': chunked mode requires parallel >= 1 (got 0)",
936 export.name
937 );
938 }
939 if let Some(0) = export.chunk_count {
940 crate::config_bail!(
941 crate::error::codes::CONFIG_CHUNK_COUNT_INVALID,
942 "export '{}': chunk_count must be >= 1",
943 export.name
944 );
945 }
946 if export.chunk_count.is_some() && export.chunk_dense {
947 anyhow::bail!(
948 "export '{}': chunk_count and chunk_dense are mutually exclusive. \
949 Use chunk_count for equal-sized chunks over a sparse key; \
950 use chunk_dense only when the key has no gaps.",
951 export.name
952 );
953 }
954 if export.chunk_count.is_some() && export.chunk_by_days.is_some() {
955 anyhow::bail!(
956 "export '{}': chunk_count and chunk_by_days are mutually exclusive. \
957 Use chunk_count: N to split an integer range into N chunks; \
958 use chunk_by_days: D to bucket a date/timestamp column by D-day windows.",
959 export.name
960 );
961 }
962 }
963 ExportMode::TimeWindow => {
964 if export.time_column.is_none() {
965 anyhow::bail!(
966 "export '{}': time_window mode requires time_column",
967 export.name
968 );
969 }
970 if export.days_window.is_none() {
971 anyhow::bail!(
972 "export '{}': time_window mode requires days_window",
973 export.name
974 );
975 }
976 }
977 ExportMode::Full => {}
978 ExportMode::Cdc => {
979 match (&export.table, &export.tables) {
980 (None, None) => anyhow::bail!(
981 "export '{}': cdc mode requires `table:` (or `tables:` for a \
982 multi-table stream)",
983 export.name
984 ),
985 (Some(_), Some(_)) => anyhow::bail!(
986 "export '{}': `table:` and `tables:` are mutually exclusive — \
987 use `tables: [a, b]` for a multi-table stream",
988 export.name
989 ),
990 (None, Some(ts)) => {
991 if ts.is_empty() {
992 anyhow::bail!(
993 "export '{}': `tables:` must list at least one table",
994 export.name
995 );
996 }
997 let mut seen = std::collections::HashSet::new();
998 for t in ts {
999 if !seen.insert(t.as_str()) {
1000 anyhow::bail!(
1001 "export '{}': duplicate table '{}' in `tables:`",
1002 export.name,
1003 t
1004 );
1005 }
1006 }
1007 if self.source.source_type == SourceType::Mssql {
1008 anyhow::bail!(
1009 "export '{}': `tables:` is not yet supported for SQL Server — \
1010 its capture instances are per-table; use one cdc export per \
1011 table (capture_instance each)",
1012 export.name
1013 );
1014 }
1015 }
1016 (Some(_), None) => {}
1017 }
1018 if export.query.is_some() || export.query_file.is_some() {
1019 anyhow::bail!(
1020 "export '{}': cdc mode reads the transaction log, not a query — \
1021 remove query/query_file and use `table:`",
1022 export.name
1023 );
1024 }
1025 }
1026 }
1027
1028 if export.tables.is_some() && export.mode != ExportMode::Cdc {
1029 anyhow::bail!(
1030 "export '{}': `tables:` is only valid with `mode: cdc` (batch exports \
1031 are one query/table per export)",
1032 export.name
1033 );
1034 }
1035
1036 if export.chunk_dense && export.mode != ExportMode::Chunked {
1037 anyhow::bail!(
1038 "export '{}': chunk_dense is only valid with mode: chunked",
1039 export.name
1040 );
1041 }
1042
1043 if export.cdc.is_some() && export.mode != ExportMode::Cdc {
1044 anyhow::bail!(
1045 "export '{}': a `cdc:` block is only valid with `mode: cdc`",
1046 export.name
1047 );
1048 }
1049
1050 // Table-qualified `columns:` override keys ("table.column"): the named
1051 // table must be one this export captures — a typo must fail at load,
1052 // never silently miss its target. Bare keys stay export-wide.
1053 {
1054 for key in export.columns.keys() {
1055 let Some((tbl, _col)) = key.split_once('.') else {
1056 continue;
1057 };
1058 if export.query.is_some() || export.query_file.is_some() {
1059 anyhow::bail!(
1060 "export '{}': qualified column override '{key}' needs a table-shaped \
1061 export (`table:`/`tables:`), not a query",
1062 export.name
1063 );
1064 }
1065 let bare = |t: &str| t.rsplit('.').next().unwrap_or(t).to_string();
1066 let captures = export
1067 .table
1068 .as_deref()
1069 .map(bare)
1070 .into_iter()
1071 .chain(export.tables.iter().flatten().map(|t| bare(t)))
1072 .any(|t| t == tbl);
1073 if !captures {
1074 anyhow::bail!(
1075 "export '{}': column override '{key}' names table '{tbl}', which this \
1076 export does not capture — fix the table name or use a bare column key \
1077 to apply it to every captured table",
1078 export.name
1079 );
1080 }
1081 }
1082 }
1083
1084 // `initial: snapshot` writes each table's snapshot under the reserved
1085 // sub-prefix `snapshot/` — a table actually NAMED "snapshot" would share
1086 // a prefix with another table's marker. Refuse the collision at load.
1087 if let Some(cdc) = &export.cdc
1088 && cdc.initial == Some(CdcInitialMode::Snapshot)
1089 {
1090 let clashes = |t: &str| t.rsplit('.').next().unwrap_or(t) == "snapshot";
1091 if export.table.as_deref().is_some_and(clashes)
1092 || export.tables.iter().flatten().any(|t| clashes(t))
1093 {
1094 anyhow::bail!(
1095 "export '{}': a table named 'snapshot' collides with the reserved \
1096 `snapshot/` sub-prefix that `cdc.initial: snapshot` writes — rename \
1097 the table or use a separate export without `initial:`",
1098 export.name
1099 );
1100 }
1101 }
1102
1103 // `initial: snapshot` needs a durable anchor BEFORE the snapshot reads.
1104 // PostgreSQL pins server-side (the slot); MySQL / SQL Server have no
1105 // server-side anchor, so the checkpoint file is mandatory there.
1106 if let Some(cdc) = &export.cdc
1107 && cdc.initial == Some(CdcInitialMode::Snapshot)
1108 && self.source.source_type != SourceType::Postgres
1109 && cdc.checkpoint.is_none()
1110 {
1111 anyhow::bail!(
1112 "export '{}': `cdc.initial: snapshot` on {:?} requires `cdc.checkpoint:` — \
1113 the checkpoint file is the anchor that makes snapshot-then-stream gap-free",
1114 export.name,
1115 self.source.source_type
1116 );
1117 }
1118
1119 // MongoDB change streams have NO server-side resume anchor (unlike a
1120 // PostgreSQL slot): the checkpoint file IS the anchor. Without it, every
1121 // run re-anchors at "now" and silently loses every change since the last
1122 // run — so `mode: cdc` on mongo requires `cdc.checkpoint:` ALWAYS, not
1123 // only under `initial: snapshot` (bug-hunt find).
1124 if export.mode == ExportMode::Cdc
1125 && self.source.source_type == SourceType::Mongo
1126 && export
1127 .cdc
1128 .as_ref()
1129 .and_then(|c| c.checkpoint.as_ref())
1130 .is_none()
1131 {
1132 anyhow::bail!(
1133 "export '{}': MongoDB `mode: cdc` requires `cdc.checkpoint:` — a change \
1134 stream has no server-side resume anchor, so without the checkpoint file \
1135 each run re-anchors at the current time and silently loses every change \
1136 between runs.",
1137 export.name
1138 );
1139 }
1140
1141 // A collection whose name contains a dot does not route through the
1142 // change-stream capture — its events are silently dropped. Refuse loudly
1143 // rather than report 0-row success forever (bug-hunt find). Batch handles
1144 // dotted names fine, so this is CDC-only.
1145 if export.mode == ExportMode::Cdc && self.source.source_type == SourceType::Mongo {
1146 let dotted = |t: &str| t.contains('.');
1147 if export.table.as_deref().is_some_and(dotted)
1148 || export.tables.iter().flatten().any(|t| dotted(t))
1149 {
1150 anyhow::bail!(
1151 "export '{}': MongoDB `mode: cdc` does not support a collection name \
1152 containing a dot — the change-stream router cannot address it and its \
1153 events would be silently dropped. Rename the collection, or capture it \
1154 with `mode: full` (batch handles dotted names).",
1155 export.name
1156 );
1157 }
1158 }
1159
1160 if let Some(days) = export.chunk_by_days {
1161 if export.mode != ExportMode::Chunked {
1162 anyhow::bail!(
1163 "export '{}': chunk_by_days requires mode: chunked",
1164 export.name
1165 );
1166 }
1167 if export.chunk_dense {
1168 anyhow::bail!(
1169 "export '{}': chunk_by_days cannot be combined with chunk_dense",
1170 export.name
1171 );
1172 }
1173 if days == 0 {
1174 crate::config_bail!(
1175 crate::error::codes::CONFIG_CHUNK_BY_DAYS_INVALID,
1176 "export '{}': chunk_by_days must be at least 1",
1177 export.name
1178 );
1179 }
1180 }
1181 Ok(())
1182 }
1183}
1184
1185/// True when a single YAML line carries a mapping value (text after `key:`)
1186/// that contains a `{` outside of any quotes — the unquoted-template-brace
1187/// shape (`prefix: {partition}`, `path: {date}/out`).
1188///
1189/// Quote-aware so a properly quoted value (`prefix: "exports/{partition}/"`)
1190/// does *not* match, and `$`-prefixed braces (`${VAR}` env placeholders) are
1191/// ignored — they are resolved before the parse and are not the footgun.
1192fn line_has_unquoted_brace_value(line: &str) -> bool {
1193 // Whole-line comments never carry a value — skip before splitting.
1194 if line.trim_start().starts_with('#') {
1195 return false;
1196 }
1197 // Split key from value at the first `": "` / `":\t"` / trailing `:`.
1198 // A YAML plain-key separator is a colon followed by whitespace or EOL.
1199 let bytes = line.as_bytes();
1200 let mut sep = None;
1201 let mut i = 0;
1202 while i < bytes.len() {
1203 if bytes[i] == b':' && (i + 1 == bytes.len() || bytes[i + 1].is_ascii_whitespace()) {
1204 sep = Some(i + 1);
1205 break;
1206 }
1207 i += 1;
1208 }
1209 let Some(value_start) = sep else {
1210 return false;
1211 };
1212 let value = line[value_start..].trim_start();
1213 // A trailing `#` after the value starts an inline comment; an empty or
1214 // comment-only value carries no brace to flag.
1215 if value.is_empty() || value.starts_with('#') {
1216 return false;
1217 }
1218
1219 let mut in_single = false;
1220 let mut in_double = false;
1221 let vbytes = value.as_bytes();
1222 for (j, &c) in vbytes.iter().enumerate() {
1223 match c {
1224 b'\'' if !in_double => in_single = !in_single,
1225 b'"' if !in_single => in_double = !in_double,
1226 b'{' if !in_single && !in_double => {
1227 // Ignore `${...}` env placeholders (resolved pre-parse).
1228 if j > 0 && vbytes[j - 1] == b'$' {
1229 continue;
1230 }
1231 return true;
1232 }
1233 _ => {}
1234 }
1235 }
1236 false
1237}
1238
1239/// Extract the lower-cased host from a `scheme://host[:port][/path]` endpoint,
1240/// or `None` when it does not look like a URL.
1241///
1242/// SecOps helper for the cloud-`endpoint` exfiltration guard (V2/V12): the host
1243/// decides whether a custom endpoint is a local emulator (loopback) or a remote
1244/// redirect target. We reject every non-loopback custom endpoint regardless of
1245/// scheme (covering both the exfil and the cleartext-`http` gaps), so only the
1246/// host is needed. We hand-parse rather than pull in a URL crate — the inputs
1247/// are operator-typed endpoints, not arbitrary URIs. A bracketed IPv6 literal
1248/// authority (`http://[::1]:9000`) keeps its address so it compares against the
1249/// loopback list.
1250fn endpoint_host(endpoint: &str) -> Option<String> {
1251 let (scheme, rest) = endpoint.split_once("://")?;
1252 if scheme.is_empty() {
1253 return None;
1254 }
1255 // Authority ends at the first `/` (path), `?` (query), or `#` (fragment);
1256 // any `user[:pass]@` userinfo head is dropped (host is after the last `@`).
1257 let authority = rest
1258 .split(['/', '?', '#'])
1259 .next()
1260 .unwrap_or("")
1261 .rsplit('@')
1262 .next()
1263 .unwrap_or("");
1264 let host = if let Some(stripped) = authority.strip_prefix('[') {
1265 // Bracketed IPv6 literal: take up to the closing `]`.
1266 stripped.split(']').next().unwrap_or("")
1267 } else {
1268 // host[:port] — strip the port suffix.
1269 authority.split(':').next().unwrap_or("")
1270 };
1271 if host.is_empty() {
1272 return None;
1273 }
1274 Some(host.to_ascii_lowercase())
1275}
1276
1277/// True when `host` names the local machine — the legitimate cloud-emulator
1278/// target (Minio / Azurite / fake-gcs on `127.0.0.1`). Anything else is a
1279/// remote host and a potential exfiltration redirect.
1280fn is_loopback_host(host: &str) -> bool {
1281 // `localhost` is the only non-IP host that counts as loopback. Everything
1282 // else must PARSE as an IP literal in the loopback range — a lexical
1283 // `starts_with("127.")` would accept attacker-controlled DNS like
1284 // `127.attacker.com` or `127.0.0.1.evil.com` (both resolve off-box), turning
1285 // the credential-exfil gate into a bypass (V2/V12). Parse strictly: only a
1286 // real `127.0.0.0/8` / `::1` address is loopback; a hostname is not.
1287 if host == "localhost" {
1288 return true;
1289 }
1290 // Tolerate a bracketed IPv6 literal (`[::1]`) in case a caller forwards one.
1291 let h = host
1292 .strip_prefix('[')
1293 .and_then(|s| s.strip_suffix(']'))
1294 .unwrap_or(host);
1295 h.parse::<std::net::IpAddr>()
1296 .is_ok_and(|ip| ip.is_loopback())
1297}
1298
1299/// True when `name` is filename-safe: rejects path-traversal (`..`), absolute
1300/// or slash-bearing names (`/`, `\`), a leading `.` (hidden / current-dir), and
1301/// embedded NULs. `ExportConfig.name` is keyed into output paths and on-disk
1302/// state, so a `../../etc/x` or absolute name escapes the output tree (V5).
1303fn is_filename_safe_name(name: &str) -> bool {
1304 !name.is_empty()
1305 && !name.starts_with('.')
1306 && !name.contains('/')
1307 && !name.contains('\\')
1308 && !name.contains("..")
1309 && !name.contains('\0')
1310}
1311
1312#[cfg(test)]
1313mod tests;
1314
1315#[cfg(test)]
1316mod audit_csv_compression {
1317 //! Finding #10: `format: csv` + a compression codec is a silent no-op
1318 //! (the file stays uncompressed but the manifest records the codec). The
1319 //! combo must be rejected at config-validate time. These tests encode the
1320 //! new rule, so reverting the fix turns them red.
1321 use super::*;
1322
1323 fn yaml(format: &str, compression_line: &str) -> String {
1324 format!(
1325 "source:\n type: postgres\n url: \"postgresql://localhost/test\"\n\
1326 exports:\n - name: t\n query: \"SELECT 1\"\n format: {format}\n\
1327 {compression_line} destination:\n type: local\n path: ./out\n"
1328 )
1329 }
1330
1331 #[test]
1332 fn audit_csv_compression_is_rejected() {
1333 // csv + gzip → rejected, with an actionable message.
1334 let err = Config::from_yaml(&yaml("csv", " compression: gzip\n")).unwrap_err();
1335 let msg = format!("{err:#}");
1336 assert!(
1337 msg.contains("CSV output does not support compression") && msg.contains("gzip"),
1338 "csv+gzip must be rejected with an actionable message; got: {msg}"
1339 );
1340 assert!(
1341 msg.contains("parquet") && msg.contains("none"),
1342 "message must point to the real options (parquet / none); got: {msg}"
1343 );
1344
1345 // Guard the boundaries: parquet+gzip and csv+none still validate.
1346 Config::from_yaml(&yaml("parquet", " compression: gzip\n"))
1347 .expect("parquet+gzip must validate");
1348 Config::from_yaml(&yaml("csv", " compression: none\n")).expect("csv+none must validate");
1349 }
1350
1351 #[test]
1352 fn audit_csv_every_real_codec_is_rejected() {
1353 // Each non-None codec is a silent no-op for CSV — none may slip through.
1354 for codec in ["zstd", "snappy", "gzip", "lz4"] {
1355 let err = Config::from_yaml(&yaml("csv", &format!(" compression: {codec}\n")))
1356 .unwrap_err();
1357 let msg = format!("{err:#}");
1358 assert!(
1359 msg.contains("CSV output does not support compression") && msg.contains(codec),
1360 "csv+{codec} must be rejected; got: {msg}"
1361 );
1362 }
1363 }
1364
1365 #[test]
1366 fn audit_csv_compression_profile_is_rejected() {
1367 // A `compression_profile:` other than `none` resolves to a real codec,
1368 // so it is the same silent no-op for CSV.
1369 for profile in ["fast", "balanced", "compact"] {
1370 let err = Config::from_yaml(&yaml(
1371 "csv",
1372 &format!(" compression_profile: {profile}\n"),
1373 ))
1374 .unwrap_err();
1375 let msg = format!("{err:#}");
1376 assert!(
1377 msg.contains("CSV output does not support compression_profile")
1378 && msg.contains(profile),
1379 "csv+profile {profile} must be rejected; got: {msg}"
1380 );
1381 }
1382 // profile: none is a no-op request and is fine.
1383 Config::from_yaml(&yaml("csv", " compression_profile: none\n"))
1384 .expect("csv + compression_profile: none must validate");
1385 }
1386
1387 #[test]
1388 fn audit_csv_default_compression_still_validates() {
1389 // Regression guard: a bare `format: csv` (no explicit codec) must keep
1390 // validating. `CompressionType::default()` is `Zstd`, but the user did
1391 // not *ask* for it — only an explicit codec is a no-op request. This
1392 // pins that the fix scans for explicit intent, not the struct default
1393 // (which would break ~60 existing csv configs).
1394 Config::from_yaml(&yaml("csv", "")).expect("bare format: csv must validate");
1395 }
1396
1397 #[test]
1398 fn audit_compression_supported_predicate() {
1399 // `compression_supported` is re-exported via `pub use format::*`.
1400 // Parquet supports every codec; CSV supports only None.
1401 for ct in [
1402 CompressionType::Zstd,
1403 CompressionType::Snappy,
1404 CompressionType::Gzip,
1405 CompressionType::Lz4,
1406 CompressionType::None,
1407 ] {
1408 assert!(compression_supported(FormatType::Parquet, ct));
1409 }
1410 assert!(compression_supported(
1411 FormatType::Csv,
1412 CompressionType::None
1413 ));
1414 for ct in [
1415 CompressionType::Zstd,
1416 CompressionType::Snappy,
1417 CompressionType::Gzip,
1418 CompressionType::Lz4,
1419 ] {
1420 assert!(
1421 !compression_supported(FormatType::Csv, ct),
1422 "CSV must not claim to support {}",
1423 ct.label()
1424 );
1425 }
1426 }
1427}
1428
1429#[cfg(test)]
1430mod reserved_load_extension {
1431 //! A downstream first-party loader carries its warehouse target in a
1432 //! top-level `load:` block so ONE config drives export + load. OSS must
1433 //! accept and ignore it — otherwise `check`/`run`/`apply` would reject the
1434 //! single-file config. Reverting the reserved field turns this red.
1435 use super::*;
1436
1437 #[test]
1438 fn reserved_top_level_load_block_is_accepted_and_ignored() {
1439 let yaml = "source:\n type: postgres\n url: \"postgresql://localhost/test\"\n\
1440 exports:\n - name: t\n query: \"SELECT 1\"\n format: parquet\n destination:\n type: gcs\n bucket: b\n prefix: \"t/\"\n\
1441 load:\n project: p\n dataset: d\n cleanup_source: true\n";
1442 let cfg = Config::from_yaml(yaml).expect("config with a reserved `load:` block must parse");
1443 assert!(cfg.load.is_some(), "the load block is captured opaquely");
1444 }
1445}
1446
1447#[cfg(test)]
1448mod audit_unquoted_template_brace {
1449 //! yaml-hint: an unquoted `{partition}` (or `{date}`) in a path/prefix
1450 //! value trips serde_yaml_ng's flow-mapping parser with a cryptic message
1451 //! that gives no clue the brace needs quoting. Since `{partition}` is the
1452 //! required token for `partition_by`, this is a common copy-paste footgun.
1453 //! `Config::from_yaml` augments the parser error with a quoting hint; these
1454 //! tests pin that behavior (and guard that valid configs are untouched).
1455 use super::*;
1456
1457 /// A full, otherwise-valid config whose `prefix:` value is whatever the
1458 /// caller passes verbatim (quoted or not). Only the `prefix:` line varies,
1459 /// so any parse error is attributable to the brace under test.
1460 fn yaml_with_prefix(prefix_value: &str) -> String {
1461 format!(
1462 "source:\n\
1463 \x20 type: postgres\n\
1464 \x20 url: \"postgresql://localhost/test\"\n\
1465 exports:\n\
1466 \x20 - name: t\n\
1467 \x20 query: \"SELECT 1\"\n\
1468 \x20 format: parquet\n\
1469 \x20 partition_by: created_date\n\
1470 \x20 destination:\n\
1471 \x20 type: local\n\
1472 \x20 path: ./out\n\
1473 \x20 prefix: {prefix_value}\n"
1474 )
1475 }
1476
1477 const HINT_FRAGMENT: &str =
1478 "a YAML value containing { } (such as {partition} or {date}) must be quoted";
1479
1480 #[test]
1481 fn bare_partition_token_gets_quoting_hint() {
1482 // `prefix: {partition}` parses as a YAML map, so serde rejects it with
1483 // `invalid type: map, expected a string` — no clue it's a quoting bug.
1484 let err = Config::from_yaml(&yaml_with_prefix("{partition}")).unwrap_err();
1485 let msg = format!("{err:#}");
1486 assert!(
1487 msg.contains(HINT_FRAGMENT),
1488 "bare {{partition}} must carry the quoting hint; got: {msg}"
1489 );
1490 // The original parser detail (type + location) is preserved.
1491 assert!(
1492 msg.contains("invalid type: map") || msg.contains("line"),
1493 "the original parser error must be kept; got: {msg}"
1494 );
1495 }
1496
1497 #[test]
1498 fn trailing_text_after_brace_gets_quoting_hint() {
1499 // `prefix: {date}/{partition}/` runs the flow map into block context:
1500 // serde emits `did not find expected key ... while parsing a block
1501 // mapping`. Same footgun, different libyaml symptom.
1502 let err = Config::from_yaml(&yaml_with_prefix("{date}/{partition}/")).unwrap_err();
1503 let msg = format!("{err:#}");
1504 assert!(
1505 msg.contains(HINT_FRAGMENT),
1506 "{{date}}/{{partition}}/ must carry the quoting hint; got: {msg}"
1507 );
1508 }
1509
1510 #[test]
1511 fn unclosed_brace_gets_quoting_hint() {
1512 // `prefix: {partition` (unclosed) is the canonical flow-mapping scanner
1513 // error: `did not find expected ',' or '}' ... while parsing a flow
1514 // mapping`. The hint must still fire.
1515 let err = Config::from_yaml(&yaml_with_prefix("{partition")).unwrap_err();
1516 let msg = format!("{err:#}");
1517 assert!(
1518 msg.contains(HINT_FRAGMENT),
1519 "unclosed brace must carry the quoting hint; got: {msg}"
1520 );
1521 }
1522
1523 #[test]
1524 fn quoted_brace_value_loads_ok() {
1525 // The fix itself, applied: a properly quoted brace value parses and
1526 // validates. This is the guard that the hint never reaches a valid
1527 // config and the success path is unchanged.
1528 let cfg = Config::from_yaml(&yaml_with_prefix("\"exports/{partition}/\""))
1529 .expect("quoted {partition} prefix must load");
1530 assert_eq!(
1531 cfg.exports[0].destination.prefix.as_deref(),
1532 Some("exports/{partition}/")
1533 );
1534 }
1535
1536 #[test]
1537 fn config_without_braces_is_untouched() {
1538 // No brace anywhere: a plain valid config still loads, and an unrelated
1539 // YAML error elsewhere must not pick up a spurious quoting hint.
1540 Config::from_yaml(&yaml_with_prefix("exports/data/"))
1541 .expect("a brace-free prefix must load");
1542 }
1543
1544 // ── line_has_unquoted_brace_value() unit coverage ──────────────────────
1545
1546 #[test]
1547 fn unquoted_brace_value_is_detected() {
1548 assert!(line_has_unquoted_brace_value(" prefix: {partition}"));
1549 assert!(line_has_unquoted_brace_value(" path: {date}/out"));
1550 assert!(line_has_unquoted_brace_value("prefix: {partition")); // unclosed
1551 }
1552
1553 #[test]
1554 fn quoted_brace_value_is_not_flagged() {
1555 // Quotes around the value hide the brace from the scanner — not a bug.
1556 assert!(!line_has_unquoted_brace_value(
1557 " prefix: \"exports/{partition}/\""
1558 ));
1559 assert!(!line_has_unquoted_brace_value(" prefix: 'data/{date}/'"));
1560 }
1561
1562 #[test]
1563 fn env_placeholder_and_plain_values_are_not_flagged() {
1564 // `${VAR}` placeholders are resolved before the parse and are not the
1565 // footgun; plain brace-free values are obviously fine.
1566 assert!(!line_has_unquoted_brace_value(" url: ${DATABASE_URL}"));
1567 assert!(!line_has_unquoted_brace_value(" path: ./out"));
1568 assert!(!line_has_unquoted_brace_value(" # prefix: {partition}")); // comment
1569 assert!(!line_has_unquoted_brace_value(" prefix:")); // no value
1570 }
1571}
1572
1573#[cfg(test)]
1574mod sec_config_validation_regression {
1575 //! Regression edge-cases that pin the *compat boundaries* of the
1576 //! config-validation security fixes — the cases that distinguish a real
1577 //! attack from a legitimate loopback / dev-container / Docker pattern.
1578 //! These complement the RED tests in `sec_config_validation`: the RED
1579 //! tests assert the attack is rejected; these assert the fix stays narrow
1580 //! enough not to break local-dev usage (see CRITICAL COMPAT).
1581 use super::*;
1582
1583 /// A full, otherwise-valid config whose single export's `destination:`
1584 /// block is whatever the caller passes verbatim.
1585 fn yaml_with_destination(dest_block: &str) -> String {
1586 format!(
1587 "source:\n type: postgres\n url: \"postgresql://localhost/test\"\n\
1588 exports:\n - name: t\n query: \"SELECT 1\"\n format: parquet\n\
1589 {dest_block}"
1590 )
1591 }
1592
1593 // ── endpoint_host / is_loopback_host helpers ─────────────────────────────
1594
1595 #[test]
1596 fn endpoint_host_parses_forms() {
1597 assert_eq!(
1598 endpoint_host("https://attacker.example.com").as_deref(),
1599 Some("attacker.example.com")
1600 );
1601 // Port and path are stripped from the host.
1602 assert_eq!(
1603 endpoint_host("http://127.0.0.1:10000/devstoreaccount1").as_deref(),
1604 Some("127.0.0.1")
1605 );
1606 // userinfo head is dropped (host is after the last `@`).
1607 assert_eq!(
1608 endpoint_host("http://user:pass@127.0.0.1:9000").as_deref(),
1609 Some("127.0.0.1")
1610 );
1611 // Bracketed IPv6 literal keeps its address.
1612 assert_eq!(endpoint_host("http://[::1]:9000").as_deref(), Some("::1"));
1613 // Not a URL → None (treated as a non-loopback custom endpoint upstream).
1614 assert_eq!(endpoint_host("not-a-url"), None);
1615 assert_eq!(endpoint_host("://nohost"), None);
1616 }
1617
1618 #[test]
1619 fn loopback_host_classification() {
1620 for h in ["127.0.0.1", "127.0.0.53", "localhost", "::1"] {
1621 assert!(is_loopback_host(h), "{h} must be loopback");
1622 }
1623 for h in ["attacker.example.com", "evil.com", "10.0.0.1", "::2"] {
1624 assert!(!is_loopback_host(h), "{h} must be remote");
1625 }
1626 }
1627
1628 // ── V2/V12 endpoint: loopback accepted regardless of allow_anonymous ─────
1629
1630 #[test]
1631 fn loopback_endpoint_without_allow_anonymous_still_accepted() {
1632 // A loopback emulator endpoint with credentials (no allow_anonymous) is
1633 // the Minio-with-keys local-dev pattern and must stay accepted — the
1634 // exfil guard targets *remote* hosts, not localhost.
1635 let cfg = yaml_with_destination(
1636 " destination:\n type: s3\n bucket: b\n region: us-east-1\n\
1637 \x20 endpoint: http://127.0.0.1:9000\n access_key_env: AK\n secret_key_env: SK\n",
1638 );
1639 Config::from_yaml(&cfg).expect("loopback endpoint with creds must stay accepted");
1640 }
1641
1642 #[test]
1643 fn remote_https_endpoint_with_allow_anonymous_is_the_only_remote_escape() {
1644 // The documented escape hatch: an explicit anonymous (emulator) opt-in
1645 // permits a non-loopback endpoint (no credentials are sent). Without
1646 // allow_anonymous the same endpoint is rejected (covered by the RED
1647 // test); with it, accepted.
1648 let cfg = yaml_with_destination(
1649 " destination:\n type: gcs\n bucket: b\n\
1650 \x20 endpoint: https://emulator.example.com\n allow_anonymous: true\n",
1651 );
1652 Config::from_yaml(&cfg).expect("remote endpoint + allow_anonymous opt-in must be accepted");
1653 }
1654
1655 // ── V15 local path: absolute allowed (Docker mount), `..` rejected ───────
1656
1657 #[test]
1658 fn absolute_local_path_is_allowed() {
1659 // `path: /output` is a legitimate Docker volume-mount pattern
1660 // (examples/rivet.yaml) and must keep validating — only `..` climbs are
1661 // the traversal footgun.
1662 let cfg =
1663 yaml_with_destination(" destination:\n type: local\n path: /output\n");
1664 Config::from_yaml(&cfg).expect("absolute local path (Docker mount) must validate");
1665 }
1666
1667 #[test]
1668 fn dotdot_in_local_prefix_is_rejected() {
1669 // `prefix` is guarded the same as `path`.
1670 let cfg = yaml_with_destination(
1671 " destination:\n type: local\n path: ./out\n prefix: a/../b\n",
1672 );
1673 let err = Config::from_yaml(&cfg).unwrap_err();
1674 let msg = format!("{err:#}");
1675 assert!(
1676 msg.contains("prefix") && msg.contains(".."),
1677 "a '..' in the local prefix must be rejected naming prefix/..; got: {msg}"
1678 );
1679 }
1680
1681 // ── V13 TLS: explicit enforced mode + knob rejected; default-mode kept ───
1682
1683 #[test]
1684 fn tls_danger_knob_without_explicit_mode_still_accepted() {
1685 // The dev-container pattern `tls: { accept_invalid_certs: true }` against
1686 // a loopback self-signed cert (e.g. the MSSQL docker container) omits
1687 // `mode:` — there is no *explicit* mode to contradict, so it must keep
1688 // validating. The RED test rejects only the explicit-mode contradiction.
1689 let yaml = "source:\n type: mssql\n url: \"sqlserver://sa:pw@127.0.0.1:1433/db\"\n \
1690 tls:\n accept_invalid_certs: true\n\
1691 exports:\n - name: t\n query: \"SELECT 1\"\n format: parquet\n \
1692 destination:\n type: local\n path: ./out\n";
1693 Config::from_yaml(yaml)
1694 .expect("dev-container default-mode + accept_invalid_certs must stay accepted");
1695 }
1696
1697 #[test]
1698 fn tls_explicit_verify_ca_plus_invalid_hostnames_rejected() {
1699 // The hostname knob is flagged too, against any explicit enforced mode.
1700 let yaml = "source:\n type: postgres\n url: \"postgresql://localhost/test\"\n \
1701 tls:\n mode: verify-ca\n accept_invalid_hostnames: true\n\
1702 exports:\n - name: t\n query: \"SELECT 1\"\n format: parquet\n \
1703 destination:\n type: local\n path: ./out\n";
1704 let err = Config::from_yaml(yaml).unwrap_err();
1705 let msg = format!("{err:#}");
1706 assert!(
1707 msg.contains("accept_invalid_hostnames") && msg.contains("verify-ca"),
1708 "explicit verify-ca + accept_invalid_hostnames must be rejected; got: {msg}"
1709 );
1710 }
1711
1712 #[test]
1713 fn tls_explicit_disable_with_knob_is_not_flagged() {
1714 // `mode: disable` carries no verification promise to contradict, so the
1715 // danger knob is a no-op there and must not be rejected.
1716 let yaml = "source:\n type: postgres\n url: \"postgresql://localhost/test\"\n \
1717 tls:\n mode: disable\n accept_invalid_certs: true\n\
1718 exports:\n - name: t\n query: \"SELECT 1\"\n format: parquet\n \
1719 destination:\n type: local\n path: ./out\n";
1720 Config::from_yaml(yaml).expect("mode: disable + knob is a no-op and must validate");
1721 }
1722
1723 // ── V5 name: filename-safe predicate boundaries ──────────────────────────
1724
1725 #[test]
1726 fn filename_safe_name_boundaries() {
1727 for ok in ["t", "orders", "daily_events", "v2-2024", "name.with.dots"] {
1728 assert!(is_filename_safe_name(ok), "{ok:?} must be accepted");
1729 }
1730 for bad in [
1731 "",
1732 "..",
1733 "../x",
1734 "/abs",
1735 "sub/dir",
1736 "back\\slash",
1737 ".hidden",
1738 "with\u{0000}nul",
1739 ] {
1740 assert!(!is_filename_safe_name(bad), "{bad:?} must be rejected");
1741 }
1742 }
1743}
1744
1745#[cfg(test)]
1746mod sec_config_validation {
1747 //! RED security tests for config-load validation gaps (cluster:
1748 //! config-validation). Each asserts the SECURE behavior through the
1749 //! stable `Config::from_yaml` seam: a malicious config that is accepted
1750 //! today must be REJECTED (or, for warn-only knobs, surfaced as an
1751 //! error/loud warning) at config-load. These are expected to FAIL until
1752 //! the corresponding production fix lands.
1753 //!
1754 //! The pattern mirrors the existing `query_file` `..`/absolute-path guard
1755 //! in `validate_export` (see `config/tests/validation.rs`): a syntactic
1756 //! check that runs at config-validate time so `rivet check` / `rivet
1757 //! doctor` catch the problem before any connect/plan/upload step.
1758 use super::*;
1759
1760 /// A full, otherwise-valid config whose single export's `destination:`
1761 /// block is whatever the caller passes verbatim. Only the destination
1762 /// varies, so any rejection is attributable to the destination under test.
1763 fn yaml_with_destination(dest_block: &str) -> String {
1764 format!(
1765 "source:\n type: postgres\n url: \"postgresql://localhost/test\"\n\
1766 exports:\n - name: t\n query: \"SELECT 1\"\n format: parquet\n\
1767 {dest_block}"
1768 )
1769 }
1770
1771 // ── V2/V12: cloud-endpoint exfiltration + http cleartext ────────────────
1772 //
1773 // `destination.endpoint` is passed straight to the opendal S3/GCS/Azure
1774 // builder with no validation (see `src/destination/{s3,gcs,azure}.rs`),
1775 // so a committed config can silently redirect every export to an
1776 // attacker-controlled host. Two distinct gaps:
1777 // V2 — a custom *non-loopback* endpoint (data exfiltration target).
1778 // V12 — an `http://` (plaintext) endpoint (credentials + data on the
1779 // wire in cleartext).
1780 // The secure behavior is to reject (or require explicit opt-in) at
1781 // config-load. Loopback/emulator endpoints (Minio/Azurite/fake-gcs on
1782 // 127.0.0.1) MUST stay accepted — that path is exercised by the existing
1783 // `gcs_allow_anonymous_parses` test and the guard test below.
1784
1785 #[test]
1786 fn sec_s3_custom_endpoint_rejected() {
1787 // SEC-RED V2: a non-loopback custom S3 endpoint is an exfiltration
1788 // target — every part upload goes to attacker.example.com. Must be
1789 // rejected (or require explicit opt-in) at config-load. Accepted today.
1790 let cfg = yaml_with_destination(
1791 " destination:\n type: s3\n bucket: my-bucket\n region: us-east-1\n\
1792 \x20 endpoint: https://attacker.example.com\n",
1793 );
1794 let res = Config::from_yaml(&cfg);
1795 assert!(
1796 res.is_err(),
1797 "a non-loopback custom S3 endpoint (https://attacker.example.com) must be \
1798 rejected at config-load (data-exfiltration target); got Ok"
1799 );
1800 let msg = format!("{:#}", res.unwrap_err());
1801 assert!(
1802 msg.contains("endpoint"),
1803 "rejection must name the offending 'endpoint' field; got: {msg}"
1804 );
1805 }
1806
1807 #[test]
1808 fn sec_http_endpoint_rejected() {
1809 // SEC-RED V12: a plaintext http:// endpoint to a *remote* host sends
1810 // credentials and exported rows over the wire in cleartext. Must be
1811 // rejected (or require explicit opt-in) at config-load. Accepted today.
1812 // Use a non-loopback host so this is distinct from the Minio/Azurite
1813 // loopback emulator case (guarded below).
1814 let cfg = yaml_with_destination(
1815 " destination:\n type: s3\n bucket: my-bucket\n region: us-east-1\n\
1816 \x20 endpoint: http://evil.com\n",
1817 );
1818 let res = Config::from_yaml(&cfg);
1819 assert!(
1820 res.is_err(),
1821 "a plaintext http:// endpoint to a remote host (http://evil.com) must be \
1822 rejected at config-load (cleartext credentials + data); got Ok"
1823 );
1824 let msg = format!("{:#}", res.unwrap_err());
1825 assert!(
1826 msg.contains("endpoint") || msg.to_lowercase().contains("http"),
1827 "rejection must name the endpoint / cleartext problem; got: {msg}"
1828 );
1829 }
1830
1831 #[test]
1832 fn sec_loopback_endpoint_still_accepted_guard() {
1833 // SEC-RED V2/V12 (guard): a loopback emulator endpoint
1834 // (`http://127.0.0.1:9000` Minio, with allow_anonymous) is the
1835 // legitimate local-dev path and MUST stay accepted after the fix.
1836 // This pins that the endpoint rejection targets *remote* hosts, not
1837 // localhost — otherwise the fix breaks every Minio/Azurite/fake-gcs
1838 // integration test (see `gcs_allow_anonymous_parses`).
1839 let cfg = yaml_with_destination(
1840 " destination:\n type: s3\n bucket: my-bucket\n region: us-east-1\n\
1841 \x20 endpoint: http://127.0.0.1:9000\n allow_anonymous: true\n",
1842 );
1843 Config::from_yaml(&cfg)
1844 .expect("a loopback emulator endpoint with allow_anonymous must stay accepted");
1845 }
1846
1847 // ── V5: export `name` path traversal ────────────────────────────────────
1848 //
1849 // `ExportConfig.name` is a free-form `String` keyed into state tracking,
1850 // file logs, and (via the destination layout) output paths — yet it is
1851 // never validated. A name like `../../../etc/x`, an absolute `/abs/x`, a
1852 // bare slash, or an embedded NUL can escape the intended output tree.
1853 // Mirror the `query_file` `..`/absolute guard: reject at config-load.
1854
1855 #[test]
1856 fn sec_export_name_traversal_rejected() {
1857 // SEC-RED V5: a traversal / absolute / slash / NUL export name escapes
1858 // the output tree (and corrupts name-keyed state). Must be rejected at
1859 // config-load. Accepted today.
1860 for bad in ["../../../etc/x", "/abs/x", "sub/dir", "with\u{0000}nul"] {
1861 // `name:` is JSON-encoded so embedded slashes / NULs survive the
1862 // YAML parse verbatim and reach validation.
1863 let name_yaml = serde_json::to_string(bad).expect("encode name");
1864 let cfg = format!(
1865 "source:\n type: postgres\n url: \"postgresql://localhost/test\"\n\
1866 exports:\n - name: {name_yaml}\n query: \"SELECT 1\"\n format: parquet\n\
1867 \x20 destination:\n type: local\n path: ./out\n"
1868 );
1869 let res = Config::from_yaml(&cfg);
1870 assert!(
1871 res.is_err(),
1872 "export name {bad:?} (traversal/absolute/slash/NUL) must be rejected at \
1873 config-load; got Ok"
1874 );
1875 let msg = format!("{:#}", res.unwrap_err());
1876 assert!(
1877 msg.contains("name"),
1878 "rejection of name {bad:?} must name the offending 'name' field; got: {msg}"
1879 );
1880 }
1881 }
1882
1883 #[test]
1884 fn sec_export_name_normal_still_accepted_guard() {
1885 // SEC-RED V5 (guard): a plain, well-formed export name must keep
1886 // loading after the fix. Pins that the traversal check is narrow.
1887 let cfg = yaml_with_destination(" destination:\n type: local\n path: ./out\n");
1888 Config::from_yaml(&cfg).expect("a normal export name ('t') must stay accepted");
1889 }
1890
1891 // ── V15: local destination `path` traversal ─────────────────────────────
1892 //
1893 // `destination.path` for a `type: local` export is written verbatim to the
1894 // filesystem. A relative `../../../../tmp/x` or absolute path lets a
1895 // committed config write outside the intended output directory. Must be
1896 // rejected (or at minimum loudly surfaced) at config-load. Accepted today.
1897
1898 #[test]
1899 fn sec_local_dest_path_traversal_rejected() {
1900 // SEC-RED V15: a traversal local-destination path writes outside the
1901 // intended output tree. Must be rejected at config-load. Accepted today.
1902 let cfg = yaml_with_destination(
1903 " destination:\n type: local\n path: ../../../../tmp/x\n",
1904 );
1905 let res = Config::from_yaml(&cfg);
1906 assert!(
1907 res.is_err(),
1908 "a local destination path containing '..' (../../../../tmp/x) must be rejected \
1909 at config-load (writes outside the output tree); got Ok"
1910 );
1911 let msg = format!("{:#}", res.unwrap_err());
1912 assert!(
1913 msg.contains("path") || msg.contains(".."),
1914 "rejection must name the offending 'path' / traversal; got: {msg}"
1915 );
1916 }
1917
1918 // ── V13: dangerous TLS cert-knob combination ─────────────────────────────
1919 //
1920 // `tls: { mode: verify-full, accept_invalid_certs: true }` silently
1921 // *downgrades* the strongest mode to "accept any cert" — `verify-full`
1922 // promises chain + hostname verification, but the danger knob disables
1923 // chain verification (see `src/source/tls.rs::build_native_tls`). The
1924 // comment at `src/source/tls.rs:55-56` claims "Each one emits a warning at
1925 // config-time (see `Config::validate`)" — but `Config::validate` emits no
1926 // such warning today. The secure behavior is a LOUD error (or surfaced
1927 // warning) at config-load. No `Err`/warning is produced today, so this is
1928 // RED.
1929
1930 #[test]
1931 fn sec_accept_invalid_certs_warns() {
1932 // SEC-RED V13: verify-full + accept_invalid_certs: true is a silent
1933 // security downgrade that contradicts the chosen mode. It must be
1934 // loudly surfaced at config-load. The only stable secure seam is an
1935 // `Err` from `Config::from_yaml` (validate returns Ok today, and there
1936 // is no captured-warning seam exposed from here — see notes). Asserting
1937 // `Err` is the strongest secure assertion and is RED against current
1938 // code.
1939 let cfg = yaml_with_destination(" destination:\n type: local\n path: ./out\n");
1940 // Splice the TLS block into the source rather than the destination so
1941 // the rest of the config stays valid.
1942 let cfg = cfg.replace(
1943 " url: \"postgresql://localhost/test\"\n",
1944 " url: \"postgresql://localhost/test\"\n tls:\n mode: verify-full\n accept_invalid_certs: true\n",
1945 );
1946 let res = Config::from_yaml(&cfg);
1947 assert!(
1948 res.is_err(),
1949 "tls mode: verify-full with accept_invalid_certs: true is a silent security \
1950 downgrade and must be loudly surfaced (error) at config-load; got Ok"
1951 );
1952 let msg = format!("{:#}", res.unwrap_err());
1953 assert!(
1954 msg.contains("accept_invalid_certs") || msg.to_lowercase().contains("verify"),
1955 "the surfaced error must name the dangerous knob / mode contradiction; got: {msg}"
1956 );
1957 }
1958}