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/// Evaluate every rule against every labelled column heading into `sink`.
150/// Deterministic order: rules in declaration order, columns as given.
151pub fn evaluate(
152    policy: &CompiledPolicy,
153    sink: &SinkFacts,
154    columns: &[ColumnFacts],
155) -> Vec<Violation> {
156    let mut out = Vec::new();
157    for rule in policy.rules() {
158        for col in columns {
159            if !col.labels.iter().any(|l| rule_applies(rule, l, sink)) {
160                continue;
161            }
162            if let Some(action) = &col.masked
163                && rule.mask.iter().any(|m| m == action)
164            {
165                continue;
166            }
167            let base = |kind: ViolationKind| Violation {
168                rule: rule.name.clone(),
169                label: rule.when.label.clone(),
170                column: col.name.clone(),
171                sink: sink.id.clone(),
172                sink_kind: sink.kind.clone(),
173                kind,
174                conservative: col.conservative,
175                satisfied_by_mask: rule.mask.clone(),
176            };
177            if rule.deny {
178                out.push(base(ViolationKind::Denied));
179                continue;
180            }
181            if rule.require.is_empty() {
182                // A mask-only rule: reaching the sink unmasked is the breach.
183                out.push(base(ViolationKind::Unmasked {
184                    allowed: rule.mask.clone(),
185                }));
186                continue;
187            }
188            for (attr, allowed) in &rule.require {
189                match sink.attributes.get(attr) {
190                    None => out.push(base(ViolationKind::MissingAttribute {
191                        attribute: attr.clone(),
192                    })),
193                    Some(v) if !allowed.iter().any(|a| a == v) => {
194                        out.push(base(ViolationKind::AttributeNotAllowed {
195                            attribute: attr.clone(),
196                            value: v.clone(),
197                            allowed: allowed.clone(),
198                        }))
199                    }
200                    Some(_) => {}
201                }
202            }
203        }
204    }
205    out
206}
207
208#[cfg(test)]
209mod tests {
210    use super::*;
211    use crate::policy::PolicySpec;
212    use serde_json::json;
213
214    fn policy() -> CompiledPolicy {
215        let spec: PolicySpec = serde_json::from_value(json!({
216            "classifications": [
217                {"label": "pii", "fields": ["email"]},
218                {"label": "finance", "fields": ["iban"]}
219            ],
220            "rules": [
221                {"name": "pii-eu", "when": {"label": "pii"}, "require": {"residency": ["eu"]}, "mask": ["hash", "redact"]},
222                {"name": "no-finance-prod-files", "when": {"label": "finance", "sink_kind": ["jsonl"], "sink": {"environment": ["prod"]}}, "deny": true}
223            ]
224        }))
225        .unwrap();
226        CompiledPolicy::compile(&spec).unwrap()
227    }
228
229    fn sink(kind: &str, attrs: &[(&str, &str)]) -> SinkFacts {
230        SinkFacts {
231            id: "default".into(),
232            kind: kind.into(),
233            attributes: attrs
234                .iter()
235                .map(|(k, v)| (k.to_string(), v.to_string()))
236                .collect(),
237        }
238    }
239
240    fn col(name: &str, labels: &[&str], masked: Option<&str>) -> ColumnFacts {
241        ColumnFacts {
242            name: name.into(),
243            labels: labels.iter().map(|s| s.to_string()).collect(),
244            masked: masked.map(String::from),
245            conservative: false,
246            via: None,
247        }
248    }
249
250    #[test]
251    fn require_checks_presence_and_value_and_masking_satisfies() {
252        let p = policy();
253        let cols = [col("email", &["pii"], None), col("id", &[], None)];
254        let v = evaluate(&p, &sink("postgres", &[("residency", "us")]), &cols);
255        assert_eq!(v.len(), 1);
256        assert!(
257            matches!(&v[0].kind, ViolationKind::AttributeNotAllowed { value, .. } if value == "us")
258        );
259        assert_eq!(v[0].satisfied_by_mask, vec!["hash", "redact"]);
260        let text = v[0].to_string();
261        assert!(
262            text.contains("rule `pii-eu`") && text.contains("masking it with hash or redact"),
263            "{text}"
264        );
265
266        let v = evaluate(&p, &sink("postgres", &[]), &cols);
267        assert!(
268            matches!(&v[0].kind, ViolationKind::MissingAttribute { attribute } if attribute == "residency")
269        );
270        assert!(v[0].to_string().contains("does not declare"));
271
272        assert!(evaluate(&p, &sink("postgres", &[("residency", "eu")]), &cols).is_empty());
273        let masked = [col("email", &["pii"], Some("hash"))];
274        assert!(evaluate(&p, &sink("postgres", &[("residency", "us")]), &masked).is_empty());
275        let partial = [col("email", &["pii"], Some("partial"))];
276        assert_eq!(
277            evaluate(&p, &sink("postgres", &[("residency", "us")]), &partial).len(),
278            1
279        );
280    }
281
282    #[test]
283    fn deny_rules_scope_by_sink_kind_and_attributes() {
284        let p = policy();
285        let cols = [col("iban", &["finance"], None)];
286        let hit = evaluate(&p, &sink("jsonl", &[("environment", "prod")]), &cols);
287        assert_eq!(hit.len(), 1);
288        assert_eq!(hit[0].kind, ViolationKind::Denied);
289        assert!(hit[0].to_string().contains("may not reach sink"));
290        assert!(evaluate(&p, &sink("jsonl", &[("environment", "dev")]), &cols).is_empty());
291        assert!(evaluate(&p, &sink("postgres", &[("environment", "prod")]), &cols).is_empty());
292        assert!(
293            evaluate(&p, &sink("jsonl", &[]), &cols).is_empty(),
294            "a `when.sink` attribute the sink lacks means the rule does not apply"
295        );
296    }
297
298    #[test]
299    fn conservative_columns_are_flagged_as_such() {
300        let p = policy();
301        let mut c = col("email", &["pii"], None);
302        c.conservative = true;
303        let v = evaluate(&p, &sink("postgres", &[("residency", "us")]), &[c]);
304        assert!(v[0].conservative);
305        assert!(v[0].to_string().contains("opaque transform"));
306        let j = serde_json::to_value(&v[0]).unwrap();
307        assert_eq!(j["kind"], "attribute_not_allowed");
308        assert_eq!(j["conservative"], true);
309    }
310
311    #[test]
312    fn mask_only_rule_requires_the_mask() {
313        let spec: PolicySpec = serde_json::from_value(json!({
314            "classifications": [{"label": "finance", "fields": ["salary"]}],
315            "rules": [{"name": "hashed-only", "when": {"label": "finance"}, "mask": ["hash"]}]
316        }))
317        .unwrap();
318        let p = CompiledPolicy::compile(&spec).unwrap();
319        let sink = SinkFacts {
320            id: "s".into(),
321            kind: "jsonl".into(),
322            attributes: BTreeMap::new(),
323        };
324        let col = |masked: Option<&str>| ColumnFacts {
325            name: "salary".into(),
326            labels: ["finance".to_string()].into(),
327            masked: masked.map(String::from),
328            conservative: false,
329            via: None,
330        };
331        let v = evaluate(&p, &sink, &[col(None)]);
332        assert_eq!(v.len(), 1);
333        assert!(matches!(&v[0].kind, ViolationKind::Unmasked { allowed } if allowed == &["hash"]));
334        assert!(
335            v[0].to_string().contains("only masked with hash"),
336            "{}",
337            v[0]
338        );
339        assert!(evaluate(&p, &sink, &[col(Some("hash"))]).is_empty());
340        assert_eq!(evaluate(&p, &sink, &[col(Some("redact"))]).len(), 1);
341        // A rule with none of deny / require / mask is still refused.
342        let bad: PolicySpec = serde_json::from_value(json!({
343            "classifications": [{"label": "finance", "fields": ["salary"]}],
344            "rules": [{"name": "empty", "when": {"label": "finance"}}]
345        }))
346        .unwrap();
347        assert!(bad.validate().unwrap_err().contains("mask"));
348    }
349}
faucet_core/policy/evaluate.rs

faucet_core/policy/
evaluate.rs
