rivet/config/
source.rs

1//! Source-database connection config: URL/structured fields, TLS, environment hints.
2
3use percent_encoding::{AsciiSet, NON_ALPHANUMERIC, utf8_percent_encode};
4use schemars::JsonSchema;
5use serde::{Deserialize, Serialize};
6
7use super::resolve::resolve_env_vars;
8use crate::tuning::{TuningConfig, TuningProfile};
9
10/// Chars to percent-encode in a URL userinfo (user / password) component: encode
11/// everything except the RFC 3986 unreserved set (ALPHA / DIGIT / `- . _ ~`), so
12/// no credential byte can be mistaken for a URL delimiter. Over-encoding is
13/// harmless — the driver percent-decodes it back.
14const USERINFO: &AsciiSet = &NON_ALPHANUMERIC
15    .remove(b'-')
16    .remove(b'.')
17    .remove(b'_')
18    .remove(b'~');
19
20#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
21#[serde(deny_unknown_fields)]
22pub struct SourceConfig {
23    #[serde(rename = "type")]
24    pub source_type: SourceType,
25
26    pub url: Option<String>,
27    pub url_env: Option<String>,
28    pub url_file: Option<String>,
29
30    pub host: Option<String>,
31    pub port: Option<u16>,
32    pub user: Option<String>,
33    pub password: Option<String>,
34    pub password_env: Option<String>,
35    pub database: Option<String>,
36
37    /// Operational profile of the source database.
38    ///
39    /// Selects the **default** tuning profile when none is explicitly set in
40    /// `source.tuning.profile` or `export.tuning.profile`:
41    ///
42    /// | `environment`           | default profile |
43    /// |-------------------------|------------------|
44    /// | `production` (default)  | `balanced` (50 ms throttle, 10 k batch, retries) |
45    /// | `replica`               | `balanced` |
46    /// | `local`                 | `fast` (no throttle, 50 k batch — saves ~30% wall on localhost) |
47    ///
48    /// Explicit `tuning.profile:` always wins over this hint.
49    #[serde(default)]
50    pub environment: Option<SourceEnvironment>,
51
52    #[serde(default)]
53    pub tuning: Option<TuningConfig>,
54
55    /// Transport security settings (ADR: SecOps). When absent, Rivet connects
56    /// without TLS — a warning is emitted so operators are aware. See [`TlsConfig`].
57    #[serde(default)]
58    pub tls: Option<TlsConfig>,
59
60    /// MongoDB-specific read options (`source.mongo:`). Honoured only when
61    /// `type: mongo`; ignored by the SQL engines. See [`MongoConfig`].
62    #[serde(default)]
63    pub mongo: Option<MongoConfig>,
64}
65
66/// MongoDB read-path knobs. All three address completeness/fidelity of a `full`
67/// collection export that the SQL engines get for free from SQL semantics.
68#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
69#[serde(deny_unknown_fields)]
70pub struct MongoConfig {
71    /// JSON rendering of the `document` column. `relaxed` (default) keeps common
72    /// scalars native (`42`, `"x"`); `canonical` wraps every number
73    /// (`{"$numberLong":"…"}`) so Int64/Double round-trip losslessly through a
74    /// JSON-number parser that would otherwise clamp values beyond 2^53.
75    #[serde(default)]
76    pub json: MongoJsonMode,
77
78    /// Read concern for the collection scan. `snapshot` gives a **point-in-time
79    /// consistent** full export (no doc missed/double-read under concurrent
80    /// writes) — requires MongoDB 5.0+ on a replica set; a standalone rejects it.
81    /// Default (`server`) uses the server's default read concern.
82    #[serde(default)]
83    pub read_concern: MongoReadConcern,
84
85    /// Keep the scan cursor alive past the server's idle timeout (default 10 min)
86    /// so a slow destination cannot let the server reap the cursor mid-scan and
87    /// silently drop the tail of a large collection. Default: `true`.
88    #[serde(default = "default_true")]
89    pub no_cursor_timeout: bool,
90
91    /// When set, read the collection with **keyset (seek) pagination** on `_id`
92    /// instead of one long-held cursor: each page is a bounded
93    /// `find({_id: {$gt: last}}).sort({_id: 1}).limit(page_size)` — an indexed
94    /// range scan that becomes one output part file. Bounds longest-query time
95    /// (no 35-minute cursor to hit a timeout / snapshot window) and is the base
96    /// for parallel `_id`-range reads. Requires an **ObjectId** `_id` (the
97    /// default); a non-ObjectId key errors with a clear message. Unset ⇒ the
98    /// single-cursor full scan.
99    #[serde(default)]
100    pub page_size: Option<usize>,
101
102    /// With keyset paging (`page_size`), persist the last committed `_id` and
103    /// **resume** from it next run — a crashed export continues where it left
104    /// off, and a re-run captures only documents inserted since (ObjectId `_id`
105    /// is time-ordered). Default `false` re-reads the whole collection each run
106    /// (plain `mode: full` semantics). No effect without `page_size`.
107    #[serde(default)]
108    pub resume: bool,
109}
110
111impl Default for MongoConfig {
112    fn default() -> Self {
113        Self {
114            json: MongoJsonMode::default(),
115            read_concern: MongoReadConcern::default(),
116            no_cursor_timeout: true,
117            page_size: None,
118            resume: false,
119        }
120    }
121}
122
123fn default_true() -> bool {
124    true
125}
126
127/// Extended-JSON rendering mode for the `document` column.
128#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone, Copy, PartialEq, Eq, Default)]
129#[serde(rename_all = "lowercase")]
130pub enum MongoJsonMode {
131    /// Native scalars where possible; `{"$oid":…}`/`{"$date":…}` only for exotic
132    /// BSON. Friendliest for `PARSE_JSON`, but Int64 renders as a bare number.
133    #[default]
134    Relaxed,
135    /// Every value type-tagged (`{"$numberLong":…}`, `{"$numberInt":…}`) — a
136    /// lossless, bulkier form that survives a double-based JSON-number parser.
137    Canonical,
138}
139
140/// Read concern applied to the collection scan.
141#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone, Copy, PartialEq, Eq, Default)]
142#[serde(rename_all = "lowercase")]
143pub enum MongoReadConcern {
144    /// The server's default read concern — no point-in-time guarantee across the
145    /// scan (a plain `find`, like today).
146    #[default]
147    Server,
148    /// `snapshot` — a consistent point-in-time read for the whole cursor
149    /// (MongoDB 5.0+ replica set only).
150    Snapshot,
151}
152
153/// Operational environment of the source database — drives the default tuning
154/// profile when none is explicitly set. Opt-in: existing configs without
155/// `environment:` continue to use `balanced` as today.
156#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone, Copy, PartialEq, Eq)]
157#[serde(rename_all = "lowercase")]
158pub enum SourceEnvironment {
159    /// Localhost / Docker compose / read-only container — no throttle by default
160    /// (compiles to `fast` profile defaults). Use when DB load is not a concern.
161    Local,
162    /// Read replica — `balanced` default. Same throttle as production, but free
163    /// to dial up `tuning.batch_size`.
164    Replica,
165    /// Live production primary — `balanced` default. Bias toward source-safety.
166    Production,
167}
168
169impl SourceEnvironment {
170    /// Default tuning profile selected by this environment when the user has
171    /// not set `tuning.profile:` explicitly.
172    pub fn default_profile(self) -> TuningProfile {
173        match self {
174            SourceEnvironment::Local => TuningProfile::Fast,
175            SourceEnvironment::Replica | SourceEnvironment::Production => TuningProfile::Balanced,
176        }
177    }
178}
179
180/// Transport security for the source database connection.
181///
182/// Credentials and exported data cross the wire on every connection; without TLS
183/// they are visible to anyone on the network path (cloud inter-VPC, cross-AZ, or
184/// a compromised upstream). The default for all new connections is
185/// [`TlsMode::Require`] when `tls:` is present; setting `tls: { mode: disable }`
186/// is explicit opt-out.
187///
188/// ```yaml
189/// source:
190///   type: postgres
191///   url_env: DATABASE_URL
192///   tls:
193///     mode: verify-full
194///     ca_file: /etc/ssl/certs/rds-ca-2019-root.pem
195/// ```
196#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone, Default)]
197#[serde(deny_unknown_fields)]
198pub struct TlsConfig {
199    /// Enforcement level. See [`TlsMode`].
200    #[serde(default)]
201    pub mode: TlsMode,
202    /// PEM-encoded CA certificate to trust for server verification. Required
203    /// for [`TlsMode::VerifyCa`] and [`TlsMode::VerifyFull`] against a private CA.
204    pub ca_file: Option<String>,
205    /// Accept certificates not chained to a trusted CA. Dangerous — disables
206    /// server authentication — and only honored when explicitly `true`.
207    #[serde(default)]
208    pub accept_invalid_certs: bool,
209    /// Accept certificates whose subjectAltName does not match the connection
210    /// hostname. Dangerous — disables hostname verification.
211    #[serde(default)]
212    pub accept_invalid_hostnames: bool,
213}
214
215/// TLS enforcement mode, mirroring libpq's `sslmode` semantics where possible.
216#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone, Copy, PartialEq, Eq, Default)]
217#[serde(rename_all = "kebab-case")]
218pub enum TlsMode {
219    /// Plaintext. Use only inside trusted networks (loopback, cgroup-private).
220    Disable,
221    /// Require a TLS handshake; accept the server certificate without verifying
222    /// issuer or hostname. Protects against passive sniffing, not MITM.
223    Require,
224    /// TLS + verify certificate chains to the configured / system trust store.
225    /// Does not check hostname (useful for IP-addressed or internal names).
226    VerifyCa,
227    /// TLS + verify chain **and** hostname against the server cert's SAN/CN.
228    /// Recommended default for production.
229    #[default]
230    VerifyFull,
231}
232
233impl TlsMode {
234    pub fn is_enforced(self) -> bool {
235        !matches!(self, TlsMode::Disable)
236    }
237}
238
239impl SourceConfig {
240    /// Return a copy of this config with **all plaintext credential material stripped**,
241    /// safe to embed in a persisted [`crate::plan::PlanArtifact`] (ADR-0005 PA9).
242    ///
243    /// Redaction rules:
244    /// - `password` → always `None` (plaintext password never leaves the process).
245    /// - `url` containing `user[:password]@` → userinfo segment replaced with `"REDACTED"`.
246    /// - `url_env`, `url_file`, `password_env` — kept (env var **names** and file paths
247    ///   are references, not secrets; `apply` needs them to re-resolve credentials).
248    /// - `host`, `port`, `user`, `database` — kept (structured connection metadata).
249    ///
250    /// If a plaintext `password` or `url` is redacted, callers should surface a warning
251    /// to the operator so env/file-based auth is available at apply time.
252    pub fn redact_for_artifact(&self) -> (Self, bool) {
253        let mut out = self.clone();
254        let mut redacted = false;
255
256        if out.password.is_some() {
257            out.password = None;
258            redacted = true;
259        }
260
261        if let Some(ref raw) = out.url
262            && let Some((userinfo_end, scheme_end)) = find_userinfo(raw)
263        {
264            let mut s = String::with_capacity(raw.len());
265            s.push_str(&raw[..scheme_end]); // "postgresql://"
266            s.push_str("REDACTED");
267            s.push_str(&raw[userinfo_end..]); // "@host:port/db…"
268            out.url = Some(s);
269            redacted = true;
270        }
271
272        (out, redacted)
273    }
274
275    pub(crate) fn has_structured_fields(&self) -> bool {
276        self.host.is_some()
277            || self.user.is_some()
278            || self.database.is_some()
279            || self.password.is_some()
280            || self.password_env.is_some()
281    }
282
283    pub(crate) fn has_url_fields(&self) -> bool {
284        self.url.is_some() || self.url_env.is_some() || self.url_file.is_some()
285    }
286
287    fn build_url_from_fields(&self) -> crate::error::Result<String> {
288        // First-user-friendly errors: name the missing field, suggest a
289        // concrete value, and remind the operator that `url_env` is the
290        // alternative path so they don't bounce.  See
291        // `docs/getting-started.md` for the full onboarding flow.
292        let host = self.host.as_deref().ok_or_else(|| {
293            anyhow::anyhow!(
294                "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`."
295            )
296        })?;
297        let user = self.user.as_deref().ok_or_else(|| {
298            anyhow::anyhow!(
299                "source: structured config is missing 'user'.\n  Hint: add `user: <username>` under `source:` in rivet.yaml."
300            )
301        })?;
302        let database = self.database.as_deref().ok_or_else(|| {
303            anyhow::anyhow!(
304                "source: structured config is missing 'database'.\n  Hint: add `database: <dbname>` under `source:` in rivet.yaml."
305            )
306        })?;
307
308        // SecOps: keep the plaintext password inside a `Zeroizing<String>` until it
309        // is spliced into the final URL, so the standalone password buffer is
310        // wiped on drop (the final URL still lives as a plain String but is
311        // shorter-lived and dropped by the driver constructor).
312        let password: zeroize::Zeroizing<String> =
313            zeroize::Zeroizing::new(match (&self.password, &self.password_env) {
314                (Some(_), Some(_)) => {
315                    anyhow::bail!("source: specify 'password' or 'password_env', not both");
316                }
317                (Some(p), None) => {
318                    static WARNED: std::sync::Once = std::sync::Once::new();
319                    WARNED.call_once(|| {
320                        log::warn!(
321                            "source config contains plaintext password -- consider using password_env"
322                        );
323                    });
324                    resolve_env_vars(p)?
325                }
326                (None, Some(env)) => std::env::var(env).map_err(|_| {
327                    anyhow::anyhow!(
328                        "source: env var '{0}' is not set (referenced by password_env).\n  Hint: export the value before running, e.g.\n      export {0}='your-database-password'",
329                        env
330                    )
331                })?,
332                (None, None) => String::new(),
333            });
334
335        let default_port = match self.source_type {
336            SourceType::Postgres => 5432,
337            SourceType::Mysql => 3306,
338            SourceType::Mssql => 1433,
339            SourceType::Mongo => 27017,
340        };
341        let port = self.port.unwrap_or(default_port);
342
343        let scheme = match self.source_type {
344            SourceType::Postgres => "postgresql",
345            SourceType::Mysql => "mysql",
346            SourceType::Mssql => "sqlserver",
347            SourceType::Mongo => "mongodb",
348        };
349
350        // Percent-encode the userinfo so a credential containing `/ @ : ? #` can't
351        // make the URL ambiguous (breaking the driver) or defeat redaction.
352        let user_enc = utf8_percent_encode(user, USERINFO);
353        if password.is_empty() {
354            Ok(format!(
355                "{}://{}@{}:{}/{}",
356                scheme, user_enc, host, port, database
357            ))
358        } else {
359            let pw_enc = utf8_percent_encode(password.as_str(), USERINFO);
360            Ok(format!(
361                "{}://{}:{}@{}:{}/{}",
362                scheme, user_enc, pw_enc, host, port, database
363            ))
364        }
365    }
366
367    pub fn resolve_url(&self) -> crate::error::Result<String> {
368        if self.has_url_fields() && self.has_structured_fields() {
369            anyhow::bail!(
370                "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."
371            );
372        }
373
374        if self.has_structured_fields() {
375            return self.build_url_from_fields();
376        }
377
378        // Capture *where* the URL came from so the password warning below
379        // can be specific: scolding an operator who already used
380        // `url_env:` (the recommendation!) for "considering url_env" is
381        // misleading and trains them to tune out our warnings.
382        //
383        // The `EnvVar(&str)` / `File(&str)` payloads are retained for
384        // future use (e.g. mentioning the env-var name in a richer
385        // diagnostic later) — `#[allow(dead_code)]` keeps clippy quiet
386        // while we keep the slot open. Renaming the variants to unit
387        // would lose the documentation that "this came from <name>".
388        #[allow(dead_code)]
389        enum UrlSource<'a> {
390            InlineYaml,
391            EnvVar(&'a str),
392            File(&'a str),
393        }
394        let (raw, source) = match (&self.url, &self.url_env, &self.url_file) {
395            (Some(u), None, None) => (u.clone(), UrlSource::InlineYaml),
396            (None, Some(env), None) => (
397                std::env::var(env).map_err(|_| {
398                    anyhow::anyhow!(
399                        "source: env var '{0}' is not set (referenced by url_env).\n  Hint: export the value before running, e.g.\n      export {0}='postgresql://user:pass@host:5432/dbname'\n  Or change `url_env: {0}` in your config to a different env var name.",
400                        env
401                    )
402                })?,
403                UrlSource::EnvVar(env),
404            ),
405            (None, None, Some(file)) => (
406                std::fs::read_to_string(file)
407                    .map_err(|e| {
408                        anyhow::anyhow!(
409                            "source: cannot read url_file '{}': {}.\n  Hint: ensure the file exists and is readable; the file should contain only the URL on a single line.",
410                            file,
411                            e
412                        )
413                    })?
414                    .trim()
415                    .to_string(),
416                UrlSource::File(file),
417            ),
418            _ => anyhow::bail!(
419                "source: configure exactly one connection method:\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:`)"
420            ),
421        };
422
423        let resolved = resolve_env_vars(&raw)?;
424
425        if resolved.contains('@')
426            && resolved.contains(':')
427            && let Some(userinfo) = resolved.split('@').next()
428            && userinfo.contains(':')
429            && !userinfo.ends_with(':')
430        {
431            // `resolve_url` is called from many places per run (plan build,
432            // doctor, every export, every chunk worker). Fire each variant
433            // of this warning exactly once per process so operators see
434            // one clean nudge, not 3-4 stacked copies in stderr.
435            //
436            // Only the InlineYaml case is a real misconfiguration to flag:
437            // the password is sitting in a committed file. EnvVar / File
438            // sources are explicitly the recommended forms — scolding an
439            // operator who already uses them for "considering url_env"
440            // would be a false alarm.
441            match source {
442                UrlSource::InlineYaml => {
443                    static WARNED: std::sync::Once = std::sync::Once::new();
444                    WARNED.call_once(|| {
445                        log::warn!(
446                            "source: inline `url:` in YAML contains a plaintext password — \
447                             move it to `url_env: DATABASE_URL` (or `url_file:`) to keep \
448                             credentials out of committed configs"
449                        );
450                    });
451                }
452                UrlSource::EnvVar(_) | UrlSource::File(_) => {
453                    // The recommended forms — no warning. Operator hygiene
454                    // for shell history / file permissions is out of scope.
455                }
456            }
457        }
458
459        Ok(resolved)
460    }
461}
462
463#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone, Copy, PartialEq, Eq)]
464#[serde(rename_all = "lowercase")]
465pub enum SourceType {
466    Postgres,
467    Mysql,
468    Mssql,
469    /// Document store. Unlike the three SQL engines, MongoDB has no SQL, no
470    /// fixed per-collection schema, and no `information_schema` — so the
471    /// SQL-shaped read seam (chunked/keyset planning, incremental predicate
472    /// building, catalog introspection) does not apply. The OSS source is the
473    /// JSON-blob model: each document exports as `_id` + a `document` JSON
474    /// column, typing punted downstream. Change streams (CDC) map onto the
475    /// canonical `ChangeStream` seam separately.
476    Mongo,
477}
478
479impl SourceType {
480    /// Whether this engine speaks SQL (the relational read seam: `SELECT`
481    /// queries, `information_schema` introspection, chunked/keyset/incremental
482    /// SQL builders). `false` for document stores like MongoDB, whose adapter
483    /// reads via the driver's native query API instead. Used at the match sites
484    /// that would otherwise have to special-case every non-SQL engine.
485    pub fn is_sql(self) -> bool {
486        !matches!(self, SourceType::Mongo)
487    }
488}
489
490/// Locate `user[:password]@` userinfo inside a standard URL.
491///
492/// Returns `(userinfo_end_index, scheme_end_index)` where:
493/// - `scheme_end_index` points right after `"://"` (start of userinfo)
494/// - `userinfo_end_index` points at the `@` separator (exclusive of `@`)
495///
496/// Returns `None` if the URL has no userinfo segment.
497fn find_userinfo(raw: &str) -> Option<(usize, usize)> {
498    let scheme = raw.find("://")? + 3;
499    let rest = &raw[scheme..];
500    // The userinfo `@` lives in the AUTHORITY, before the first `/ ? #` (a later
501    // `@` belongs to the path/query — e.g. `?filter=a@b` — and must NOT be treated
502    // as userinfo, or a credential-free URL is falsely redacted). Within the
503    // authority, take the LAST `@` (a password may contain `@`). This is correct
504    // for a well-formed URL; a RAW `/` in the password would make the URL
505    // ambiguous — which is why `build_url_from_fields` percent-encodes the userinfo
506    // (a `/` becomes `%2F`), so a Rivet-constructed URL never hits that case.
507    let authority_end = rest.find(['/', '?', '#']).unwrap_or(rest.len());
508    let at = rest[..authority_end].rfind('@')?;
509    Some((scheme + at, scheme))
510}
511
512#[cfg(test)]
513mod tests {
514    use super::*;
515
516    // ── TlsMode::is_enforced ────────────────────────────────────────────────
517
518    #[test]
519    fn tls_mode_disable_not_enforced() {
520        assert!(!TlsMode::Disable.is_enforced());
521    }
522
523    #[test]
524    fn tls_mode_require_is_enforced() {
525        assert!(TlsMode::Require.is_enforced());
526        assert!(TlsMode::VerifyCa.is_enforced());
527        assert!(TlsMode::VerifyFull.is_enforced());
528    }
529
530    // ── SourceConfig::redact_for_artifact ───────────────────────────────────
531
532    fn make_source(source_type: SourceType) -> SourceConfig {
533        SourceConfig {
534            source_type,
535            url: None,
536            url_env: None,
537            url_file: None,
538            host: None,
539            port: None,
540            user: None,
541            password: None,
542            password_env: None,
543            database: None,
544            environment: None,
545            tuning: None,
546            tls: None,
547            mongo: None,
548        }
549    }
550
551    #[test]
552    fn redact_plaintext_password() {
553        let mut src = make_source(SourceType::Postgres);
554        src.password = Some("s3cr3t".into());
555        let (redacted, flag) = src.redact_for_artifact();
556        assert!(flag, "redaction should be flagged");
557        assert!(
558            redacted.password.is_none(),
559            "plaintext password must be stripped"
560        );
561    }
562
563    #[test]
564    fn redact_url_with_password() {
565        let mut src = make_source(SourceType::Postgres);
566        src.url = Some("postgresql://user:hunter2@db.example.com:5432/app".into());
567        let (redacted, flag) = src.redact_for_artifact();
568        assert!(flag, "URL redaction flagged");
569        let url = redacted.url.unwrap();
570        assert!(!url.contains("hunter2"), "password must not appear: {url}");
571        assert!(url.contains("REDACTED"), "placeholder must appear: {url}");
572        assert!(url.contains("@db.example.com"), "host retained: {url}");
573    }
574
575    #[test]
576    fn redact_url_without_at_sign_not_flagged() {
577        let mut src = make_source(SourceType::Postgres);
578        src.url = Some("postgresql://db.example.com:5432/app".into());
579        let (_, flag) = src.redact_for_artifact();
580        assert!(!flag, "URL with no userinfo must not be flagged");
581    }
582
583    #[test]
584    fn redact_url_with_user_but_no_password_is_flagged() {
585        let mut src = make_source(SourceType::Postgres);
586        src.url = Some("postgresql://user@db.example.com:5432/app".into());
587        let (redacted, flag) = src.redact_for_artifact();
588        assert!(flag, "bare user@ is still userinfo and gets redacted");
589        let url = redacted.url.unwrap();
590        assert!(url.contains("REDACTED"), "userinfo replaced: {url}");
591        assert!(!url.contains("user@"), "bare username removed: {url}");
592    }
593
594    #[test]
595    fn redact_env_var_reference_kept_intact() {
596        let mut src = make_source(SourceType::Mysql);
597        src.url_env = Some("DB_URL".into());
598        src.password_env = Some("DB_PASS".into());
599        let (redacted, flag) = src.redact_for_artifact();
600        assert!(!flag, "env var references are not secrets");
601        assert_eq!(redacted.url_env.as_deref(), Some("DB_URL"));
602        assert_eq!(redacted.password_env.as_deref(), Some("DB_PASS"));
603    }
604
605    #[test]
606    fn redact_mysql_url_with_password() {
607        let mut src = make_source(SourceType::Mysql);
608        src.url = Some("mysql://root:pass@127.0.0.1:3306/mydb".into());
609        let (redacted, flag) = src.redact_for_artifact();
610        assert!(flag);
611        let url = redacted.url.unwrap();
612        assert!(url.contains("REDACTED"), "{url}");
613        assert!(!url.contains("pass"), "{url}");
614    }
615
616    // ── SourceConfig::resolve_url (structured fields) ───────────────────────
617
618    #[test]
619    fn resolve_url_from_structured_fields_postgres() {
620        let mut src = make_source(SourceType::Postgres);
621        src.host = Some("pg.internal".into());
622        src.user = Some("alice".into());
623        src.database = Some("warehouse".into());
624        src.port = Some(5433);
625        let url = src.resolve_url().unwrap();
626        assert_eq!(url, "postgresql://alice@pg.internal:5433/warehouse");
627    }
628
629    #[test]
630    fn resolve_url_from_structured_fields_defaults_port() {
631        let mut src = make_source(SourceType::Mysql);
632        src.host = Some("my.internal".into());
633        src.user = Some("bob".into());
634        src.database = Some("orders".into());
635        let url = src.resolve_url().unwrap();
636        assert_eq!(url, "mysql://bob@my.internal:3306/orders");
637    }
638
639    #[test]
640    fn resolve_url_direct_url_passthrough() {
641        let mut src = make_source(SourceType::Postgres);
642        src.url = Some("postgresql://carol@pg.example.com:5432/db".into());
643        let url = src.resolve_url().unwrap();
644        assert_eq!(url, "postgresql://carol@pg.example.com:5432/db");
645    }
646
647    #[test]
648    fn resolve_url_rejects_mixed_url_and_structured() {
649        let mut src = make_source(SourceType::Postgres);
650        src.url = Some("postgresql://carol@pg.example.com/db".into());
651        src.host = Some("other".into());
652        let err = src.resolve_url().unwrap_err();
653        let msg = format!("{err:#}");
654        assert!(
655            msg.contains("URL-based") || msg.contains("structured"),
656            "{msg}"
657        );
658    }
659
660    #[test]
661    fn resolve_url_rejects_missing_host() {
662        let mut src = make_source(SourceType::Postgres);
663        src.user = Some("alice".into());
664        src.database = Some("warehouse".into());
665        let err = src.resolve_url().unwrap_err();
666        let msg = format!("{err:#}");
667        assert!(msg.contains("host"), "{msg}");
668    }
669
670    // ── find_userinfo ────────────────────────────────────────────────────────
671
672    #[test]
673    fn find_userinfo_detects_password_in_url() {
674        let url = "postgresql://user:pass@host/db";
675        let result = find_userinfo(url);
676        assert!(result.is_some(), "should detect user:pass@");
677    }
678
679    #[test]
680    fn find_userinfo_no_password_no_at_returns_none() {
681        assert!(find_userinfo("postgresql://host/db").is_none());
682    }
683
684    #[test]
685    fn find_userinfo_user_only_at_sign_matches() {
686        let url = "postgresql://user@host/db";
687        assert!(find_userinfo(url).is_some(), "bare user@ should match");
688    }
689
690    #[test]
691    fn find_userinfo_no_at_sign_returns_none() {
692        assert!(find_userinfo("postgresql://db.example.com:5432/app").is_none());
693    }
694
695    // ── SEC-RED: embedded `@` in password must not leak to plan artifact ──────
696
697    #[test]
698    fn sec_artifact_redaction_password_with_at() {
699        // SEC-RED V7: find_userinfo (used by redact_for_artifact when building
700        // the persisted plan JSON) splits userinfo at the FIRST `@` via
701        // `rest.find('@')`, leaking the password tail after an embedded `@`.
702        // For `postgresql://rivet:p@ssw0rd@host/db` the first `@` sits right
703        // after `p`, so `userinfo_end` lands before `ssw0rd@host/db` and the
704        // rewrite emits `postgresql://REDACTED@ssw0rd@host/db` — the password
705        // tail `ssw0rd` round-trips into the artifact. The terminator must be
706        // the LAST `@` before the path (rfind semantics, as already used by
707        // redact_pg_url in state/mod.rs:564).
708        let mut src = make_source(SourceType::Postgres);
709        src.url = Some("postgresql://rivet:p@ssw0rd@db.example.com:5432/orders".into());
710        let (redacted, flag) = src.redact_for_artifact();
711        assert!(flag, "URL with userinfo must be flagged as redacted");
712        let url = redacted.url.expect("url retained after redaction");
713        assert!(
714            !url.contains("ssw0rd"),
715            "password tail after embedded @ must not leak into artifact: {url}"
716        );
717        assert!(
718            !url.contains("p@ssw0rd"),
719            "full password must not leak into artifact: {url}"
720        );
721        assert!(url.contains("REDACTED"), "placeholder must appear: {url}");
722        assert!(
723            url.contains("@db.example.com:5432/orders"),
724            "host and path must be retained: {url}"
725        );
726    }
727
728    #[test]
729    fn build_url_percent_encodes_userinfo_so_delimiters_cant_leak() {
730        // A2 (audit root fix): a password with URL delimiters (`/ @ : ? #` — `/` is
731        // a common base64 char) is percent-encoded when Rivet builds the connection
732        // URL from structured fields. Un-encoded, the raw `/`/`@` made the URL
733        // ambiguous: it broke the driver AND stopped the password-redaction scan
734        // early, leaking the tail. Encoding is the unambiguous root fix (redaction
735        // of a well-formed URL is then correct). RED before the encoding.
736        let mut src = make_source(SourceType::Postgres);
737        src.host = Some("db.example.com".into());
738        src.user = Some("rivet".into());
739        src.password = Some("pa/s:s@w?rd#x".into());
740        src.database = Some("orders".into());
741        let url = src.resolve_url().expect("url built from fields");
742        assert!(
743            !url.contains("pa/s:s@w?rd#x"),
744            "raw password with delimiters must not appear un-encoded: {url}"
745        );
746        assert!(
747            url.contains("pa%2Fs%3As%40w%3Frd%23x"),
748            "userinfo must be percent-encoded: {url}"
749        );
750    }
751
752    #[test]
753    fn mssql_url_from_fields_roundtrips_through_parse_mssql_url() {
754        // Round-2 audit #1: the encoding above is only safe if the consumer
755        // decodes it. PG/MySQL/Mongo delegate to a driver URL parser that
756        // percent-decodes; MSSQL's hand-rolled `parse_mssql_url` is the outlier
757        // that must decode itself. Prove the two halves agree: a password with
758        // every URL delimiter survives encode→parse losslessly. RED before the
759        // decode landed in `parse_mssql_url` (it echoed the literal %-bytes).
760        let mut src = make_source(SourceType::Mssql);
761        src.host = Some("db.internal".into());
762        src.user = Some(r"dom\svc".into());
763        src.password = Some("p:a/s@s?w#d!%x".into());
764        src.database = Some("rivet".into());
765        let url = src.resolve_url().expect("mssql url built from fields");
766        let parsed =
767            crate::source::mssql::parse_mssql_url(&url).expect("built mssql url must parse back");
768        assert_eq!(parsed.user, r"dom\svc", "user round-trips");
769        assert_eq!(
770            parsed.password, "p:a/s@s?w#d!%x",
771            "every URL-delimiter char must survive the encode→decode round-trip"
772        );
773        assert_eq!(parsed.database, "rivet");
774    }
775}
rivet/config/source.rs

rivet/config/
source.rs
