faucet_core/policy/
evaluate.rs

1//! Pure policy evaluation (#702): given a sink's attributes and the labelled
2//! columns heading into it, which rules are violated. No I/O; shared by the
3//! static pass (`validate` / `plan` / `doctor` / `run` / serve submit) and the
4//! runtime backstop.
5
6use super::compile::CompiledPolicy;
7use super::spec::PolicyRule;
8use schemars::JsonSchema;
9use serde::{Deserialize, Serialize};
10use std::collections::{BTreeMap, BTreeSet};
11use std::fmt;
12
13/// The destination side of an evaluation.
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct SinkFacts {
16    /// The sink template name (`default` for the singular `pipeline.sink`).
17    pub id: String,
18    /// Connector kind (`postgres`, `jsonl`, …).
19    pub kind: String,
20    /// Declared `attributes` (`residency`, `environment`, `region`, …).
21    pub attributes: BTreeMap<String, String>,
22}
23
24/// One column as the policy sees it.
25#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
26pub struct ColumnFacts {
27    /// Column dot-path as it lands in the sink.
28    pub name: String,
29    /// Labels the column carries.
30    pub labels: BTreeSet<String>,
31    /// The mask action applied to it before the sink (`hash`, `redact`, …),
32    /// when the masking policy provably covers this column.
33    #[serde(default, skip_serializing_if = "Option::is_none")]
34    pub masked: Option<String>,
35    /// The column may or may not reach the sink under this name — an opaque
36    /// transform hides the mapping, so the label is carried conservatively.
37    #[serde(default)]
38    pub conservative: bool,
39    /// How the label was assigned (`name`, `value`, `lineage`), for reports.
40    #[serde(default, skip_serializing_if = "Option::is_none")]
41    pub via: Option<String>,
42}
43
44/// Why a rule was violated.
45#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
46#[serde(tag = "kind", rename_all = "snake_case")]
47pub enum ViolationKind {
48    /// The rule denies the label at this sink outright.
49    Denied,
50    /// The sink declares no such attribute.
51    MissingAttribute { attribute: String },
52    /// The sink's attribute value is not in the allowed set.
53    AttributeNotAllowed {
54        attribute: String,
55        value: String,
56        allowed: Vec<String>,
57    },
58    /// The rule lets the label reach the sink only masked, and the column is
59    /// not masked with one of the listed actions.
60    Unmasked { allowed: Vec<String> },
61}
62
63/// One violated rule for one column at one sink.
64#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
65pub struct Violation {
66    pub rule: String,
67    pub label: String,
68    pub column: String,
69    /// The sink template name.
70    pub sink: String,
71    pub sink_kind: String,
72    #[serde(flatten)]
73    pub kind: ViolationKind,
74    /// The column's presence is inferred conservatively (opaque transform).
75    #[serde(default)]
76    pub conservative: bool,
77    /// Mask actions that would have satisfied the rule.
78    #[serde(default, skip_serializing_if = "Vec::is_empty")]
79    pub satisfied_by_mask: Vec<String>,
80}
81
82impl fmt::Display for Violation {
83    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
84        let what = match &self.kind {
85            ViolationKind::Denied => format!(
86                "label `{}` may not reach sink `{}` ({})",
87                self.label, self.sink, self.sink_kind
88            ),
89            ViolationKind::MissingAttribute { attribute } => format!(
90                "label `{}` requires sink attribute `{attribute}`, which sink `{}` ({}) does not declare",
91                self.label, self.sink, self.sink_kind
92            ),
93            ViolationKind::AttributeNotAllowed {
94                attribute,
95                value,
96                allowed,
97            } => format!(
98                "label `{}` requires sink `{attribute}` in [{}]; sink `{}` ({}) has `{value}`",
99                self.label,
100                allowed.join(", "),
101                self.sink,
102                self.sink_kind
103            ),
104            ViolationKind::Unmasked { allowed } => format!(
105                "label `{}` may reach sink `{}` ({}) only masked with {}",
106                self.label,
107                self.sink,
108                self.sink_kind,
109                allowed.join(" or ")
110            ),
111        };
112        write!(
113            f,
114            "rule `{}`: column `{}`{} — {what}",
115            self.rule,
116            self.column,
117            if self.conservative {
118                " (may be present: opaque transform)"
119            } else {
120                ""
121            }
122        )?;
123        if !self.satisfied_by_mask.is_empty() {
124            write!(
125                f,
126                "; masking it with {} would satisfy the rule",
127                self.satisfied_by_mask.join(" or ")
128            )?;
129        }
130        Ok(())
131    }
132}
133
134/// Whether `rule` governs a column carrying `label` heading into `sink`.
135pub fn rule_applies(rule: &PolicyRule, label: &str, sink: &SinkFacts) -> bool {
136    if rule.when.label != label {
137        return false;
138    }
139    if !rule.when.sink_kind.is_empty() && !rule.when.sink_kind.iter().any(|k| k == &sink.kind) {
140        return false;
141    }
142    rule.when.sink.iter().all(|(attr, allowed)| {
143        sink.attributes
144            .get(attr)
145            .is_some_and(|v| allowed.iter().any(|a| a == v))
146    })
147}
148
149/// The violations a quarantine would cause at the DLQ: every labelled column
150/// of the record (`columns`), evaluated with `dlq` as their destination, when
151/// anything is `quarantined`. The quarantined record reaches the DLQ in full,
152/// so every rule must hold there for every column it carries — not only for
153/// the column that broke the main sink's rule (#862 M-020).
154pub fn dlq_violations(
155    policy: &CompiledPolicy,
156    dlq: &SinkFacts,
157    columns: &[ColumnFacts],
158    quarantined: &[Violation],
159) -> Vec<Violation> {
160    if quarantined.is_empty() {
161        return Vec::new();
162    }
163    evaluate(policy, dlq, columns)
164}
165
166/// Evaluate every rule against every labelled column heading into `sink`.
167/// Deterministic order: rules in declaration order, columns as given.
168pub fn evaluate(
169    policy: &CompiledPolicy,
170    sink: &SinkFacts,
171    columns: &[ColumnFacts],
172) -> Vec<Violation> {
173    let mut out = Vec::new();
174    for rule in policy.rules() {
175        for col in columns {
176            if !col.labels.iter().any(|l| rule_applies(rule, l, sink)) {
177                continue;
178            }
179            if let Some(action) = &col.masked
180                && rule.mask.iter().any(|m| m == action)
181            {
182                continue;
183            }
184            let base = |kind: ViolationKind| Violation {
185                rule: rule.name.clone(),
186                label: rule.when.label.clone(),
187                column: col.name.clone(),
188                sink: sink.id.clone(),
189                sink_kind: sink.kind.clone(),
190                kind,
191                conservative: col.conservative,
192                satisfied_by_mask: rule.mask.clone(),
193            };
194            if rule.deny {
195                out.push(base(ViolationKind::Denied));
196                continue;
197            }
198            if rule.require.is_empty() {
199                // A mask-only rule: reaching the sink unmasked is the breach.
200                out.push(base(ViolationKind::Unmasked {
201                    allowed: rule.mask.clone(),
202                }));
203                continue;
204            }
205            for (attr, allowed) in &rule.require {
206                match sink.attributes.get(attr) {
207                    None => out.push(base(ViolationKind::MissingAttribute {
208                        attribute: attr.clone(),
209                    })),
210                    Some(v) if !allowed.iter().any(|a| a == v) => {
211                        out.push(base(ViolationKind::AttributeNotAllowed {
212                            attribute: attr.clone(),
213                            value: v.clone(),
214                            allowed: allowed.clone(),
215                        }))
216                    }
217                    Some(_) => {}
218                }
219            }
220        }
221    }
222    out
223}
224
225#[cfg(test)]
226mod tests {
227    use super::*;
228    use crate::policy::PolicySpec;
229    use serde_json::json;
230
231    fn policy() -> CompiledPolicy {
232        let spec: PolicySpec = serde_json::from_value(json!({
233            "classifications": [
234                {"label": "pii", "fields": ["email"]},
235                {"label": "finance", "fields": ["iban"]}
236            ],
237            "rules": [
238                {"name": "pii-eu", "when": {"label": "pii"}, "require": {"residency": ["eu"]}, "mask": ["hash", "redact"]},
239                {"name": "no-finance-prod-files", "when": {"label": "finance", "sink_kind": ["jsonl"], "sink": {"environment": ["prod"]}}, "deny": true}
240            ]
241        }))
242        .unwrap();
243        CompiledPolicy::compile(&spec).unwrap()
244    }
245
246    fn sink(kind: &str, attrs: &[(&str, &str)]) -> SinkFacts {
247        SinkFacts {
248            id: "default".into(),
249            kind: kind.into(),
250            attributes: attrs
251                .iter()
252                .map(|(k, v)| (k.to_string(), v.to_string()))
253                .collect(),
254        }
255    }
256
257    fn col(name: &str, labels: &[&str], masked: Option<&str>) -> ColumnFacts {
258        ColumnFacts {
259            name: name.into(),
260            labels: labels.iter().map(|s| s.to_string()).collect(),
261            masked: masked.map(String::from),
262            conservative: false,
263            via: None,
264        }
265    }
266
267    #[test]
268    fn require_checks_presence_and_value_and_masking_satisfies() {
269        let p = policy();
270        let cols = [col("email", &["pii"], None), col("id", &[], None)];
271        let v = evaluate(&p, &sink("postgres", &[("residency", "us")]), &cols);
272        assert_eq!(v.len(), 1);
273        assert!(
274            matches!(&v[0].kind, ViolationKind::AttributeNotAllowed { value, .. } if value == "us")
275        );
276        assert_eq!(v[0].satisfied_by_mask, vec!["hash", "redact"]);
277        let text = v[0].to_string();
278        assert!(
279            text.contains("rule `pii-eu`") && text.contains("masking it with hash or redact"),
280            "{text}"
281        );
282
283        let v = evaluate(&p, &sink("postgres", &[]), &cols);
284        assert!(
285            matches!(&v[0].kind, ViolationKind::MissingAttribute { attribute } if attribute == "residency")
286        );
287        assert!(v[0].to_string().contains("does not declare"));
288
289        assert!(evaluate(&p, &sink("postgres", &[("residency", "eu")]), &cols).is_empty());
290        let masked = [col("email", &["pii"], Some("hash"))];
291        assert!(evaluate(&p, &sink("postgres", &[("residency", "us")]), &masked).is_empty());
292        let partial = [col("email", &["pii"], Some("partial"))];
293        assert_eq!(
294            evaluate(&p, &sink("postgres", &[("residency", "us")]), &partial).len(),
295            1
296        );
297    }
298
299    #[test]
300    fn deny_rules_scope_by_sink_kind_and_attributes() {
301        let p = policy();
302        let cols = [col("iban", &["finance"], None)];
303        let hit = evaluate(&p, &sink("jsonl", &[("environment", "prod")]), &cols);
304        assert_eq!(hit.len(), 1);
305        assert_eq!(hit[0].kind, ViolationKind::Denied);
306        assert!(hit[0].to_string().contains("may not reach sink"));
307        assert!(evaluate(&p, &sink("jsonl", &[("environment", "dev")]), &cols).is_empty());
308        assert!(evaluate(&p, &sink("postgres", &[("environment", "prod")]), &cols).is_empty());
309        assert!(
310            evaluate(&p, &sink("jsonl", &[]), &cols).is_empty(),
311            "a `when.sink` attribute the sink lacks means the rule does not apply"
312        );
313    }
314
315    #[test]
316    fn conservative_columns_are_flagged_as_such() {
317        let p = policy();
318        let mut c = col("email", &["pii"], None);
319        c.conservative = true;
320        let v = evaluate(&p, &sink("postgres", &[("residency", "us")]), &[c]);
321        assert!(v[0].conservative);
322        assert!(v[0].to_string().contains("opaque transform"));
323        let j = serde_json::to_value(&v[0]).unwrap();
324        assert_eq!(j["kind"], "attribute_not_allowed");
325        assert_eq!(j["conservative"], true);
326    }
327
328    #[test]
329    fn mask_only_rule_requires_the_mask() {
330        let spec: PolicySpec = serde_json::from_value(json!({
331            "classifications": [{"label": "finance", "fields": ["salary"]}],
332            "rules": [{"name": "hashed-only", "when": {"label": "finance"}, "mask": ["hash"]}]
333        }))
334        .unwrap();
335        let p = CompiledPolicy::compile(&spec).unwrap();
336        let sink = SinkFacts {
337            id: "s".into(),
338            kind: "jsonl".into(),
339            attributes: BTreeMap::new(),
340        };
341        let col = |masked: Option<&str>| ColumnFacts {
342            name: "salary".into(),
343            labels: ["finance".to_string()].into(),
344            masked: masked.map(String::from),
345            conservative: false,
346            via: None,
347        };
348        let v = evaluate(&p, &sink, &[col(None)]);
349        assert_eq!(v.len(), 1);
350        assert!(matches!(&v[0].kind, ViolationKind::Unmasked { allowed } if allowed == &["hash"]));
351        assert!(
352            v[0].to_string().contains("only masked with hash"),
353            "{}",
354            v[0]
355        );
356        assert!(evaluate(&p, &sink, &[col(Some("hash"))]).is_empty());
357        assert_eq!(evaluate(&p, &sink, &[col(Some("redact"))]).len(), 1);
358        // A rule with none of deny / require / mask is still refused.
359        let bad: PolicySpec = serde_json::from_value(json!({
360            "classifications": [{"label": "finance", "fields": ["salary"]}],
361            "rules": [{"name": "empty", "when": {"label": "finance"}}]
362        }))
363        .unwrap();
364        assert!(bad.validate().unwrap_err().contains("mask"));
365    }
366}
faucet_core/policy/evaluate.rs

faucet_core/policy/
evaluate.rs
