faucet-core 1.14.0

//! PII detection + column-level masking policies (issue #206): a declarative,
//! destination-scoped policy that classifies sensitive fields — by field-name
//! pattern, by value detector (email / card-with-Luhn / SSN / phone / IPv4), or
//! by explicit field list — and rewrites them per action (`redact` / `hash` /
//! `tokenize` / `partial`).
//!
//! **Ordering.** The masking pass runs *first* in the page loop — before the
//! quality, contract, and schema-drift passes and before every sink write. So
//! every downstream consumer (those passes, the sink, the DLQ, and the
//! sink-side lineage sample) observes only masked values; PII never leaks to a
//! sink, a dead-letter queue, or a lineage facet. Masking is value-only and
//! key-preserving, and it keeps determinism (equal inputs → equal masked
//! outputs), non-null-ness, and per-key uniqueness, so the later quality /
//! contract checks stay meaningful over masked data.
//!
//! Masking never fails a run or quarantines a record: matching fields are
//! rewritten in place. The pipeline wires metrics in `instrumented_apply_masking`.

pub mod compile;
pub mod config;
pub mod detect;
mod hash;

pub use compile::CompiledMasking;
pub use config::{Detector, MaskAction, MaskRule, MaskingSpec, MatchSpec};

use compile::CompiledRule;
use serde_json::Value;

/// One field-masking event, for metrics/observability.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MaskHit {
    /// The rule's stable label.
    pub rule: String,
    /// The action applied (`redact` / `hash` / `tokenize` / `partial`).
    pub action: &'static str,
    /// The value detector that matched, if the match was value-based.
    pub detector: Option<&'static str>,
}

/// Result of masking one page.
#[derive(Debug, Clone, Default)]
pub struct MaskingOutcome {
    /// The page with every matching field rewritten in place.
    pub records: Vec<Value>,
    /// One entry per masked field (order = discovery order).
    pub hits: Vec<MaskHit>,
}

/// Apply the masking policy to one page. Pure: no metrics, no I/O, infallible.
///
/// Each record is walked depth-first; for every field (including nested object
/// keys and array elements, addressed by dot-path such as `user.contacts.0.email`)
/// the first rule that matches wins and its action rewrites the value. A
/// name-based match (`field_pattern` / `fields`) can target a whole subtree:
/// `redact` replaces it wholesale and the hash/tokenize/partial actions rewrite
/// every scalar leaf inside it. Value detectors operate on scalar leaves.
pub fn apply_masking(records: Vec<Value>, masking: &CompiledMasking) -> MaskingOutcome {
    let mut out = MaskingOutcome {
        records: Vec::with_capacity(records.len()),
        hits: Vec::new(),
    };
    for mut rec in records {
        mask_node(&mut rec, "", masking, &mut out.hits);
        out.records.push(rec);
    }
    out
}

/// Recursively mask `node`, addressed by `path`. Returns nothing; mutates in
/// place and appends any hits.
fn mask_node(node: &mut Value, path: &str, m: &CompiledMasking, hits: &mut Vec<MaskHit>) {
    // 1. Does a rule match this node as a whole?
    if let Some((rule, detector)) = first_match(node, path, m) {
        match &rule.action {
            MaskAction::Redact { mask } => {
                *node = mask.clone();
                hits.push(hit(rule, detector));
                return; // replaced wholesale — do not recurse
            }
            action => {
                if let Some(s) = scalar_to_string(node) {
                    *node = Value::String(apply_scalar_action(action, &s, &m.hasher));
                    hits.push(hit(rule, detector));
                    return;
                }
                if matches!(node, Value::Object(_) | Value::Array(_)) {
                    mask_leaves(node, rule, action, &m.hasher, hits);
                    return;
                }
            }
        }
    }

    // 2. No terminal match — recurse into children.
    match node {
        Value::Object(map) => {
            for (k, v) in map.iter_mut() {
                // No name rule → the path is never read, so build none.
                if m.needs_paths {
                    mask_node(v, &child_path(path, k), m, hits);
                } else {
                    mask_node(v, UNTRACKED_PATH, m, hits);
                }
            }
        }
        Value::Array(items) => {
            for (i, v) in items.iter_mut().enumerate() {
                if m.needs_paths {
                    mask_node(v, &child_path(path, &i.to_string()), m, hits);
                } else {
                    mask_node(v, UNTRACKED_PATH, m, hits);
                }
            }
        }
        Value::Number(n) if m.stringify_numbers => *node = Value::String(n.to_string()),
        _ => {}
    }
}

