dre_protocol/
connection.rs

1//! Static checks of a profile entry's connection settings, from the fields a plugin declares:
2//! a required field missing, an unknown key, a value of the wrong [`FieldKind`] or not one of
3//! its `choices`. The SDK runs them for `validate_connection` and before `open` and `deliver`,
4//! then the plugin's own rules, so `dre validate` and `dre run` say the same thing. Nothing here
5//! touches the network.
6
7use serde_json::{Map, Value};
8
9use crate::Kind;
10use crate::msg::{ConnectionField, FieldKind};
11
12/// Keys of a profile entry that core reads, which a plugin may ignore.
13pub const CORE_KEYS: &[&str] = &["threads"];
14
15/// What the checks found: errors stop a run; warnings (an unknown key) don't.
16#[derive(Debug, Default, Clone, PartialEq, Eq)]
17pub struct Checked {
18    pub errors: Vec<String>,
19    pub warnings: Vec<String>,
20}
21
22/// Check `connection` against `fields`. Keys in `unresolved` (an unset `env_var()`) count as set
23/// but aren't checked. Messages name fields, never values.
24pub fn check(
25    kind: Kind,
26    name: &str,
27    fields: &[ConnectionField],
28    connection: &Map<String, Value>,
29    unresolved: &[String],
30) -> Checked {
31    let mut out = Checked::default();
32    for (key, v) in connection {
33        let Some(f) = fields.iter().find(|f| &f.name == key) else {
34            // Core's own keys on a connection entry: `threads` (how many Bindings run at once).
35            if CORE_KEYS.contains(&key.as_str()) {
36                continue;
37            }
38            let hint = closest(key, fields.iter().map(|f| f.name.as_str()))
39                .map(|c| format!("; did you mean `{c}`?"))
40                .unwrap_or_default();
41            out.warnings
42                .push(format!("unknown key `{key}` for {kind} `{name}`, ignored{hint}"));
43            continue;
44        };
45        if v.is_null() || unresolved.iter().any(|u| u == key) {
46            continue;
47        }
48        // Jinja left in a value is rendered by core before it arrives; a literal `{{` isn't checked.
49        if v.as_str().is_some_and(crate::options::is_template) {
50            continue;
51        }
52        if let Some(e) = check_value(f, v) {
53            out.errors.push(format!("`{key}` {e}"));
54        }
55    }
56    for f in fields.iter().filter(|f| f.required && f.default.is_none()) {
57        let set =
58            connection.get(&f.name).is_some_and(|v| !v.is_null()) || unresolved.iter().any(|u| u == &f.name);
59        if !set {
60            out.errors.push(format!("`{}` is required", f.name));
61        }
62    }
63    out
64}
65
66fn check_value(f: &ConnectionField, v: &Value) -> Option<String> {
67    let ok = match f.kind {
68        None => true,
69        Some(FieldKind::String) => v.is_string() || v.is_number() || v.is_boolean(),
70        Some(FieldKind::Integer) => {
71            v.is_i64() || v.is_u64() || v.as_str().is_some_and(|s| s.trim().parse::<i64>().is_ok())
72        }
73        Some(FieldKind::Boolean) => v.is_boolean(),
74        Some(FieldKind::Duration) => crate::delivery::parse_duration(v).is_ok(),
75        Some(FieldKind::Map) => v.is_object(),
76        Some(FieldKind::Path) => match v.as_str() {
77            Some(p) => {
78                let path = expand_home(p);
79                if !path.exists() {
80                    return Some(format!("names a file that doesn't exist ({})", path.display()));
81                }
82                true
83            }
84            None => false,
85        },
86    };
87    if !ok {
88        return Some(match f.kind {
89            Some(FieldKind::Integer) => "must be a whole number".into(),
90            Some(FieldKind::Boolean) => "must be true or false".into(),
91            Some(FieldKind::Duration) => "must be a duration such as `30s` or `2m`, or seconds".into(),
92            Some(FieldKind::Map) => "must be a block of settings".into(),
93            Some(FieldKind::Path) => "must be a file path".into(),
94            _ => "must be a string".into(),
95        });
96    }
97    if !f.choices.is_empty() {
98        let s = match v {
99            Value::String(s) => s.clone(),
100            other => other.to_string(),
101        };
102        if !f.choices.contains(&s) {
103            let shown: Vec<String> = f.choices.iter().map(|c| format!("`{c}`")).collect();
104            return Some(format!("must be one of {}", shown.join(", ")));
105        }
106    }
107    None
108}
109
110/// `~/` as the home directory, as the plugins read paths.
111fn expand_home(p: &str) -> std::path::PathBuf {
112    match p.strip_prefix("~/").or_else(|| p.strip_prefix("~\\")) {
113        Some(rest) => std::env::home_dir().unwrap_or_default().join(rest),
114        None => std::path::PathBuf::from(p),
115    }
116}
117
118/// The declared name closest to `key`, if it's near enough to be a typo.
119pub fn closest<'a>(key: &str, names: impl Iterator<Item = &'a str>) -> Option<&'a str> {
120    names
121        .map(|n| (distance(key, n), n))
122        .filter(|(d, n)| *d <= 2.max(n.len() / 4))
123        .min_by_key(|(d, _)| *d)
124        .map(|(_, n)| n)
125}
126
127/// Levenshtein distance.
128fn distance(a: &str, b: &str) -> usize {
129    let b: Vec<char> = b.chars().collect();
130    // `row` has one more entry than `b`, so `j + 1` and `b.len()` are in range.
131    let mut row: Vec<usize> = (0..=b.len()).collect();
132    for (i, ca) in a.chars().enumerate() {
133        let mut prev = row[0];
134        row[0] = i + 1;
135        for (j, cb) in b.iter().enumerate() {
136            let cur = row[j + 1];
137            row[j + 1] = (prev + usize::from(ca != *cb))
138                .min(row[j] + 1)
139                .min(row[j + 1] + 1);
140            prev = cur;
141        }
142    }
143    row[b.len()]
144}
145
146/// `messages` with every secret field's value replaced by `*****`, so a plugin's rule that
147/// quotes a value can't leak it.
148#[must_use]
149pub fn redact(
150    messages: Vec<String>,
151    fields: &[ConnectionField],
152    connection: &Map<String, Value>,
153) -> Vec<String> {
154    let secrets: Vec<&str> = fields
155        .iter()
156        .filter(|f| f.secret)
157        .filter_map(|f| connection.get(&f.name).and_then(Value::as_str))
158        .filter(|s| s.len() >= 3)
159        .collect();
160    messages
161        .into_iter()
162        .map(|m| secrets.iter().fold(m, |m, s| m.replace(s, "*****")))
163        .collect()
164}
165
166#[cfg(test)]
167mod tests {
168    use super::*;
169    use serde_json::json;
170
171    fn fields() -> Vec<ConnectionField> {
172        vec![
173            ConnectionField::new("host", "").required(),
174            ConnectionField::new("port", "")
175                .kind(FieldKind::Integer)
176                .default(5432),
177            ConnectionField::new("password", "").secret(),
178            ConnectionField::new("sslmode", "").choices(&["disable", "require"]),
179            ConnectionField::new("key_path", "").kind(FieldKind::Path),
180            ConnectionField::new("timeout", "").kind(FieldKind::Duration),
181            ConnectionField::new("passive", "").kind(FieldKind::Boolean),
182        ]
183    }
184
185    fn run(c: Value, unresolved: &[&str]) -> Checked {
186        let u: Vec<String> = unresolved.iter().map(|s| s.to_string()).collect();
187        check(Kind::Source, "pg", &fields(), c.as_object().unwrap(), &u)
188    }
189
190    #[test]
191    fn required_kind_choices_and_unknown_keys() {
192        let c = run(
193            json!({"port": "abc", "sslmode": "verify", "timeout": "soon", "passive": "yes", "hots": "x"}),
194            &[],
195        );
196        assert_eq!(
197            c.errors,
198            [
199                "`port` must be a whole number",
200                "`sslmode` must be one of `disable`, `require`",
201                "`timeout` must be a duration such as `30s` or `2m`, or seconds",
202                "`passive` must be true or false",
203                "`host` is required",
204            ]
205        );
206        assert_eq!(
207            c.warnings,
208            ["unknown key `hots` for source `pg`, ignored; did you mean `host`?"]
209        );
210        let ok = run(
211            json!({"host": "h", "port": "5432", "timeout": "30s", "passive": true}),
212            &[],
213        );
214        assert_eq!(ok, Checked::default());
215    }
216
217    #[test]
218    fn a_path_must_exist_and_unresolved_values_count_as_set() {
219        let c = run(json!({"host": "h", "key_path": "/no/such/key"}), &[]);
220        assert!(
221            c.errors[0].starts_with("`key_path` names a file that doesn't exist"),
222            "{c:?}"
223        );
224        let c = run(json!({"host": null, "port": null}), &["host", "port"]);
225        assert_eq!(c, Checked::default());
226    }
227
228    #[test]
229    fn secrets_are_redacted() {
230        let c = json!({"password": "hunter22"});
231        let m = redact(
232            vec!["bad password hunter22".into()],
233            &fields(),
234            c.as_object().unwrap(),
235        );
236        assert_eq!(m, ["bad password *****"]);
237    }
238}
dre_protocol/connection.rs

dre_protocol/
connection.rs
