rivet/config/
source.rs

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

rivet/config/
source.rs