/// Apply a scalar-only action to every scalar leaf of a name-matched subtree.
fn mask_leaves(
    node: &mut Value,
    rule: &CompiledRule,
    action: &MaskAction,
    hasher: &hash::Hasher,
    hits: &mut Vec<MaskHit>,
) {
    match node {
        Value::Object(map) => {
            for v in map.values_mut() {
                mask_leaves(v, rule, action, hasher, hits);
            }
        }
        Value::Array(items) => {
            for v in items.iter_mut() {
                mask_leaves(v, rule, action, hasher, hits);
            }
        }
        leaf => {
            if let Some(s) = scalar_to_string(leaf) {
                *leaf = Value::String(apply_scalar_action(action, &s, hasher));
                hits.push(hit(rule, None));
            }
        }
    }
}

/// Find the first rule that matches `node` at `path`. Returns the rule and, if
/// the match was value-based, the detector that fired.
fn first_match<'a>(
    node: &Value,
    path: &str,
    m: &'a CompiledMasking,
) -> Option<(&'a CompiledRule, Option<&'static str>)> {
    for rule in &m.rules {
        // Name-based: field pattern over the dot-path, or explicit field list.
        // The empty root path never matches a name rule.
        if !path.is_empty() {
            if let Some(re) = &rule.field_pattern
                && re.is_match(path)
            {
                return Some((rule, None));
            }
            if rule.fields.contains(path) {
                return Some((rule, None));
            }
        }
        // Value-based: strings through every detector, numbers through the
        // card detector only.
        if let Some(det) = rule.value_detector
            && detect::detects_value(det, node)
        {
            return Some((rule, Some(detector_label(det))));
        }
    }
    None
}

fn apply_scalar_action(action: &MaskAction, value: &str, hasher: &hash::Hasher) -> String {
    match action {
        MaskAction::Hash => hasher.digest_hex(value),
        MaskAction::Tokenize { prefix } => hasher.token(value, prefix.as_deref().unwrap_or("")),
        MaskAction::Partial {
            keep_last,
            mask_char,
        } => partial(value, *keep_last, mask_char.unwrap_or('*')),
        // Redact is handled at the node level (works on any type).
        MaskAction::Redact { .. } => "***".to_string(),
    }
}

/// Reveal only the last `keep_last` characters; replace the rest with
/// `mask_char`. Counts by Unicode scalar so multibyte strings mask correctly.
fn partial(value: &str, keep_last: usize, mask_char: char) -> String {
    let chars: Vec<char> = value.chars().collect();
    let n = chars.len();
    // Reveal at most `keep_last` trailing chars, but never the whole value
    // (a short secret with keep_last ≥ len would otherwise leak in the clear).
    let revealed = if keep_last >= n { 0 } else { keep_last };
    let masked_len = n - revealed;
    let mut s = String::with_capacity(n);
    for _ in 0..masked_len {
        s.push(mask_char);
    }
    s.extend(&chars[masked_len..]);
    s
}

/// String form of a scalar JSON value (`string` / `number` / `bool`); `None`
/// for null, objects, and arrays (nothing to mask as a scalar).
fn scalar_to_string(v: &Value) -> Option<String> {
    match v {
        Value::String(s) => Some(s.clone()),
        Value::Number(n) => Some(n.to_string()),
        Value::Bool(b) => Some(b.to_string()),
        _ => None,
    }
}

/// The path of a non-root node when no rule matches by name.
const UNTRACKED_PATH: &str = "*";

fn child_path(prefix: &str, key: &str) -> String {
    if prefix.is_empty() {
        key.to_string()
    } else {
        format!("{prefix}.{key}")
    }
}

fn hit(rule: &CompiledRule, detector: Option<&'static str>) -> MaskHit {
    MaskHit {
        rule: rule.label.clone(),
        action: rule.action_label,
        detector,
    }
}

