faucet_core/policy/
spec.rs

1//! Config types for data-flow policies (#702): classifications that tag
2//! columns with labels, and rules that say where a labelled column may land.
3
4use crate::masking::{Detector, MaskAction};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7use std::collections::BTreeMap;
8
9/// Mask actions a rule may name in `mask:`.
10pub const MASK_ACTIONS: [&str; 4] = ["redact", "hash", "tokenize", "partial"];
11
12/// A data-flow policy: **classifications** (which columns carry which label)
13/// and **rules** (what a labelled column requires of the sink it lands in).
14/// Declared once — in a policy file passed with `--policy`, in a deployment
15/// overlay, or as a top-level `policy:` block — and evaluated against every
16/// pipeline before any data moves; a value detector re-checks at run time.
17#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
18#[serde(deny_unknown_fields)]
19pub struct PolicySpec {
20    /// Policy grammar version. Only `1` exists.
21    #[serde(default = "default_version")]
22    pub version: u32,
23
24    /// Free-form description shown by `faucet policy`.
25    #[serde(default, skip_serializing_if = "Option::is_none")]
26    pub description: Option<String>,
27
28    /// How columns get their labels.
29    #[serde(default)]
30    pub classifications: Vec<Classification>,
31
32    /// What each label demands of a destination.
33    #[serde(default)]
34    pub rules: Vec<PolicyRule>,
35}
36
37fn default_version() -> u32 {
38    1
39}
40
41/// One way a column earns a label: by exact name, by a regex over the column
42/// name, or by a value detector (run at run time over the real values, and
43/// statically over a sample when one is available).
44#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
45#[serde(deny_unknown_fields)]
46pub struct Classification {
47    /// The label this classification assigns (`pii`, `finance`, `health`, …).
48    pub label: String,
49
50    #[serde(default, skip_serializing_if = "Option::is_none")]
51    pub description: Option<String>,
52
53    /// Exact column names (top-level or dot-path) that carry the label.
54    #[serde(default, skip_serializing_if = "Vec::is_empty")]
55    pub fields: Vec<String>,
56
57    /// Regex over the column's dot-path; a match assigns the label.
58    #[serde(default, skip_serializing_if = "Option::is_none")]
59    pub field_pattern: Option<String>,
60
61    /// A value detector (`email`, `credit_card`, `ssn`, `phone`, `ipv4`): a
62    /// column whose values look like the class carries the label. Reuses the
63    /// masking detectors, so a masked column no longer matches.
64    #[serde(default, skip_serializing_if = "Option::is_none")]
65    pub value_detector: Option<Detector>,
66}
67
68/// A rule over one label: which sinks it applies to, and what satisfies it.
69#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
70#[serde(deny_unknown_fields)]
71pub struct PolicyRule {
72    /// Stable rule name — named in every violation, audit entry and metric.
73    pub name: String,
74
75    #[serde(default, skip_serializing_if = "Option::is_none")]
76    pub description: Option<String>,
77
78    /// What the rule applies to.
79    pub when: RuleWhen,
80
81    /// Sink attributes a labelled column's destination must carry, each with
82    /// its allowed values: `{ residency: [eu], environment: [prod, staging] }`.
83    /// A sink missing the attribute violates the rule.
84    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
85    pub require: BTreeMap<String, Vec<String>>,
86
87    /// Mask actions that satisfy the rule instead: a column masked by one of
88    /// these (`redact` / `hash` / `tokenize` / `partial`) may land anywhere the
89    /// rule would otherwise refuse. Empty = masking does not satisfy the rule.
90    #[serde(default, skip_serializing_if = "Vec::is_empty")]
91    pub mask: Vec<String>,
92
93    /// The labelled column may not reach a matching sink at all (unless a
94    /// listed `mask` action applies).
95    #[serde(default)]
96    pub deny: bool,
97
98    /// What a run does when a value detector finds the label at run time in a
99    /// column the static pass could not see: fail the run before the page is
100    /// written, or quarantine the offending records to the DLQ.
101    #[serde(default)]
102    pub on_runtime: RuntimeAction,
103}
104
105/// The scope of a rule.
106#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
107#[serde(deny_unknown_fields)]
108pub struct RuleWhen {
109    /// The label the rule governs.
110    pub label: String,
111
112    /// Only sinks of these connector kinds (empty = every sink).
113    #[serde(default, skip_serializing_if = "Vec::is_empty")]
114    pub sink_kind: Vec<String>,
115
116    /// Only sinks whose attributes match all of these (empty = every sink) —
117    /// e.g. `{ environment: [prod] }` to govern production destinations only.
118    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
119    pub sink: BTreeMap<String, Vec<String>>,
120}
121
122/// Runtime consequence of a value-detector hit that violates a rule.
123#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
124#[serde(rename_all = "snake_case")]
125pub enum RuntimeAction {
126    /// Fail the run before the page reaches the sink.
127    #[default]
128    Fail,
129    /// Route the offending records to the DLQ; the rest of the page is written.
130    Quarantine,
131}
132
133impl RuntimeAction {
134    pub fn as_str(self) -> &'static str {
135        match self {
136            RuntimeAction::Fail => "fail",
137            RuntimeAction::Quarantine => "quarantine",
138        }
139    }
140}
141
142impl PolicySpec {
143    /// Fail-fast validation, surfaced at config load. A policy that does not
144    /// validate is refused outright — a broken policy must fail closed.
145    pub fn validate(&self) -> Result<(), String> {
146        if self.version != 1 {
147            return Err(format!("unsupported policy version {}", self.version));
148        }
149        let mut labels = std::collections::BTreeSet::new();
150        for (i, c) in self.classifications.iter().enumerate() {
151            if c.label.trim().is_empty() {
152                return Err(format!("classifications[{i}]: label is empty"));
153            }
154            if c.fields.is_empty() && c.field_pattern.is_none() && c.value_detector.is_none() {
155                return Err(format!(
156                    "classification '{}': set fields, field_pattern, or value_detector",
157                    c.label
158                ));
159            }
160            if let Some(p) = &c.field_pattern {
161                regex::Regex::new(p).map_err(|e| {
162                    format!("classification '{}': invalid field_pattern: {e}", c.label)
163                })?;
164            }
165            if let Some(f) = c.fields.iter().find(|f| f.trim().is_empty()) {
166                return Err(format!(
167                    "classification '{}': fields contains an empty name {f:?}",
168                    c.label
169                ));
170            }
171            labels.insert(c.label.as_str());
172        }
173        let mut names = std::collections::BTreeSet::new();
174        for (i, r) in self.rules.iter().enumerate() {
175            if r.name.trim().is_empty() {
176                return Err(format!("rules[{i}]: name is empty"));
177            }
178            if !names.insert(r.name.as_str()) {
179                return Err(format!("rule '{}' is declared twice", r.name));
180            }
181            if !labels.contains(r.when.label.as_str()) {
182                return Err(format!(
183                    "rule '{}': label '{}' has no classification",
184                    r.name, r.when.label
185                ));
186            }
187            if !r.deny && r.require.is_empty() && r.mask.is_empty() {
188                return Err(format!(
189                    "rule '{}': set `deny: true`, at least one `require` attribute, or a `mask` \
190                     list (the column may reach the sink only masked)",
191                    r.name
192                ));
193            }
194            for (attr, allowed) in &r.require {
195                if attr.trim().is_empty() {
196                    return Err(format!("rule '{}': require has an empty attribute", r.name));
197                }
198                if allowed.is_empty() {
199                    return Err(format!(
200                        "rule '{}': require.{attr} lists no allowed values",
201                        r.name
202                    ));
203                }
204            }
205            for (attr, allowed) in &r.when.sink {
206                if attr.trim().is_empty() || allowed.is_empty() {
207                    return Err(format!(
208                        "rule '{}': when.sink.{attr} needs an attribute and at least one value",
209                        r.name
210                    ));
211                }
212            }
213            for m in &r.mask {
214                if !MASK_ACTIONS.contains(&m.as_str()) {
215                    return Err(format!(
216                        "rule '{}': mask action {m:?} is not one of {}",
217                        r.name,
218                        MASK_ACTIONS.join(", ")
219                    ));
220                }
221            }
222        }
223        Ok(())
224    }
225
226    /// Merge another policy in (classifications and rules appended). A rule
227    /// name present on both sides is an error — two sources must not silently
228    /// override each other.
229    pub fn merge(mut self, other: PolicySpec) -> Result<PolicySpec, String> {
230        for r in &other.rules {
231            if self.rules.iter().any(|m| m.name == r.name) {
232                return Err(format!(
233                    "rule '{}' is declared by two policy sources",
234                    r.name
235                ));
236            }
237        }
238        self.classifications.extend(other.classifications);
239        self.rules.extend(other.rules);
240        if self.description.is_none() {
241            self.description = other.description;
242        }
243        Ok(self)
244    }
245
246    /// Whether any classification uses a value detector (so the runtime
247    /// backstop has something to do).
248    pub fn has_value_detectors(&self) -> bool {
249        self.classifications
250            .iter()
251            .any(|c| c.value_detector.is_some())
252    }
253}
254
255/// The action label a masking action satisfies a rule with.
256pub fn mask_action_label(action: &MaskAction) -> &'static str {
257    action.label()
258}
259
260#[cfg(test)]
261mod tests {
262    use super::*;
263    use serde_json::json;
264
265    fn spec(v: serde_json::Value) -> PolicySpec {
266        serde_json::from_value(v).unwrap()
267    }
268
269    fn valid() -> PolicySpec {
270        spec(json!({
271            "classifications": [
272                {"label": "pii", "fields": ["email"], "value_detector": "email"},
273                {"label": "finance", "field_pattern": "^(amount|iban)$"}
274            ],
275            "rules": [
276                {"name": "pii-eu", "when": {"label": "pii"}, "require": {"residency": ["eu"]}, "mask": ["hash", "redact"]},
277                {"name": "no-finance-to-files", "when": {"label": "finance", "sink_kind": ["jsonl"]}, "deny": true, "on_runtime": "quarantine"}
278            ]
279        }))
280    }
281
282    #[test]
283    fn a_valid_policy_parses_with_defaults() {
284        let p = valid();
285        assert!(p.validate().is_ok());
286        assert_eq!(p.version, 1);
287        assert!(p.has_value_detectors());
288        assert_eq!(p.rules[0].on_runtime, RuntimeAction::Fail);
289        assert_eq!(p.rules[1].on_runtime, RuntimeAction::Quarantine);
290        assert_eq!(RuntimeAction::Quarantine.as_str(), "quarantine");
291        assert_eq!(mask_action_label(&MaskAction::Hash), "hash");
292    }
293
294    #[test]
295    fn validation_names_each_defect() {
296        let bad = |f: fn(&mut PolicySpec)| {
297            let mut p = valid();
298            f(&mut p);
299            p.validate().unwrap_err()
300        };
301        assert!(bad(|p| p.version = 2).contains("version"));
302        assert!(bad(|p| p.classifications[0].label = " ".into()).contains("label is empty"));
303        assert!(
304            bad(|p| p.classifications[1] = Classification {
305                label: "x".into(),
306                ..Default::default()
307            })
308            .contains("set fields")
309        );
310        assert!(
311            bad(|p| p.classifications[1].field_pattern = Some("(".into()))
312                .contains("invalid field_pattern")
313        );
314        assert!(bad(|p| p.classifications[0].fields = vec!["".into()]).contains("empty name"));
315        assert!(bad(|p| p.rules[0].name = "".into()).contains("name is empty"));
316        assert!(bad(|p| p.rules[1].name = "pii-eu".into()).contains("declared twice"));
317        assert!(bad(|p| p.rules[0].when.label = "nope".into()).contains("no classification"));
318        assert!(
319            bad(|p| {
320                p.rules[0].require.clear();
321                p.rules[0].mask.clear();
322            })
323            .contains("deny: true")
324        );
325        assert!(
326            bad(|p| {
327                p.rules[0].require.insert("".into(), vec!["x".into()]);
328            })
329            .contains("empty attribute")
330        );
331        assert!(
332            bad(|p| {
333                p.rules[0].require.insert("region".into(), vec![]);
334            })
335            .contains("no allowed values")
336        );
337        assert!(
338            bad(|p| {
339                p.rules[0].when.sink.insert("env".into(), vec![]);
340            })
341            .contains("when.sink")
342        );
343        assert!(bad(|p| p.rules[0].mask = vec!["encrypt".into()]).contains("not one of"));
344    }
345
346    #[test]
347    fn merge_appends_and_refuses_duplicate_rule_names() {
348        let a = valid();
349        let b = spec(json!({
350            "description": "team policy",
351            "classifications": [{"label": "health", "fields": ["diagnosis"]}],
352            "rules": [{"name": "health-masked", "when": {"label": "health"}, "deny": true, "mask": ["redact"]}]
353        }));
354        let merged = a.clone().merge(b.clone()).unwrap();
355        assert_eq!(merged.classifications.len(), 3);
356        assert_eq!(merged.rules.len(), 3);
357        assert_eq!(merged.description.as_deref(), Some("team policy"));
358        assert!(merged.validate().is_ok());
359        let err = a.merge(valid()).unwrap_err();
360        assert!(err.contains("two policy sources"), "{err}");
361    }
362
363    #[test]
364    fn unknown_keys_are_rejected() {
365        assert!(serde_json::from_value::<PolicySpec>(json!({"rulez": []})).is_err());
366        assert!(
367            serde_json::from_value::<PolicyRule>(
368                json!({"name": "r", "when": {"label": "x"}, "bogus": 1})
369            )
370            .is_err()
371        );
372    }
373}
faucet_core/policy/spec.rs

faucet_core/policy/
spec.rs
