feat(proxy): accept inbound EQL ciphertext payloads

CHANGELOG.md

@@ -6,6 +6,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Added
+
+- **Pre-encrypted EQL values in statements**: SQL literals and bound parameters may now carry EQL v3 storage payloads produced by an application. Proxy validates their wire shape, version, destination column and required SEM terms, authenticates their ciphertext with the connection's active keyset, and forwards them without encrypting them again. Invalid payloads fail closed with one generic error so validation details cannot be used as an oracle.
+
 ## [3.0.1] - 2026-08-05
 
 ### Added

packages/cipherstash-proxy-integration/src/inbound_ciphertext.rs

@@ -0,0 +1,140 @@
+//! End-to-end coverage for application-encrypted EQL payloads entering Proxy.
+
+#[cfg(test)]
+mod tests {
+    use crate::common::{clear_with_client, connect_with_tls, random_id, PROXY};
+    use cipherstash_client::{
+        encryption::{Plaintext, ScopedCipher},
+        eql::{
+            encrypt_eql_v3, EqlEncryptOpts, EqlOperation, EqlOutputV3, Identifier,
+            PreparedPlaintext,
+        },
+        schema::{column::Index, ColumnConfig, ColumnType},
+        zerokms::{ClientKey, ZeroKMSBuilder},
+        AutoStrategy, IdentifiedBy,
+    };
+    use std::{borrow::Cow, sync::Arc};
+    use uuid::Uuid;
+
+    async fn cipher() -> Arc<ScopedCipher<AutoStrategy>> {
+        let client_id = env("CS_CLIENT_ID", "CS_ENCRYPT__CLIENT_ID")
+            .parse()
+            .expect("CS_CLIENT_ID must be a UUID");
+        let client_key =
+            ClientKey::from_hex_v1(client_id, &env("CS_CLIENT_KEY", "CS_ENCRYPT__CLIENT_KEY"))
+                .expect("CS_CLIENT_KEY must be valid");
+        let zerokms = ZeroKMSBuilder::auto()
+            .expect("ZeroKMS credentials must be configured")
+            .with_client_key(client_key)
+            .build()
+            .expect("ZeroKMS client must initialize");
+        let keyset_id: Uuid = env("CS_DEFAULT_KEYSET_ID", "CS_ENCRYPT__DEFAULT_KEYSET_ID")
+            .parse()
+            .expect("CS_DEFAULT_KEYSET_ID must be a UUID");
+        Arc::new(
+            ScopedCipher::init(Arc::new(zerokms), Some(IdentifiedBy::Uuid(keyset_id)))
+                .await
+                .expect("scoped cipher must initialize"),
+        )
+    }
+
+    fn env(primary: &str, nested: &str) -> String {
+        std::env::var(primary)
+            .or_else(|_| std::env::var(nested))
+            .unwrap_or_else(|_| panic!("{primary} must be configured"))
+    }
+
+    fn text_search_config(column: &str) -> ColumnConfig {
+        ColumnConfig::build(column)
+            .casts_as(ColumnType::Text)
+            .add_index(Index::new_unique())
+            .add_index(Index::new_ope())
+            .add_index(Index::new_match())
+    }
+
+    async fn encrypt_text(table: &str, column: &str, plaintext: &str) -> String {
+        let prepared = PreparedPlaintext::new(
+            Cow::Owned(text_search_config(column)),
+            Identifier::new(table, column),
+            Plaintext::from(plaintext),
+            EqlOperation::Store,
+        );
+        let mut outputs =
+            encrypt_eql_v3(cipher().await, vec![prepared], &EqlEncryptOpts::default())
+                .await
+                .expect("application-side encryption must succeed");
+        let EqlOutputV3::Store(ciphertext) = outputs.remove(0) else {
+            panic!("store encryption must return a stored payload");
+        };
+        serde_json::to_string(&ciphertext).unwrap()
+    }
+
+    #[tokio::test]
+    async fn accepts_pre_encrypted_parameter_for_storage_and_search() {
+        let client = connect_with_tls(*PROXY).await;
+        clear_with_client(&client).await;
+        let id = random_id();
+        let plaintext = "encrypted in the application";
+        let payload = encrypt_text("encrypted", "encrypted_text", plaintext).await;
+
+        client
+            .execute(
+                "INSERT INTO encrypted (id, encrypted_text) VALUES ($1, $2)",
+                &[&id, &payload],
+            )
+            .await
+            .unwrap();
+
+        let rows = client
+            .query(
+                "SELECT encrypted_text FROM encrypted WHERE encrypted_text = $1",
+                &[&payload],
+            )
+            .await
+            .unwrap();
+        assert_eq!(rows[0].get::<_, String>(0), plaintext);
+    }
+
+    #[tokio::test]
+    async fn accepts_pre_encrypted_literal_for_storage() {
+        let client = connect_with_tls(*PROXY).await;
+        clear_with_client(&client).await;
+        let id = random_id();
+        let plaintext = "application encrypted literal";
+        let payload = encrypt_text("encrypted", "encrypted_text", plaintext).await;
+        let payload = payload.replace('\'', "''");
+
+        client
+            .simple_query(&format!(
+                "INSERT INTO encrypted (id, encrypted_text) VALUES ({id}, '{payload}')"
+            ))
+            .await
+            .unwrap();
+
+        let row = client
+            .query_one("SELECT encrypted_text FROM encrypted WHERE id = $1", &[&id])
+            .await
+            .unwrap();
+        assert_eq!(row.get::<_, String>(0), plaintext);
+    }
+
+    #[tokio::test]
+    async fn rejects_payload_for_a_different_destination_with_generic_error() {
+        let client = connect_with_tls(*PROXY).await;
+        clear_with_client(&client).await;
+        let id = random_id();
+        let payload = encrypt_text("some_other_table", "encrypted_text", "secret").await;
+
+        let error = client
+            .execute(
+                "INSERT INTO encrypted (id, encrypted_text) VALUES ($1, $2)",
+                &[&id, &payload],
+            )
+            .await
+            .expect_err("destination mismatch must fail closed");
+        assert_eq!(
+            error.as_db_error().unwrap().message(),
+            "Invalid encrypted value"
+        );
+    }
+}

