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