validation: surface write-schema redact annotations on projections of split-schema collections

crates/assemble/src/lib.rs

@@ -5,6 +5,15 @@ use proto_flow::flow;
 use proto_gazette::{broker, consumer};
 use std::time::Duration;
 
+pub fn inference_redact(redact: &shape::Redact) -> flow::inference::Redact {
+    match redact {
+        shape::Redact::Multiple => flow::inference::Redact::Multiple,
+        shape::Redact::Strategy(redact::Strategy::Block) => flow::inference::Redact::Block,
+        shape::Redact::Strategy(redact::Strategy::Sha256) => flow::inference::Redact::Sha256,
+        shape::Redact::Unset => flow::inference::Redact::Unset,
+    }
+}
+
 pub fn inference(shape: &Shape, exists: Exists) -> flow::Inference {
     let default_json: bytes::Bytes = shape
         .default
@@ -48,12 +57,7 @@ pub fn inference(shape: &Shape, exists: Exists) -> flow::Inference {
         }
         shape::Reduce::Unset => flow::inference::Reduce::Unset,
     };
-    let redact = match shape.redact {
-        shape::Redact::Multiple => flow::inference::Redact::Multiple,
-        shape::Redact::Strategy(redact::Strategy::Block) => flow::inference::Redact::Block,
-        shape::Redact::Strategy(redact::Strategy::Sha256) => flow::inference::Redact::Sha256,
-        shape::Redact::Unset => flow::inference::Redact::Unset,
-    };
+    let redact = inference_redact(&shape.redact);
 
     let content_media_type = shape
         .content_media_type

crates/flow-web/Cargo.toml

@@ -2,7 +2,7 @@
 name = "flow-web"
 
 # wasm-pack isn't yet fully compatible with workspace inheritance, so we can't use that for these fields
-version = "0.6.10"
+version = "0.6.11"
 authors = ["Estuary developers  <engineering@estuary.dev>"]
 edition = "2021"
 license = "BSL"

crates/flow-web/tests/web.rs

@@ -90,6 +90,50 @@ fn test_skim_projections() {
     assert!(result.errors.is_empty());
 }
 
