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

faucet_core/policy/
evaluate.rs