fn detector_label(d: Detector) -> &'static str {
    match d {
        Detector::Email => "email",
        Detector::CreditCard => "credit_card",
        Detector::Ssn => "ssn",
        Detector::Phone => "phone",
        Detector::Ipv4 => "ipv4",
        Detector::PhoneLoose => "phone_loose",
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    fn compile(v: Value) -> CompiledMasking {
        let spec: MaskingSpec = serde_json::from_value(v).unwrap();
        CompiledMasking::compile(&spec).unwrap()
    }

    #[test]
    fn redact_replaces_named_field() {
        let m = compile(json!({
            "rules": [{ "match": { "field_pattern": "^email$" }, "action": { "type": "redact" } }]
        }));
        let out = apply_masking(vec![json!({"email": "a@b.com", "name": "Al"})], &m);
        assert_eq!(out.records[0], json!({"email": "***", "name": "Al"}));
        assert_eq!(out.hits.len(), 1);
        assert_eq!(out.hits[0].action, "redact");
        assert_eq!(out.hits[0].detector, None);
    }

    #[test]
    fn redact_custom_mask_value() {
        // Explicit null mask nulls the field (e.g. for a nullable DB column).
        let m = compile(json!({
            "rules": [{ "match": { "fields": ["ssn"] },
                        "action": { "type": "redact", "mask": null } }]
        }));
        let out = apply_masking(vec![json!({"ssn": "123-45-6789"})], &m);
        assert_eq!(out.records[0], json!({"ssn": null}));

        // A custom scalar mask replaces the value with that scalar.
        let m2 = compile(json!({
            "rules": [{ "match": { "fields": ["x"] },
                        "action": { "type": "redact", "mask": "REDACTED" } }]
        }));
        let out2 = apply_masking(vec![json!({"x": "secret"})], &m2);
        assert_eq!(out2.records[0], json!({"x": "REDACTED"}));
    }

    #[test]
    fn value_detector_matches_across_field_names() {
        let m = compile(json!({
            "rules": [{ "match": { "value_detector": "email" }, "action": { "type": "redact" } }]
        }));
        let out = apply_masking(
            vec![json!({"contact": "a@b.com", "note": "hi", "n": 5})],
            &m,
        );
        assert_eq!(
            out.records[0],
            json!({"contact": "***", "note": "hi", "n": 5})
        );
        assert_eq!(out.hits[0].detector, Some("email"));
    }

    #[test]
    fn numbers_are_not_phones_but_can_be_cards() {
        let m = compile(json!({
            "rules": [
                { "match": { "value_detector": "phone" }, "action": { "type": "redact" } },
                { "match": { "value_detector": "credit_card" }, "action": { "type": "redact" } }
            ]
        }));
        let out = apply_masking(
            vec![json!({
                "ts": 1717200000000u64, "amount": 4155552671u64, "flag": true,
                "tel": "415-555-2671", "cc": 4111111111111111u64
            })],
            &m,
        );
        assert_eq!(
            out.records[0],
            json!({"ts": "1717200000000", "amount": "4155552671", "flag": true,
                   "tel": "***", "cc": "***"}),
            "numbers are not phones (kept, as strings because a card rule can match numbers)"
        );
    }

    #[test]
    fn hash_is_deterministic_and_joinable() {
        let m = compile(json!({
            "key": "k",
            "rules": [{ "match": { "fields": ["uid"] }, "action": { "type": "hash" } }]
        }));
        let out = apply_masking(
            vec![
                json!({"uid": "u1"}),
                json!({"uid": "u1"}),
                json!({"uid": "u2"}),
            ],
            &m,
        );
        let h1 = out.records[0]["uid"].as_str().unwrap();
        assert_eq!(
            h1,
            out.records[1]["uid"].as_str().unwrap(),
            "same input → same hash"
        );
        assert_ne!(h1, out.records[2]["uid"].as_str().unwrap());
        assert_eq!(h1.len(), 64);
    }

    #[test]
    fn tokenize_with_prefix() {
        let m = compile(json!({
            "key": "k",
            "rules": [{ "match": { "fields": ["id"] },
                        "action": { "type": "tokenize", "prefix": "tok_" } }]
        }));
        let out = apply_masking(vec![json!({"id": "abc"})], &m);
        assert!(out.records[0]["id"].as_str().unwrap().starts_with("tok_"));
    }

    #[test]
    fn partial_keeps_last_n() {
        let m = compile(json!({
            "rules": [{ "match": { "fields": ["card"] },
                        "action": { "type": "partial", "keep_last": 4 } }]
        }));
        let out = apply_masking(vec![json!({"card": "4111111111111111"})], &m);
        assert_eq!(out.records[0]["card"], json!("************1111"));
    }

    #[test]
    fn partial_masks_all_when_keep_ge_len() {
        let m = compile(json!({
            "rules": [{ "match": { "fields": ["pin"] },
                        "action": { "type": "partial", "keep_last": 10 } }]
        }));
        let out = apply_masking(vec![json!({"pin": "1234"})], &m);
        assert_eq!(
            out.records[0]["pin"],
            json!("****"),
            "never reveal a short value whole"
        );
    }

    #[test]
    fn hash_coerces_number_to_string() {
        let m = compile(json!({
            "rules": [{ "match": { "fields": ["n"] }, "action": { "type": "hash" } }]
        }));
        let out = apply_masking(vec![json!({"n": 42})], &m);
        assert!(out.records[0]["n"].is_string());
    }

    #[test]
    fn nested_object_and_array_paths_masked() {
        let m = compile(json!({
            "rules": [{ "match": { "field_pattern": r"contacts\.\d+\.email" },
                        "action": { "type": "redact" } }]
        }));
        let out = apply_masking(
            vec![json!({
                "user": {"name": "Al"},
                "contacts": [ {"email": "a@b.com"}, {"email": "c@d.com"} ]
            })],
            &m,
        );
        assert_eq!(out.records[0]["contacts"][0]["email"], json!("***"));
        assert_eq!(out.records[0]["contacts"][1]["email"], json!("***"));
        assert_eq!(out.records[0]["user"]["name"], json!("Al"));
        assert_eq!(out.hits.len(), 2);
    }

    #[test]
    fn redact_whole_subtree_by_name() {
        let m = compile(json!({
            "rules": [{ "match": { "fields": ["address"] }, "action": { "type": "redact" } }]
        }));
        let out = apply_masking(
            vec![json!({"address": {"street": "1 Main", "zip": "94103"}, "id": 1})],
            &m,
        );
        assert_eq!(out.records[0], json!({"address": "***", "id": 1}));
    }

    #[test]
    fn a_number_matching_detector_writes_every_number_as_a_string() {
        let m = compile(json!({
            "key": "k",
            "rules": [{ "match": { "value_detector": "credit_card" }, "action": { "type": "hash" } }]
        }));
        let page1 = apply_masking(
            vec![json!({"acct": 12345, "amount": -1.5, "n": {"x": 7}, "ok": true, "z": null})],
            &m,
        );
        let page2 = apply_masking(vec![json!({"acct": 4111111111111111u64})], &m);
        assert_eq!(
            page1.records[0],
            json!({"acct": "12345", "amount": "-1.5", "n": {"x": "7"}, "ok": true, "z": null})
        );
        assert!(
            page1.hits.is_empty(),
            "a coerced number is not a masking hit"
        );
        let masked = page2.records[0]["acct"].as_str().unwrap();
        assert_eq!(masked.len(), 64, "the card is hashed, still a string");

        let plain = compile(json!({
            "rules": [{ "match": { "value_detector": "email" }, "action": { "type": "redact" } }]
        }));
        let out = apply_masking(vec![json!({"acct": 12345})], &plain);
        assert_eq!(out.records[0]["acct"], json!(12345));
    }

    #[test]
    fn first_matching_rule_wins() {
        let m = compile(json!({
            "rules": [
                { "name": "first", "match": { "fields": ["x"] }, "action": { "type": "redact" } },
                { "name": "second", "match": { "fields": ["x"] }, "action": { "type": "hash" } }
            ]
        }));
        let out = apply_masking(vec![json!({"x": "v"})], &m);
        assert_eq!(out.records[0]["x"], json!("***"));
        assert_eq!(out.hits[0].rule, "first");
    }

    #[test]
    fn scalar_action_on_container_masks_every_leaf_before_later_rules() {
        let m = compile(json!({
            "key": "k",
            "rules": [
                { "match": { "field_pattern": "^blob$" }, "action": { "type": "hash" } },
                { "match": { "value_detector": "email" }, "action": { "type": "redact" } }
            ]
        }));
        let out = apply_masking(vec![json!({"blob": {"e": "a@b.com"}})], &m);
        let e = out.records[0]["blob"]["e"].as_str().unwrap();
        assert_eq!(
            e.len(),
            64,
            "the name-matched hash rule wins over the detector"
        );
        assert_eq!(out.hits.len(), 1);
        assert_eq!(out.hits[0].action, "hash");
    }

    #[test]
    fn hash_on_name_matched_array_masks_each_element() {
        let m = compile(json!({
            "key": "k",
            "rules": [{ "match": { "fields": ["phones"] }, "action": { "type": "hash" } }]
        }));
        let out = apply_masking(
            vec![json!({"phones": ["+14155552671", "+14155550000", null], "id": 1})],
            &m,
        );
        let phones = out.records[0]["phones"].as_array().unwrap();
        for p in &phones[..2] {
            let s = p.as_str().unwrap();
            assert_eq!(s.len(), 64);
            assert!(!s.contains("415"));
        }
        assert_eq!(phones[2], json!(null));
        assert_eq!(out.records[0]["id"], json!(1));
        assert_eq!(out.hits.len(), 2);
    }

    #[test]
    fn partial_on_anchored_pattern_masks_nested_object_leaves() {
        let m = compile(json!({
            "rules": [{ "match": { "field_pattern": "^card$" },
                        "action": { "type": "partial", "keep_last": 4 } }]
        }));
        let out = apply_masking(
            vec![json!({"card": {"number": "4111111111111111", "exp": {"m": 12, "ok": true}}})],
            &m,
        );
        let card = &out.records[0]["card"];
        assert_eq!(card["number"], json!("************1111"));
        assert_eq!(card["exp"]["m"], json!("**"));
        assert_eq!(card["exp"]["ok"], json!("****"));
        assert_eq!(out.hits.len(), 3);
    }

    #[test]
    fn tokenize_on_anchored_pattern_masks_nested_object_leaves() {
        let m = compile(json!({
            "key": "k",
            "rules": [{ "match": { "field_pattern": "^card$" },
                        "action": { "type": "tokenize", "prefix": "tok_" } }]
        }));
        let out = apply_masking(vec![json!({"card": {"number": "4111111111111111"}})], &m);
        let n = out.records[0]["card"]["number"].as_str().unwrap();
        assert!(n.starts_with("tok_"));
        assert!(!n.contains("4111"));
        assert_eq!(out.hits.len(), 1);
    }

    #[test]
    fn null_is_not_masked() {
        let m = compile(json!({
            "rules": [{ "match": { "fields": ["x"] }, "action": { "type": "hash" } }]
        }));
        let out = apply_masking(vec![json!({"x": null})], &m);
        assert_eq!(out.records[0]["x"], json!(null));
        assert!(out.hits.is_empty());
    }

    #[test]
    fn non_object_record_untouched() {
        let m = compile(json!({
            "rules": [{ "match": { "value_detector": "email" }, "action": { "type": "redact" } }]
        }));
        // top-level string that is an email — root path is empty so name rules
        // don't fire, but value detector does.
        let out = apply_masking(vec![json!("a@b.com")], &m);
        assert_eq!(out.records[0], json!("***"));
    }

    #[test]
    fn no_match_leaves_page_untouched() {
        let m = compile(json!({
            "rules": [{ "match": { "fields": ["absent"] }, "action": { "type": "redact" } }]
        }));
        let page = vec![json!({"a": 1, "b": "x"})];
        let out = apply_masking(page.clone(), &m);
        assert_eq!(out.records, page);
        assert!(out.hits.is_empty());
    }
}