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    let mut row: Vec<usize> = (0..=b.len()).collect();
131    for (i, ca) in a.chars().enumerate() {
132        let mut prev = row[0];
133        row[0] = i + 1;
134        for (j, cb) in b.iter().enumerate() {
135            let cur = row[j + 1];
136            row[j + 1] = (prev + usize::from(ca != *cb))
137                .min(row[j] + 1)
138                .min(row[j + 1] + 1);
139            prev = cur;
140        }
141    }
142    row[b.len()]
143}
144
145/// `messages` with every secret field's value replaced by `*****`, so a plugin's rule that
146/// quotes a value can't leak it.
147pub fn redact(
148    messages: Vec<String>,
149    fields: &[ConnectionField],
150    connection: &Map<String, Value>,
151) -> Vec<String> {
152    let secrets: Vec<&str> = fields
153        .iter()
154        .filter(|f| f.secret)
155        .filter_map(|f| connection.get(&f.name).and_then(Value::as_str))
156        .filter(|s| s.len() >= 3)
157        .collect();
158    messages
159        .into_iter()
160        .map(|m| secrets.iter().fold(m, |m, s| m.replace(s, "*****")))
161        .collect()
162}
163
164#[cfg(test)]
165mod tests {
166    use super::*;
167    use serde_json::json;
168
169    fn fields() -> Vec<ConnectionField> {
170        vec![
171            ConnectionField::new("host", "").required(),
172            ConnectionField::new("port", "")
173                .kind(FieldKind::Integer)
174                .default(5432),
175            ConnectionField::new("password", "").secret(),
176            ConnectionField::new("sslmode", "").choices(&["disable", "require"]),
177            ConnectionField::new("key_path", "").kind(FieldKind::Path),
178            ConnectionField::new("timeout", "").kind(FieldKind::Duration),
179            ConnectionField::new("passive", "").kind(FieldKind::Boolean),
180        ]
181    }
182
183    fn run(c: Value, unresolved: &[&str]) -> Checked {
184        let u: Vec<String> = unresolved.iter().map(|s| s.to_string()).collect();
185        check(Kind::Source, "pg", &fields(), c.as_object().unwrap(), &u)
186    }
187
188    #[test]
189    fn required_kind_choices_and_unknown_keys() {
190        let c = run(
191            json!({"port": "abc", "sslmode": "verify", "timeout": "soon", "passive": "yes", "hots": "x"}),
192            &[],
193        );
194        assert_eq!(
195            c.errors,
196            [
197                "`port` must be a whole number",
198                "`sslmode` must be one of `disable`, `require`",
199                "`timeout` must be a duration such as `30s` or `2m`, or seconds",
200                "`passive` must be true or false",
201                "`host` is required",
202            ]
203        );
204        assert_eq!(
205            c.warnings,
206            ["unknown key `hots` for source `pg`, ignored; did you mean `host`?"]
207        );
208        let ok = run(
209            json!({"host": "h", "port": "5432", "timeout": "30s", "passive": true}),
210            &[],
211        );
212        assert_eq!(ok, Checked::default());
213    }
214
215    #[test]
216    fn a_path_must_exist_and_unresolved_values_count_as_set() {
217        let c = run(json!({"host": "h", "key_path": "/no/such/key"}), &[]);
218        assert!(
219            c.errors[0].starts_with("`key_path` names a file that doesn't exist"),
220            "{c:?}"
221        );
222        let c = run(json!({"host": null, "port": null}), &["host", "port"]);
223        assert_eq!(c, Checked::default());
224    }
225
226    #[test]
227    fn secrets_are_redacted() {
228        let c = json!({"password": "hunter22"});
229        let m = redact(
230            vec!["bad password hunter22".into()],
231            &fields(),
232            c.as_object().unwrap(),
233        );
234        assert_eq!(m, ["bad password *****"]);
235    }
236}
dre_protocol/connection.rs

dre_protocol/
connection.rs