packages/cipherstash-proxy-integration/src/lib.rs

@@ -7,6 +7,7 @@ mod empty_result;
 mod encryption_sanity;
 mod eql_regression;
 mod extended_protocol_error_messages;
+mod inbound_ciphertext;
 mod insert;
 mod legacy_v2_column;
 mod map_concat;

packages/cipherstash-proxy/src/error.rs

@@ -75,6 +75,7 @@ impl Error {
             // stores plaintext in a column its operator believes is encrypted
             // (CIP-3688). No configuration may turn that back on.
             Error::Mapping(MappingError::UnmappableEncryptedColumn { .. })
+                | Error::Encrypt(EncryptError::InvalidInboundCiphertext)
         )
     }
 }
@@ -255,6 +256,12 @@ pub enum TlsConfigError {
 
 #[derive(Error, Debug)]
 pub enum EncryptError {
+    /// Deliberately contains no payload or validation detail: inbound
+    /// ciphertext failures are attacker-controlled and detailed responses can
+    /// become an oracle.
+    #[error("Invalid encrypted value")]
+    InvalidInboundCiphertext,
+
     #[error(transparent)]
     CiphertextCouldNotBeSerialised(#[from] serde_json::Error),
 

packages/cipherstash-proxy/src/postgresql/frontend.rs

@@ -21,6 +21,7 @@ use crate::postgresql::context::Portal;
 use crate::postgresql::data::{
     compose_json_selector_path, json_value_selector_plaintext, literal_from_sql, literal_json_value,
 };
+use crate::postgresql::inbound_eql;
 use crate::postgresql::messages::close::Close;
 use crate::postgresql::messages::error_response::ErrorResponseCode;
 use crate::postgresql::messages::ready_for_query::ReadyForQuery;
@@ -648,7 +649,18 @@ where
             return Ok(vec![]);
         }
 
-        let plaintexts = literals_to_plaintext(typed_statement, literal_columns)?;
+        let inbound = literal_values
+            .iter()
+            .zip(literal_columns)
+            .map(|((_, literal), column)| {
+                let (Some(column), Some(value)) = (column, (*literal).clone().into_string()) else {
+                    return Ok(None);
+                };
+                inbound_eql::parse(value.as_bytes(), column).map_err(Error::from)
+            })
+            .collect::<Result<Vec<_>, Error>>()?;
+        let skip = inbound.iter().map(Option::is_some).collect::<Vec<_>>();
+        let plaintexts = literals_to_plaintext_skipping(typed_statement, literal_columns, &skip)?;
 
         let start = Instant::now();
 
@@ -660,6 +672,9 @@ where
                 counter!(ENCRYPTION_ERROR_TOTAL).increment(1);
             })?;
 
+        self.authenticate_and_merge_inbound(&mut encrypted, inbound)
+            .await?;
+
         for ((_, literal), encrypted) in literal_values.iter().zip(encrypted.iter_mut()) {
             project_query_operand(
                 typed_statement.query_operands.contains_literal(literal),
@@ -1156,8 +1171,13 @@ where
         bind: &Bind,
         statement: &Statement,
     ) -> Result<Vec<Option<crate::EqlOutput>>, Error> {
-        let plaintexts =
-            bind.to_plaintext(&statement.output_params, &statement.postgres_param_types)?;
+        let inbound = bind.inbound_ciphertexts(&statement.output_params)?;
+        let skip = inbound.iter().map(Option::is_some).collect::<Vec<_>>();
+        let plaintexts = bind.to_plaintext_skipping(
+            &statement.output_params,
+            &statement.postgres_param_types,
+            &skip,
+        )?;
 
         // Encryption is positional over the OUTPUT params — the values actually
         // sent — not over what the client bound.
@@ -1179,6 +1199,9 @@ where
                 counter!(ENCRYPTION_ERROR_TOTAL).increment(1);
             })?;
 