+#[wasm_bindgen_test]
+fn test_skim_projections_write_schema_redaction() {
+    // A redact annotation on the write schema of a collection with a
+    // standalone read schema (no $ref to flow://write-schema). Redaction runs
+    // at write time against the write schema, so the annotation must surface
+    // on the projection's inference.
+    let input: JsValue = to_js_value(&json!({
+        "collection": "acmeCo/test",
+        "model": {
+            "key": ["/id"],
+            "writeSchema": {
+                "type": "object",
+                "properties": {
+                    "id": {"type": "string"},
+                    "secret": {"type": "string", "redact": {"strategy": "block"}}
+                },
+                "required": ["id"]
+            },
+            "readSchema": {
+                "type": "object",
+                "properties": {
+                    "id": {"type": "string"},
+                    "secret": {"type": "string"}
+                },
+                "required": ["id"]
+            }
+        }
+    }));
+    let result = flow_web::skim_collection_projections(input).unwrap();
+    let result: flow_web::collection::CollectionProjectionsResult =
+        serde_wasm_bindgen::from_value(result).unwrap();
+
+    let secret = result
+        .projections
+        .iter()
+        .find(|p| p.ptr == "/secret")
+        .expect("a /secret projection is skimmed");
+    assert_eq!(
+        secret.inference.as_ref().unwrap().redact(),
+        flow::inference::Redact::Block,
+    );
+    assert!(result.errors.is_empty());
+}
+
 #[wasm_bindgen_test]
 fn test_field_selection() {
     let input: JsValue = to_js_value(&json!({

crates/validation/src/collection.rs

@@ -701,6 +701,25 @@ fn walk_collection_projections(
     // Now de-duplicate on field, taking the first entry. Recall that user projections are first.
     specs.dedup_by(|l, r| l.field.cmp(&r.field).is_eq());
 
+    // Redaction runs at write time against the write schema, so a `redact`
+    // annotation reachable only through the write schema is in force even when
+    // the read schema doesn't carry it. Surface it on the built inference of
+    // every projection whose read-schema inference has no strategy of its own.
+    if let Some(write_spec) = write_spec {
+        for spec in specs.iter_mut() {
+            let Some(inference) = spec.inference.as_mut() else {
+                continue;
+            };
+            if inference.redact != flow::inference::Redact::Unset as i32 {
+                continue;
+            }
+            let (w_shape, _) = write_spec
+                .shape
+                .locate(&json::Pointer::from(spec.ptr.as_str()));
+            inference.redact = assemble::inference_redact(&w_shape.redact) as i32;
+        }
+    }
+
     (models, specs)
 }
 

crates/validation/tests/skim_projections_tests.rs

@@ -0,0 +1,135 @@
+use serde_json::json;
+
+// Run skim_projections over a collection model, returning a compact view of
+// each projection's redact inference plus any validation errors. This mirrors
+// the flow-web `skim_collection_projections` entry point used by the dashboard.
+fn skim(model: serde_json::Value) -> (Vec<(String, String, String)>, Vec<String>) {
+    let model: models::CollectionDef = serde_json::from_value(model).unwrap();
+
+    let scope_url = url::Url::parse("flow://collection/acmeCo/test").unwrap();
+    let scope = json::Scope::new(&scope_url);
+    let collection = models::Collection::new("acmeCo/test");
+    let mut errors = tables::Errors::new();
+
+    let projections =
+        validation::collection::skim_projections(scope, &collection, &model, &mut errors);
+
+    let projections = projections
+        .into_iter()
+        .map(|p| {
+            let redact = p
+                .inference
+                .as_ref()
+                .map(|i| i.redact().as_str_name().to_string())
+                .unwrap_or_default();
+            (p.field, p.ptr, redact)
+        })
+        .collect();
+
+    let errors = errors
+        .into_iter()
+        .map(|err| format!("{:#}", err.error))
+        .collect();
+
+    (projections, errors)
+}
+
+// The repro from the issue: a redact annotation on the write schema of a
+// collection with a standalone read schema (no $ref to flow://write-schema).
+// Redaction runs at write time against the write schema, so the annotation is
+// in force and must surface on the projection's inference.
+#[test]
+fn test_write_schema_redact_with_standalone_read_schema() {
+    let (projections, errors) = skim(json!({
+        "key": ["/id"],
+        "writeSchema": {
+            "type": "object",
+            "properties": {
+                "id": {"type": "string"},
+                "secret": {"type": "string", "redact": {"strategy": "block"}}
+            },
+            "required": ["id"]
+        },
+        "readSchema": {
+            "type": "object",
+            "properties": {
+                "id": {"type": "string"},
+                "secret": {"type": "string"}
+            },
+            "required": ["id"]
+        }
+    }));
+
+    insta::assert_debug_snapshot!((projections, errors));
+}
+
+// A single schema plays both roles: the annotation already surfaces today.
+#[test]
+fn test_single_schema_redact() {
+    let (projections, errors) = skim(json!({
+        "key": ["/id"],
+        "schema": {
+            "type": "object",
+            "properties": {
+                "id": {"type": "string"},
+                "secret": {"type": "string", "redact": {"strategy": "sha256"}}
+            },
+            "required": ["id"]
+        }
+    }));
+
+    insta::assert_debug_snapshot!((projections, errors));
+}
+
+// A read schema which $refs flow://write-schema: the annotation flows through
+// the inlined reference and already surfaces today.
+#[test]
+fn test_ref_style_read_schema_redact() {
+    let (projections, errors) = skim(json!({
+        "key": ["/id"],
+        "writeSchema": {
+            "type": "object",
+            "properties": {
+                "id": {"type": "string"},
+                "secret": {"type": "string", "redact": {"strategy": "block"}}
+            },
+            "required": ["id"]
+        },
+        "readSchema": {
+            "allOf": [{"$ref": "flow://write-schema"}]
+        }
+    }));
+
+    insta::assert_debug_snapshot!((projections, errors));
+}
+
+// `block` at a location which must exist is reported for a write-schema
+// annotation just as it is for a single schema. A redact strategy on a key
+// location, however, is checked against the read schema only: connector
+// -managed write schemas may annotate a key which flow://relaxed-write-schema
+// then strips from the read path, so `sha256` on key /id here is deliberately
+// not an error.
+#[test]
+fn test_write_schema_redact_errors_with_standalone_read_schema() {
+    let (_projections, errors) = skim(json!({
+        "key": ["/id"],
+        "writeSchema": {
+            "type": "object",
+            "properties": {
+                "id": {"type": "string", "redact": {"strategy": "sha256"}},
+                "message": {"type": "string", "redact": {"strategy": "block"}}
+            },
+            "required": ["id", "message"]
+        },
+        "readSchema": {
+            "type": "object",
+            "properties": {
+                "id": {"type": "string"},
+                "message": {"type": "string"}
+            },
+            "required": ["id"]
+        }
+    }));
+
+    insta::assert_debug_snapshot!(errors);
+}

crates/validation/tests/snapshots/skim_projections_tests__ref_style_read_schema_redact.snap

@@ -0,0 +1,34 @@
+---
+source: crates/validation/tests/skim_projections_tests.rs
+expression: "(projections, errors)"
+---
+(
+    [
+        (
+            "_meta/flow_truncated",
+            "/_meta/flow_truncated",
+            "REDACT_UNSET",
+        ),
+        (
+            "flow_document",
+            "",
+            "REDACT_UNSET",
+        ),
+        (
+            "flow_published_at",
+            "/_meta/uuid",
+            "REDACT_UNSET",
+        ),
+        (
+            "id",
+            "/id",
+            "REDACT_UNSET",
+        ),
+        (
+            "secret",
+            "/secret",
+            "REDACT_BLOCK",
+        ),
+    ],
+    [],
+)

crates/validation/tests/snapshots/skim_projections_tests__single_schema_redact.snap

@@ -0,0 +1,34 @@
+---
+source: crates/validation/tests/skim_projections_tests.rs
+expression: "(projections, errors)"
+---
+(
+    [
+        (
+            "_meta/flow_truncated",
+            "/_meta/flow_truncated",
+            "REDACT_UNSET",
+        ),
+        (
+            "flow_document",
+            "",
+            "REDACT_UNSET",
+        ),
+        (
+            "flow_published_at",
+            "/_meta/uuid",
+            "REDACT_UNSET",
+        ),
+        (
+            "id",
+            "/id",
+            "REDACT_UNSET",
+        ),
+        (
+            "secret",
+            "/secret",
+            "REDACT_SHA256",
+        ),
+    ],
+    [],
+)

crates/validation/tests/snapshots/skim_projections_tests__write_schema_redact_errors_with_standalone_read_schema.snap

@@ -0,0 +1,7 @@
+---
+source: crates/validation/tests/skim_projections_tests.rs
+expression: errors
+---
+[
+    "`block` redact strategy cannot be applied at '/message' because it must exist",
+]

crates/validation/tests/snapshots/skim_projections_tests__write_schema_redact_with_standalone_read_schema.snap

@@ -0,0 +1,34 @@
+---
+source: crates/validation/tests/skim_projections_tests.rs
+expression: "(projections, errors)"
+---
+(
+    [
+        (
+            "_meta/flow_truncated",
+            "/_meta/flow_truncated",
+            "REDACT_UNSET",
+        ),
+        (
+            "flow_document",
+            "",
+            "REDACT_UNSET",
+        ),
+        (
+            "flow_published_at",
+            "/_meta/uuid",
+            "REDACT_UNSET",
+        ),
+        (
+            "id",
+            "/id",
+            "REDACT_UNSET",
+        ),
+        (
+            "secret",
+            "/secret",
+            "REDACT_BLOCK",
+        ),
+    ],
+    [],
+)