capture+materialize: Support Nonsensitive Field Overlays

crates/doc/src/annotation.rs

@@ -14,6 +14,11 @@ pub enum Annotation {
     Redact(redact::Strategy),
     /// "secret" or "airbyte_secret" annotation keyword.
     Secret(bool),
+    /// "nonsensitive" annotation keyword. Marks a location (or subtree) as not
+    /// security-relevant, meaning it may be carried in a plaintext `sops.overlay`
+    /// and modified without re-encrypting the configuration. This annotation must
+    /// only be added after human review, un-annotated locations are treated as sensitive.
+    Nonsensitive(bool),
     /// "multiline" annotation keyword marks fields that should have a multiline text input in the
     /// UI. This is from the airbyte spec.
     Multiline(bool),
@@ -41,6 +46,7 @@ impl schema::Annotation for Annotation {
             Annotation::Reduce(_) => "reduce",
             Annotation::Redact(_) => "redact",
             Annotation::Secret(_) => "secret",
+            Annotation::Nonsensitive(_) => "nonsensitive",
             Annotation::Multiline(_) => "multiline",
             Annotation::Advanced(_) => "advanced",
             Annotation::Order(_) => "order",
@@ -54,8 +60,8 @@ impl schema::Annotation for Annotation {
             // `x-str-minimum` / `x-str-maximum` are enforced validation keywords
             // not annotations so we need to preempt the starts_with("x-") check.
             schema::keywords::X_STR_MINIMUM | schema::keywords::X_STR_MAXIMUM => false,
-            "reduce" | "redact" | "secret" | "airbyte_secret" | "multiline" | "advanced"
-            | "order" | "discriminator" => true,
+            "reduce" | "redact" | "secret" | "airbyte_secret" | "nonsensitive" | "multiline"
+            | "advanced" | "order" | "discriminator" => true,
             key if key.starts_with("x-") || key.starts_with("X-") => true,
             _ => schema::CoreAnnotation::uses_keyword(keyword),
         }
@@ -69,6 +75,7 @@ impl schema::Annotation for Annotation {
             "redact" => Ok(Annotation::Redact(redact::Strategy::try_from(value)?)),
             "order" => Ok(Annotation::Order(i32::deserialize(value)?)),
             "secret" | "airbyte_secret" => Ok(Annotation::Secret(bool::deserialize(value)?)),
+            "nonsensitive" => Ok(Annotation::Nonsensitive(bool::deserialize(value)?)),
             "multiline" => Ok(Annotation::Multiline(bool::deserialize(value)?)),
             "advanced" => Ok(Annotation::Advanced(bool::deserialize(value)?)),
             "discriminator" => Ok(Annotation::Discriminator(value.clone())),

crates/doc/src/shape/inference.rs

@@ -241,6 +241,7 @@ impl Shape {
                     // but explicitly mentioned so that a compiler error will force us to check
                     // here as new annotations are added.
                     Annotation::Secret(b) => shape.secret = Some(*b),
+                    Annotation::Nonsensitive(b) => shape.nonsensitive = Some(*b),
                     Annotation::Multiline(_) => {}
                     Annotation::Advanced(_) => {}
                     Annotation::Order(_) => {}
@@ -556,6 +557,7 @@ mod test {
                     None,
                 ))),
                 secret: Some(true),
+                nonsensitive: None,
                 content_media_type: Some("some/thing".into()),
                 string: StringShape {
                     content_encoding: Some("base64".into()),
@@ -771,6 +773,7 @@ mod test {
                 provenance: Inline,
                 default: None,
                 secret: None,
+                nonsensitive: None,
                 content_media_type: None,
                 annotations: {},
                 array: ArrayShape {

crates/doc/src/shape/intersect.rs

@@ -268,6 +268,7 @@ impl Shape {
         let provenance = lhs.provenance.intersect(rhs.provenance);
         let default = lhs.default.or(rhs.default);
         let secret = lhs.secret.or(rhs.secret);
+        let nonsensitive = lhs.nonsensitive.or(rhs.nonsensitive);
 
         let content_media_type = lhs.content_media_type.or(rhs.content_media_type);
 
@@ -313,6 +314,7 @@ impl Shape {
             provenance,
             default,
             secret,
+            nonsensitive,
             content_media_type,
             annotations,
             string,

crates/doc/src/shape/mod.rs

@@ -36,6 +36,10 @@ pub struct Shape {
     pub default: Option<Box<(Value, Option<super::FailedValidation>)>>,
     /// Is this location sensitive? For example, a password or credential.
     pub secret: Option<bool>,
+    /// Is this location explicitly marked as not security-relevant? Such locations
+    /// (and their subtrees) may be carried in a plaintext `sops.overlay` and modified
+    /// without re-encrypting the configuration. `None` is treated as sensitive.
+    pub nonsensitive: Option<bool>,
     /// Annotated `contentMediaType`. The JSON-Schema specification defines
     /// this annotation only for strings; Flow extends it to apply to any
     /// type, so it lives at the top level rather than nested in a typed
@@ -196,6 +200,7 @@ impl Shape {
             provenance: Provenance::Unset,
             default: None,
             secret: None,
+            nonsensitive: None,
             content_media_type: None,
             annotations: BTreeMap::new(),
             array: ArrayShape::new(),
@@ -218,6 +223,7 @@ impl Shape {
             provenance: Provenance::Inline,
             default: None,
             secret: None,
+            nonsensitive: None,
             content_media_type: None,
             annotations: BTreeMap::new(),
             array: ArrayShape::new(),

crates/doc/src/shape/schema.rs

@@ -18,6 +18,7 @@ fn to_sub_schema(shape: Shape) -> Schema {
         provenance: _, // Not mapped to a schema.
         default,
         secret,
+        nonsensitive,
         content_media_type,
         annotations,
         array,
@@ -202,6 +203,9 @@ fn to_sub_schema(shape: Shape) -> Schema {
         if let Some(true) = secret {
             out.insert("secret".to_string(), serde_json::json!(true));
         }
+        if let Some(true) = nonsensitive {
+            out.insert("nonsensitive".to_string(), serde_json::json!(true));
+        }
 
         match reduce {
             Reduce::Unset | Reduce::Multiple => {}

crates/doc/src/shape/snapshots/doc__shape__inference__test__annotation_collection.snap

@@ -12,6 +12,7 @@ Shape {
     provenance: Inline,
     default: None,
     secret: None,
+    nonsensitive: None,
     content_media_type: None,
     annotations: {
         "x-test-top-level": Bool(true),
@@ -44,6 +45,7 @@ Shape {
                     provenance: Inline,
                     default: None,
                     secret: None,
+                    nonsensitive: None,
                     content_media_type: None,
                     annotations: {
                         "X-bar-top-level": Bool(true),
@@ -90,6 +92,7 @@ Shape {
                     provenance: Inline,
                     default: None,
                     secret: None,
+                    nonsensitive: None,
                     content_media_type: None,
                     annotations: {
                         "x-conflicting-ann": String("yes please"),
@@ -133,6 +136,7 @@ Shape {
                     provenance: Inline,
                     default: None,
                     secret: None,
+                    nonsensitive: None,
                     content_media_type: None,
                     annotations: {
                         "X-foo-top-level": Bool(false),

crates/doc/src/shape/union.rs

@@ -283,6 +283,7 @@ impl Shape {
         let provenance = lhs.provenance.union(rhs.provenance);
         let default = union_option(lhs.default, rhs.default);
         let secret = union_option(lhs.secret, rhs.secret);
+        let nonsensitive = union_option(lhs.nonsensitive, rhs.nonsensitive);
         let content_media_type = union_option(lhs.content_media_type, rhs.content_media_type);
 
         // Union of annotations is actually an _intersection_, which yields only
@@ -333,6 +334,7 @@ impl Shape {
             provenance,
             default,
             secret,
+            nonsensitive,
             content_media_type,
             annotations,
             string,

crates/flowctl/src/raw/materialize_fixture.rs

@@ -72,6 +72,7 @@ pub async fn do_materialize_fixture(
             }),
             state_json: checkpoint.to_string().into(),
             version: "test".to_string(),
+            sealed_config_json: Default::default(),
         }),
         ..Default::default()
     });

crates/proto-flow/src/capture.rs

@@ -147,6 +147,13 @@ pub mod request {
         /// Last-persisted connector checkpoint state from a previous invocation.
         #[prost(bytes = "bytes", tag = "4")]
         pub state_json: ::prost::bytes::Bytes,
+        /// Sealed (encrypted) endpoint configuration of this capture.
+        /// This is the SOPS-encrypted document from which the runtime derived the
+        /// decrypted `capture.config_json`, including any `sops` metadata stanza.
+        /// Connectors may reuse it to emit `configUpdate`s that modify nonsensitive
+        /// overlay fields without re-encrypting the configuration.
+        #[prost(bytes = "bytes", tag = "5")]
+        pub sealed_config_json: ::prost::bytes::Bytes,
     }
     /// Tell the connector that some number of its preceding Checkpoints have
     /// committed to the Flow recovery log.

crates/proto-flow/src/capture.serde.rs

@@ -624,6 +624,9 @@ impl serde::Serialize for request::Open {
         if !self.state_json.is_empty() {
             len += 1;
         }
+        if !self.sealed_config_json.is_empty() {
+            len += 1;
+        }
         let mut struct_ser = serializer.serialize_struct("capture.Request.Open", len)?;
         if let Some(v) = self.capture.as_ref() {
             struct_ser.serialize_field("capture", v)?;
@@ -639,6 +642,11 @@ impl serde::Serialize for request::Open {
             #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("state", &crate::as_raw_json(&self.state_json)?)?;
         }
+        if !self.sealed_config_json.is_empty() {
+            #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
+            struct_ser.serialize_field("sealedConfig", &crate::as_raw_json(&self.sealed_config_json)?)?;
+        }
         struct_ser.end()
     }
 }
@@ -654,6 +662,8 @@ impl<'de> serde::Deserialize<'de> for request::Open {
             "range",
             "state_json",
             "state",
+            "sealed_config_json",
+            "sealedConfig",
         ];
 
         #[allow(clippy::enum_variant_names)]
@@ -662,6 +672,7 @@ impl<'de> serde::Deserialize<'de> for request::Open {
             Version,
             Range,
             StateJson,
+            SealedConfigJson,
             __SkipField__,
         }
         impl<'de> serde::Deserialize<'de> for GeneratedField {
@@ -688,6 +699,7 @@ impl<'de> serde::Deserialize<'de> for request::Open {
                             "version" => Ok(GeneratedField::Version),
                             "range" => Ok(GeneratedField::Range),
                             "state" | "state_json" => Ok(GeneratedField::StateJson),
+                            "sealedConfig" | "sealed_config_json" => Ok(GeneratedField::SealedConfigJson),
                             _ => Ok(GeneratedField::__SkipField__),
                         }
                     }
@@ -711,6 +723,7 @@ impl<'de> serde::Deserialize<'de> for request::Open {
                 let mut version__ = None;
                 let mut range__ = None;
                 let mut state_json__ = None;
+                let mut sealed_config_json__ = None;
                 while let Some(k) = map_.next_key()? {
                     match k {
                         GeneratedField::Capture => {
@@ -739,6 +752,14 @@ impl<'de> serde::Deserialize<'de> for request::Open {
                                 Some(map_.next_value::<crate::RawJSONDeserialize>()?.0)
                             ;
                         }
+                        GeneratedField::SealedConfigJson => {
+                            if sealed_config_json__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("sealedConfig"));
+                            }
+                            sealed_config_json__ = 
+                                Some(map_.next_value::<crate::RawJSONDeserialize>()?.0)
+                            ;
+                        }
                         GeneratedField::__SkipField__ => {
                             let _ = map_.next_value::<serde::de::IgnoredAny>()?;
                         }
@@ -749,6 +770,7 @@ impl<'de> serde::Deserialize<'de> for request::Open {
                     version: version__.unwrap_or_default(),
                     range: range__,
                     state_json: state_json__.unwrap_or_default(),
+                    sealed_config_json: sealed_config_json__.unwrap_or_default(),
                 })
             }
         }

crates/proto-flow/src/materialize.rs

@@ -160,6 +160,13 @@ pub mod request {
         /// Last-persisted connector checkpoint state from a previous session.
         #[prost(bytes = "bytes", tag = "4")]
         pub state_json: ::prost::bytes::Bytes,
+        /// Sealed (encrypted) endpoint configuration of this materialization.
+        /// This is the SOPS-encrypted document from which the runtime derived the
+        /// decrypted `materialization.config_json`, including any `sops` metadata stanza.
+        /// Connectors may reuse it to emit `configUpdate`s that modify nonsensitive
+        /// overlay fields without re-encrypting the configuration.
+        #[prost(bytes = "bytes", tag = "5")]
+        pub sealed_config_json: ::prost::bytes::Bytes,
     }
     /// Load a document identified by its key. The given key may have never before been stored,
     /// but a given key will be sent in a transaction Load just one time.

crates/proto-flow/src/materialize.serde.rs

@@ -1083,6 +1083,9 @@ impl serde::Serialize for request::Open {
         if !self.state_json.is_empty() {
             len += 1;
         }
+        if !self.sealed_config_json.is_empty() {
+            len += 1;
+        }
         let mut struct_ser = serializer.serialize_struct("materialize.Request.Open", len)?;
         if let Some(v) = self.materialization.as_ref() {
             struct_ser.serialize_field("materialization", v)?;
@@ -1098,6 +1101,11 @@ impl serde::Serialize for request::Open {
             #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("state", &crate::as_raw_json(&self.state_json)?)?;
         }
+        if !self.sealed_config_json.is_empty() {
+            #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
+            struct_ser.serialize_field("sealedConfig", &crate::as_raw_json(&self.sealed_config_json)?)?;
+        }
         struct_ser.end()
     }
 }
@@ -1113,6 +1121,8 @@ impl<'de> serde::Deserialize<'de> for request::Open {
             "range",
             "state_json",
             "state",
+            "sealed_config_json",
+            "sealedConfig",
         ];
 
         #[allow(clippy::enum_variant_names)]
@@ -1121,6 +1131,7 @@ impl<'de> serde::Deserialize<'de> for request::Open {
             Version,
             Range,
             StateJson,
+            SealedConfigJson,
             __SkipField__,
         }
         impl<'de> serde::Deserialize<'de> for GeneratedField {
@@ -1147,6 +1158,7 @@ impl<'de> serde::Deserialize<'de> for request::Open {
                             "version" => Ok(GeneratedField::Version),
                             "range" => Ok(GeneratedField::Range),
                             "state" | "state_json" => Ok(GeneratedField::StateJson),
+                            "sealedConfig" | "sealed_config_json" => Ok(GeneratedField::SealedConfigJson),
                             _ => Ok(GeneratedField::__SkipField__),
                         }
                     }
@@ -1170,6 +1182,7 @@ impl<'de> serde::Deserialize<'de> for request::Open {
                 let mut version__ = None;
                 let mut range__ = None;
                 let mut state_json__ = None;
+                let mut sealed_config_json__ = None;
                 while let Some(k) = map_.next_key()? {
                     match k {
                         GeneratedField::Materialization => {
@@ -1198,6 +1211,14 @@ impl<'de> serde::Deserialize<'de> for request::Open {
                                 Some(map_.next_value::<crate::RawJSONDeserialize>()?.0)
                             ;
                         }
+                        GeneratedField::SealedConfigJson => {
+                            if sealed_config_json__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("sealedConfig"));
+                            }
+                            sealed_config_json__ = 
+                                Some(map_.next_value::<crate::RawJSONDeserialize>()?.0)
+                            ;
+                        }
                         GeneratedField::__SkipField__ => {
                             let _ = map_.next_value::<serde::de::IgnoredAny>()?;
                         }
@@ -1208,6 +1229,7 @@ impl<'de> serde::Deserialize<'de> for request::Open {
                     version: version__.unwrap_or_default(),
                     range: range__,
                     state_json: state_json__.unwrap_or_default(),
+                    sealed_config_json: sealed_config_json__.unwrap_or_default(),
                 })
             }
         }

crates/proto-flow/tests/regression.rs

@@ -440,6 +440,9 @@ fn ex_capture_request() -> capture::Request {
             version: "11:22:33:44".to_string(),
             range: Some(ex_range()),
             state_json: json!({"connector": {"state": 42}}).to_string().into(),
+            sealed_config_json: json!({"encrypted": "c2VjcmV0", "sops": {"mac": "abc"}})
+                .to_string()
+                .into(),
         }),
         acknowledge: Some(capture::request::Acknowledge { checkpoints: 32 }),
         internal: ex_internal(),
@@ -625,6 +628,9 @@ fn ex_materialize_request() -> materialize::Request {
             version: "11:22:33:44".to_string(),
             range: Some(ex_range()),
             state_json: json!({"connector": {"state": 42}}).to_string().into(),
+            sealed_config_json: json!({"encrypted": "c2VjcmV0", "sops": {"mac": "abc"}})
+                .to_string()
+                .into(),
         }),
         acknowledge: Some(materialize::request::Acknowledge {
             state_patches_json: json!([{"acked": true}]).to_string().into(),