+        self.authenticate_and_merge_inbound(&mut encrypted, inbound)
+            .await?;
+
         for (output, encrypted) in statement.output_params.iter().zip(encrypted.iter_mut()) {
             project_query_operand(output.query_operand, encrypted);
         }
@@ -1205,6 +1228,37 @@ where
         Ok(encrypted)
     }
 
+    /// Authenticate inbound ciphertext with this connection's scoped cipher.
+    /// Any parse, metadata, key or AEAD failure is collapsed to one response.
+    async fn authenticate_and_merge_inbound(
+        &self,
+        encrypted: &mut [Option<EqlOutput>],
+        inbound: Vec<Option<crate::EqlCiphertext>>,
+    ) -> Result<(), Error> {
+        let positions = inbound
+            .iter()
+            .enumerate()
+            .filter_map(|(index, ciphertext)| ciphertext.as_ref().map(|ct| (index, ct.clone())))
+            .collect::<Vec<_>>();
+        if positions.is_empty() {
+            return Ok(());
+        }
+
+        let ciphertexts = positions
+            .iter()
+            .map(|(_, ciphertext)| Some(ciphertext.clone()))
+            .collect();
+        self.context
+            .decrypt(ciphertexts)
+            .await
+            .map_err(|_| EncryptError::InvalidInboundCiphertext)?;
+
+        for (index, ciphertext) in positions {
+            encrypted[index] = Some(EqlOutput::Store(ciphertext));
+        }
+        Ok(())
+    }
+
     fn type_check<'a>(
         &self,
         statement: &'a ast::Statement,
@@ -1391,45 +1445,59 @@ fn project_query_operand(query_operand: bool, encrypted: &mut Option<EqlOutput>)
 fn literals_to_plaintext(
     typed_statement: &TypeCheckedStatement<'_>,
     literal_columns: &Vec<Option<Column>>,
+) -> Result<Vec<Option<Plaintext>>, Error> {
+    literals_to_plaintext_skipping(typed_statement, literal_columns, &[])
+}
+
+fn literals_to_plaintext_skipping(
+    typed_statement: &TypeCheckedStatement<'_>,
+    literal_columns: &Vec<Option<Column>>,
+    skip: &[bool],
 ) -> Result<Vec<Option<Plaintext>>, Error> {
     let literals = typed_statement.literal_values();
 
     let plaintexts = literals
         .iter()
         .zip(literal_columns)
-        .map(|((eql_term, val), col)| match col {
-            Some(col) => {
-                let plaintext = match eql_term.variant() {
-                    EqlTermVariant::JsonValueSelector => {
-                        json_value_selector_literal_plaintext(typed_statement, val)
-                    }
-                    // A selector that carries a collapsed chain keys the composed
-                    // path, not the one segment it spells. Only a selector the
-                    // mapper recorded a chain for: a single access has no record
-                    // and takes the ordinary single-segment route below.
-                    EqlTermVariant::JsonAccessor
-                        if typed_statement
-                            .json_accessor_paths
-                            .for_literal(val)
-                            .is_some() =>
-                    {
-                        json_accessor_path_literal_plaintext(typed_statement, val)
-                    }
-                    _ => literal_from_sql(val, col.eql_term(), col.cast_type()),
-                };
+        .enumerate()
+        .map(|(index, ((eql_term, val), col))| {
+            if skip.get(index).copied().unwrap_or(false) {
+                return Ok(None);
+            }
+            match col {
+                Some(col) => {
+                    let plaintext = match eql_term.variant() {
+                        EqlTermVariant::JsonValueSelector => {
+                            json_value_selector_literal_plaintext(typed_statement, val)
+                        }
+                        // A selector that carries a collapsed chain keys the composed
+                        // path, not the one segment it spells. Only a selector the
+                        // mapper recorded a chain for: a single access has no record
+                        // and takes the ordinary single-segment route below.
+                        EqlTermVariant::JsonAccessor
+                            if typed_statement
+                                .json_accessor_paths
+                                .for_literal(val)
+                                .is_some() =>
+                        {
+                            json_accessor_path_literal_plaintext(typed_statement, val)
+                        }
+                        _ => literal_from_sql(val, col.eql_term(), col.cast_type()),
+                    };
 
-                plaintext.map_err(|err| {
-                    debug!(
-                        target: MAPPER,
-                        msg = "Could not convert literal value",
-                        value = ?val,
-                        cast_type = ?col.cast_type(),
-                        error = err.to_string()
-                    );
-                    MappingError::InvalidParameter(Box::new(col.to_owned())).into()
-                })
+                    plaintext.map_err(|err| {
+                        debug!(
+                            target: MAPPER,
+                            msg = "Could not convert literal value",
+                            value = ?val,
+                            cast_type = ?col.cast_type(),
+                            error = err.to_string()
+                        );
+                        MappingError::InvalidParameter(Box::new(col.to_owned())).into()
+                    })
+                }
+                None => Ok(None),
             }
-            None => Ok(None),
         })
         .collect::<Result<Vec<_>, Error>>()?;
     Ok(plaintexts)