feat(core): add masked_connector_response with per-connector key allowlist

.github/workflows/ci.yml

@@ -547,15 +547,15 @@ jobs:
             done
             echo "🔍 Connector-only PR — scoping tests to: $CONNECTOR_NAMES"
             cargo nextest run \
-              --features grpc-server/connector-request-kafka \
+              --features grpc-server/connector-request-kafka,grpc-server/connector-response-masking \
               --config-file .nextest.toml \
               --profile ci \
               --no-tests=warn \
               -E "($FILTER) & ($SCHEMA_EXCLUDE)"
           else
             echo "🔍 Running full test suite (event=$EVENT)"
             cargo nextest run \
-              --features grpc-server/connector-request-kafka \
+              --features grpc-server/connector-request-kafka,grpc-server/connector-response-masking \
               --config-file .nextest.toml \
               --profile ci \
               -E "$SCHEMA_EXCLUDE"

Dockerfile

@@ -55,6 +55,8 @@ RUN apt-get update \
     && rm -rf /var/lib/apt/lists/*
 
 # Build only the binary shipped by the runtime stage; skips test/SDK crates.
+# `connector-response-masking` is deliberately absent: the code that turns a connector response
+# into `masked_connector_response` is then not compiled in, so no config setting can enable it here.
 COPY . .
 RUN --mount=type=cache,target=/sccache \
     cargo build --release --features kafka,connector-request-kafka,otel -p grpc-server

config/development.toml

@@ -237,3 +237,37 @@ enqueue_timeout_ms = 5000
 
 [unmasked_headers]
 keys = ["x-request-id","x-merchant-id","x-lineage-ids","x-reference-id","x-connector","x-tenant-id","x-shadow-mode","x-proxy-name"]
+
+# Selectively-masked view of the connector response, exposed as
+# `masked_connector_response`. Every key is preserved; a value is shown only if
+# that connector's list below names it. Gated separately from
+# `common.return_raw_connector_data`, so this can stay on where raw capture is off.
+#
+# This whole section only takes effect in a build compiled with
+# `--features connector-response-masking`. Without it the code is not in the binary and every
+# setting below is ignored, `enabled` included.
+[connector_response_masking]
+enabled = true
+
+# Whether to ALSO write the masked view to our own logs (`response.masked_body`).
+# `enabled` above already returns it to the caller; this is the extra copy we retain,
+# so it stays off outside development.
+log_to_span = true
+
+# Per-connector unmask lists, comma-separated and case-insensitive.
+# A connector with no entry gets every value masked (keys still visible).
+# Naming a key here reveals only that key's own value: an object below it is re-decided key by
+# key, and an array below it stays masked, since its elements have no key you could name.
+# A key whose name looks like a full PAN, CVV, expiry or credential stays masked even if listed.
+# Truncated values (cardSummary, last4, cardBin) are not covered — name them if needed.
+# A body that is not JSON, XML or form-encoded has no keys to gate on, so it is replaced
+# wholesale by a stub carrying only its size.
+#
+# Names are validated against every connector enum — payment, surcharge, payout, FRM
+# and authenticator — so interpayments, deutschebank and plaid are valid keys here.
+#
+# Only one entry is seeded, as a worked example for testing. Add a line per
+# connector as you need its fields visible; an unknown connector name here will
+# abort startup rather than be ignored.
+[connector_response_masking.connector_keys]
+adyen = "pspreference,resultcode,merchantreference,refusalreason,eventcode,success"

config/production.toml

@@ -178,3 +178,36 @@ keys = ["x-request-id","x-merchant-id","x-lineage-ids","x-reference-id","x-conne
 # Connectors that require an external API call for webhook source verification
 [webhook_source_verification_call] # comma-separated list of connector names (case-insensitive)
 connectors_with_webhook_source_verification_call = "paypal, truelayer"
+
+# Selectively-masked view of the connector response, exposed as
+# `masked_connector_response`. Every key is preserved; a value is shown only if
+# that connector's list below names it. Gated separately from
+# `common.return_raw_connector_data`, so this can stay on where raw capture is off.
+#
+# This whole section only takes effect in a build compiled with
+# `--features connector-response-masking`. The release image does not enable it, so the code is
+# absent and every setting below is ignored, `enabled` included.
+[connector_response_masking]
+# Off by default: turning this on returns connector response bytes to the caller, so each
+# deployment should opt in once it has chosen the key lists below.
+enabled = false
+
+# Whether to ALSO write the masked view to our own logs (`response.masked_body`).
+# `enabled` above already returns it to the caller; this is the extra copy we retain,
+# so a mistaken allowlist entry stays contained to whoever configured it.
+log_to_span = false
+
+# Per-connector unmask lists, comma-separated and case-insensitive.
+# A connector with no entry gets every value masked (keys still visible).
+# Naming a key here reveals only that key's own value: an object below it is re-decided key by
+# key, and an array below it stays masked, since its elements have no key you could name.
+# A key whose name looks like a full PAN, CVV, expiry or credential stays masked even if listed.
+# Truncated values (cardSummary, last4, cardBin) are not covered — name them if needed.
+# A body that is not JSON, XML or form-encoded has no keys to gate on, so it is replaced
+# wholesale by a stub carrying only its size.
+#
+# Only one entry is seeded, as a worked example for testing. Add a line per
+# connector as you need its fields visible; an unknown connector name here will
+# abort startup rather than be ignored.
+[connector_response_masking.connector_keys]
+adyen = "pspreference,resultcode,merchantreference,refusalreason,eventcode,success"

config/sandbox.toml

@@ -179,3 +179,36 @@ keys = ["x-request-id","x-merchant-id","x-lineage-ids","x-reference-id","x-conne
 # Connectors that require an external API call for webhook source verification
 [webhook_source_verification_call] # comma-separated list of connector names (case-insensitive)
 connectors_with_webhook_source_verification_call = "paypal, truelayer"
+
+# Selectively-masked view of the connector response, exposed as
+# `masked_connector_response`. Every key is preserved; a value is shown only if
+# that connector's list below names it. Gated separately from
+# `common.return_raw_connector_data`, so this can stay on where raw capture is off.
+#
+# This whole section only takes effect in a build compiled with
+# `--features connector-response-masking`. The release image does not enable it, so the code is
+# absent and every setting below is ignored, `enabled` included.
+[connector_response_masking]
+# Off by default: turning this on returns connector response bytes to the caller, so each
+# deployment should opt in once it has chosen the key lists below.
+enabled = false
+
+# Whether to ALSO write the masked view to our own logs (`response.masked_body`).
+# `enabled` above already returns it to the caller; this is the extra copy we retain,
+# so a mistaken allowlist entry stays contained to whoever configured it.
+log_to_span = false
+
+# Per-connector unmask lists, comma-separated and case-insensitive.
+# A connector with no entry gets every value masked (keys still visible).
+# Naming a key here reveals only that key's own value: an object below it is re-decided key by
+# key, and an array below it stays masked, since its elements have no key you could name.
+# A key whose name looks like a full PAN, CVV, expiry or credential stays masked even if listed.
+# Truncated values (cardSummary, last4, cardBin) are not covered — name them if needed.
+# A body that is not JSON, XML or form-encoded has no keys to gate on, so it is replaced
+# wholesale by a stub carrying only its size.
+#
+# Only one entry is seeded, as a worked example for testing. Add a line per
+# connector as you need its fields visible; an unknown connector name here will
+# abort startup rather than be ignored.
+[connector_response_masking.connector_keys]
+adyen = "pspreference,resultcode,merchantreference,refusalreason,eventcode,success"

crates/common/common_utils/src/bytes_utils.rs

@@ -0,0 +1,16 @@
+//! Small helpers over raw byte slices.
+
+const UTF8_BOM: &[u8] = &[0xEF, 0xBB, 0xBF];
+
+/// Strip any leading UTF-8 BOMs from `bytes`.
+///
+/// Several gateways (Authorize.Net among them) prefix responses with a BOM, which every body
+/// parser rejects. Takes bytes rather than a decoded string so callers can strip before deciding
+/// whether the body is even UTF-8. Repeated BOMs are all removed.
+pub fn strip_utf8_bom(bytes: &[u8]) -> &[u8] {
+    let mut rest = bytes;
+    while let Some(stripped) = rest.strip_prefix(UTF8_BOM) {
+        rest = stripped;
+    }
+    rest
+}

crates/common/common_utils/src/events.rs

@@ -1,5 +1,6 @@
 use hyperswitch_masking::ErasedMaskSerialize;
 use serde::{Deserialize, Serialize};
+use std::borrow::Cow;
 use std::collections::HashMap;
 
 use crate::errors::EventPublisherError;
@@ -320,6 +321,24 @@ impl Event {
         });
     }
 
+    /// Carry the connector's reply, every value masked except the ones that connector's allowlist
+    /// names, so a consumer can read it off the event stream.
+    ///
+    /// Recorded on the connector-call event because that is where it is produced, next to the
+    /// request and response it describes. Attached whenever it exists: delivery to consumers is
+    /// the point of the field and is deliberately independent of whether we log it ourselves —
+    /// see [`MASKED_CONNECTOR_RESPONSE_KEY`].
+    pub fn add_masked_connector_response(&mut self, masked_response: &str) {
+        MaskedSerdeValue::from_masked_optional(
+            &masked_response.to_string(),
+            MASKED_CONNECTOR_RESPONSE_KEY,
+        )
+        .map(|masked| {
+            self.additional_fields
+                .insert(MASKED_CONNECTOR_RESPONSE_KEY.to_string(), masked);
+        });
+    }
+
     pub fn set_grpc_error_response(&mut self, tonic_error: &tonic::Status) {
         self.status_code = Some(tonic_error.code().into());
         let error_body = serde_json::json!({
@@ -515,16 +534,73 @@ impl Default for EventConfig {
     }
 }
 
+/// Serde key of the connector's masked reply, wherever it appears in an event.
+///
+/// Named once because two things must agree on it: [`Event::add_masked_connector_response`], which
+/// writes it, and the callers that ask [`emit_event_with_config_redacting`] to keep it out of our
+/// logs.
+pub const MASKED_CONNECTOR_RESPONSE_KEY: &str = "masked_connector_response";
+
+/// Drop `keys` from `value` wherever they occur, at any depth.
+///
+/// Recursive rather than top-level: the same key sits at different depths depending on which event
+/// carries it — top level on the connector event, under `response_data` on the gRPC event, and
+/// nested again inside `event_content.content` for webhook responses. Borrows unless something
+/// actually matches, so the usual empty-`keys` call costs nothing.
+pub fn without_keys<'a>(value: &'a serde_json::Value, keys: &[&str]) -> Cow<'a, serde_json::Value> {
+    fn contains(value: &serde_json::Value, keys: &[&str]) -> bool {
+        match value {
+            serde_json::Value::Object(map) => map
+                .iter()
+                .any(|(key, nested)| keys.contains(&key.as_str()) || contains(nested, keys)),
+            serde_json::Value::Array(items) => items.iter().any(|item| contains(item, keys)),
+            _ => false,
+        }
+    }
+
+    fn strip(value: &mut serde_json::Value, keys: &[&str]) {
+        match value {
+            serde_json::Value::Object(map) => {
+                map.retain(|key, _| !keys.contains(&key.as_str()));
+                map.values_mut().for_each(|nested| strip(nested, keys));
+            }
+            serde_json::Value::Array(items) => items.iter_mut().for_each(|item| strip(item, keys)),
+            _ => {}
+        }
+    }
+
+    if keys.is_empty() || !contains(value, keys) {
+        return Cow::Borrowed(value);
+    }
+    let mut owned = value.clone();
+    strip(&mut owned, keys);
+    Cow::Owned(owned)
+}
+
 /// Emit an event: always processes and logs; publishes to Kafka only when kafka feature is enabled.
 pub fn emit_event_with_config(event: Event, config: &EventConfig) {
+    emit_event_with_config_redacting(event, config, &[])
+}
+
+/// As [`emit_event_with_config`], but omits `redacted_from_log` from the copy written to our own
+/// logs.
+///
+/// The published payload is deliberately untouched. A field can be required by a downstream
+/// consumer and still be something we decline to retain ourselves, which is exactly the shape
+/// `connector_response_masking.log_to_span` asks for.
+pub fn emit_event_with_config_redacting(
+    event: Event,
+    config: &EventConfig,
+    redacted_from_log: &[&str],
+) {
     let processed_event = match process_event_with_config(&event, config) {
         Ok(processed) => processed,
         Err(e) => {
             tracing::error!(error = ?e, "Failed to process event");
             return;
         }
     };
-    let event_json = serde_json::to_string(&processed_event)
+    let event_json = serde_json::to_string(&without_keys(&processed_event, redacted_from_log))
         .unwrap_or_else(|e| format!("{{\"error\":\"Failed to serialize event: {}\"}}", e));
     tracing::info!(
         events_enabled = config.enabled,
@@ -779,3 +855,95 @@ mod runtime_metadata_tests {
         assert_eq!(value.get("version").and_then(|v| v.as_str()), Some("v"));
     }
 }
+
+#[cfg(test)]
+#[allow(clippy::expect_used)]
+mod log_redaction_tests {
+    use super::{without_keys, MASKED_CONNECTOR_RESPONSE_KEY};
+    use serde_json::json;
+    use std::borrow::Cow;
+
+    const KEYS: &[&str] = &[MASKED_CONNECTOR_RESPONSE_KEY];
+
+    #[test]
+    fn removes_the_key_at_the_top_level() {
+        // Where it lands on the connector-call event: `additional_fields` is flattened.
+        let value = json!({"masked_connector_response": "{\"id\":\"1\"}", "connector": "adyen"});
+        let out = without_keys(&value, KEYS);
+        assert!(out.get(MASKED_CONNECTOR_RESPONSE_KEY).is_none());
+        assert_eq!(out.get("connector").and_then(|v| v.as_str()), Some("adyen"));
+    }
+
+    #[test]
+    fn removes_the_key_when_nested() {
+        // Where it lands on the gRPC event (`response_data`) and, deeper, inside a webhook
+        // response (`event_content.content`). A top-level-only strip passes the first and
+        // silently leaks the second.
+        let value = json!({
+            "response_data": {
+                "status": "CHARGED",
+                "masked_connector_response": "{\"pan\":\"***\"}",
+                "event_content": {
+                    "content": {"masked_connector_response": "{\"deep\":\"***\"}"}
+                }
+            }
+        });
+        let out = without_keys(&value, KEYS);
+        assert!(!out.to_string().contains(MASKED_CONNECTOR_RESPONSE_KEY));
+        assert_eq!(
+            out.pointer("/response_data/status")
+                .and_then(|v| v.as_str()),
+            Some("CHARGED"),
+            "siblings must survive"
+        );
+    }
+
+    #[test]
+    fn removes_the_key_inside_arrays() {
+        let value = json!({"items": [{"masked_connector_response": "x", "keep": 1}]});
+        let out = without_keys(&value, KEYS);
+        assert!(!out.to_string().contains(MASKED_CONNECTOR_RESPONSE_KEY));
+        assert_eq!(
+            out.pointer("/items/0/keep").and_then(|v| v.as_u64()),
+            Some(1)
+        );
+    }
+
+    #[test]
+    fn borrows_when_nothing_matches() {
+        // The `log_to_span = true` and no-redaction paths must not pay for a deep clone.
+        let value = json!({"response_data": {"status": "CHARGED"}});
+        assert!(matches!(without_keys(&value, KEYS), Cow::Borrowed(_)));
+        assert!(matches!(without_keys(&value, &[]), Cow::Borrowed(_)));
+    }
+
+    #[test]
+    fn an_empty_key_list_is_a_no_op_even_when_the_key_is_present() {
+        // This is the `log_to_span = true` case: the value stays in the logged copy.
+        let value = json!({"masked_connector_response": "kept"});
+        let out = without_keys(&value, &[]);
+        assert_eq!(
+            out.get(MASKED_CONNECTOR_RESPONSE_KEY)
+                .and_then(|v| v.as_str()),
+            Some("kept")
+        );
+    }
+
+    #[test]
+    fn the_published_value_is_never_mutated() {
+        // The property the whole change rests on: redaction produces a separate copy for the log,
+        // leaving the payload handed to Kafka intact.
+        let published =
+            json!({"masked_connector_response": "{\"id\":\"1\"}", "connector": "adyen"});
+        let logged = without_keys(&published, KEYS).into_owned();
+
+        assert!(logged.get(MASKED_CONNECTOR_RESPONSE_KEY).is_none());
+        assert_eq!(
+            published
+                .get(MASKED_CONNECTOR_RESPONSE_KEY)
+                .and_then(|v| v.as_str()),
+            Some("{\"id\":\"1\"}"),
+            "the published copy must still carry it"
+        );
+    }
+}

crates/common/common_utils/src/lib.rs

@@ -2,6 +2,7 @@
 
 extern crate self as common_utils;
 
+pub mod bytes_utils;
 pub mod config_patch;
 pub mod crypto;
 pub mod custom_serde;
@@ -23,7 +24,10 @@ pub mod types;
 pub use errors::{CustomResult, EventPublisherError, ParsingError, ValidationError};
 #[cfg(feature = "kafka")]
 pub use event_publisher::init_event_publisher;
-pub use events::emit_event_with_config;
+pub use events::{
+    emit_event_with_config, emit_event_with_config_redacting, without_keys,
+    MASKED_CONNECTOR_RESPONSE_KEY,
+};
 
 #[cfg(not(feature = "kafka"))]
 pub fn init_event_publisher(_config: &events::EventConfig) {}